nrdk.tss.utils

Data marshalling utilities.

nrdk.tss.utils.cut_trace

cut_trace(
    timestamps: Float[ndarray, N], values: TValue, gap: float = 20.0
) -> list[TValue]

Cut trace into multiple sub-traces based on gaps in timestamps.

Tip

Set gap to a duration, in seconds, which is greater than (but of similar magnitude) than the expected effective sampling rate of the time series.

Parameters:

Name	Type	Description	Default
timestamps	Float[ndarray, N]	measurement timestamps.	required
values	TValue	time series measurement values. Is expected to be a PyTree[Shaped[np.ndarray, "N ..."]], where the length of each node is equal to the number of timestamps passed.	required
gap	float	timestamp gap, nominally in seconds, which denotes a new trace.	20.0

Returns:

Type	Description
list[TValue]	A list of metrics for each sub-trace, with the same structure and type as the inputs.

Source code in src/nrdk/tss/utils.py

def cut_trace(
    timestamps: Float[np.ndarray, "N"],
    values: TValue,
    gap: float = 20.
) -> list[TValue]:
    """Cut trace into multiple sub-traces based on gaps in timestamps.

    !!! tip

        Set `gap` to a duration, in seconds, which is greater than (but of
        similar magnitude) than the expected effective sampling rate of the
        time series.

    Args:
        timestamps: measurement timestamps.
        values: time series measurement values. Is expected to be a
            `PyTree[Shaped[np.ndarray, "N ..."]]`, where the length of each
            node is equal to the number of `timestamps` passed.
        gap: timestamp gap, nominally in seconds, which denotes a new trace.

    Returns:
        A list of metrics for each sub-trace, with the same structure and type
            as the inputs.
    """
    cuts, = np.where(np.diff(timestamps) > gap)
    cuts = np.concatenate([[0], cuts, [timestamps.shape[0]]])

    return [
        optree.tree_map(lambda x: x[start:stop], values)  # type: ignore
        for start, stop in zip(cuts[:-1], cuts[1:])]

nrdk.tss.utils.intersect_difference

intersect_difference(
    y1: Num[ndarray, N1],
    y2: Num[ndarray, N2],
    t1: Num[ndarray, N1] | None = None,
    t2: Num[ndarray, N2] | None = None,
) -> Num[ndarray, N]

Compute the difference between two time series at common timestamps.

Info

If t1 and t2 are not provided, the two time series are assumed to be synchronized.

Parameters:

Name	Type	Description	Default
y1	Num[ndarray, N1]	first time series.	required
y2	Num[ndarray, N2]	second time series.	required
t1	Num[ndarray, N1] | None	timestamps for the first time series.	None
t2	Num[ndarray, N2] | None	timestamps for the second time series.	None

Returns:

Type	Description
Num[ndarray, N]	Differences y1 - y2 at common timestamps, sorted in increasing order.

Source code in src/nrdk/tss/utils.py

def intersect_difference(
    y1: Num[np.ndarray, "N1"], y2: Num[np.ndarray, "N2"],
    t1: Num[np.ndarray, "N1"] | None = None,
    t2: Num[np.ndarray, "N2"] | None = None
) -> Num[np.ndarray, "N"]:
    """Compute the difference between two time series at common timestamps.

    !!! info

        If `t1` and `t2` are not provided, the two time series are assumed to
        be synchronized.

    Args:
        y1: first time series.
        y2: second time series.
        t1: timestamps for the first time series.
        t2: timestamps for the second time series.

    Returns:
        Differences `y1 - y2` at common timestamps, sorted in increasing order.
    """
    if t1 is None or t2 is None:
        if y1.shape != y2.shape:
            raise ValueError(
                f"If no timestamps are provided, both time series must have "
                f"the same shape: y1:{y1.shape} != y2:{y2.shape}.")
        return y1 - y2

    _, i1, i2 = np.intersect1d(t1, t2, return_indices=True)
    return y1[i1] - y2[i2]

nrdk.tss.utils.tree_flatten

tree_flatten(
    tree: MetricTree[LeafType], _path: list[str] = []
) -> dict[str, LeafType]

Flatten a nested structure into a dictionary with path-like keys.

Inverse of tree_unflatten.

Parameters:

Name	Type	Description	Default
tree	MetricTree[LeafType]	input tree structure. Must consist only of dictionaries, where each leaf value is a numpy array with the same leading batch size.	required

Returns:

Type	Description
dict[str, LeafType]	A dictionary, where keys correspond to /-joined strings of the key paths to each leaf.

Source code in src/nrdk/tss/utils.py

def tree_flatten(
    tree: MetricTree[LeafType], _path: list[str] = []
) -> dict[str, LeafType]:
    """Flatten a nested structure into a dictionary with path-like keys.

    Inverse of [`tree_unflatten`][..].

    Args:
        tree: input tree structure. Must consist only of dictionaries, where
            each leaf value is a numpy array with the same leading batch size.

    Returns:
        A dictionary, where keys correspond to `/`-joined strings of the key
            paths to each leaf.
    """
    if isinstance(tree, dict):
        out = {}
        for key, value in tree.items():
            out.update(tree_flatten(value, _path=_path + [key]))
        return out
    else:
        return {"/".join(_path): tree}

nrdk.tss.utils.tree_unflatten

tree_unflatten(flattened: dict[str, LeafType]) -> MetricTree[LeafType]

Unflatten a dictionary with path-like keys into a nested structure.

Inverse of tree_flatten.

Parameters:

Name	Type	Description	Default
flattened	dict[str, LeafType]	input dictionary, where keys are /-joined strings of the key paths to each leaf.	required

Returns:

Type	Description
MetricTree[LeafType]	Expanded dictionary with arbitrary nesting.

Source code in src/nrdk/tss/utils.py

def tree_unflatten(flattened: dict[str, LeafType]) -> MetricTree[LeafType]:
    """Unflatten a dictionary with path-like keys into a nested structure.

    Inverse of [`tree_flatten`][..].

    Args:
        flattened: input dictionary, where keys are `/`-joined strings of the
            key paths to each leaf.

    Returns:
        Expanded dictionary with arbitrary nesting.
    """
    tree = {}
    for key, value in flattened.items():
        fullpath = key.split("/")
        current = tree
        for subpath in fullpath[:-1]:
            current = current.setdefault(subpath, {})
        current[fullpath[-1]] = value
    return tree