Skip to content

Data in, one sample at a time

Robot data comes in many forms: files, recordings, arrays, live sensors. This page turns all of them into the same stream of readings, which every tool in the library then reads one at a time.

Every estimator in the library learns from a stream of samples: a Signal (named channels with units and a timestamp) and, when there is one, a target y. Robot data arrive in many shapes: log files, ROS bags, arrays, live sensors at different rates. This page turns all of them into the same stream, merges sensors in time order, makes test streams whose changes happen at a known moment, and ends with a model that refuses to learn from a fault.

1. A robot log file becomes a stream

A CSV file with the columns of the CASPER recording as it is distributed on Kaggle (Kayan et al. 2023), one row per sample, with a timestamp column and a column holding the six joint velocities as a list; here three joints and three rows.

from pathlib import Path
from dense_armor.utility.stream import iter_csv

p = Path("arm.csv")
p.write_text(
    "Timestamp,Actual Joint Velocities\n"
    '0.00,"[0.1, -0.2, 0.0]"\n'
    '0.05,"[0.2, -0.1, 0.0]"\n'
    '0.10,"[0.3, 0.0, 0.1]"\n'
)
for sig, y in iter_csv(p, time_col="Timestamp", array_cols=["Actual Joint Velocities"]):
    print(sig.t, {k: round(float(v), 2) for k, v in sig.items()})

0.0 {'Actual_Joint_Velocities_0': 0.1, 'Actual_Joint_Velocities_1': -0.2, 'Actual_Joint_Velocities_2': 0.0}
0.05 {'Actual_Joint_Velocities_0': 0.2, 'Actual_Joint_Velocities_1': -0.1, 'Actual_Joint_Velocities_2': 0.0}
0.1 {'Actual_Joint_Velocities_0': 0.3, 'Actual_Joint_Velocities_1': 0.0, 'Actual_Joint_Velocities_2': 0.1}

Each row becomes one Signal with its timestamp; the list column is split into one channel per joint. The file is read one row at a time, so a 24-hour log uses the same memory as a three-row one; start and stop read only a time window and stop at the end of it. An empty or unreadable cell becomes a missing channel, not an error.

2. Two sensors, one timeline

A camera at 30 frames per second and joint encoders at 100 Hz give two streams; an estimator needs them as one stream, in time order.

import numpy as np
from dense_armor.utility.stream import iter_array, merge_by_time

cam = iter_array(np.array([[1.0], [2.0]]), t=np.array([0.0, 1 / 30]), names=["frame"])
joints = iter_array(np.array([[0.1], [0.2], [0.3], [0.4]]), t=np.arange(4) / 100, names=["q0"])
for sig, _ in merge_by_time(cam, joints):
    print(round(sig.t, 3), sig.names[0])

0.0 frame
0.0 q0
0.01 q0
0.02 q0
0.03 q0
0.033 frame

The samples come out sorted by time across the two sensors; at equal times (0.0 here) the order of the arguments decides, camera first. The merge keeps only one sample per stream in memory, so it works on live sensors as well as on files.

3. Choosing and reshaping channels

A sample with a joint position, a torque and a temperature; the model needs only the first two, with the torque scaled.

from dense_armor.roles import Signal
from dense_armor.utility.compose import FuncTransformer, Select

sig = Signal(values=[0.5, 2.0, 9.0], names=["q0", "tau0", "temp"], units=["rad", "N m", "C"], t=1.0)
pipe = Select(keys=("q0", "tau0")) | FuncTransformer(lambda v: v * 2.0, keys=("tau0",))
out = pipe.transform_one(sig)
print({k: float(v) for k, v in out.items()}, out.t, out.units)

{'q0': 0.5, 'tau0': 4.0} 1.0 ['rad', 'N m']

Select keeps the named channels (Discard drops them), FuncTransformer applies a function to some channels; joined with | they form a pipeline like any other in the library. The timestamp and the units of the kept channels travel with the sample.

4. A test stream whose concept changes

To test a drift detector you need a stream whose rule changes at a known place. This binary stream changes its concept once, half way, gradually: near the change each sample comes from the old or the new concept with a probability that follows a sigmoid.

from dense_armor.utility.datasets import DriftStream

ds = DriftStream(n_samples=20000, n_features=2, n_drifts=1, kind="gradual", spacing=20.0, seed=0)
ds.data()
c = ds.concepts_
print([round(float(c[a:a + 1000].mean()), 2) for a in range(7000, 13000, 1000)])

[0.07, 0.22, 0.42, 0.61, 0.82, 0.92]

A concept is the rule that ties the input features to the class. In this generator each concept is two Gaussian clusters, one per class, in a random direction; the paper's own StreamGenerator builds them from make_classification and the Madelon rules (section 3, page 7). The stream visits one concept, then the next; the change sits at sample 10000. Each number in the output is the share of samples from the new concept in a window of 1000, from sample 7000 to 13000; it rises smoothly through the change (0.42 just before, 0.61 just after).

The weight of the new concept at position $p \in [0, 1]$ of the stream is the sigmoid that Ksieniewicz and Zyblewski (2020) introduce for all three drift types in section 3.2.1 (page 8). The paper does not give its closed form; the generator uses the standard logistic form,

$$s(p) = \frac{1}{1 + e^{-\text{spacing}\,(p - d)}}$$

with $p$ the position of the sample as a fraction of the stream, $d = 0.5$ the position of the drift (half way) and spacing how sharp the change is. At $p = d$ the sigmoid is exactly $0.5$; with spacing $= 20$ it goes from $0.12$ at $p = 0.4$ to $0.88$ at $p = 0.6$. The same sigmoid drives the three drift types of the paper (section 3.2, pages 8–9): "sudden" is the limit for a large spacing and swaps the concept at once, "gradual" draws the old or the new concept at each sample with probability s(p), "incremental" moves the clusters continuously with the same weight.

