diff --git a/docs/api/data.md b/docs/api/data.md index ab15e37..7359302 100644 --- a/docs/api/data.md +++ b/docs/api/data.md @@ -6,6 +6,7 @@ ::: harp.data.read ::: harp.data.DatasetReader ::: harp.data.default_file_resolver +::: harp.data.FileNameResolver ::: harp.data.parse_to_dataframe ::: harp.data.payload_to_dataframe ::: harp.data.to_file diff --git a/docs/api/device.md b/docs/api/device.md index 8dc7c0a..668ac0d 100644 --- a/docs/api/device.md +++ b/docs/api/device.md @@ -3,6 +3,8 @@ --- ::: harp.device.client.Device +::: harp.device.client.attach_writer +::: harp.device.client.DeviceWriter ::: harp.device.client.DeviceError ::: harp.device.client.Subscription ::: harp.device.client.EventHandler @@ -14,4 +16,5 @@ ::: harp.device.schema.ConverterContext ::: harp.device.schema.DeviceModule ::: harp.device.schema.DeviceModuleLike +::: harp.device.schema.DEVICE_SCHEMA_FILENAME ::: harp.device.core diff --git a/docs/examples/record-dataset.md b/docs/examples/record-dataset.md new file mode 100644 index 0000000..13fa90c --- /dev/null +++ b/docs/examples/record-dataset.md @@ -0,0 +1,15 @@ +# Record a dataset folder + +`harp.device.client.attach_writer` records everything a device emits into a **de-multiplexed dataset folder**, the layout [Read a dataset folder](dataset.md) reads back: one binary file per register, named `_
.bin`, next to a copy of the `device.yml` the module was built from. + +Each message is appended to the file of its own register as the complete Harp frame it arrived as, header and checksum included, which is what the reference C# writer records. The file of a register is created the first time a message for it arrives, so the folder holds exactly the registers that were seen. + +The writer owns its subscription, so leaving the `with` block both detaches it from the device and closes the files. All three message types are recorded by default: the register dump a device performs on request arrives as `Read` messages, so recording only `Event` would drop the configuration the session ran under. + +{% include-markdown "includes/serial-port.md" %} + + +```python +[](./record_dataset.py) +``` + diff --git a/docs/examples/record_dataset.py b/docs/examples/record_dataset.py new file mode 100644 index 0000000..2b456f1 --- /dev/null +++ b/docs/examples/record_dataset.py @@ -0,0 +1,35 @@ +from pathlib import Path + +from harp import data +from harp import serial +from harp.device import client, core, schema + +SERIAL_PORT = "/dev/ttyUSB0" # or "COMx" in Windows, where "x" is the serial port number + +behavior = schema.create_device_module(Path("device.yml").read_bytes()) + +with serial.open_device(behavior, port=SERIAL_PORT) as device: + # The writer owns its subscription, so leaving this block detaches it from the + # device and closes the files. `device.yml` is copied into the folder as it opens. + with client.attach_writer(device, "session.harp") as writer: + # Ask the device to report every register, so the folder records the + # configuration the session ran under and not only its events. Do this once the + # writer is attached, or the reply burst is missed. + device.write( + core.OperationControl, + core.OperationControlPayload( + operation_mode=core.OperationMode.ACTIVE, + dump_registers=True, + heartbeat=core.EnableFlag.ENABLED, + mute_replies=False, + operation_led=core.EnableFlag.ENABLED, + visual_indicators=core.EnableFlag.ENABLED, + ), + ) + + input("Recording. Press Enter to stop.\n") + print("recorded:", {address: path.name for address, path in writer.paths.items()}) + +# The folder carries its own schema, so it reads back without a module in hand. +reader = data.open_dataset("session.harp") +print(reader.contents) diff --git a/mkdocs.yml b/mkdocs.yml index 1045a5f..42eb057 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -76,6 +76,7 @@ nav: - Read and write registers: examples/read-write-registers.md - Subscribe to events: examples/events.md - Read a dataset folder: examples/dataset.md + - Record a dataset folder: examples/record-dataset.md - Read a single register file: examples/register-file.md - Registers from a schema: examples/registers-from-schema.md - Guides: diff --git a/src/packages/harp-data/README.md b/src/packages/harp-data/README.md index faf537d..cbac063 100644 --- a/src/packages/harp-data/README.md +++ b/src/packages/harp-data/README.md @@ -59,6 +59,8 @@ The `` prefix comes from the `DEVICE_NAME` declared by the device mo When a device module declaring an identity is supplied and the folder carries a `device.yml`, their `whoAmI` values are checked against each other. Reusing a module across sessions and opening the wrong folder then fails on construction rather than decoding the files against the wrong register map. Pass `validate=False` to turn off every check the reader performs, so a folder whose `device.yml` is damaged can be read with a module obtained elsewhere. +Recordings are produced by `harp.device.client.attach_writer`, on the other side of the same file format. See the [harp-device README](https://github.com/harp-tech/python/tree/main/src/packages/harp-device). + ## Read a single register file `parse_to_dataframe` takes a register and a source, either a path, bytes, or an open binary file, and returns one row per frame: diff --git a/src/packages/harp-data/src/harp/data/__init__.py b/src/packages/harp-data/src/harp/data/__init__.py index ea4b57f..427d8d3 100644 --- a/src/packages/harp-data/src/harp/data/__init__.py +++ b/src/packages/harp-data/src/harp/data/__init__.py @@ -1,4 +1,4 @@ -from ._dataset import DatasetReader, default_file_resolver, open_dataset +from ._dataset import DatasetReader, FileNameResolver, default_file_resolver, open_dataset from ._read import read from ._reader import REFERENCE_EPOCH, parse_to_dataframe, payload_to_dataframe from ._write import to_buffer, to_file @@ -12,5 +12,6 @@ "DatasetReader", "open_dataset", "default_file_resolver", + "FileNameResolver", "REFERENCE_EPOCH", ] diff --git a/src/packages/harp-data/src/harp/data/_dataset.py b/src/packages/harp-data/src/harp/data/_dataset.py index 976c531..855b058 100644 --- a/src/packages/harp-data/src/harp/data/_dataset.py +++ b/src/packages/harp-data/src/harp/data/_dataset.py @@ -7,6 +7,7 @@ import pandas as pd from harp.device.schema import ( + DEVICE_SCHEMA_FILENAME, DeviceModule, DeviceModuleLike, create_device_module, @@ -22,9 +23,6 @@ FileNameResolver = Callable[[Path, str], Mapping[int, list[Path]]] -DEVICE_SCHEMA_FILENAME = "device.yml" -"""Default filename of the device schema looked up inside a dataset folder.""" - def default_file_resolver(root: Path, name: str) -> dict[int, list[Path]]: """Harp file format resolver: map address -> sorted ``_
...`` files.""" @@ -70,7 +68,7 @@ class DatasetReader(Generic[M]): 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`) + they are concatenated in filename order. Pass ``resolver`` (a :data:`FileNameResolver`) to support an alternative on-disk layout. ``epoch`` anchors the time index of every read to absolute time, so one dataset is diff --git a/src/packages/harp-device/README.md b/src/packages/harp-device/README.md index 5a7f900..8db90d8 100644 --- a/src/packages/harp-device/README.md +++ b/src/packages/harp-device/README.md @@ -25,7 +25,9 @@ A device is described by a module. Downstream, often generated, packages record ```python from harp.device.core import REGISTER_MAP as _CORE_REGISTER_MAP +DEVICE_NAME: str = "Behavior" WHO_AM_I: int = 1216 +DEVICE_METADATA: bytes = ... # the device.yml the package was built from REGISTER_MAP = {**_CORE_REGISTER_MAP, 32: DigitalInputState, ...} ``` @@ -33,6 +35,8 @@ This is the same structure `create_device_module` builds from a schema, so a dev A device module names only what its schema declares, the registers beside the enums and payload classes built from them. The core registers and any core mask reused by the schema have a single definition, in `harp.device.core`, and are accessed from there rather than through the device module. The core register set is not a device, so it carries no `WHO_AM_I`. `REGISTER_MAP` covers the complete device address space, including both core and application registers. +`DEVICE_METADATA` is the `device.yml` the module was built from, as bytes, so the schema travels with the module and a recording can carry a copy of it without the original file being at hand. `DeviceWriter` is what usually reads it, copying it into the folder it records. + Pass the module to `Device`, or to `open_device`, to validate identity on open: ```python diff --git a/src/packages/harp-device/src/harp/device/client/__init__.py b/src/packages/harp-device/src/harp/device/client/__init__.py index 15e23bd..2b54495 100644 --- a/src/packages/harp-device/src/harp/device/client/__init__.py +++ b/src/packages/harp-device/src/harp/device/client/__init__.py @@ -1,8 +1,10 @@ -"""Talking to a Harp device: the device itself, its transport and the framer.""" +"""Talking to a Harp device: the device itself, its transport, the framer, and the +writer that records what it emits.""" from ._device import Device, DeviceError, EventHandler, Subscription from ._framer import HarpFramer from ._transport import ITransport, TransportError +from ._writer import DeviceWriter, attach_writer __all__ = [ "Device", @@ -12,4 +14,6 @@ "HarpFramer", "ITransport", "TransportError", + "DeviceWriter", + "attach_writer", ] diff --git a/src/packages/harp-device/src/harp/device/client/_writer.py b/src/packages/harp-device/src/harp/device/client/_writer.py new file mode 100644 index 0000000..57fb11a --- /dev/null +++ b/src/packages/harp-device/src/harp/device/client/_writer.py @@ -0,0 +1,287 @@ +"""Record what a device emits as a de-multiplexed dataset folder. + +The layout the Harp file format standard defines, one binary file per register beside +the schema of the device:: + + session.harp/ + Behavior_0.bin + Behavior_44.bin + ... + device.yml + +Each message is appended to the file of its own register as the complete Harp frame it +arrived as, header and checksum included, so a file is a run of frames that needs nothing +else to be read. This is what the reference C# writer records. + +Writing is a device concern rather than an analysis one, so it lives here and costs no +pandas. :class:`~harp.data.DatasetReader` is the other end, reading such a folder into +DataFrames. +""" + +import threading +from collections.abc import Callable, Iterable +from functools import partial +from os import PathLike +from pathlib import Path +from typing import IO, Any, Self + +from harp.device.schema import DEVICE_SCHEMA_FILENAME, DeviceModuleLike +from harp.protocol import HarpMessage, MessageType + +from ._device import Device, Subscription + +_ALL_MESSAGE_TYPES: frozenset[MessageType] = frozenset(MessageType) +"""Every message type, which is what a recording captures by default. + +A Harp device answers a read of one register with a ``Read`` message and a write with a +``Write`` message, and the register dump a device performs on request arrives as a burst +of ``Read`` messages. Capturing only ``Event`` would drop the configuration the device +was running under, so a recording captures all three. +""" + + +_FileNameFormatter = Callable[[int], str] +"""The file one register is written to, named relative to the dataset folder. +""" + + +def _default_file_formatter(device_name: str, address: int) -> str: + """Harp file format name for one register: ``_
.bin``. + + The standard also allows a trailing ``_`` field, which a custom formatter + supplies; this names the plain form. :func:`~harp.data.default_file_resolver` reads + either back. :class:`DeviceWriter` binds ``device_name`` to reach the + shape a formatter has, so what it calls per register takes the address alone. + """ + return f"{device_name}_{address}.bin" + + +class DeviceWriter: + """De-multiplexing sink writing Harp messages into a dataset folder. + + Hand it messages with :meth:`write`, or let :func:`attach_writer` feed it from a live + device. The file of a register is created the first time a message for that register + arrives, so the folder holds exactly the registers that were seen:: + + with DeviceWriter(behavior, "session.harp") as writer: + writer.write(message) + + ``device_module`` supplies the ```` prefix of every file and the + ``device.yml`` written into the folder. That copy is not optional: a recording + describes itself or it is not written at all, so a module carrying no + ``DEVICE_METADATA`` raises here rather than leaving a folder nothing can be decoded + against later. + + The ```` prefix is the ``DEVICE_NAME`` of that module and nothing else, + so a recording is named after the device that produced it and a reader matching files + by that name cannot be pointed at the wrong prefix. + + ``overwrite`` decides what happens when a file is already there. By default the first + write to an existing path raises :class:`FileExistsError`, so a folder is never + silently half-overwritten by a second recording. A register whose file is created + late in the run therefore fails late in the run, rather than at construction. + + File naming defaults to the Harp file format, ``_
.bin``. Pass + ``formatter`` -- any callable taking an address and returning a name relative to the + folder -- for any other layout, paired with the + :data:`~harp.data.FileNameResolver` that reads it back. It names one file from an + address alone, so the device name is what the default is bound with rather than + something a custom formatter is handed. A custom one holds whatever else it needs -- + the standard trailing ``_`` field, a timestamp, a subfolder -- and a module + that names no device is then enough to record with:: + + DeviceWriter(behavior, root, formatter=lambda address: f"Behavior_{address}_0.bin") + + The folder itself is created eagerly, along with its ``device.yml``, so a path that + cannot be written fails before any message is taken. Writes are serialized and may + come from any thread, and the frames of one register keep the order they were written + in. Files stay open and buffered until :meth:`flush` or :meth:`close`, so use the + writer as a context manager or close it yourself, or a recording ends short of its + last frames. + """ + + def __init__( + self, + device_module: DeviceModuleLike, + root: str | PathLike[str], + *, + formatter: _FileNameFormatter | None = None, + overwrite: bool = False, + ) -> None: + self._root = Path(root) + self._name = self._resolve_name(device_module, required=formatter is None) + self._formatter = formatter or partial(_default_file_formatter, self._name) + self._overwrite = overwrite + self._lock = threading.Lock() + self._streams: dict[int, IO[bytes]] = {} + self._paths: dict[int, Path] = {} + self._subscription: Subscription | None = None + self._closed = False + self._root.mkdir(parents=True, exist_ok=True) + self._schema_path = self._write_schema(device_module) + + def _resolve_name(self, module: DeviceModuleLike, *, required: bool) -> str: + """The device name, demanded only when the default formatter has to build on it.""" + declared = module.DEVICE_NAME + if declared or not required: + return declared + raise ValueError( + f"No name for the files under {self._root}: this module declares an empty " + f"DEVICE_NAME, so only a formatter of your own can name them." + ) + + def _write_schema(self, module: DeviceModuleLike) -> Path: + """Copy the schema of ``module`` into the folder, so the folder describes itself.""" + schema = getattr(module, "DEVICE_METADATA", b"") + if not schema: + raise ValueError( + f"No DEVICE_METADATA to describe {self._root} with. Record with a module " + f"declaring one." + ) + path = self._root / DEVICE_SCHEMA_FILENAME + if path.exists() and not self._overwrite: + raise FileExistsError(f"'{path}' already exists. Pass overwrite=True to replace it.") + path.write_bytes(schema) + return path + + @property + def root(self) -> Path: + """The dataset folder being written.""" + return self._root + + @property + def name(self) -> str: + """The ```` prefix every file is written under.""" + return self._name + + @property + def paths(self) -> dict[int, Path]: + """The file written for each register address seen so far.""" + with self._lock: + return dict(sorted(self._paths.items())) + + @property + def schema_path(self) -> Path: + """The ``device.yml`` written into the folder.""" + return self._schema_path + + @property + def closed(self) -> bool: + """Whether :meth:`close` has run.""" + return self._closed + + def write(self, message: HarpMessage[Any]) -> None: + """Append the complete frame of ``message`` to the file of its register. + + Raises :class:`ValueError` once the writer is closed, and + :class:`FileExistsError` when this is the first message for a register whose file + is already on disk and ``overwrite`` is off. + """ + frame = message.bytes + address = message.address + with self._lock: + if self._closed: + raise ValueError(f"This writer for {self._root} is closed.") + stream = self._streams.get(address) + if stream is None: + stream = self._open(address) + stream.write(frame) + + def _open(self, address: int) -> IO[bytes]: + path = self._root / self._formatter(address) + path.parent.mkdir(parents=True, exist_ok=True) # a formatter may name a subfolder + stream = open(path, "wb" if self._overwrite else "xb") + self._streams[address] = stream + self._paths[address] = path + return stream + + def _deliver(self, message: HarpMessage[Any]) -> None: + """Take a message off a subscription, dropping one still in flight at close. + + Unlike :meth:`write` this cannot raise on a closed writer, since a message may + already be on its way to a handler when the subscription is cancelled, and there + is nothing wrong with the recording having ended first. + """ + if self._closed: + return + self.write(message) + + def flush(self) -> None: + """Push every buffered frame to the operating system.""" + with self._lock: + streams = list(self._streams.values()) + for stream in streams: + stream.flush() + + def close(self) -> None: + """Detach from the device, if attached, and close every open file. Idempotent.""" + subscription = self._subscription + if subscription is not None: + subscription.unsubscribe() + with self._lock: + if self._closed: + return + self._closed = True + self._subscription = None + streams = list(self._streams.values()) + self._streams.clear() + for stream in streams: + stream.close() + + def __enter__(self) -> Self: + return self + + def __exit__(self, *args: object) -> None: + self.close() + + +def attach_writer( + device: Device[Any], + root: str | PathLike[str], + *, + formatter: _FileNameFormatter | None = None, + message_types: MessageType | Iterable[MessageType] = _ALL_MESSAGE_TYPES, + overwrite: bool = False, +) -> DeviceWriter: + """Record everything ``device`` emits into a dataset folder at ``root``. + + Builds a :class:`DeviceWriter` from the module the device was constructed with and + subscribes it to the whole message stream, returning the writer so the recording can + be flushed, inspected and ended:: + + with Device(transport, behavior) as dev, attach_writer(dev, "session.harp"): + ... # every message the device emits is recorded + + The writer owns the subscription, so closing it, by leaving the ``with`` block or by + calling :meth:`DeviceWriter.close`, both detaches from the device and closes the + files. Nothing else has to be unsubscribed, and the device outlives the recording, so + several recordings may be made over one open device. + + ``message_types`` narrows what is recorded, and records all of them by default: a + Harp device answers reads and writes with ``Read`` and ``Write`` messages, and the + register dump it performs on request arrives as ``Read``, so recording only ``Event`` + would drop the configuration the device was running under. Requesting that dump is + the caller's to do, by writing to :class:`~harp.device.core.OperationControl` with + ``dump_registers`` set, once the writer is attached. + + The remaining arguments are those of :class:`DeviceWriter`. The device has to have + been constructed with a module, since that module is what names the files and + describes the recording; one opened without it cannot be recorded from. + """ + if device.module is None: + raise ValueError( + "This device was opened without a module, so nothing names or describes a " + "recording of it. Construct it with the device module to record." + ) + writer = DeviceWriter( + device.module, + root, + formatter=formatter, + overwrite=overwrite, + ) + try: + writer._subscription = device.subscribe_all(writer._deliver, message_types=message_types) + except Exception: + writer.close() + raise + return writer diff --git a/src/packages/harp-device/src/harp/device/schema/__init__.py b/src/packages/harp-device/src/harp/device/schema/__init__.py index de66397..ba3b5ff 100644 --- a/src/packages/harp-device/src/harp/device/schema/__init__.py +++ b/src/packages/harp-device/src/harp/device/schema/__init__.py @@ -1,12 +1,18 @@ """Building a device interface from a Harp ``device.yml`` at runtime.""" from ._emit import ConverterContext, parse_device_schema -from ._module import DeviceModule, DeviceModuleLike, create_device_module +from ._module import ( + DEVICE_SCHEMA_FILENAME, + DeviceModule, + DeviceModuleLike, + create_device_module, +) __all__ = [ "create_device_module", "DeviceModule", "DeviceModuleLike", + "DEVICE_SCHEMA_FILENAME", "parse_device_schema", "ConverterContext", ] diff --git a/src/packages/harp-device/src/harp/device/schema/_module.py b/src/packages/harp-device/src/harp/device/schema/_module.py index 4b8231a..8ef855a 100644 --- a/src/packages/harp-device/src/harp/device/schema/_module.py +++ b/src/packages/harp-device/src/harp/device/schema/_module.py @@ -9,7 +9,6 @@ import types from typing import Any, Mapping, Optional, Protocol, runtime_checkable - from harp.protocol import RegisterBase from harp.device.core import REGISTER_MAP as CORE_REGISTER_MAP @@ -18,6 +17,9 @@ _DEFAULT_NAME = "Device" """Module name used when the schema carries no ``device`` header.""" +DEVICE_SCHEMA_FILENAME = "device.yml" +"""Conventional name of the schema file, inside a device repository or a recording.""" + @runtime_checkable class DeviceModuleLike(Protocol): @@ -31,11 +33,16 @@ class DeviceModuleLike(Protocol): ``DEVICE_NAME`` is required rather than optional, so a generated package always states the name used for its recordings. A schema declaring none still builds, since :class:`DeviceModule` declares the member and leaves it empty. + + ``DEVICE_METADATA`` is the ``device.yml`` the module was built from, so the schema + travels with the module and can be re-parsed, re-emitted or copied into a recording + without the file it came from being at hand. """ DEVICE_NAME: str WHO_AM_I: int REGISTER_MAP: dict[int, type[RegisterBase[Any]]] + DEVICE_METADATA: bytes class DeviceModule(types.ModuleType): @@ -55,8 +62,11 @@ class DeviceModule(types.ModuleType): REGISTER_MAP: dict[int, type[RegisterBase[Any]]] """Address -> register class, the core Harp registers merged with those of the schema.""" + DEVICE_METADATA: bytes + """The ``device.yml`` text this module was built from.""" + __all__: list[str] - """The declarations of the schema, beside ``REGISTER_MAP`` and ``WHO_AM_I``.""" + """The declarations of the schema, beside ``REGISTER_MAP``, ``WHO_AM_I`` and ``DEVICE_METADATA``.""" def create_device_module( @@ -82,6 +92,8 @@ def create_device_module( * ``DEVICE_NAME``, the ``device`` name of the schema, or ``name`` when given, and empty for a header-less register fragment. Recordings are written under this name, so :class:`~harp.data.DatasetReader` matches files by it; + * ``DEVICE_METADATA``, ``text`` itself as bytes, so the schema travels with the module + and a recording can carry a copy of it; * ``__name__``, the same name, falling back to ``"Device"`` so the module is never anonymous. This names the module rather than the device, and is not part of what a device module promises; @@ -100,6 +112,7 @@ def create_device_module( behavior = create_device_module(Path("device.yml").read_bytes()) behavior.AnalogData """ + schema = text.encode() if isinstance(text, str) else bytes(text) device = parse_device_schema(text) emitter = _Emitter(device, converters, require_converters) registers = emitter.emit() @@ -119,6 +132,13 @@ def create_device_module( DEVICE_NAME=device_name, REGISTER_MAP=register_map, WHO_AM_I=int(device.whoAmI or 0), - __all__=[*sorted(contents), "DEVICE_NAME", "REGISTER_MAP", "WHO_AM_I"], + DEVICE_METADATA=schema, + __all__=[ + *sorted(contents), + "DEVICE_NAME", + "DEVICE_METADATA", + "REGISTER_MAP", + "WHO_AM_I", + ], ) return module diff --git a/tests/conformance.py b/tests/conformance.py index 8e24b4a..e0fb4e2 100644 --- a/tests/conformance.py +++ b/tests/conformance.py @@ -136,6 +136,7 @@ class Hybrid(Device[None]): DEVICE_NAME: ClassVar[str] = "Hybrid" WHO_AM_I: ClassVar[int] = 1216 REGISTER_MAP: ClassVar[dict[int, type[RegisterBase[Any]]]] = {} + DEVICE_METADATA: ClassVar[bytes] = b"" device = open_device(Hybrid, port="COM3") assert_type(device, Hybrid) diff --git a/tests/device/expected_device.py b/tests/device/expected_device.py index 0f027f3..983dcfc 100644 --- a/tests/device/expected_device.py +++ b/tests/device/expected_device.py @@ -2,6 +2,7 @@ # To make changes, edit the device metadata and regenerate the interface. import enum +from pathlib import Path from typing import Any, ClassVar import numpy as np @@ -32,6 +33,7 @@ __all__ = [ "DEVICE_NAME", + "DEVICE_METADATA", "WHO_AM_I", "PortDigitalIOS", "PwmPort", @@ -68,6 +70,9 @@ DEVICE_NAME: str = "Tests" WHO_AM_I: int = 0 +_SCHEMA_PATH = Path(__file__).parent.parent / "assets" / "device.yml" +DEVICE_METADATA: bytes = _SCHEMA_PATH.read_bytes() + class PortDigitalIOS(enum.IntFlag): DIO0 = 0x1 diff --git a/tests/device/test_create_device_module.py b/tests/device/test_create_device_module.py index 02cfac4..428b5f3 100644 --- a/tests/device/test_create_device_module.py +++ b/tests/device/test_create_device_module.py @@ -17,7 +17,7 @@ from .converters import DataConverter CONVERTERS = {"DataConverter": DataConverter()} -MODULE_CONSTANTS = {"DEVICE_NAME", "REGISTER_MAP", "WHO_AM_I"} +MODULE_CONSTANTS = {"DEVICE_NAME", "DEVICE_METADATA", "REGISTER_MAP", "WHO_AM_I"} @pytest.fixture diff --git a/tests/device/test_writer.py b/tests/device/test_writer.py new file mode 100644 index 0000000..f228254 --- /dev/null +++ b/tests/device/test_writer.py @@ -0,0 +1,416 @@ +import queue +import threading +import types +from collections.abc import Callable, Iterable +from functools import partial + +import pytest +from harp.data import DatasetReader, open_dataset, parse_to_dataframe +from harp.device.client import ( + Device, + DeviceWriter, + TransportError, + attach_writer, +) +from harp.device.client._writer import _default_file_formatter +from harp.device.core import WhoAmI +from harp.device.schema import create_device_module +from harp.protocol import HarpMessage, MessageType + +from tests.fixtures import make_frame_from_raw + +_U16 = 0x02 +"""Payload-type byte of a U16 payload, as the size nibble alone.""" + + +def default_file_formatter_with_suffix(device_name: str, address: int, suffix: str) -> str: + """The standard name carrying its optional trailing field, as a caller would write it.""" + return f"{device_name}_{address}_{suffix}.bin" + + +class _ScriptedTransport: + """A transport replying with whatever ``on_write`` returns for each request.""" + + def __init__(self) -> None: + self.on_write: Callable[[bytes], Iterable[bytes]] | None = None + self.failing = False + self._inbox: queue.SimpleQueue[bytes] = queue.SimpleQueue() + + def open(self) -> None: ... + + def close(self) -> None: ... + + def write(self, data: bytes) -> None: + if self.on_write is not None: + for frame in self.on_write(data): + self._inbox.put(frame) + + def read(self) -> bytes: + if self.failing: + raise TransportError("simulated transport failure") + try: + return self._inbox.get(timeout=0.01) + except queue.Empty: + return b"" + + def inject(self, frame: bytes) -> None: + self._inbox.put(frame) + + +def _event(address: int, value: int) -> bytes: + return make_frame_from_raw( + MessageType.Event, + address, + 255, + _U16, + value.to_bytes(2, "little"), + timestamp=b"\x01\x00\x00\x00\x00\x00", + ) + + +@pytest.fixture +def nameless_module(): + """A header-less register fragment: it declares no device, so DEVICE_NAME is empty.""" + return create_device_module("registers:\n Foo: {address: 40, type: U16, access: Read}\n") + + +@pytest.fixture +def module(device_yml): + # require_converters=False: the test device.yml declares a custom DataConverter that + # is not injected here, and native decoding is enough to exercise the file layout. + return create_device_module(device_yml, require_converters=False) + + +# --------------------------------------------------------------------------- +# File naming and layout +# --------------------------------------------------------------------------- + + +def test_default_formatter_names_the_plain_standard_form(): + assert _default_file_formatter("Behavior", 44) == "Behavior_44.bin" + + +def test_writes_one_file_per_register_named_after_the_device(module, tmp_path): + root = tmp_path / "session.harp" + with DeviceWriter(module, root) as writer: + writer.write(HarpMessage.parse(_event(32, 7))) + writer.write(HarpMessage.parse(_event(33, 9))) + writer.write(HarpMessage.parse(_event(32, 8))) + written = {p.name for p in root.glob("*.bin")} + assert written == {"Tests_32.bin", "Tests_33.bin"} + assert writer.paths == {32: root / "Tests_32.bin", 33: root / "Tests_33.bin"} + + +def test_writes_the_complete_frame_in_arrival_order(module, tmp_path): + frames = [_event(32, value) for value in (1, 2, 3)] + with DeviceWriter(module, tmp_path) as writer: + for frame in frames: + writer.write(HarpMessage.parse(frame)) + # The whole frame is recorded, header and checksum included, so the file is what a + # register parser reads without consulting anything else. + assert (tmp_path / "Tests_32.bin").read_bytes() == b"".join(frames) + + +def test_a_custom_formatter_supplies_the_trailing_suffix_field(module, tmp_path): + # The standard's optional field is one thing a custom formatter is for. + formatter = partial(default_file_formatter_with_suffix, "Tests", suffix="0") + with DeviceWriter(module, tmp_path, formatter=formatter) as writer: + writer.write(HarpMessage.parse(_event(32, 1))) + assert (tmp_path / "Tests_32_0.bin").is_file() + + +def test_a_custom_formatter_decides_the_whole_layout(module, tmp_path): + def by_folder(address): + return f"{address}/Tests.bin" + + with DeviceWriter(module, tmp_path, formatter=by_folder) as writer: + writer.write(HarpMessage.parse(_event(32, 1))) + # A formatter naming a subfolder gets it created rather than failing on the open. + assert (tmp_path / "32" / "Tests.bin").read_bytes() == _event(32, 1) + assert writer.paths == {32: tmp_path / "32" / "Tests.bin"} + + +def test_a_custom_formatter_reads_back_through_the_matching_resolver(module, tmp_path): + def by_folder(address): + return f"{address}/Tests.bin" + + def resolve(root, name): + return {int(p.parent.name): [p] for p in sorted(root.glob(f"*/{name}.bin"))} + + frames = [_event(32, value) for value in (1, 2)] + with DeviceWriter(module, tmp_path, formatter=by_folder) as writer: + for frame in frames: + writer.write(HarpMessage.parse(frame)) + reader = DatasetReader(module, tmp_path, resolver=resolve) + expected = parse_to_dataframe(module.REGISTER_MAP[32], b"".join(frames)) + assert reader.read(32).equals(expected) + + +def test_the_prefix_is_the_device_name_of_the_module(module, tmp_path): + # Nothing can override it, so files are always named after the device that made them. + with DeviceWriter(module, tmp_path) as writer: + writer.write(HarpMessage.parse(_event(32, 1))) + assert writer.name == module.DEVICE_NAME == "Tests" + assert (tmp_path / "Tests_32.bin").is_file() + + +def test_a_module_without_a_name_needs_one(nameless_module, tmp_path): + # A header-less register fragment declares no device, so nothing names the files. + assert nameless_module.DEVICE_NAME == "" + with pytest.raises(ValueError, match="DEVICE_NAME"): + DeviceWriter(nameless_module, tmp_path) + + +def test_a_custom_formatter_lifts_the_name_requirement(nameless_module, tmp_path): + # The name only feeds the default formatter, so a custom one needs no device name. + def by_address(address): + return f"{address}.bin" + + with DeviceWriter(nameless_module, tmp_path, formatter=by_address) as writer: + writer.write(HarpMessage.parse(_event(32, 1))) + assert (tmp_path / "32.bin").is_file() + + +def test_the_folder_is_created(module, tmp_path): + root = tmp_path / "nested" / "session.harp" + DeviceWriter(module, root).close() + assert root.is_dir() + + +# --------------------------------------------------------------------------- +# The schema copied into the folder +# --------------------------------------------------------------------------- + + +def test_the_schema_of_the_module_is_copied_into_the_folder(module, device_yml, tmp_path): + writer = DeviceWriter(module, tmp_path) + writer.close() + assert writer.schema_path == tmp_path / "device.yml" + assert writer.schema_path.read_text() == device_yml + + +def test_a_module_carrying_no_schema_raises(tmp_path): + # DeviceModuleLike requires DEVICE_METADATA, so a module without one is broken, not a + # folder to write undescribed. + broken = types.ModuleType("Broken") + broken.DEVICE_NAME = "Broken" + broken.WHO_AM_I = 0 + broken.REGISTER_MAP = {} + with pytest.raises(ValueError, match="DEVICE_METADATA"): + DeviceWriter(broken, tmp_path) + + +def test_a_module_carrying_an_empty_schema_raises(module, tmp_path): + module.DEVICE_METADATA = b"" + with pytest.raises(ValueError, match="DEVICE_METADATA"): + DeviceWriter(module, tmp_path) + + +# --------------------------------------------------------------------------- +# Overwrite +# --------------------------------------------------------------------------- + + +def test_an_existing_schema_is_not_silently_replaced(module, tmp_path): + DeviceWriter(module, tmp_path).close() + with pytest.raises(FileExistsError, match="device.yml"): + DeviceWriter(module, tmp_path) + + +def test_an_existing_register_file_is_not_silently_replaced(module, tmp_path): + with DeviceWriter(module, tmp_path) as first: + first.write(HarpMessage.parse(_event(32, 1))) + # overwrite= lets the schema through, so the register file is what is under test. + (tmp_path / "device.yml").unlink() + with DeviceWriter(module, tmp_path) as second: + with pytest.raises(FileExistsError): + second.write(HarpMessage.parse(_event(32, 2))) + + +def test_overwrite_replaces_both_the_schema_and_the_files(module, tmp_path): + with DeviceWriter(module, tmp_path) as first: + first.write(HarpMessage.parse(_event(32, 1))) + first.write(HarpMessage.parse(_event(32, 2))) + with DeviceWriter(module, tmp_path, overwrite=True) as second: + second.write(HarpMessage.parse(_event(32, 3))) + assert (tmp_path / "Tests_32.bin").read_bytes() == _event(32, 3) + + +# --------------------------------------------------------------------------- +# Lifecycle +# --------------------------------------------------------------------------- + + +def test_writing_to_a_closed_writer_raises(module, tmp_path): + writer = DeviceWriter(module, tmp_path) + writer.close() + assert writer.closed + with pytest.raises(ValueError, match="closed"): + writer.write(HarpMessage.parse(_event(32, 1))) + + +def test_close_is_idempotent(module, tmp_path): + writer = DeviceWriter(module, tmp_path) + writer.close() + writer.close() + + +def test_flush_exposes_frames_before_close(module, tmp_path): + with DeviceWriter(module, tmp_path) as writer: + writer.write(HarpMessage.parse(_event(32, 1))) + writer.flush() + assert (tmp_path / "Tests_32.bin").read_bytes() == _event(32, 1) + + +def test_concurrent_writers_do_not_interleave_frames(module, tmp_path): + frame = _event(32, 1) + with DeviceWriter(module, tmp_path) as writer: + + def run() -> None: + for _ in range(50): + writer.write(HarpMessage.parse(frame)) + + threads = [threading.Thread(target=run) for _ in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert (tmp_path / "Tests_32.bin").read_bytes() == frame * 200 + + +# --------------------------------------------------------------------------- +# attach_writer +# --------------------------------------------------------------------------- + + +def _wait_for(predicate, timeout: float = 2.0) -> None: + deadline = threading.Event() + for _ in range(int(timeout / 0.01)): + if predicate(): + return + deadline.wait(0.01) + raise AssertionError("timed out waiting for the recording") + + +def test_attach_writer_records_the_whole_stream(module, tmp_path): + transport = _ScriptedTransport() + root = tmp_path / "session.harp" + with Device(transport, module) as device: + with attach_writer(device, root) as writer: + transport.inject(_event(32, 1)) + transport.inject(_event(33, 2)) + _wait_for(lambda: set(writer.paths) == {32, 33}) + assert writer.closed + assert (root / "Tests_32.bin").read_bytes() == _event(32, 1) + assert (root / "Tests_33.bin").read_bytes() == _event(33, 2) + + +def test_attach_writer_records_read_replies_by_default(module, tmp_path): + # The register dump a device performs on request arrives as Read messages, so a + # recording that only kept Event would lose the configuration it ran under. + transport = _ScriptedTransport() + reply = make_frame_from_raw(MessageType.Read, 0, 255, _U16, (1216).to_bytes(2, "little")) + transport.on_write = lambda _data: [reply] + with Device(transport, module) as device: + with attach_writer(device, tmp_path) as writer: + assert int(device.read(WhoAmI).payload) == 1216 + _wait_for(lambda: 0 in writer.paths) + assert (tmp_path / "Tests_0.bin").read_bytes() == reply + + +def test_attach_writer_narrows_to_the_requested_message_types(module, tmp_path): + transport = _ScriptedTransport() + reply = make_frame_from_raw(MessageType.Read, 0, 255, _U16, (1216).to_bytes(2, "little")) + transport.on_write = lambda _data: [reply] + with Device(transport, module) as device: + with attach_writer(device, tmp_path, message_types=MessageType.Event) as writer: + device.read(WhoAmI) + transport.inject(_event(32, 1)) + _wait_for(lambda: 32 in writer.paths) + assert set(writer.paths) == {32} + + +def test_attach_writer_takes_the_name_and_schema_from_the_device_module( + module, device_yml, tmp_path +): + transport = _ScriptedTransport() + with Device(transport, module) as device: + with attach_writer(device, tmp_path) as writer: + assert writer.name == "Tests" + assert (tmp_path / "device.yml").read_text() == device_yml + + +def test_closing_the_writer_detaches_it_from_the_device(module, tmp_path): + transport = _ScriptedTransport() + with Device(transport, module) as device: + writer = attach_writer(device, tmp_path) + transport.inject(_event(32, 1)) + _wait_for(lambda: 32 in writer.paths) + writer.close() + transport.inject(_event(33, 2)) + transport.inject(_event(32, 2)) + # Nothing after the close is recorded, and no handler error is raised for it. + _wait_for(lambda: True) + assert set(writer.paths) == {32} + assert (tmp_path / "Tests_32.bin").read_bytes() == _event(32, 1) + + +def test_several_recordings_over_one_open_device(module, tmp_path): + transport = _ScriptedTransport() + with Device(transport, module) as device: + with attach_writer(device, tmp_path / "first.harp") as first: + transport.inject(_event(32, 1)) + _wait_for(lambda: 32 in first.paths) + with attach_writer(device, tmp_path / "second.harp") as second: + transport.inject(_event(32, 2)) + _wait_for(lambda: 32 in second.paths) + assert (tmp_path / "first.harp" / "Tests_32.bin").read_bytes() == _event(32, 1) + assert (tmp_path / "second.harp" / "Tests_32.bin").read_bytes() == _event(32, 2) + + +def test_attach_writer_forwards_the_formatter(module, tmp_path): + def by_folder(address): + return f"{address}/Tests.bin" + + transport = _ScriptedTransport() + with Device(transport, module) as device: + with attach_writer(device, tmp_path, formatter=by_folder) as writer: + transport.inject(_event(32, 1)) + _wait_for(lambda: 32 in writer.paths) + assert (tmp_path / "32" / "Tests.bin").is_file() + + +def test_attach_writer_refuses_a_device_opened_without_a_module(tmp_path): + # The module is what names the files and describes the folder, so there is nothing + # to record with when a device was opened without one. + with Device(_ScriptedTransport()) as device: + with pytest.raises(ValueError, match="without a module"): + attach_writer(device, tmp_path) + + +# --------------------------------------------------------------------------- +# Round trip +# --------------------------------------------------------------------------- + + +def test_a_recorded_folder_reads_back_through_open_dataset(module, tmp_path): + root = tmp_path / "session.harp" + frames = [_event(32, value) for value in (1, 2, 3)] + with DeviceWriter(module, root) as writer: + for frame in frames: + writer.write(HarpMessage.parse(frame)) + # The folder carries its own device.yml, so it opens without a module in hand. + reader = open_dataset(root, require_converters=False) + assert reader.name == "Tests" + assert reader.contents["DigitalInputs"] == 32 + assert reader.read(32).equals(parse_to_dataframe(module.REGISTER_MAP[32], b"".join(frames))) + + +def test_a_suffixed_recording_reads_back_by_suffix(module, tmp_path): + root = tmp_path / "session.harp" + formatter = partial(default_file_formatter_with_suffix, "Tests", suffix="0") + with DeviceWriter(module, root, formatter=formatter) as writer: + writer.write(HarpMessage.parse(_event(32, 1))) + reader = DatasetReader(module, root) + assert reader.paths[32] == [root / "Tests_32_0.bin"] + assert len(reader.read(32, suffix="0")) == 1