Recording API

skelarm.recording

Time-series logging of robot states with TOML-described .sklog.npz files.

A :class:StateLog records a sequence of frames, each a timestamp plus a set of named channels (q, dq, tau and any extra signals a producer wants to keep, e.g. a controller's target or tracking error). The container is schema-agnostic: it stores whatever channels it is given, so new signals need no format change.

The canonical on-disk format is a numpy .npz archive holding one array per channel plus a __meta__ member with TOML metadata (schema version, producer, the embedded robot geometry, per-channel units/labels, and any free-form extra metadata such as the scenario config that produced the run). A standalone human-readable .toml export is also available for small logs.

StateLog

A time series of robot states, savable to and loadable from a file.

Parameters:

Name	Type	Description	Default
skeleton	Skeleton	The robot whose geometry is embedded in the log so it can be replayed or analyzed without the original config. Required for :meth:build_skeleton.	None
producer	str	A free-form name of whatever generated the log (e.g. "dynamics_simulator").	''
channel_meta	Mapping[str, Mapping[str, Any]]	Optional per-channel descriptors (unit, label, columns) used to document the physical meaning of each channel and label analysis plots.	None
extra	Mapping[str, Any]	Optional extra metadata stored under an [extra] table.	None

Source code in src/skelarm/recording.py

class StateLog:
    """A time series of robot states, savable to and loadable from a file.

    Parameters
    ----------
    skeleton : Skeleton, optional
        The robot whose geometry is embedded in the log so it can be replayed or
        analyzed without the original config. Required for :meth:`build_skeleton`.
    producer : str, optional
        A free-form name of whatever generated the log (e.g. ``"dynamics_simulator"``).
    channel_meta : Mapping[str, Mapping[str, Any]], optional
        Optional per-channel descriptors (``unit``, ``label``, ``columns``) used to
        document the physical meaning of each channel and label analysis plots.
    extra : Mapping[str, Any], optional
        Optional extra metadata stored under an ``[extra]`` table.
    """

    def __init__(
        self,
        skeleton: Skeleton | None = None,
        *,
        producer: str = "",
        channel_meta: Mapping[str, Mapping[str, Any]] | None = None,
        extra: Mapping[str, Any] | None = None,
    ) -> None:
        """Initialize an empty log."""
        self._skeleton_dict: dict[str, Any] | None = skeleton.to_dict() if skeleton is not None else None
        self._producer = producer
        self._channel_meta: dict[str, dict[str, Any]] = {k: dict(v) for k, v in (channel_meta or {}).items()}
        self._extra: dict[str, Any] = dict(extra or {})
        self._created_at = datetime.now(UTC).isoformat()
        self._records: list[dict[str, NDArray[np.float64]]] = []
        self._channel_order: list[str] | None = None

    def record(self, time: float, **channels: ArrayLike) -> None:
        """Append one frame at ``time`` with the given named channels.

        The set of channel names and each channel's width are fixed by the first
        call; later calls must match.

        Parameters
        ----------
        time : float
            The frame timestamp.
        **channels : ArrayLike
            Named channel values for this frame (scalars or 1-D arrays).

        Raises
        ------
        ValueError
            If the channel names or a channel's shape differ from the first frame.
        """
        values = {name: np.asarray(value, dtype=np.float64) for name, value in channels.items()}
        if self._channel_order is None:
            self._channel_order = list(values)
        elif set(values) != set(self._channel_order):
            msg = f"channel set {sorted(values)} does not match the log's channels {sorted(self._channel_order)}"
            raise ValueError(msg)
        else:
            first = self._records[0]
            for name in self._channel_order:
                if values[name].shape != first[name].shape:
                    msg = f"channel {name!r} has shape {values[name].shape}, expected {first[name].shape}"
                    raise ValueError(msg)
        row = {_TIME: np.asarray(time, dtype=np.float64)}
        for name in self._channel_order:
            row[name] = values[name]
        self._records.append(row)

    def record_skeleton(self, skeleton: Skeleton, time: float, **extra: ArrayLike) -> None:
        """Record the canonical robot state (``q``, ``dq``, ``tau``) plus extra channels."""
        self.record(time, q=skeleton.q, dq=skeleton.dq, tau=skeleton.tau, **extra)

    def __len__(self) -> int:
        """Return the number of recorded frames."""
        return len(self._records)

    @property
    def producer(self) -> str:
        """The name of whatever produced the log."""
        return self._producer

    @property
    def created_at(self) -> str:
        """ISO-8601 timestamp of when the log was created."""
        return self._created_at

    @property
    def channel_meta(self) -> dict[str, dict[str, Any]]:
        """Per-channel descriptors (``unit`` / ``label`` / ``columns``)."""
        return self._channel_meta

    @property
    def extra(self) -> dict[str, Any]:
        """Free-form extra metadata stored under the ``[extra]`` table."""
        return self._extra

    @property
    def channel_names(self) -> list[str]:
        """The recorded channel names (excluding ``time``), in record order."""
        return list(self._channel_order) if self._channel_order is not None else []

    @property
    def times(self) -> NDArray[np.float64]:
        """The frame timestamps as a 1-D array."""
        return np.array([record[_TIME] for record in self._records], dtype=np.float64)

    def channel(self, name: str) -> NDArray[np.float64]:
        """Return channel ``name`` stacked over frames (shape ``(N,)`` or ``(N, k)``)."""
        if name == _TIME:
            return self.times
        if self._channel_order is None or name not in self._channel_order:
            msg = f"unknown channel {name!r}; available: {self.channel_names}"
            raise KeyError(msg)
        return np.stack([record[name] for record in self._records])

    def to_arrays(self) -> dict[str, NDArray[np.float64]]:
        """Return every channel (including ``time``) as a mapping of stacked arrays."""
        arrays = {_TIME: self.times}
        for name in self.channel_names:
            arrays[name] = self.channel(name)
        return arrays

    def build_skeleton(self) -> Skeleton:
        """Reconstruct the robot embedded in the log.

        Raises
        ------
        ValueError
            If the log carries no embedded robot geometry.
        """
        if self._skeleton_dict is None:
            msg = "this log has no embedded skeleton"
            raise ValueError(msg)
        return Skeleton.from_dict(self._skeleton_dict)

    def save(self, path: str | Path) -> None:
        """Save the log as a ``.sklog.npz`` archive (channel arrays + TOML metadata)."""
        meta_toml = dump_toml(self._meta_dict())
        # dict[str, Any] so the meta string array and the float arrays can share one
        # ``**`` spread without tripping savez_compressed's typed ``allow_pickle`` kwarg.
        payload: dict[str, Any] = {**self.to_arrays(), _META_KEY: np.array(meta_toml)}
        np.savez_compressed(path, **payload)

    def export_toml(self, path: str | Path) -> None:
        """Write a standalone, human-readable TOML file (metadata + data arrays).

        Practical for small logs; for long runs prefer :meth:`save` (``.npz``).
        """
        document = self._meta_dict()
        document["data"] = {name: array.tolist() for name, array in self.to_arrays().items()}
        from pathlib import Path as _Path

        _Path(path).write_text(dump_toml(document), encoding="utf-8")

    @classmethod
    def load(cls, path: str | Path) -> StateLog:
        """Load a log previously written by :meth:`save`."""
        with np.load(path, allow_pickle=False) as archive:
            meta = tomllib.loads(str(archive[_META_KEY]))
            arrays = {name: archive[name] for name in archive.files if name != _META_KEY}
        log = cls()
        log._populate(meta, arrays)
        return log

    def _populate(self, meta: Mapping[str, Any], arrays: dict[str, NDArray[np.float64]]) -> None:
        """Fill this (freshly constructed) log from loaded metadata and channel arrays."""
        self._skeleton_dict = meta.get("skeleton")
        self._producer = meta.get("producer", "")
        self._channel_meta = meta.get("channels", {})
        self._extra = meta.get("extra", {})
        self._created_at = meta.get("created_at", "")

        columns = {name: array for name, array in arrays.items() if name != _TIME}
        times = arrays[_TIME]
        self._channel_order = list(columns)
        self._records = [
            {_TIME: np.asarray(times[i], dtype=np.float64), **{name: columns[name][i] for name in columns}}
            for i in range(len(times))
        ]

    def _meta_dict(self) -> dict[str, Any]:
        """Assemble the TOML-serializable metadata document."""
        meta: dict[str, Any] = {
            "schema_version": _SCHEMA_VERSION,
            "created_at": self._created_at,
            "producer": self._producer,
        }
        if self._skeleton_dict is not None:
            meta["skeleton"] = self._skeleton_dict
        if self._channel_meta:
            meta["channels"] = self._channel_meta
        if self._extra:
            meta["extra"] = self._extra
        return meta

channel_meta property

Per-channel descriptors (unit / label / columns).

channel_names property

The recorded channel names (excluding time), in record order.

created_at property

ISO-8601 timestamp of when the log was created.

extra property

Free-form extra metadata stored under the [extra] table.

producer property

The name of whatever produced the log.

times property

The frame timestamps as a 1-D array.

__init__(skeleton=None, *, producer='', channel_meta=None, extra=None)

Initialize an empty log.

Source code in src/skelarm/recording.py

def __init__(
    self,
    skeleton: Skeleton | None = None,
    *,
    producer: str = "",
    channel_meta: Mapping[str, Mapping[str, Any]] | None = None,
    extra: Mapping[str, Any] | None = None,
) -> None:
    """Initialize an empty log."""
    self._skeleton_dict: dict[str, Any] | None = skeleton.to_dict() if skeleton is not None else None
    self._producer = producer
    self._channel_meta: dict[str, dict[str, Any]] = {k: dict(v) for k, v in (channel_meta or {}).items()}
    self._extra: dict[str, Any] = dict(extra or {})
    self._created_at = datetime.now(UTC).isoformat()
    self._records: list[dict[str, NDArray[np.float64]]] = []
    self._channel_order: list[str] | None = None

__len__()

Return the number of recorded frames.

Source code in src/skelarm/recording.py

def __len__(self) -> int:
    """Return the number of recorded frames."""
    return len(self._records)

build_skeleton()

Reconstruct the robot embedded in the log.

Raises:

Type	Description
ValueError	If the log carries no embedded robot geometry.

Source code in src/skelarm/recording.py

def build_skeleton(self) -> Skeleton:
    """Reconstruct the robot embedded in the log.

    Raises
    ------
    ValueError
        If the log carries no embedded robot geometry.
    """
    if self._skeleton_dict is None:
        msg = "this log has no embedded skeleton"
        raise ValueError(msg)
    return Skeleton.from_dict(self._skeleton_dict)

channel(name)

Return channel name stacked over frames (shape (N,) or (N, k)).

Source code in src/skelarm/recording.py

def channel(self, name: str) -> NDArray[np.float64]:
    """Return channel ``name`` stacked over frames (shape ``(N,)`` or ``(N, k)``)."""
    if name == _TIME:
        return self.times
    if self._channel_order is None or name not in self._channel_order:
        msg = f"unknown channel {name!r}; available: {self.channel_names}"
        raise KeyError(msg)
    return np.stack([record[name] for record in self._records])

export_toml(path)

Write a standalone, human-readable TOML file (metadata + data arrays).

Practical for small logs; for long runs prefer :meth:save (.npz).

Source code in src/skelarm/recording.py

def export_toml(self, path: str | Path) -> None:
    """Write a standalone, human-readable TOML file (metadata + data arrays).

    Practical for small logs; for long runs prefer :meth:`save` (``.npz``).
    """
    document = self._meta_dict()
    document["data"] = {name: array.tolist() for name, array in self.to_arrays().items()}
    from pathlib import Path as _Path

    _Path(path).write_text(dump_toml(document), encoding="utf-8")

load(path) classmethod

Load a log previously written by :meth:save.

Source code in src/skelarm/recording.py

@classmethod
def load(cls, path: str | Path) -> StateLog:
    """Load a log previously written by :meth:`save`."""
    with np.load(path, allow_pickle=False) as archive:
        meta = tomllib.loads(str(archive[_META_KEY]))
        arrays = {name: archive[name] for name in archive.files if name != _META_KEY}
    log = cls()
    log._populate(meta, arrays)
    return log

record(time, **channels)

Append one frame at time with the given named channels.

The set of channel names and each channel's width are fixed by the first call; later calls must match.

Parameters:

Name	Type	Description	Default
time	float	The frame timestamp.	required
**channels	ArrayLike	Named channel values for this frame (scalars or 1-D arrays).	{}

Raises:

Type	Description
ValueError	If the channel names or a channel's shape differ from the first frame.

Source code in src/skelarm/recording.py

def record(self, time: float, **channels: ArrayLike) -> None:
    """Append one frame at ``time`` with the given named channels.

    The set of channel names and each channel's width are fixed by the first
    call; later calls must match.

    Parameters
    ----------
    time : float
        The frame timestamp.
    **channels : ArrayLike
        Named channel values for this frame (scalars or 1-D arrays).

    Raises
    ------
    ValueError
        If the channel names or a channel's shape differ from the first frame.
    """
    values = {name: np.asarray(value, dtype=np.float64) for name, value in channels.items()}
    if self._channel_order is None:
        self._channel_order = list(values)
    elif set(values) != set(self._channel_order):
        msg = f"channel set {sorted(values)} does not match the log's channels {sorted(self._channel_order)}"
        raise ValueError(msg)
    else:
        first = self._records[0]
        for name in self._channel_order:
            if values[name].shape != first[name].shape:
                msg = f"channel {name!r} has shape {values[name].shape}, expected {first[name].shape}"
                raise ValueError(msg)
    row = {_TIME: np.asarray(time, dtype=np.float64)}
    for name in self._channel_order:
        row[name] = values[name]
    self._records.append(row)

record_skeleton(skeleton, time, **extra)

Record the canonical robot state (q, dq, tau) plus extra channels.

Source code in src/skelarm/recording.py

def record_skeleton(self, skeleton: Skeleton, time: float, **extra: ArrayLike) -> None:
    """Record the canonical robot state (``q``, ``dq``, ``tau``) plus extra channels."""
    self.record(time, q=skeleton.q, dq=skeleton.dq, tau=skeleton.tau, **extra)

save(path)

Save the log as a .sklog.npz archive (channel arrays + TOML metadata).

Source code in src/skelarm/recording.py

def save(self, path: str | Path) -> None:
    """Save the log as a ``.sklog.npz`` archive (channel arrays + TOML metadata)."""
    meta_toml = dump_toml(self._meta_dict())
    # dict[str, Any] so the meta string array and the float arrays can share one
    # ``**`` spread without tripping savez_compressed's typed ``allow_pickle`` kwarg.
    payload: dict[str, Any] = {**self.to_arrays(), _META_KEY: np.array(meta_toml)}
    np.savez_compressed(path, **payload)

to_arrays()

Return every channel (including time) as a mapping of stacked arrays.

Source code in src/skelarm/recording.py

def to_arrays(self) -> dict[str, NDArray[np.float64]]:
    """Return every channel (including ``time``) as a mapping of stacked arrays."""
    arrays = {_TIME: self.times}
    for name in self.channel_names:
        arrays[name] = self.channel(name)
    return arrays

dump_toml(data)

Serialize a constrained nested mapping to TOML text.

Supports scalars (bool, int, float, str), (possibly nested) lists of scalars, sub-tables (dict), and arrays of tables (list of dict). This is enough for :class:StateLog metadata and exports; stdlib tomllib reads the result back.

Source code in src/skelarm/recording.py

def dump_toml(data: Mapping[str, Any]) -> str:
    """Serialize a constrained nested mapping to TOML text.

    Supports scalars (``bool``, ``int``, ``float``, ``str``), (possibly nested)
    lists of scalars, sub-tables (``dict``), and arrays of tables (``list`` of
    ``dict``). This is enough for :class:`StateLog` metadata and exports; stdlib
    ``tomllib`` reads the result back.
    """
    lines: list[str] = []
    _emit_table(data, [], lines)
    return "\n".join(lines) + "\n"