5. A robot arm with a fault at a known time

A labelled joint stream from the real URDF of a Panda arm. Torque is the turning force the motors apply at each joint; on a moving arm it is not free, it is fixed by the motion through the arm's own dynamics. The stream has a periodic motion, torques from the library's inverse dynamics, sensor noise, and from second 2 a 5 kg payload on the last link.

import numpy as np
from dense_armor.utility.datasets import SyntheticArm

ds = SyntheticArm("test/fixtures/urdf/panda.urdf", period_s=2.0, rate_hz=50.0, n_cycles=2,
                  fault_at_s=2.0, fault="payload", payload_mass=5.0, seed=0)
r0, r1 = [], []
for sig, y in ds.stream():
    q, qd, qdd = ds.trajectory(sig.t)
    r = float(sig["tau_1"] - ds.nominal_torque(q, qd, qdd)[1])
    (r1 if y else r0).append(r)
print(ds.payload_link, round(np.mean(r0), 3), round(np.std(r0), 3), round(np.mean(r1), 2))

panda_link7 -0.004 0.051 -12.54

The residual is what is left of the measured torque once the expected one is subtracted:

$$r = \tau_{\text{meas}} - \tau_{\text{model}}.$$

On a healthy arm it is noise: mean -0.004 N·m and standard deviation 0.051 N·m, exactly the torque noise the stream adds. On a faulty arm it carries the fault: it jumps to a mean of -12.54 N·m.

The expected torque comes from the arm's inverse dynamics,

$$\tau = M(q)\,\ddot q + C(q, \dot q)\,\dot q + g(q),$$

