diff --git a/README.md b/README.md index 39c9235..c7b3645 100644 --- a/README.md +++ b/README.md @@ -52,42 +52,48 @@ pip install harp-data ## Quickstart -Have only a device's `device.yml`? `create_device` compiles it into a typed -`Device` at runtime — no code-generation step — giving you the device's registers -(keyed by address) and its identity: +There are two ways you'll typically use `harp`: talking to a **live device** over a +serial connection, or reading **data recorded to disk**. + +**Talk to a live device.** Open a connection and read/write registers by class: ```python -from pathlib import Path -from harp.device import create_device +from harp.device import Device, WhoAmI, OperationControl, OperationControlPayload, OperationMode +from harp.serial import open_serial_device -Behavior = create_device(Path("device.yml").read_text()) -Behavior.__whoami__ # device identity from the schema -AnalogData = Behavior.REGISTER_MAP[44] # registers are reached by address +# Use "COMx" on Windows, "/dev/ttyUSBx" on Linux. +with open_serial_device(Device, port="/dev/ttyUSB0") as device: + print("WhoAmI:", device.read(WhoAmI).parsed) + device.write(OperationControl, OperationControlPayload(operation_mode=OperationMode.ACTIVE)) ``` -The generated device works like any other. **Talk to hardware** over a serial -transport — `read`/`write` take a register class: +**Read a recorded session.** Point a `DatasetReader` at a dataset folder and read +registers into pandas DataFrames — no hardware required: ```python -from harp.serial import open_serial_device +from harp.data import create_dataset_reader -# Use "COMx" on Windows, "/dev/ttyUSBx" on Linux. -with open_serial_device(Behavior, port="/dev/ttyUSB0") as device: - print(device.read(AnalogData).parsed) +# Finds device.yml in the folder, builds the device, returns a ready-to-use reader. +reader = create_dataset_reader("session.harp") +df = reader.read(44) # one register, by address (or pass its class) +everything = reader.read_all() # {register_name: DataFrame} ``` -...or use the same register classes to **decode recorded data** into a pandas -DataFrame: +Both paths are driven by a device schema. If you have only a `device.yml` and no +pre-generated package, `create_device` compiles it into a typed `Device` at runtime — +no code-generation step — which is exactly what `create_dataset_reader` does under +the hood: ```python -from harp.data import parse_to_dataframe +from pathlib import Path +from harp.device import create_device -df = parse_to_dataframe(AnalogData, "Behavior_44.bin") +Behavior = create_device(Path("device.yml").read_text()) +AnalogData = Behavior.REGISTER_MAP[44] # registers are reached by address ``` -See the [Examples](https://harp-tech.org/pyharp/examples/) for full walkthroughs, -including reading device info, subscribing to events, and working with custom -interface-type converters. +See the [Examples](https://harp-tech.org/pyharp/examples/) for the full walkthroughs, +including subscribing to device events and working with custom interface-type converters. ## Contributing diff --git a/docs/api/data.md b/docs/api/data.md index b79926a..c66a43f 100644 --- a/docs/api/data.md +++ b/docs/api/data.md @@ -2,5 +2,11 @@ --- +::: harp.data.create_dataset_reader +::: harp.data.DatasetReader +::: harp.data.default_file_resolver ::: harp.data.parse_to_dataframe ::: harp.data.payload_to_dataframe +::: harp.data.to_file +::: harp.data.to_buffer +::: harp.data.REFERENCE_EPOCH diff --git a/docs/examples/index.md b/docs/examples/index.md index 492d90a..ff83850 100644 --- a/docs/examples/index.md +++ b/docs/examples/index.md @@ -14,4 +14,5 @@ Talking to a device: Reading recorded data: -- [Reading Data into a DataFrame](./read_data_to_dataframe/read_data_to_dataframe.md) - load a register's binary data file into a pandas DataFrame with `harp.data`. +- [Reading a Whole Dataset Folder](./read_dataset/read_dataset.md) - load an entire recorded session folder into pandas DataFrames with `DatasetReader`. +- [Reading Data into a DataFrame](./read_data_to_dataframe/read_data_to_dataframe.md) - decode a single register's binary file into a pandas DataFrame. diff --git a/docs/examples/read_data_to_dataframe/read_data_to_dataframe.md b/docs/examples/read_data_to_dataframe/read_data_to_dataframe.md index a205b79..d0637b4 100644 --- a/docs/examples/read_data_to_dataframe/read_data_to_dataframe.md +++ b/docs/examples/read_data_to_dataframe/read_data_to_dataframe.md @@ -1,8 +1,14 @@ # Reading Data into a DataFrame -This example demonstrates how to load a Harp register's binary data file into a -pandas DataFrame using `harp.data`. The register definition tells `parse_to_dataframe` -how to decode each frame, so you get named columns (and decoded enums) for free. +This example demonstrates how to load a **single** Harp register's binary data +file into a pandas DataFrame using `harp.data`. The register definition tells +`parse_to_dataframe` how to decode each frame, so you get named columns (and +decoded enums) for free. + +!!! tip + Have a whole recorded session folder rather than one loose file? Use + [`DatasetReader`](../read_dataset/read_dataset.md), which reads every register + in a dataset folder driven by the device schema. ```python diff --git a/docs/examples/read_data_to_dataframe/read_data_to_dataframe.py b/docs/examples/read_data_to_dataframe/read_data_to_dataframe.py index fda5a2d..df67c2c 100644 --- a/docs/examples/read_data_to_dataframe/read_data_to_dataframe.py +++ b/docs/examples/read_data_to_dataframe/read_data_to_dataframe.py @@ -1,11 +1,20 @@ from harp.data import parse_to_dataframe from harp.device import OperationControl -# Parse a register's binary dump into a pandas DataFrame — one row per frame, -# one column per field, plus a leading "timestamp" column. -df = parse_to_dataframe(OperationControl, "OperationControl.bin", timestamp=True) +# Parse a single register's binary dump into a pandas DataFrame — one row per +# frame, one column per field. The register class tells `parse_to_dataframe` how +# to decode each frame, so you get named columns (and decoded enums) for free. +df = parse_to_dataframe(OperationControl, "OperationControl.bin") print(df.head()) +# When the frames are timestamped (the default), the Harp time becomes the +# DataFrame index, named "Time" — float seconds from device start. +print(df.index.name, df.index[:3].to_list()) + # `parse_to_dataframe` also accepts raw bytes or an open binary file object: with open("OperationControl.bin", "rb") as f: df = parse_to_dataframe(OperationControl, f) + +# To read a whole recorded session folder at once (many registers, driven by the +# device schema) use `harp.data.DatasetReader` — see the "Reading a Whole Dataset +# Folder" example. diff --git a/docs/examples/read_dataset/read_dataset.md b/docs/examples/read_dataset/read_dataset.md new file mode 100644 index 0000000..2cd4ea6 --- /dev/null +++ b/docs/examples/read_dataset/read_dataset.md @@ -0,0 +1,25 @@ +# Reading a Whole Dataset Folder + +A Harp acquisition is usually saved as a **de-multiplexed dataset folder**: one +binary file per register, named `_
.bin`, next to the device's +`device.yml` schema. `harp.data.DatasetReader` reads that whole folder into pandas +DataFrames, driven by a [generated device](../../api/device.md) that describes how +to decode each register. + +This is the recommended entry point when you have a recorded session on disk. To +decode a single loose `.bin` file instead, see +[Reading Data into a DataFrame](../read_data_to_dataframe/read_data_to_dataframe.md). + +The quickest way in is `create_dataset_reader(folder)`: it finds the `device.yml` +inside the folder, builds the device for you, and returns a reader ready to go. +(If you already have a device class — e.g. from a pre-generated package — construct +`DatasetReader(Device, folder)` directly instead.) You then read a register by +class or by address, or read every register at once with `read_all()`. Timestamps +are detected automatically and placed on the `"Time"` index (float seconds, or an +absolute `DatetimeIndex` when you pass an `epoch`). + + +```python +[](./read_dataset.py) +``` + diff --git a/docs/examples/read_dataset/read_dataset.py b/docs/examples/read_dataset/read_dataset.py new file mode 100644 index 0000000..1e9b834 --- /dev/null +++ b/docs/examples/read_dataset/read_dataset.py @@ -0,0 +1,46 @@ +from harp.data import REFERENCE_EPOCH, create_dataset_reader +from harp.device import OperationControl + +# A Harp acquisition is usually saved as a de-multiplexed dataset folder — one +# `.bin` file per register, named "_
.bin", next to the +# device's `device.yml` schema: +# +# 📦 session.harp +# ┣ 📜 Behavior_0.bin +# ┣ 📜 Behavior_44.bin +# ┣ ... +# ┗ 📜 device.yml +# +# `create_dataset_reader` does the right thing: it finds `device.yml` inside the +# folder, builds the device that knows how to decode each register, and hands back +# a reader ready to go. +reader = create_dataset_reader("session.harp") + +# Read one register into a DataFrame — by register class (any register in the +# device's map, including the common ones like `OperationControl`)... +df = reader.read(OperationControl) + +# ...or by address. Timestamps are auto-detected from the frames, and when present +# they become the DataFrame index, named "Time" (float seconds from device start). +df = reader.read(44) +print(df.head()) + +# Read every register that has a file on disk at once, keyed by register name. +everything = reader.read_all() +print(list(everything)) + +# Pass an epoch to turn the "Time" index into an absolute `DatetimeIndex` instead +# of float seconds. `REFERENCE_EPOCH` is time zero of the Harp clock (UTC). +absolute = reader.read(44, epoch=REFERENCE_EPOCH) +print(absolute.index[:3]) + +# --- Already have a device class? ------------------------------------------- +# A pre-generated device package, or one you built yourself with `create_device`, +# can drive the reader directly — construct `DatasetReader(Device, folder)`: +# +# from harp.data import DatasetReader +# from harp.device import create_device +# from pathlib import Path +# +# Behavior = create_device((Path("session.harp") / "device.yml").read_text()) +# reader = DatasetReader(Behavior, "session.harp") diff --git a/mkdocs.yml b/mkdocs.yml index 6cf012d..c4a63b4 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -77,6 +77,7 @@ nav: - Getting Device Info: examples/get_info/get_info.md - Read and Write from Registers: examples/read_and_write_from_registers/read_and_write_from_registers.md - Subscribing to Events: examples/subscribing_to_events/subscribing_to_events.md + - Reading a Whole Dataset Folder: examples/read_dataset/read_dataset.md - Reading Data into a DataFrame: examples/read_data_to_dataframe/read_data_to_dataframe.md - API: - Protocol: api/protocol.md diff --git a/src/packages/harp-data/README.md b/src/packages/harp-data/README.md index abfdbde..80045b8 100644 --- a/src/packages/harp-data/README.md +++ b/src/packages/harp-data/README.md @@ -4,7 +4,59 @@ Load Harp register data into pandas DataFrames. This is the package that pulls in `pandas` — [`harp-protocol`](../harp-protocol) stays numpy-only and exposes a pandas-free `ColumnData` view that this package assembles into a DataFrame. -## Read a register from a file +There are two ways in, depending on what you have on disk: + +- a whole **dataset folder** (many registers) → `DatasetReader` +- a single **register file** or buffer → `parse_to_dataframe` + +## Read a whole dataset folder + +A Harp acquisition is usually saved as a de-multiplexed folder — one binary file +per register, named `_
.bin`, alongside the device's +`device.yml` schema: + +```text +📦 session.harp + ┣ 📜 Behavior_0.bin + ┣ 📜 Behavior_44.bin + ┣ ... + ┗ 📜 device.yml +``` + +Reading is driven by a generated +[`harp.device.Device`](../harp-device) that describes how to decode each register. +`create_dataset_reader` does that for you — it finds the `device.yml` in the folder, +builds the device, and returns a ready-to-use reader: + +```python +from harp.data import create_dataset_reader + +reader = create_dataset_reader("session.harp") +df = reader.read(AnalogData) # by register class +df = reader.read(44) # by address +everything = reader.read_all() # {register_name: DataFrame} +``` + +Already have a device class (e.g. a pre-generated package, or one built with +`create_device`)? Drive `DatasetReader` with it directly: + +```python +from pathlib import Path +from harp.data import DatasetReader +from harp.device import create_device + +Behavior = create_device((Path("session.harp") / "device.yml").read_text()) +reader = DatasetReader(Behavior, "session.harp") +``` + +Timestamps are auto-detected per register and placed on the DataFrame index +(named `"Time"`): float seconds by default, or an absolute `DatetimeIndex` when +you pass `epoch=REFERENCE_EPOCH`. Multi-chunk registers logged as +`_
_.bin` are concatenated in filename order; pass a +`resolver` to support an alternative on-disk layout, or `name=` to override the +file prefix. + +## Read a single register file `parse_to_dataframe` takes a register and a source (path, bytes, or open binary file) and returns one row per frame: @@ -17,7 +69,10 @@ df = parse_to_dataframe(AnalogData, "AnalogData.bin") df = parse_to_dataframe(AnalogData, raw, timestamp=True, message_type=False, decode_enums=True) ``` -Enum fields decode to `pd.Categorical` (`decode_enums=False` keeps raw codes). +With `timestamp=True` (the default) the Harp time becomes the DataFrame index, +named `"Time"` — float seconds, or an absolute `DatetimeIndex` when you also pass +`epoch=REFERENCE_EPOCH`. Enum fields decode to `pd.Categorical` +(`decode_enums=False` keeps raw codes). ## From an already-parsed payload @@ -30,3 +85,14 @@ from harp.data import payload_to_dataframe _data, timestamps, _msg, payload = AnalogData.parse_bulk(raw) df = payload_to_dataframe(payload) ``` + +## Write data back out + +`to_file` / `to_buffer` are the inverse of the readers — encode values as Harp +frames. Handy for round-tripping data or generating test corpora: + +```python +from harp.data import to_file + +to_file(AnalogData, values, "AnalogData.bin", timestamps=seconds) +``` diff --git a/src/packages/harp-data/pyproject.toml b/src/packages/harp-data/pyproject.toml index 47bdfff..118792e 100644 --- a/src/packages/harp-data/pyproject.toml +++ b/src/packages/harp-data/pyproject.toml @@ -5,6 +5,7 @@ description = "Load Harp device data into pandas DataFrames" requires-python = ">=3.11" dependencies = [ "harp-protocol", + "harp-device", "numpy>=1.24", "pandas>=2.0", ] diff --git a/src/packages/harp-data/src/harp/data/__init__.py b/src/packages/harp-data/src/harp/data/__init__.py index ec595a5..39adede 100644 --- a/src/packages/harp-data/src/harp/data/__init__.py +++ b/src/packages/harp-data/src/harp/data/__init__.py @@ -1,5 +1,6 @@ +from ._dataset import DatasetReader, create_dataset_reader, default_file_resolver from ._read import read -from ._reader import parse_to_dataframe, payload_to_dataframe +from ._reader import REFERENCE_EPOCH, parse_to_dataframe, payload_to_dataframe from ._write import to_buffer, to_file __all__ = [ @@ -8,4 +9,8 @@ "payload_to_dataframe", "to_buffer", "to_file", + "DatasetReader", + "create_dataset_reader", + "default_file_resolver", + "REFERENCE_EPOCH", ] diff --git a/src/packages/harp-data/src/harp/data/_dataset.py b/src/packages/harp-data/src/harp/data/_dataset.py new file mode 100644 index 0000000..bad29a6 --- /dev/null +++ b/src/packages/harp-data/src/harp/data/_dataset.py @@ -0,0 +1,219 @@ +import re +from collections.abc import Callable, Mapping +from datetime import datetime +from os import PathLike +from pathlib import Path +from typing import Any + +import pandas as pd +from harp.device import Device, create_device +from harp.protocol import RegisterBase +from harp.protocol._constants import _TIMESTAMP_FLAG + +from ._reader import parse_to_dataframe + +RegisterKey = type[RegisterBase[Any]] | int + +FileNameResolver = Callable[[Path, str], Mapping[int, list[Path]]] + +#: Default filename of the device schema looked up inside a dataset folder. +DEVICE_SCHEMA_FILENAME = "device.yml" + + +def default_file_resolver(root: Path, name: str) -> dict[int, list[Path]]: + """Harp file format resolver: map address -> sorted ``_
...`` files.""" + pattern = re.compile(rf"^{re.escape(name)}_(\d+)(?:_.*)?$") + files: dict[int, list[Path]] = {} + for path in sorted(root.glob("*.bin")): + match = pattern.match(path.stem) + if match is not None: + files.setdefault(int(match.group(1)), []).append(path) + return files + + +class DatasetReader: + """Reader over a de-multiplexed Harp dataset folder. + + Construct from a generated device and a dataset folder, then read a register's + frames into a DataFrame by register class or by address:: + + reader = DatasetReader(Behavior, "session.harp") + df = reader.read(AnalogData) # by register class + df = reader.read(44) # by address + everything = reader.read_all() # {register_name: DataFrame} + + ``device`` is a generated :class:`~harp.device.Device` subclass; its + ``REGISTER_MAP`` and class name are read on demand. ``name`` overrides the + ```` file prefix, which defaults to the device class name. + + File resolution defaults to the Harp file format: ``_
.bin`` and, + when a register was logged as several ``_
_.bin`` chunks, + they are concatenated in filename order. Pass ``resolver`` (a :data:`FileResolver`) + to support an alternative on-disk layout. + """ + + def __init__( + self, + device: type[Device], + root: str | PathLike[str], + *, + name: str | None = None, + resolver: FileNameResolver = default_file_resolver, + ) -> None: + self._device = device + self._root = Path(root) + self._name_override = name + self._resolver = resolver + self._files = dict(self._resolver(self._root, self.name)) + + @property + def root(self) -> Path: + """The dataset folder being read.""" + return self._root + + @property + def device(self) -> type[Device]: + """The generated device this reader parses against.""" + return self._device + + @property + def name(self) -> str: + """The ```` prefix used to match binary files.""" + return self._name_override or self._device.__name__ + + @property + def registers(self) -> Mapping[int, type[RegisterBase[Any]]]: + """The device's address -> register-class map.""" + return self._device.REGISTER_MAP + + @property + def files(self) -> Mapping[int, list[Path]]: + """The discovered address -> binary file(s) present under :attr:`root`.""" + return self._files + + def read( + self, + register: RegisterKey, + *, + suffix: str | None = None, + timestamp: bool | None = None, + epoch: datetime | None = None, + message_type: bool = False, + decode_enums: bool = True, + demux_bit_masks: bool = False, + ) -> pd.DataFrame: + """Read one register's data into a DataFrame. + + ``register`` is a register class or an address. ``suffix`` selects a single + ``_
_.bin`` chunk (default: concatenate every chunk + for the address). ``timestamp`` defaults to ``None`` — auto-detect from the + frame's payload-type bit; pass ``True``/``False`` to force. ``epoch`` makes + the ``"Time"`` index absolute (e.g. :data:`~harp.data.REFERENCE_EPOCH`). The + remaining options match :func:`~harp.data.parse_to_dataframe`. + """ + cls, address = self._resolve(register) + paths = self._resolve_files(address, suffix) + raw = b"".join(p.read_bytes() for p in paths) + ts = self._first_frame_timestamped(raw) if timestamp is None else timestamp + return parse_to_dataframe( + cls, + raw, + timestamp=ts, + epoch=epoch, + message_type=message_type, + decode_enums=decode_enums, + demux_bit_masks=demux_bit_masks, + ) + + def read_all( + self, + *, + timestamp: bool | None = None, + epoch: datetime | None = None, + message_type: bool = False, + decode_enums: bool = True, + demux_bit_masks: bool = False, + ) -> dict[str, pd.DataFrame]: + """Read every register that has a file present, keyed by register name. + + Files whose address is not in the device's registers are skipped. + Options are forwarded to :meth:`read`. + """ + registers = self.registers + out: dict[str, pd.DataFrame] = {} + for address in sorted(self._files): + cls = registers.get(address) + if cls is None: + continue + out[cls.__name__] = self.read( + address, + timestamp=timestamp, + epoch=epoch, + message_type=message_type, + decode_enums=decode_enums, + demux_bit_masks=demux_bit_masks, + ) + return out + + def _resolve(self, register: RegisterKey) -> tuple[type[RegisterBase[Any]], int]: + if isinstance(register, type): + return register, register.address + cls = self.registers.get(register) + if cls is None: + raise KeyError(f"No register at address {register} in this device's map.") + return cls, register + + def _resolve_files(self, address: int, suffix: str | None) -> list[Path]: + paths = self._files.get(address) + if not paths: + raise FileNotFoundError( + f"No data file for register address {address} under {self._root} " + f"(expected '{self.name}_{address}[_].bin')." + ) + if suffix is not None: + paths = [p for p in paths if p.stem.endswith(f"_{suffix}")] + if not paths: + raise FileNotFoundError( + f"No '_{suffix}' chunk for register address {address} under {self._root}." + ) + return paths + + @staticmethod + def _first_frame_timestamped(raw: bytes) -> bool: + """Whether the first frame carries a timestamp (payload-type bit ``0x10``).""" + return len(raw) > 4 and bool(raw[4] & _TIMESTAMP_FLAG) + + +def create_dataset_reader( + root: str | PathLike[str], + *, + schema: str | PathLike[str] | None = None, + name: str | None = None, + resolver: FileNameResolver = default_file_resolver, + converters: Mapping[str, Any] | None = None, + strict: bool = True, +) -> DatasetReader: + """Build a :class:`DatasetReader` for a dataset folder, device and all. + + Convenience wrapper that finds the device schema inside ``root`` (``device.yml`` + by default), generates a device from it with :func:`~harp.device.create_device`, + and returns a reader ready to :meth:`~DatasetReader.read`:: + + reader = create_dataset_reader("session.harp") + df = reader.read(44) + + ``schema`` points at the schema file explicitly when it isn't ``root/device.yml``. + ``converters`` and ``strict`` are forwarded to :func:`~harp.device.create_device` + for custom ``interfaceType`` decoding; ``name`` and ``resolver`` are forwarded to + :class:`DatasetReader`. Use ``DatasetReader(device, root)`` directly when you + already have a (e.g. pre-generated) device class. + """ + root_path = Path(root) + schema_path = Path(schema) if schema is not None else root_path / DEVICE_SCHEMA_FILENAME + if not schema_path.is_file(): + raise FileNotFoundError( + f"No device schema at '{schema_path}'. Pass schema= to point at a device.yml, " + f"or build the device yourself and use DatasetReader(device, root)." + ) + device = create_device(schema_path.read_text(), converters=converters, strict=strict) + return DatasetReader(device, root_path, name=name, resolver=resolver) diff --git a/src/packages/harp-data/src/harp/data/_reader.py b/src/packages/harp-data/src/harp/data/_reader.py index 8ff9ae3..f596a7b 100644 --- a/src/packages/harp-data/src/harp/data/_reader.py +++ b/src/packages/harp-data/src/harp/data/_reader.py @@ -1,16 +1,32 @@ """Load Harp register data into pandas DataFrames.""" +from datetime import datetime from pathlib import Path from typing import Any, BinaryIO, Union import numpy as np import pandas as pd from harp.protocol import RegisterBase +from numpy.typing import NDArray Source = Union[str, Path, bytes, bytearray, memoryview, BinaryIO] _MSG_NAMES = np.array(["_NONE", "Read", "Write", "Event"]) +#: Harp reference epoch — time zero of the Harp clock (UTC). +REFERENCE_EPOCH = datetime(1904, 1, 1) + +_TIME_INDEX_NAME = "Time" + + +def _time_index(seconds: NDArray[np.float64], epoch: datetime | None) -> pd.Index: + """The Harp time axis: float seconds, or absolute datetime when ``epoch`` is set.""" + if epoch is None: + return pd.Index(seconds, name=_TIME_INDEX_NAME) + return pd.DatetimeIndex( + pd.Timestamp(epoch) + pd.to_timedelta(seconds, unit="s"), name=_TIME_INDEX_NAME + ) + def _read_bytes(source: Source) -> bytes: if isinstance(source, (bytes, bytearray, memoryview)): @@ -53,17 +69,21 @@ def parse_to_dataframe( source: Source, *, timestamp: bool = True, + epoch: Union[datetime, None] = None, message_type: bool = False, decode_enums: bool = True, demux_bit_masks: bool = False, ) -> pd.DataFrame: """Parse all frames of ``register`` from ``source`` into a DataFrame. - ``source`` may be a file path, raw bytes, or an open binary file object. - ``timestamp`` and ``message_type`` insert leading columns; ``decode_enums`` - controls whether enum fields become ``pd.Categorical`` (True) or raw codes; - ``demux_bit_masks`` expands each flag (``BitMask``) field into one boolean - column per flag member (True) or keeps it as a single raw-integer column. + ``source`` may be a file path, raw bytes, or an open binary file object. When + ``timestamp`` is set, the Harp time becomes the DataFrame index (named + ``"Time"``): float seconds by default, or an absolute ``DatetimeIndex`` when + ``epoch`` is given (e.g. :data:`REFERENCE_EPOCH`). ``message_type`` inserts a + leading column; ``decode_enums`` controls whether enum fields become + ``pd.Categorical`` (True) or raw codes; ``demux_bit_masks`` expands each flag + (``BitMask``) field into one boolean column per flag member (True) or keeps it + as a single raw-integer column. """ raw = _read_bytes(source) _data, timestamps, msg_view, payload = register.parse_bulk(raw, parse_timestamp=timestamp) @@ -80,9 +100,10 @@ def parse_to_dataframe( if len(df) > 0: raise ValueError( "Buffer contains no timestamp data; pass timestamp=False to suppress " - "the timestamp column." + "the time index." ) - # Empty buffer: no frames to timestamp — return the empty frame as-is. + seconds = np.empty(0, dtype=np.float64) # empty buffer: empty Time index else: - df.insert(0, "timestamp", timestamps) + seconds = np.asarray(timestamps, dtype=np.float64) + df.index = _time_index(seconds, epoch) return df diff --git a/tests/data/__init__.py b/tests/data/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/data/test_dataset.py b/tests/data/test_dataset.py new file mode 100644 index 0000000..581f415 --- /dev/null +++ b/tests/data/test_dataset.py @@ -0,0 +1,232 @@ +import re + +import numpy as np +import pandas as pd +import pytest +from harp.data import ( + REFERENCE_EPOCH, + DatasetReader, + create_dataset_reader, + parse_to_dataframe, +) +from harp.device import create_device + + +def _records(cls, n, seed): + dtype = cls.payload_class.dtype + rng = np.random.default_rng(seed) + raw = rng.integers(0, 128, size=n * dtype.itemsize, dtype=np.uint8) + return raw.view(dtype).copy() + + +@pytest.fixture +def emitted_device(device_yml): + # strict=False: the test device.yml uses a custom DataConverter we don't inject + # here; native decoding is enough to exercise file resolution and parsing. + return create_device(device_yml, strict=False) + + +@pytest.fixture +def dataset(emitted_device, tmp_path): + """A dataset folder with three app registers; the first is timestamped.""" + dev = emitted_device + name = dev.__name__ + addresses = [a for a in sorted(dev.REGISTER_MAP) if a >= 32][:3] + specs = {} + for i, address in enumerate(addresses): + cls = dev.REGISTER_MAP[address] + records = _records(cls, 5, seed=address) + timestamped = i == 0 + timestamps = np.arange(5, dtype=np.float64) if timestamped else None + buf = bytes(cls.format_bulk(records, timestamps=timestamps)) + (tmp_path / f"{name}_{address}.bin").write_bytes(buf) + specs[address] = (cls, timestamped, buf) + return dev, name, tmp_path, specs + + +def test_read_by_class_and_by_address(dataset): + dev, _name, root, specs = dataset + reader = DatasetReader(dev, root) + for address, (cls, timestamped, buf) in specs.items(): + expected = parse_to_dataframe(cls, buf, timestamp=timestamped) + assert reader.read(cls).equals(expected) + assert reader.read(address).equals(expected) + + +def test_timestamp_is_auto_detected(dataset): + dev, _name, root, specs = dataset + reader = DatasetReader(dev, root) + for address, (_cls, timestamped, _buf) in specs.items(): + df = reader.read(address) + # Timestamped frames get a "Time" index; untimestamped keep a plain RangeIndex. + assert (df.index.name == "Time") is timestamped + + +def test_time_index_is_float_seconds_without_epoch(dataset): + dev, _name, root, specs = dataset + reader = DatasetReader(dev, root) + address = next(a for a, (_c, ts, _b) in specs.items() if ts) # the timestamped register + df = reader.read(address) + assert df.index.name == "Time" + assert list(df.index) == [0.0, 1.0, 2.0, 3.0, 4.0] # arange(5) seconds from the fixture + + +def test_epoch_gives_absolute_datetime_index(dataset): + dev, _name, root, specs = dataset + reader = DatasetReader(dev, root) + address = next(a for a, (_c, ts, _b) in specs.items() if ts) + df = reader.read(address, epoch=REFERENCE_EPOCH) + assert isinstance(df.index, pd.DatetimeIndex) + assert df.index.name == "Time" + # Harp seconds are measured from the reference epoch (timestamps were arange(5)). + assert df.index[0] == pd.Timestamp(REFERENCE_EPOCH) + assert df.index[2] == pd.Timestamp(REFERENCE_EPOCH) + pd.Timedelta(seconds=2) + + +def test_read_all_keyed_by_register_name(dataset): + dev, _name, root, specs = dataset + reader = DatasetReader(dev, root) + frames = reader.read_all() + assert set(frames) == {cls.__name__ for cls, _ts, _buf in specs.values()} + for cls, _timestamped, _buf in specs.values(): + assert frames[cls.__name__].equals(reader.read(cls.address)) + + +def test_suffix_chunks_are_concatenated(emitted_device, tmp_path): + dev = emitted_device + name = dev.__name__ + address = next(a for a in sorted(dev.REGISTER_MAP) if a >= 32) + cls = dev.REGISTER_MAP[address] + chunk0 = bytes(cls.format_bulk(_records(cls, 3, seed=1))) + chunk1 = bytes(cls.format_bulk(_records(cls, 2, seed=2))) + (tmp_path / f"{name}_{address}_0.bin").write_bytes(chunk0) + (tmp_path / f"{name}_{address}_1.bin").write_bytes(chunk1) + + reader = DatasetReader(dev, tmp_path) + combined = parse_to_dataframe(cls, chunk0 + chunk1, timestamp=False) + assert reader.read(cls).reset_index(drop=True).equals(combined) + # A specific chunk can still be selected by suffix. + only0 = parse_to_dataframe(cls, chunk0, timestamp=False) + assert reader.read(cls, suffix="0").equals(only0) + + +def test_non_device_raises_on_register_access(dataset): + _dev, _name, root, _specs = dataset + # Registers are derived lazily; a non-Device fails when they are accessed. + reader = DatasetReader(object, root) + with pytest.raises(AttributeError, match="REGISTER_MAP"): + _ = reader.registers + + +def test_explicit_name_overrides(dataset): + dev, name, root, _specs = dataset + reader = DatasetReader(dev, root, name=name) + assert isinstance(reader, DatasetReader) + assert reader.name == name + + +def test_missing_register_file_raises(dataset): + dev, _name, root, _specs = dataset + reader = DatasetReader(dev, root) + # WhoAmI (address 0) is in the map but has no file in this dataset. + with pytest.raises(FileNotFoundError): + reader.read(0) + + +def test_unknown_address_raises(dataset): + dev, _name, root, _specs = dataset + reader = DatasetReader(dev, root) + with pytest.raises(KeyError): + reader.read(9999) + + +def test_custom_file_resolver_supports_alternative_layout(emitted_device, tmp_path): + dev = emitted_device + addresses = [a for a in sorted(dev.REGISTER_MAP) if a >= 32][:2] + expected = {} + for address in addresses: + cls = dev.REGISTER_MAP[address] + buf = bytes(cls.format_bulk(_records(cls, 3, seed=address))) + (tmp_path / f"reg{address}.bin").write_bytes(buf) # not the Harp layout + expected[cls.__name__] = parse_to_dataframe(cls, buf, timestamp=False) + + def resolver(root, _name): + found = {} + for path in sorted(root.glob("reg*.bin")): + match = re.match(r"^reg(\d+)$", path.stem) + if match is not None: + found.setdefault(int(match.group(1)), []).append(path) + return found + + reader = DatasetReader(dev, tmp_path, resolver=resolver) + assert set(reader.files) == set(addresses) + frames = reader.read_all() + assert set(frames) == set(expected) + for register_name, df in frames.items(): + assert df.equals(expected[register_name]) + + +def test_files_property_lists_discovered_bins(dataset): + dev, _name, root, specs = dataset + reader = DatasetReader(dev, root) + assert set(reader.files) == set(specs) + + +def test_read_all_registers_of_mock_device(emitted_device, tmp_path): + """Write one .bin per register of the device.yml device, then read them all back.""" + dev = emitted_device + name = dev.__name__ + expected = {} + for address, cls in dev.REGISTER_MAP.items(): + records = _records(cls, 4, seed=address) + # Alternate timestamped/untimestamped to exercise both parse paths. + timestamped = address % 2 == 0 + timestamps = np.arange(4, dtype=np.float64) if timestamped else None + buf = bytes(cls.format_bulk(records, timestamps=timestamps)) + (tmp_path / f"{name}_{address}.bin").write_bytes(buf) + expected[cls.__name__] = parse_to_dataframe(cls, buf, timestamp=timestamped) + + reader = DatasetReader(dev, tmp_path) + frames = reader.read_all() + + assert set(reader.files) == set(dev.REGISTER_MAP) + assert set(frames) == set(expected) + assert len(frames) == len(dev.REGISTER_MAP) + for register_name, df in frames.items(): + assert len(df) == 4 + assert df.equals(expected[register_name]) + + +def test_reader_derives_name_and_registers_from_device(dataset): + dev, name, root, _specs = dataset + reader = DatasetReader(dev, root) + assert reader.device is dev + assert reader.name == name + assert reader.registers == dev.REGISTER_MAP + + +def test_create_dataset_reader_builds_device_from_device_yml(dataset, device_yml): + dev, _name, root, specs = dataset + (root / "device.yml").write_text(device_yml) + # strict=False mirrors the emitted_device fixture (custom DataConverter not injected). + reader = create_dataset_reader(root, strict=False) + assert isinstance(reader, DatasetReader) + # Reads match a reader built from an explicitly-generated device. + reference = DatasetReader(dev, root) + for address, (cls, _timestamped, _buf) in specs.items(): + assert reader.read(address).equals(reference.read(cls)) + + +def test_create_dataset_reader_accepts_explicit_schema_path(dataset, device_yml, tmp_path): + _dev, _name, root, specs = dataset + schema_path = tmp_path / "elsewhere.yml" # not inside the dataset folder + schema_path.write_text(device_yml) + reader = create_dataset_reader(root, schema=schema_path, strict=False) + address = next(iter(specs)) + assert not reader.read(address).empty + + +def test_create_dataset_reader_missing_schema_raises(dataset): + _dev, _name, root, _specs = dataset # no device.yml written into the folder + with pytest.raises(FileNotFoundError, match="device.yml"): + create_dataset_reader(root) diff --git a/uv.lock b/uv.lock index 5a2cb84..bff724e 100644 --- a/uv.lock +++ b/uv.lock @@ -400,6 +400,7 @@ requires-dist = [ name = "harp-data" source = { editable = "src/packages/harp-data" } dependencies = [ + { name = "harp-device" }, { name = "harp-protocol" }, { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" }, { name = "numpy", version = "2.5.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, @@ -408,6 +409,7 @@ dependencies = [ [package.metadata] requires-dist = [ + { name = "harp-device", editable = "src/packages/harp-device" }, { name = "harp-protocol", editable = "src/packages/harp-protocol" }, { name = "numpy", specifier = ">=1.24" }, { name = "pandas", specifier = ">=2.0" },