with $q$ the joint positions, $\dot q$ the joint velocities, $\ddot q$ the joint accelerations, $M(q)$ the mass matrix (how the arm's inertia couples the joints), $C(q, \dot q)\,\dot q$ the Coriolis and centrifugal torques, and $g(q)$ the gravity torques. The payload adds its own gravity torque

$$\tau_{\text{payload}} = m\,g\,J_z(q),$$

where $m$ is the payload mass, $g$ the acceleration of gravity and $J_z(q)$ the vertical row of the Jacobian of the link that carries the payload (here panda_link7, the flange): the vertical speed of that link for a unit joint velocity. y is 0 before the fault and 1 after, so the stream also scores detectors.

6. A model that does not learn the fault

A recursive least-squares model estimates the residual offset, the constant value the model adds to its prediction to best fit the samples it has seen; on a healthy arm it is about zero. Protected asks an anomaly detector before each sample and skips learning on the ones the detector flags; the detector is a Hampel scorer wrapped in AnomalyGate, which also keeps it from learning flagged samples. The same model without protection is shown next to it for comparison.

from dense_armor.roles.anomaly_detector import AnomalyGate
from dense_armor.roles.protection import Protected
from dense_armor.utility.anomaly.filters import HampelScorer
from dense_armor.utility.datasets import SyntheticArm
from dense_armor.utility.learn.online_dynamics import RecursiveLeastSquares

ds = SyntheticArm("test/fixtures/urdf/panda.urdf", period_s=2.0, rate_hz=50.0, n_cycles=2,
                  fault_at_s=2.0, fault="payload", payload_mass=5.0, seed=0)
gate = AnomalyGate(HampelScorer(radius=10, n_sigmas=5.0, feature="r"))
prot = Protected(model=RecursiveLeastSquares(feature_keys=["one"]), detector=gate)
bare = RecursiveLeastSquares(feature_keys=["one"])
for sig, y in ds.stream():
    q, qd, qdd = ds.trajectory(sig.t)
    x = {"r": float(sig["tau_1"] - ds.nominal_torque(q, qd, qdd)[1]), "one": 1.0}
    gate.learn_one(x)
    prot.learn_one(x, x["r"])
    bare.learn_one(x, x["r"])
print(round(prot.model.predict_one({"one": 1.0}), 3), round(bare.predict_one({"one": 1.0}), 3))

-0.003 -6.271

After the fault, the protected model still says the healthy offset is -0.003 N·m; the unprotected one has averaged the fault in and says -6.271 N·m, about half of the -12.5 N·m step because half of its samples came after the fault. In the test suite the same run flags 99 of the 100 faulty samples and 1 of the 100 healthy ones. A model protected this way can be used to keep estimating "normal" while the robot is in trouble.

API reference

stream

Streams: iterators of (Signal, y) pairs in time order.

iter_array 

iter_array(
    X: Any,
    y: Any = None,
    t: Any = None,
    names: Iterable[str] | None = None,
    units: Iterable[str] | None = None,
) -> Iterator[tuple[Signal, Any]]

Iterate rows of an array as (Signal, y) pairs.

Parameters:

Name Type Description Default
X Any

array of shape (n, d) or (n,).

 required
y Any

optional labels of shape (n,).

 None
t Any

optional timestamps in seconds. If None, t = float(i).

 None
names Iterable[str] | None

channel names in column order.

 None
units Iterable[str] | None

channel units.

 None

Yields:

Type Description
tuple[Signal, Any]

(signal, label) pairs in time order.

Raises:

Type Description
ValueError

if a length does not match.
Source code in dense_armor/utility/stream/array.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def iter_array(
    X: Any,
    y: Any = None,
    t: Any = None,
    names: Iterable[str] | None = None,
    units: Iterable[str] | None = None,
) -> Iterator[tuple[Signal, Any]]:
    """Iterate rows of an array as ``(Signal, y)`` pairs.

    Args:
        X: array of shape ``(n, d)`` or ``(n,)``.
        y: optional labels of shape ``(n,)``.
        t: optional timestamps in seconds. If ``None``, ``t = float(i)``.
        names: channel names in column order.
        units: channel units.

    Yields:
        ``(signal, label)`` pairs in time order.

    Raises:
        ValueError: if a length does not match.
    """
    arr = np.asarray(X)
    if arr.ndim == 1:
        arr = arr[:, None]
    if arr.ndim != 2:
        raise ValueError(f"X must be 1D or 2D, got shape {arr.shape}")
    n, d = arr.shape
    name_list = [f"x{i}" for i in range(d)] if names is None else list(names)
    if len(name_list) != d:
        raise ValueError(f"names has {len(name_list)} entries, X has {d} columns")
    unit_list = ["" for _ in range(d)] if units is None else list(units)
    if len(unit_list) != d:
        raise ValueError(f"units has {len(unit_list)} entries, X has {d} columns")
    y_arr = None if y is None else np.asarray(y)
    if y_arr is not None and len(y_arr) != n:
        raise ValueError(f"y has {len(y_arr)} entries, X has {n} rows")
    if t is None:
        t_arr = np.arange(n, dtype=float)
    else:
        t_arr = np.asarray(t, dtype=float)
        if len(t_arr) != n:
            raise ValueError(f"t has {len(t_arr)} entries, X has {n} rows")
    for i in range(n):
        sig = Signal(
            values=arr[i],
            names=name_list,
            units=unit_list,
            t=float(t_arr[i]),
        )
        yield sig, (None if y_arr is None else y_arr[i])

iter_csv 

iter_csv(
    path: str | Path,
    target: str | None = None,
    time_col: str | None = None,
    columns: list[str] | None = None,
    array_cols: list[str] | None = None,
    start: float | None = None,
    stop: float | None = None,
) -> Iterator[tuple[Signal, Any]]

Iterate rows of a CSV file as (Signal, y) pairs.

Parameters:

Name Type Description Default
path str | Path

path to the CSV file.

 required
target str | None

label column, or None.

 None
time_col str | None

timestamp column in seconds, or None.

 None
columns list[str] | None

scalar feature columns. None = every column but target, time_col and array_cols.

 None
array_cols list[str] | None

columns holding a list literal "[...]", expanded into <name>_0, ..., <name>_{k-1} channels; k from the first row.

 None
start float | None

skip rows before this time.

 None
stop float | None

stop at the first row past this time.

 None

Yields:

Type Description
tuple[Signal, Any]

(signal, label) pairs in file order.

Raises:

Type Description
ValueError

on a requested column that is not in the file.
Source code in dense_armor/utility/stream/csv.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def iter_csv(
    path: str | Path,
    target: str | None = None,
    time_col: str | None = None,
    columns: list[str] | None = None,
    array_cols: list[str] | None = None,
    start: float | None = None,
    stop: float | None = None,
) -> Iterator[tuple[Signal, Any]]:
    """Iterate rows of a CSV file as ``(Signal, y)`` pairs.

    Args:
        path: path to the CSV file.
        target: label column, or ``None``.
        time_col: timestamp column in seconds, or ``None``.
        columns: scalar feature columns. ``None`` = every column but
            ``target``, ``time_col`` and ``array_cols``.
        array_cols: columns holding a list literal ``"[...]"``, expanded
            into ``<name>_0, ..., <name>_{k-1}`` channels; ``k`` from the
            first row.
        start: skip rows before this time.
        stop: stop at the first row past this time.

    Yields:
        ``(signal, label)`` pairs in file order.

    Raises:
        ValueError: on a requested column that is not in the file.
    """
    array_cols = [] if array_cols is None else list(array_cols)
    path = Path(path)
    with path.open(newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        header = list(reader.fieldnames or [])
        for name in columns or []:
            if name not in header:
                raise ValueError(f"column {name!r} not in {header}")
        for name in array_cols:
            if name not in header:
                raise ValueError(f"column {name!r} not in {header}")
        if target is not None and target not in header:
            raise ValueError(f"target {target!r} not in {header}")
        if time_col is not None and time_col not in header:
            raise ValueError(f"time_col {time_col!r} not in {header}")
        if columns is None:
            col_list = [
                c
                for c in header
                if c != target and c != time_col and c not in array_cols
            ]
        else:
            col_list = list(columns)
        try:
            first = next(reader)
        except StopIteration:
            return
        array_lengths: dict[str, int] = {}
        for name in array_cols:
            parsed = _parse_array_cell(first.get(name))
            array_lengths[name] = 0 if parsed is None else len(parsed)
        feat_names: list[str] = list(col_list)
        for name in array_cols:
            clean = name.replace(" ", "_")
            for i in range(array_lengths[name]):
                feat_names.append(f"{clean}_{i}")
        feat_units: list[str] = ["" for _ in feat_names]
        for row in chain([first], reader):
            t_val: float | None = None
            if time_col is not None:
                t_val = _parse_scalar_cell(row.get(time_col))
            if t_val is not None:
                if start is not None and t_val < start:
                    continue
                if stop is not None and t_val > stop:
                    return
            vals: list[float] = []
            mask: list[bool] = []
            for name in col_list:
                v = _parse_scalar_cell(row.get(name))
                if v is None:
                    vals.append(float("nan"))
                    mask.append(True)
                else:
                    vals.append(v)
                    mask.append(False)
            for name in array_cols:
                k = array_lengths[name]
                parsed = _parse_array_cell(row.get(name))
                if parsed is None or len(parsed) != k:
                    for _ in range(k):
                        vals.append(float("nan"))
                        mask.append(True)
                else:
                    for v in parsed:
                        vals.append(v)
                        mask.append(False)
            if not vals:
                sig = Signal(values=np.zeros(0), names=[], units=[], t=t_val)
            else:
                sig = Signal(
                    values=np.asarray(vals, dtype=float),
                    names=feat_names,
                    units=feat_units,
                    t=t_val,
                    missing=np.asarray(mask, dtype=bool),
                )
            if target is not None:
                raw = row.get(target)
                if raw is None or str(raw).strip() == "":
                    label: Any = None
                else:
                    try:
                        label = float(raw)
                    except (TypeError, ValueError):
                        label = raw
            else:
                label = None
            yield sig, label

merge_by_time 

merge_by_time(
    *streams: Iterable[tuple[Signal, Any]],
) -> Iterator[tuple[Signal, Any]]

Merge streams in time order.

Parameters:

Name Type Description Default
*streams Iterable[tuple[Signal, Any]]

iterables of (Signal, y) pairs, each already in non-decreasing order of Signal.t.

 ()

Yields:

Type Description
Signal

(signal, y) pairs, in non-decreasing order of t. Equal
Any

timestamps keep the order of the arguments.

Raises:

Type Description
ValueError

if a sample has t=None.
Source code in dense_armor/utility/stream/merge.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def merge_by_time(
    *streams: Iterable[tuple[Signal, Any]],
) -> Iterator[tuple[Signal, Any]]:
    """Merge streams in time order.

    Args:
        *streams: iterables of ``(Signal, y)`` pairs, each already in
            non-decreasing order of ``Signal.t``.

    Yields:
        ``(signal, y)`` pairs, in non-decreasing order of ``t``. Equal
        timestamps keep the order of the arguments.

    Raises:
        ValueError: if a sample has ``t=None``.
    """
    heap: list[tuple[float, int, Signal, Any, Iterator]] = []
    for i, s in enumerate(streams):
        it = iter(s)
        try:
            sig, y = next(it)
        except StopIteration:
            continue
        if sig.t is None:
            raise ValueError("merge_by_time needs a timestamp on every sample")
        heapq.heappush(heap, (float(sig.t), i, sig, y, it))
    while heap:
        _, i, sig, y, it = heapq.heappop(heap)
        yield sig, y
        try:
            sig2, y2 = next(it)
        except StopIteration:
            continue
        if sig2.t is None:
            raise ValueError("merge_by_time needs a timestamp on every sample")
        heapq.heappush(heap, (float(sig2.t), i, sig2, y2, it))

iter_rosbag 

iter_rosbag(
    path: str | Path, topics: list[str] | None = None
) -> Iterator[tuple[Signal, Any]]

Iterate a ROS 1 or ROS 2 bag as (Signal, y) pairs.

Parameters:

Name Type Description Default
path str | Path

ROS 1 .bag file, or ROS 2 bag directory.

 required
topics list[str] | None

topic names to read. None = every topic with a sensor_msgs/msg/JointState message.

 None

Yields:

Type Description
tuple[Signal, Any]

(signal, None) pairs, in message order.

Raises:

Type Description
ImportError

if rosbags is not installed.
Source code in dense_armor/utility/stream/rosbag.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def iter_rosbag(
    path: str | Path,
    topics: list[str] | None = None,
) -> Iterator[tuple[Signal, Any]]:
    """Iterate a ROS 1 or ROS 2 bag as ``(Signal, y)`` pairs.

    Args:
        path: ROS 1 ``.bag`` file, or ROS 2 bag directory.
        topics: topic names to read. ``None`` = every topic with a
            ``sensor_msgs/msg/JointState`` message.

    Yields:
        ``(signal, None)`` pairs, in message order.

    Raises:
        ImportError: if ``rosbags`` is not installed.
    """
    AnyReader = _load_rosbags()
    path = Path(path)
    with AnyReader([path]) as reader:
        if topics is None:
            conns = [c for c in reader.connections if c.msgtype == _JOINT_STATE]
        else:
            wanted = set(topics)
            conns = [
                c
                for c in reader.connections
                if c.msgtype == _JOINT_STATE and c.topic in wanted
            ]
        if not conns:
            return
        for conn, ts_ns, raw in reader.messages(connections=conns):
            msg = reader.deserialize(raw, conn.msgtype)
            name = list(msg.name)
            pos = list(msg.position)
            vel = list(msg.velocity)
            eff = list(msg.effort)
            vals: dict[str, float] = {}
            units: dict[str, str] = {}
            for j, jname in enumerate(name):
                if j < len(pos):
                    vals[f"q_{jname}"] = float(pos[j])
                    units[f"q_{jname}"] = "rad"
                if j < len(vel):
                    vals[f"qd_{jname}"] = float(vel[j])
                    units[f"qd_{jname}"] = "rad/s"
                if j < len(eff):
                    vals[f"tau_{jname}"] = float(eff[j])
                    units[f"tau_{jname}"] = "N*m"
            if not vals:
                continue
            t_s = float(ts_ns) * 1e-9
            sig = Signal.from_dict(vals, units=units, t=t_s)
            yield sig, None

datasets

Datasets: ready-made streams for tests and experiments.

Casper 

Casper(
    path: str | Path,
    time_col: str = "Timestamp",
    file: str = "right_arm.csv",
    target_col: str | None = None,
)

CASPER UR3e joint stream.

Parameters:

Name Type Description Default
path str | Path

folder containing the CSV files.

 required
time_col str

name of the timestamp column, in seconds. Default "Timestamp".

 'Timestamp'
file str

name of the CSV file inside path. Default "right_arm.csv".

 'right_arm.csv'
target_col str | None

name of the label column, if present. Default None.

 None

Raises:

Type Description
FileNotFoundError

if the file does not exist.

Examples:

>>> from dense_armor.utility.datasets import Casper
>>> ds = Casper("/path/to/casper_folder")
Source code in dense_armor/utility/datasets/casper.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def __init__(
    self,
    path: str | Path,
    time_col: str = "Timestamp",
    file: str = "right_arm.csv",
    target_col: str | None = None,
) -> None:
    self.path = Path(path)
    self.file = file
    self.time_col = time_col
    self.target_col = target_col
    self._csv = self.path / file
    if not self._csv.exists():
        raise FileNotFoundError(
            f"{self._csv} not found; pass the folder of the Kaggle "
            f"copy of the CASPER dataset"
        )

stream 

stream(
    columns: list[str],
    start_h: float | None = None,
    stop_h: float | None = None,
) -> Iterator[tuple[Signal, Any]]

Iterate the joint stream, in time order.

Parameters:

Name Type Description Default
columns list[str]

names of the joint columns to expand; each cell holds a six-element list literal and becomes six channels.

 required
start_h float | None

skip rows before start_h hours from the first timestamp.

 None
stop_h float | None

stop as soon as a row goes past stop_h hours from the first timestamp.

 None

Yields:

Type Description
tuple[Signal, Any]

(signal, label) pairs with the expanded joint channels.
Source code in dense_armor/utility/datasets/casper.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def stream(
    self,
    columns: list[str],
    start_h: float | None = None,
    stop_h: float | None = None,
) -> Iterator[tuple[Signal, Any]]:
    """Iterate the joint stream, in time order.

    Args:
        columns: names of the joint columns to expand; each cell
            holds a six-element list literal and becomes six
            channels.
        start_h: skip rows before ``start_h`` hours from the first
            timestamp.
        stop_h: stop as soon as a row goes past ``stop_h`` hours
            from the first timestamp.

    Yields:
        ``(signal, label)`` pairs with the expanded joint channels.
    """
    start_s: float | None = None
    stop_s: float | None = None
    if start_h is not None or stop_h is not None:
        t0 = _first_timestamp(self._csv, self.time_col)
        if t0 is not None:
            if start_h is not None:
                start_s = t0 + start_h * 3600.0
            if stop_h is not None:
                stop_s = t0 + stop_h * 3600.0
    yield from iter_csv(
        self._csv,
        target=self.target_col,
        time_col=self.time_col,
        columns=[],
        array_cols=columns,
        start=start_s,
        stop=stop_s,
    )

n_rows 

n_rows() -> int

Number of data rows in the CSV file.

Source code in dense_armor/utility/datasets/casper.py 
131
132
133
134
135
136
137
138
def n_rows(self) -> int:
    """Number of data rows in the CSV file."""
    n = 0
    with self._csv.open(newline="", encoding="utf-8") as f:
        reader = csv.reader(f)
        for _ in reader:
            n += 1
    return max(0, n - 1)

DriftStream 

DriftStream(
    n_samples: int,
    n_features: int,
    n_drifts: int,
    kind: str,
    spacing: float = 10.0,
    seed: int | None = 0,
)

Synthetic binary stream with concept drift.

Parameters:

Name Type Description Default
n_samples int

number of samples in the stream.

 required
n_features int

number of input dimensions.

 required
n_drifts int

number of concept changes. Zero means stationary.

 required
kind str

one of "sudden", "gradual", "incremental".

 required
spacing float

sharpness of the sigmoid, in inverse position units. Higher values give more abrupt transitions.

 10.0
seed int | None

seed of the random generator.

 0

Raises:

Type Description
ValueError

on a bad kind or a non-positive count.

Examples:

>>> from dense_armor.utility.datasets import DriftStream
>>> ds = DriftStream(n_samples=50, n_features=3, n_drifts=2,
...                  kind="gradual", spacing=20.0, seed=1)
>>> X, y = ds.data()
>>> X.shape
(50, 3)
>>> ds.drift_positions
array([0.33333333, 0.66666667])
Source code in dense_armor/utility/datasets/drift_stream.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def __init__(
    self,
    n_samples: int,
    n_features: int,
    n_drifts: int,
    kind: str,
    spacing: float = 10.0,
    seed: int | None = 0,
) -> None:
    if n_samples < 1:
        raise ValueError(f"n_samples must be >= 1, got {n_samples}")
    if n_features < 1:
        raise ValueError(f"n_features must be >= 1, got {n_features}")
    if n_drifts < 0:
        raise ValueError(f"n_drifts must be >= 0, got {n_drifts}")
    if kind not in self._KINDS:
        raise ValueError(f"kind must be one of {self._KINDS}, got {kind!r}")
    self.n_samples = n_samples
    self.n_features = n_features
    self.n_drifts = n_drifts
    self.kind = kind
    self.spacing = float(spacing)
    self.seed = seed
    rng = np.random.default_rng(seed)
    self._mu = [_make_concept(rng, n_features, j) for j in range(n_drifts + 1)]
    self._scale = self._mu[0][1]
    if n_drifts == 0:
        self.drift_positions = np.zeros(0)
    else:
        self.drift_positions = np.array(
            [(j + 1) / (n_drifts + 1) for j in range(n_drifts)]
        )

data 

data() -> tuple[np.ndarray, np.ndarray]

Generate the whole stream.

Returns:

Type Description
ndarray

(X, y) with shapes (n_samples, n_features) and
ndarray

(n_samples,).
Source code in dense_armor/utility/datasets/drift_stream.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def data(self) -> tuple[np.ndarray, np.ndarray]:
    """Generate the whole stream.

    Returns:
        ``(X, y)`` with shapes ``(n_samples, n_features)`` and
        ``(n_samples,)``.
    """
    rng = np.random.default_rng(None if self.seed is None else self.seed + 1)
    X = np.zeros((self.n_samples, self.n_features))
    y = np.zeros(self.n_samples, dtype=int)
    concepts = np.zeros(self.n_samples, dtype=int)
    for i in range(self.n_samples):
        p = i / max(1, self.n_samples - 1)
        x, label, which = self._sample_one(rng, p)
        X[i] = x
        y[i] = label
        concepts[i] = which
    self.concepts_ = concepts
    return X, y

stream 

stream() -> Iterator[tuple[Signal, int]]

Yield (Signal, y) pairs in time order.

Source code in dense_armor/utility/datasets/drift_stream.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
def stream(self) -> Iterator[tuple[Signal, int]]:
    """Yield ``(Signal, y)`` pairs in time order."""
    X, y = self.data()
    names = [f"x{i}" for i in range(self.n_features)]
    units = ["" for _ in range(self.n_features)]
    for i in range(self.n_samples):
        sig = Signal(
            values=X[i],
            names=names,
            units=units,
            t=float(i),
        )
        yield sig, int(y[i])

SyntheticArm 

SyntheticArm(
    urdf_path: str | Path,
    period_s: float,
    rate_hz: float,
    n_cycles: int = 4,
    fault_at_s: float | None = None,
    fault: str | None = None,
    fault_joint: int | None = None,
    friction_b: float = 20.0,
    payload_mass: float = 2.0,
    payload_link: str | None = None,
    noise_std: tuple[float, float, float] = (
        0.001,
        0.005,
        0.05,
    ),
    seed: int | None = 0,
)

Periodic joint stream from a URDF with an optional fault.

Parameters:

Name Type Description Default
urdf_path str | Path

URDF readable by :class:~dense_armor.dynamics.urdf_dynamics.RigidBodyModel.

 required
period_s float

period of one sinusoidal cycle, in seconds.

 required
rate_hz float

sampling rate, in Hz.

 required
n_cycles int

number of cycles generated.

 4
fault_at_s float | None

time from the start when the fault begins, in seconds. None means no fault.

 None
fault str | None

"friction" or "payload". Required when fault_at_s is given.

 None
fault_joint int | None

index of the joint with the "friction" fault. Default: the last joint.

 None
friction_b float

viscous coefficient for "friction", in N m s.

 20.0
payload_mass float

mass added for "payload", in kg.

 2.0
payload_link str | None

link carrying the payload. Default: the last link of the longest chain.

 None
noise_std tuple[float, float, float]

standard deviation of the Gaussian sensor noise on q (rad), qd (rad/s) and tau (N m).

 (0.001, 0.005, 0.05)
seed int | None

seed of the random generator.

 0

Raises:

Type Description
ValueError

on bad arguments.

Examples:

>>> from dense_armor.utility.datasets import SyntheticArm
>>> ds = SyntheticArm(
...     "test/fixtures/urdf/panda.urdf",
...     period_s=1.0, rate_hz=50.0,
...     fault_at_s=0.5, fault="payload", seed=0,
... )
Source code in dense_armor/utility/datasets/synthetic_arm.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def __init__(
    self,
    urdf_path: str | Path,
    period_s: float,
    rate_hz: float,
    n_cycles: int = 4,
    fault_at_s: float | None = None,
    fault: str | None = None,
    fault_joint: int | None = None,
    friction_b: float = 20.0,
    payload_mass: float = 2.0,
    payload_link: str | None = None,
    noise_std: tuple[float, float, float] = (1e-3, 5e-3, 5e-2),
    seed: int | None = 0,
) -> None:
    if period_s <= 0:
        raise ValueError(f"period_s must be > 0, got {period_s}")
    if rate_hz <= 0:
        raise ValueError(f"rate_hz must be > 0, got {rate_hz}")
    if n_cycles < 1:
        raise ValueError(f"n_cycles must be >= 1, got {n_cycles}")
    if fault_at_s is not None and fault not in ("friction", "payload"):
        raise ValueError(f"fault must be 'friction' or 'payload', got {fault!r}")
    self.urdf_path = Path(urdf_path)
    self.period_s = float(period_s)
    self.rate_hz = float(rate_hz)
    self.n_cycles = n_cycles
    self.fault_at_s = fault_at_s
    self.fault = fault
    self.friction_b = float(friction_b)
    self.payload_mass = float(payload_mass)
    self.noise_std = tuple(noise_std)
    self.seed = seed
    (
        self.model,
        self.payload_link,
        self._mass_matrix,
        self._bias,
        self._gravity,
        self._jac,
    ) = _dynamics(self.urdf_path, payload_link)
    self.n_joints = int(self.model.n)
    if fault_joint is None:
        fault_joint = self.n_joints - 1
    if not 0 <= fault_joint < self.n_joints:
        raise ValueError(
            f"fault_joint must be in [0, {self.n_joints}), got {fault_joint}"
        )
    self.fault_joint = int(fault_joint)
    self.q_min, self.q_max = _joint_limits(self.urdf_path, self.n_joints)
    span = np.maximum(self.q_max - self.q_min, 1e-3)
    self.amplitude = 0.3 * span
    self.mid = 0.5 * (self.q_min + self.q_max)
    freqs = 1.0 + 0.25 * np.arange(self.n_joints)
    phases = 0.5 * np.arange(self.n_joints)
    self.freqs = freqs
    self.phases = phases
    self.n_samples = round(n_cycles * period_s * rate_hz)
    self.dt = 1.0 / rate_hz

trajectory 

trajectory(
    t: float,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]

Sinusoidal position, velocity and acceleration at t.

Parameters:

Name Type Description Default
t float

time from the start, in seconds.

 required

Returns:

Type Description
tuple[ndarray, ndarray, ndarray]

(q, qd, qdd), three arrays of shape (n_joints,).
Source code in dense_armor/utility/datasets/synthetic_arm.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def trajectory(self, t: float) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Sinusoidal position, velocity and acceleration at ``t``.

    Args:
        t: time from the start, in seconds.

    Returns:
        ``(q, qd, qdd)``, three arrays of shape ``(n_joints,)``.
    """
    w = 2.0 * np.pi / self.period_s
    phase = w * self.freqs * t + self.phases
    q = self.mid + self.amplitude * np.sin(phase)
    qd = self.amplitude * self.freqs * w * np.cos(phase)
    qdd = -self.amplitude * self.freqs**2 * w**2 * np.sin(phase)
    return q, qd, qdd

nominal_torque 

nominal_torque(
    q: ndarray, qd: ndarray, qdd: ndarray
) -> np.ndarray

Inverse dynamics of the arm, without noise or fault.

Parameters:

Name Type Description Default
q ndarray

joint positions.

 required
qd ndarray

joint velocities.

 required
qdd ndarray

joint accelerations.

 required

Returns:

Type Description
ndarray

``mass_matrix(q) @ qdd + bias_forces(q, qd)
ndarray
  • gravity_forces(q)``.
Source code in dense_armor/utility/datasets/synthetic_arm.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def nominal_torque(
    self, q: np.ndarray, qd: np.ndarray, qdd: np.ndarray
) -> np.ndarray:
    """Inverse dynamics of the arm, without noise or fault.

    Args:
        q: joint positions.
        qd: joint velocities.
        qdd: joint accelerations.

    Returns:
        ``mass_matrix(q) @ qdd + bias_forces(q, qd)
        + gravity_forces(q)``.
    """
    M = np.asarray(self._mass_matrix(q), dtype=float)
    b = np.asarray(self._bias(q, qd), dtype=float)
    g = np.asarray(self._gravity(q), dtype=float)
    return M @ qdd + b + g

payload_torque 

payload_torque(q: ndarray) -> np.ndarray

Gravity torque of the payload mass, m * g * J_z(q).

Source code in dense_armor/utility/datasets/synthetic_arm.py 
221
222
223
224
def payload_torque(self, q: np.ndarray) -> np.ndarray:
    """Gravity torque of the payload mass, ``m * g * J_z(q)``."""
    jz = np.asarray(self._jac(q), dtype=float)[2]
    return self.payload_mass * 9.81 * jz

stream 

stream() -> Iterator[tuple[Signal, int]]

Yield (Signal, y) pairs at rate_hz.

Each :class:Signal has channels q_j, qd_j, tau_j for every joint; t is the sample time in seconds; y is 0 before fault_at_s and 1 after.

Yields:

Type Description
tuple[Signal, int]

(signal, label) pairs in time order.
Source code in dense_armor/utility/datasets/synthetic_arm.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def stream(self) -> Iterator[tuple[Signal, int]]:
    """Yield ``(Signal, y)`` pairs at ``rate_hz``.

    Each :class:`Signal` has channels ``q_j``, ``qd_j``, ``tau_j``
    for every joint; ``t`` is the sample time in seconds; ``y`` is
    ``0`` before ``fault_at_s`` and ``1`` after.

    Yields:
        ``(signal, label)`` pairs in time order.
    """
    rng = np.random.default_rng(self.seed)
    names: list[str] = []
    units: list[str] = []
    for j in range(self.n_joints):
        names += [f"q_{j}", f"qd_{j}", f"tau_{j}"]
        units += ["rad", "rad/s", "N*m"]
    for i in range(self.n_samples):
        t = i * self.dt
        q, qd, qdd = self.trajectory(t)
        tau = self.nominal_torque(q, qd, qdd) + self._fault_term(q, qd, t)
        vals: list[float] = []
        for j in range(self.n_joints):
            vals.append(q[j] + rng.normal(0.0, self.noise_std[0]))
            vals.append(qd[j] + rng.normal(0.0, self.noise_std[1]))
            vals.append(tau[j] + rng.normal(0.0, self.noise_std[2]))
        sig = Signal(
            values=np.asarray(vals, dtype=float),
            names=names,
            units=units,
            t=t,
        )
        y = 0
        if self.fault_at_s is not None and t >= self.fault_at_s:
            y = 1
        yield sig, y

compose

Compose: small transformers to select, drop and reshape channels.

FuncTransformer 

FuncTransformer(
    fn: Callable,
    keys: tuple[str, ...] | None = None,
    whole: bool = False,
)

Bases: Transformer

Apply fn to channels, or to the whole dict.

Parameters:

Name Type Description Default
fn Callable

a callable taking one value (when whole is False) or the whole dict (when whole is True), and returning the new value or None to skip the channel.

 required
keys tuple[str, ...] | None

names of the channels to transform. None transforms every channel when whole is False.

 None
whole bool

if True, fn receives the whole dict and must return a dict, which replaces the output.

 False

Examples:

>>> from dense_armor.utility.compose import FuncTransformer
>>> f = FuncTransformer(
...     lambda d: {"y": sum(d.values())}, whole=True
... )
>>> f.transform_one({"a": 1.0, "b": 2.0})
{'y': 3.0}
Source code in dense_armor/utility/compose/func.py 
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    fn: Callable,
    keys: tuple[str, ...] | None = None,
    whole: bool = False,
) -> None:
    self.fn = fn
    self.keys = None if keys is None else list(keys)
    self.whole = bool(whole)

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> FuncTransformer

No-op: the transformer is stateless.

Source code in dense_armor/utility/compose/func.py 
63
64
65
66
67
68
def learn_one(
    self, x: Any, y: Any = None, t: float | None = None
) -> "FuncTransformer":
    """No-op: the transformer is stateless."""
    self._time_step(t)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> Any

Apply the function and return the new sample.

Parameters:

Name Type Description Default
x Any

a :class:Signal or a plain dict.

 required
t float | None

unused.

 None

Returns:

Type Description
Any

A new :class:Signal (or dict) with the transformed values.
Source code in dense_armor/utility/compose/func.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def transform_one(self, x: Any, t: float | None = None) -> Any:
    """Apply the function and return the new sample.

    Args:
        x: a :class:`Signal` or a plain dict.
        t: unused.

    Returns:
        A new :class:`Signal` (or dict) with the transformed values.
    """
    _ = t
    if self.whole:
        return self._transform_whole(x)
    return self._transform_channels(x)

Discard 

Discard(keys: tuple[str, ...] = ())

Bases: Transformer

Drop the named channels.

Parameters:

Name Type Description Default
keys tuple[str, ...]

names of the channels to drop. Keys not present in the input are silently ignored.

 ()

Examples:

>>> from dense_armor.roles import Signal
>>> from dense_armor.utility.compose import Discard
>>> s = Signal(values=[1.0, 2.0], names=["a", "b"],
...            units=["", ""], t=0.0)
>>> sorted(Discard(keys=("a",)).transform_one(s))
['b']
Source code in dense_armor/utility/compose/select.py 
102
103
def __init__(self, keys: tuple[str, ...] = ()) -> None:
    self.keys = list(keys)

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> Discard

No-op: the transformer is stateless.

Source code in dense_armor/utility/compose/select.py 
109
110
111
112
def learn_one(self, x: Any, y: Any = None, t: float | None = None) -> "Discard":
    """No-op: the transformer is stateless."""
    self._time_step(t)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> Any

Return a copy without the dropped keys.

Parameters:

Name Type Description Default
x Any

a :class:Signal or a plain dict.

 required
t float | None

unused.

 None

Returns:

Type Description
Any

A new :class:Signal (or dict) without the dropped keys.
Source code in dense_armor/utility/compose/select.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def transform_one(self, x: Any, t: float | None = None) -> Any:
    """Return a copy without the dropped keys.

    Args:
        x: a :class:`Signal` or a plain dict.
        t: unused.

    Returns:
        A new :class:`Signal` (or dict) without the dropped keys.
    """
    _ = t
    if isinstance(x, Signal):
        keep = [k for k in x.names if k not in self.keys]
    else:
        keep = [k for k in x if k not in self.keys]
    return _subset(x, keep)

Select 

Select(keys: tuple[str, ...] = ())

Bases: Transformer

Keep only the named channels.

Parameters:

Name Type Description Default
keys tuple[str, ...]

names of the channels to keep, in the order given. Keys not present in the input are silently ignored.

 ()

Examples:

>>> from dense_armor.roles import Signal
>>> from dense_armor.utility.compose import Select
>>> s = Signal(values=[1.0, 2.0], names=["a", "b"],
...            units=["", ""], t=0.0)
>>> sorted(Select(keys=("b",)).transform_one(s))
['b']
Source code in dense_armor/utility/compose/select.py 
60
61
def __init__(self, keys: tuple[str, ...] = ()) -> None:
    self.keys = list(keys)

learn_one 

learn_one(
    x: Any, y: Any = None, t: float | None = None
) -> Select

No-op: the transformer is stateless.

Source code in dense_armor/utility/compose/select.py 
67
68
69
70
def learn_one(self, x: Any, y: Any = None, t: float | None = None) -> "Select":
    """No-op: the transformer is stateless."""
    self._time_step(t)
    return self

transform_one 

transform_one(x: Any, t: float | None = None) -> Any

Return a copy with only the selected keys.

Parameters:

Name Type Description Default
x Any

a :class:Signal or a plain dict.

 required
t float | None

unused.

 None

Returns:

Type Description
Any

A new :class:Signal (or dict) with only the selected keys.
Source code in dense_armor/utility/compose/select.py 
72
73
74
75
76
77
78
79
80
81
82
83
def transform_one(self, x: Any, t: float | None = None) -> Any:
    """Return a copy with only the selected keys.

    Args:
        x: a :class:`Signal` or a plain dict.
        t: unused.

    Returns:
        A new :class:`Signal` (or dict) with only the selected keys.
    """
    _ = t
    return _subset(x, self.keys)

Details

  • iter_csv uses Python's csv module, one row at a time; a list cell "[a, b, c]" becomes channels <name>_0, <name>_1, ... (spaces in the name become underscores); start / stop select a window on time_col and stop reading after stop.
  • iter_rosbag(path, topics=None) reads ROS 1 and ROS 2 bags with the pure Python package rosbags (extra [ros]); sensor_msgs/msg/JointState becomes channels q_<joint>, qd_<joint>, tau_<joint> with the time of the message header.
  • merge_by_time is a k-way merge with a heap; it assumes each input stream is already in time order. shuffle(stream, buffer_size, seed) shuffles streams without timestamps in a bounded buffer.
  • Casper(path).stream(columns, start_h, stop_h) reads right_arm.csv from a local copy of the CASPER dataset (Kayan et al. 2023, section III-A: a UR3e arm logged at about 20 Hz, a 24-hour test with about half of the samples anomalous); hours are counted from the first timestamp. The code repository of the dataset is MIT (https://github.com/hkayann/CASPER-PerCom), the data are CC BY-SA 4.0 (https://www.kaggle.com/datasets/hkayan/industrial-robotic-arm-anomaly-detection); the library reads the user's copy and redistributes nothing.
  • DriftStream: each concept is two Gaussian clusters, one per class, placed along a random direction; the class of each cluster alternates between concepts, so a model trained on one concept is wrong on the next. Each sample uses the drift nearest to its position, so every transition is continuous on both sides. "sudden" is the limit of the sigmoid for a large spacing and does not use it; in "incremental" the label follows the concept with the larger weight. Implemented from the description in Ksieniewicz and Zyblewski (2020), sections 3.2.1 to 3.2.3 (pages 8–9).
  • SyntheticArm: sinusoidal joint trajectories at 30 % of each joint's range, inverse dynamics with RigidBodyModel, compiled once per URDF and process; faults: "friction" (viscous b * qd on one joint) or "payload" (a mass on the last link of the longest chain, or on payload_link).
  • Protected passes the timestamp only to models that accept it; HampelScorer.threshold is its n_sigmas, which AnomalyGate uses to decide.
  • Sources: Ksieniewicz, P., Zyblewski, P. (2020), "stream-learn — open-source Python library for difficult data stream batch analysis", arXiv:2001.11077. Kayan, H., Rana, O., Burnap, P., Perera, C. (2023), "CASPER: Context-aware anomaly detection system for industrial robotic arms".