diff --git a/README.md b/README.md index 9b35717..89199f6 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,45 @@ pip install harp-data `harp-benchmarks` (under `src/packages/`) is internal-only and is never published to PyPI. +## 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: + +```python +from pathlib import Path +from harp.device import create_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 +``` + +The generated device works like any other. **Talk to hardware** over a serial +transport — `read`/`write` take a register class: + +```python +from harp.serial import open_serial_device + +# Use "COMx" on Windows, "/dev/ttyUSBx" on Linux. +with open_serial_device(Behavior, port="/dev/ttyUSB0") as device: + print(device.read(AnalogData).parsed) +``` + +...or use the same register classes to **decode recorded data** into a pandas +DataFrame: + +```python +from harp.data import parse_to_dataframe + +df = parse_to_dataframe(AnalogData, "Behavior_44.bin") +``` + +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. + ## Contributing harp is a [uv workspace](https://docs.astral.sh/uv/concepts/workspaces/): every package under diff --git a/docs/api/device.md b/docs/api/device.md index dfcade9..652e4c7 100644 --- a/docs/api/device.md +++ b/docs/api/device.md @@ -3,6 +3,9 @@ --- ::: harp.device.Device +::: harp.device.create_device +::: harp.device.parse_device_schema +::: harp.device.ConverterContext ::: harp.device.HarpFramer ::: harp.device.ITransport ::: harp.device.TransportError diff --git a/docs/examples/create_device/create_device.md b/docs/examples/create_device/create_device.md new file mode 100644 index 0000000..15907d9 --- /dev/null +++ b/docs/examples/create_device/create_device.md @@ -0,0 +1,53 @@ +# Generating a Device from a Schema + +This example demonstrates how to turn a Harp `device.yml` into a typed +[`Device`](../../api/device.md) at runtime with `create_device`, without a +code-generation step. This is the quickest way to get started when you have only a +device's schema and no pre-generated package for it. + +The compiled device exposes its registers through `REGISTER_MAP` (keyed by address) +and carries the device's `__whoami__` identity. From there it works exactly like a +pre-generated device class — drive it over a transport to talk to hardware, or use +its register classes to decode recorded data. + +## When to use runtime generation + +`create_device` trades statically generated device packages for schema-driven +convenience. It's worth understanding what that buys you and what it costs. + +**You gain:** + +- **No build step.** A `device.yml` — even one you just pulled off a device — + becomes a working device in a single call. There's nothing to generate, install, + or keep in sync with the schema. +- **Coverage for any device.** You don't need a published package for the device; + unreleased, custom, or one-off schemas work immediately. +- **The schema stays the single source of truth.** Register, field, and enum names + come straight from the `device.yml`, verbatim. + +**You give up:** + +- **Named, typed access.** Registers are reached by address (`REGISTER_MAP[44]`), + not as importable, autocompleting classes (`from harp_behavior import AnalogData`). + You lose editor discovery and static type checking of register names. +- **Generator naming conventions.** Identifiers are kept verbatim from the yml + (`AnalogInput0`, `DIO0`) rather than the C# generator's snake_case fields and + `UPPER_SNAKE` enum members, so code written against a generated package won't line + up name-for-name. +- **Turn-key custom types.** A custom `interfaceType` must be injected yourself via + `converters=` (see below), whereas a generated package ships its own converters. + +For shipped, widely-used devices a pre-generated package from the +[Harp C# generator](https://github.com/harp-tech/generators) remains the +authoritative choice — better editor support and static typing. Reach for +`create_device` when you want to go from a schema to working code with no +generation step. + +!!! warning + Don't forget to change the `SERIAL_PORT` to the one that corresponds to your device! The `SERIAL_PORT` must be denoted as `/dev/ttyUSBx` in Linux and `COMx` in Windows, where `x` is the number of the serial port. + + +```python +[](./create_device.py) +``` + diff --git a/docs/examples/create_device/create_device.py b/docs/examples/create_device/create_device.py new file mode 100644 index 0000000..11ec1b8 --- /dev/null +++ b/docs/examples/create_device/create_device.py @@ -0,0 +1,38 @@ +from pathlib import Path + +from harp.data import parse_to_dataframe +from harp.device import create_device +from harp.serial import open_serial_device + +SERIAL_PORT = "/dev/ttyUSB0" # or "COMx" in Windows ("x" is the number of the serial port) + +# `create_device` compiles a Harp `device.yml` into a typed `Device` subclass at +# runtime — no code-generation step. This is the quickest way to work with a device +# when you don't have a pre-generated package for it: point it at the schema and you +# get the device's registers (keyed by address) plus its identity. +Behavior = create_device(Path("device.yml").read_text()) + +print("WhoAmI:", Behavior.__whoami__) # device identity, taken from the schema +AnalogData = Behavior.REGISTER_MAP[44] # registers are reached by address + +# The generated device behaves like any other `Device` class. Talk to hardware over +# a transport — `read`/`write` take a register class: +with open_serial_device(Behavior, port=SERIAL_PORT) as device: + print("AnalogData:", device.read(AnalogData).parsed) + +# ...or use the same register classes to decode a recorded binary dump into a +# pandas DataFrame (see the "Reading Data into a DataFrame" example for more): +df = parse_to_dataframe(AnalogData, "Behavior_44.bin") +print(df.head()) + + +# --- Custom interface types -------------------------------------------------- +# A register with a custom `interfaceType` needs a converter so its field decodes +# to the right Python type. Pass it via `converters=`, keyed by "Converter": +# +# Behavior = create_device(yml_text, converters={"DataConverter": DataConverter()}) +# +# An unresolved custom type raises `UnknownConverterError`; pass `strict=False` to +# decode it natively instead. `exclude_private=True` (the default) drops registers +# marked `private` in the schema. If you only want the parsed schema model rather +# than a device, `parse_device_schema(yml_text)` returns that directly. diff --git a/docs/examples/index.md b/docs/examples/index.md index 52e4245..492d90a 100644 --- a/docs/examples/index.md +++ b/docs/examples/index.md @@ -2,8 +2,16 @@ This section contains some examples to help you get started with `harp`. -Here's the complete list of available examples: +Working from a device schema: + +- [Generating a Device from a Schema](./create_device/create_device.md) - compile a `device.yml` into a typed device at runtime with `create_device`. + +Talking to a device: - [Getting Device Info](./get_info/get_info.md) - connect to a Harp device and read its information. - [Read and Write from Registers](./read_and_write_from_registers/read_and_write_from_registers.md) - connect to a Harp device and read and write its registers. +- [Subscribing to Events](./subscribing_to_events/subscribing_to_events.md) - react to messages pushed by the device without polling. + +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`. diff --git a/mkdocs.yml b/mkdocs.yml index 48b3d6b..6cf012d 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -73,10 +73,11 @@ nav: - Home: index.md - Examples: - examples/index.md + - Generating a Device from a Schema: examples/create_device/create_device.md - 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 - - Reading Data into a DataFrame: examples/read_data_to_dataframe/read_data_to_dataframe.md - Subscribing to Events: examples/subscribing_to_events/subscribing_to_events.md + - Reading Data into a DataFrame: examples/read_data_to_dataframe/read_data_to_dataframe.md - API: - Protocol: api/protocol.md - Serial: api/serial.md diff --git a/src/packages/harp-benchmarks/src/harp/benchmarks/_registers.py b/src/packages/harp-benchmarks/src/harp/benchmarks/_registers.py index 7033400..461cb58 100644 --- a/src/packages/harp-benchmarks/src/harp/benchmarks/_registers.py +++ b/src/packages/harp-benchmarks/src/harp/benchmarks/_registers.py @@ -1,14 +1,10 @@ """Shared benchmark fixtures: every register from ``harp.benchmarks.register_models`` -paired with a representative sample value. +paired with whether its frames carry a timestamp. Both ``generate.py`` (writes the .bin corpora) and ``benchmark.py`` (times parsing) -import :data:`BENCHMARK_REGISTERS` from here so the two stay in lock-step: the value -used to *format* each frame is the same one whose parsed shape we benchmark. - -The sample values mirror ``register_models.main()``'s round-trip fixtures — they -exercise the full spread of payload shapes the Harp protocol allows (trivial -scalars, struct payloads with byte gaps, masked sub-fields, custom converters, -enum/flag single-member unwrap). +import :data:`BENCHMARK_REGISTERS` from here so the two stay in lock-step. Payloads +are synthesized as random bytes per frame at generation time (see ``generate.py``), +so no sample values live here — only the register class and its frame shape. All generated artifacts live under ``./benchmark`` in the current working directory. """ @@ -16,36 +12,24 @@ from pathlib import Path from typing import Any, NamedTuple -import numpy as np - from harp.benchmarks.register_models import ( AnalogData, - AnalogDataPayload, BitmaskSplitter, - BitmaskSplitterPayload, ComplexConfiguration, - ComplexConfigurationPayload, Counter0, CustomMemberConverter, - CustomMemberConverterPayload, CustomPayload, CustomRawPayload, DigitalInputs, EncoderMode, - EncoderModeMask, - PortDigitalIOS, PortDIOSet, PulseDO0, PulseDOPort0, - PwmPort, StartPulse, - StartPulsePayload, StartPulseTrain, - StartPulseTrainPayload, Version, - VersionPayload, ) -from harp.protocol import HarpVersion, RegisterBase +from harp.protocol import RegisterBase ARTIFACTS_DIR = Path("benchmark").resolve() DATA_DIR = ARTIFACTS_DIR / "data" @@ -53,11 +37,10 @@ class BenchmarkedRegister(NamedTuple): - """A register under benchmark together with a value that ``format()`` accepts.""" + """A register under benchmark, with whether its corpus frames are timestamped.""" name: str register: type[RegisterBase[Any]] - value: Any timestamped: bool = True @property @@ -70,77 +53,28 @@ def filename(self) -> str: def _base_registers() -> list[BenchmarkedRegister]: - """One (timestamped) fixture per register — :func:`_build` derives the untimestamped twin.""" + """One (timestamped) fixture per register — :func:`_build` derives the untimestamped twin. + + The set spans the full spread of payload shapes the Harp protocol allows: trivial + scalars, struct payloads with byte gaps, masked sub-fields, custom converters, and + enum/flag single-member unwrap. + """ return [ - BenchmarkedRegister("DigitalInputs", DigitalInputs, np.uint8(0b1010)), - BenchmarkedRegister( - "AnalogData", - AnalogData, - AnalogDataPayload( - Analog0=np.float32(1.0), - Analog1=np.float32(2.0), - Analog2=np.float32(3.0), - Accelerometer=np.array([4, 5, 6], dtype=np.float32), - ), - ), - BenchmarkedRegister( - "ComplexConfiguration", - ComplexConfiguration, - ComplexConfigurationPayload( - PwmPort=PwmPort.Pwm2, - DutyCycle=np.float32(0.5), - Frequency=np.float32(1000.0), - EventsEnabled=True, - Delta=np.uint32(42), - ), - ), - BenchmarkedRegister( - "Version", - Version, - VersionPayload( - ProtocolVersion=HarpVersion(2, 0, 0), - FirmwareVersion=HarpVersion(1, 2, 3), - HardwareVersion=HarpVersion(1, 0, 0), - CoreId="abc", - InterfaceHash=np.arange(20, dtype=np.uint8), - ), - ), - BenchmarkedRegister("CustomPayload", CustomPayload, HarpVersion(3, 1, 4)), - BenchmarkedRegister("CustomRawPayload", CustomRawPayload, HarpVersion(0, 0, 1)), - BenchmarkedRegister( - "CustomMemberConverter", - CustomMemberConverter, - CustomMemberConverterPayload(Header=np.uint8(7), Data=-1234), - ), - BenchmarkedRegister( - "BitmaskSplitter", - BitmaskSplitter, - BitmaskSplitterPayload(Low=np.int32(0xA), High=np.int32(0x5)), - ), - BenchmarkedRegister("Counter0", Counter0, np.int32(-100000)), - BenchmarkedRegister( - "PortDIOSet", - PortDIOSet, - PortDigitalIOS.DIO0 | PortDigitalIOS.DIO3, - ), - BenchmarkedRegister("PulseDOPort0", PulseDOPort0, np.uint16(5)), - BenchmarkedRegister("PulseDO0", PulseDO0, np.uint16(9)), - BenchmarkedRegister( - "StartPulse", - StartPulse, - StartPulsePayload(DigitalOutput=PwmPort.Pwm1, PulseWidth=np.uint16(300)), - ), - BenchmarkedRegister( - "StartPulseTrain", - StartPulseTrain, - StartPulseTrainPayload( - DigitalOutput=PwmPort.Pwm1, - PulseWidth=np.uint16(300), - Frequency=np.uint8(200), - PulseCount=np.uint8(50), - ), - ), - BenchmarkedRegister("EncoderMode", EncoderMode, EncoderModeMask.Displacement), + BenchmarkedRegister("DigitalInputs", DigitalInputs), + BenchmarkedRegister("AnalogData", AnalogData), + BenchmarkedRegister("ComplexConfiguration", ComplexConfiguration), + BenchmarkedRegister("Version", Version), + BenchmarkedRegister("CustomPayload", CustomPayload), + BenchmarkedRegister("CustomRawPayload", CustomRawPayload), + BenchmarkedRegister("CustomMemberConverter", CustomMemberConverter), + BenchmarkedRegister("BitmaskSplitter", BitmaskSplitter), + BenchmarkedRegister("Counter0", Counter0), + BenchmarkedRegister("PortDIOSet", PortDIOSet), + BenchmarkedRegister("PulseDOPort0", PulseDOPort0), + BenchmarkedRegister("PulseDO0", PulseDO0), + BenchmarkedRegister("StartPulse", StartPulse), + BenchmarkedRegister("StartPulseTrain", StartPulseTrain), + BenchmarkedRegister("EncoderMode", EncoderMode), ] diff --git a/src/packages/harp-benchmarks/src/harp/benchmarks/generate.py b/src/packages/harp-benchmarks/src/harp/benchmarks/generate.py index 64c62ad..1567a9d 100644 --- a/src/packages/harp-benchmarks/src/harp/benchmarks/generate.py +++ b/src/packages/harp-benchmarks/src/harp/benchmarks/generate.py @@ -2,9 +2,11 @@ import sys from pathlib import Path +import numpy as np + from harp.benchmarks._registers import BENCHMARK_REGISTERS, DATA_DIR, BenchmarkedRegister -_TIMESTAMP = 42 +_SEED = 42 def corpus_path(reg: BenchmarkedRegister, data_dir: Path = DATA_DIR): @@ -12,19 +14,30 @@ def corpus_path(reg: BenchmarkedRegister, data_dir: Path = DATA_DIR): return data_dir / reg.filename -def _frame_timestamp(reg: BenchmarkedRegister) -> int | None: - return _TIMESTAMP if reg.timestamped else None +def _frames(reg: BenchmarkedRegister, entries: int) -> np.ndarray: + """Build ``entries`` frames of ``reg`` with a random per-frame payload. + + Bytes are held to the ASCII range (0..127) so every field varies while staying + valid for any ``StringConverter`` member and free of float NaN/inf — the corpus is + decoded (``to_columns`` / ``parse_to_dataframe``) during the benchmark. Timestamps, + when present, are a monotonic ramp. Returns the flat uint8 wire buffer. + """ + dtype = reg.register.payload_class.dtype + rng = np.random.default_rng(_SEED + reg.address) + records = rng.integers(0, 128, size=entries * dtype.itemsize, dtype=np.uint8).view(dtype) + timestamps = np.arange(entries, dtype=np.float64) if reg.timestamped else None + return reg.register.format_bulk(records, timestamps=timestamps) def generate_one( reg: BenchmarkedRegister, entries: int, data_dir: Path = DATA_DIR ) -> tuple[str, int, int]: """Write ``entries`` frames for ``reg``. Returns (path, frame_size, file_size).""" - frame = reg.register.format(reg.value, timestamp=_frame_timestamp(reg)) + buf = _frames(reg, entries) path = corpus_path(reg, data_dir) path.parent.mkdir(parents=True, exist_ok=True) - path.write_bytes(frame * entries) - return str(path), len(frame), path.stat().st_size + path.write_bytes(buf.tobytes()) + return str(path), len(buf) // entries, path.stat().st_size def ensure_corpus( @@ -33,13 +46,13 @@ def ensure_corpus( """Generate ``reg``'s corpus unless a matching cached file already exists. The cache is honored only when the existing file's size matches ``entries`` - exactly (frame_size * entries); a stale file (different entry count) is rebuilt. + exactly (stride * entries); a stale file (different entry count) is rebuilt. Returns (path, generated). """ path = corpus_path(reg, data_dir) if path.exists() and not force: - frame_size = len(reg.register.format(reg.value, timestamp=_frame_timestamp(reg))) - if path.stat().st_size == frame_size * entries: + stride = len(_frames(reg, 1)) + if path.stat().st_size == stride * entries: return path, False generate_one(reg, entries, data_dir) return path, True diff --git a/src/packages/harp-data/src/harp/data/__init__.py b/src/packages/harp-data/src/harp/data/__init__.py index c8208f1..5aff91a 100644 --- a/src/packages/harp-data/src/harp/data/__init__.py +++ b/src/packages/harp-data/src/harp/data/__init__.py @@ -1,6 +1,9 @@ from ._reader import parse_to_dataframe, payload_to_dataframe +from ._write import to_buffer, to_file __all__ = [ "parse_to_dataframe", "payload_to_dataframe", + "to_buffer", + "to_file", ] diff --git a/src/packages/harp-data/src/harp/data/_reader.py b/src/packages/harp-data/src/harp/data/_reader.py index b068df4..8ff9ae3 100644 --- a/src/packages/harp-data/src/harp/data/_reader.py +++ b/src/packages/harp-data/src/harp/data/_reader.py @@ -77,9 +77,12 @@ def parse_to_dataframe( ) if timestamp: if timestamps is None: - raise ValueError( - "Buffer contains no timestamp data; pass timestamp=False to suppress " - "the timestamp column." - ) - df.insert(0, "timestamp", timestamps) + if len(df) > 0: + raise ValueError( + "Buffer contains no timestamp data; pass timestamp=False to suppress " + "the timestamp column." + ) + # Empty buffer: no frames to timestamp — return the empty frame as-is. + else: + df.insert(0, "timestamp", timestamps) return df diff --git a/src/packages/harp-data/src/harp/data/_write.py b/src/packages/harp-data/src/harp/data/_write.py new file mode 100644 index 0000000..ab48c58 --- /dev/null +++ b/src/packages/harp-data/src/harp/data/_write.py @@ -0,0 +1,45 @@ +"""Write Harp register data to a binary buffer/file — the inverse of the readers. + +Thin wrappers over :meth:`RegisterBase.format_bulk` giving a pandas-package home +and a file sink. Useful for round-tripping data and generating typed test corpora. +""" + +from os import PathLike +from typing import Any + +import numpy as np +from harp.protocol import MessageType, PayloadBase, RegisterBase +from numpy.typing import ArrayLike, NDArray + + +def to_buffer( + register: type[RegisterBase[Any]], + values: PayloadBase | ArrayLike, + *, + timestamps: ArrayLike | None = None, + message_type: MessageType | ArrayLike = MessageType.Event, + port: int = 255, +) -> NDArray[np.uint8]: + """Encode ``values`` as a flat buffer of ``register`` frames. + + ``values`` is a payload (scalar or batch) or an ndarray of the register's + ``payload_class.dtype``; ``timestamps`` (length-N seconds) makes every frame + timestamped; ``message_type`` is one :class:`MessageType` or a length-N array + (e.g. the msgtype view from ``parse_bulk``). + """ + return register.format_bulk(values, timestamps=timestamps, message_type=message_type, port=port) + + +def to_file( + register: type[RegisterBase[Any]], + values: PayloadBase | ArrayLike, + file: str | PathLike, + *, + timestamps: ArrayLike | None = None, + message_type: MessageType | ArrayLike = MessageType.Event, + port: int = 255, +) -> None: + """Write ``values`` as ``register`` frames to ``file`` (see :func:`to_buffer`).""" + to_buffer(register, values, timestamps=timestamps, message_type=message_type, port=port).tofile( + file + ) diff --git a/src/packages/harp-device/README.md b/src/packages/harp-device/README.md index f4b0f01..3221dd8 100644 --- a/src/packages/harp-device/README.md +++ b/src/packages/harp-device/README.md @@ -33,3 +33,28 @@ REGISTER_MAP = {**_CORE_REGISTER_MAP, 32: DigitalInputState, ...} A new transport is just an object implementing the `ITransport` protocol (`open`/`write`/`read`/`close`). + +## Generating a device from a `device.yml` + +If you don't have a pre-generated device package, `create_device` builds a +`Device` from Harp `device.yml` text. Registers are reached by address through +`REGISTER_MAP`; field and enum names come from the yml verbatim. + +```python +from pathlib import Path +from harp.device import create_device + +Behavior = create_device(Path("device.yml").read_text()) +reg = Behavior.REGISTER_MAP[44] +``` + +For a custom `interfaceType`, pass its converter via `converters=` (keyed by +`{InterfaceType}Converter` / `{MemberName}Converter`); an unresolved custom type +raises `UnknownConverterError`, or pass `strict=False` to decode it natively: + +```python +create_device(yml_text, converters={"DataConverter": DataConverter()}) +``` + +`parse_device_schema(yml_text)` is also public if you just want the parsed +schema model (registers, masks, and optional device identity) without a `Device`. diff --git a/src/packages/harp-device/pyproject.toml b/src/packages/harp-device/pyproject.toml index 54e8578..8efea01 100644 --- a/src/packages/harp-device/pyproject.toml +++ b/src/packages/harp-device/pyproject.toml @@ -5,6 +5,8 @@ description = "Transport-agnostic Harp device protocol layer" requires-python = ">=3.11" dependencies = [ "harp-protocol", + "pydantic>=2", + "pydantic-yaml>=1", ] [build-system] diff --git a/src/packages/harp-device/src/harp/device/__init__.py b/src/packages/harp-device/src/harp/device/__init__.py index ac81527..d352a19 100644 --- a/src/packages/harp-device/src/harp/device/__init__.py +++ b/src/packages/harp-device/src/harp/device/__init__.py @@ -1,4 +1,5 @@ from ._device import Device, EventHandler, Subscription +from ._emit_device import create_device from ._framer import HarpFramer from ._registers import ( AssemblyVersion, @@ -26,12 +27,16 @@ WhoAmI, ) from ._register_map import REGISTER_MAP +from ._schema import ConverterContext, parse_device_schema from ._transport import ITransport, TransportError __all__ = [ "Device", "EventHandler", "Subscription", + "create_device", + "parse_device_schema", + "ConverterContext", "HarpFramer", "ITransport", "TransportError", diff --git a/src/packages/harp-device/src/harp/device/_device.py b/src/packages/harp-device/src/harp/device/_device.py index 69adfd7..d765da3 100644 --- a/src/packages/harp-device/src/harp/device/_device.py +++ b/src/packages/harp-device/src/harp/device/_device.py @@ -81,6 +81,9 @@ class Device: #: Expected ``WhoAmI`` of the device this class models; ``0x0`` skips the check. __whoami__: ClassVar[int] = 0x0 + #: Address -> register class; empty on the base, overridden by generated devices. + REGISTER_MAP: ClassVar[dict[int, type[RegisterBase[Any]]]] = {} + def __init__(self, transport: ITransport, *, raise_on_error: bool = True) -> None: self._transport = transport self.raise_on_error = raise_on_error diff --git a/src/packages/harp-device/src/harp/device/_emit_device.py b/src/packages/harp-device/src/harp/device/_emit_device.py new file mode 100644 index 0000000..82f43b4 --- /dev/null +++ b/src/packages/harp-device/src/harp/device/_emit_device.py @@ -0,0 +1,37 @@ +from typing import Any, Mapping, Optional, Union + +from ._device import Device +from ._register_map import REGISTER_MAP as CORE_REGISTER_MAP +from ._schema import create_registers, parse_device_schema +from ._schema._emit import ConverterValue +from ._schema._model import DeviceModel + + +def create_device( + source: Union[str, DeviceModel], + *, + name: Optional[str] = None, + converters: Optional[Mapping[str, ConverterValue]] = None, + strict: bool = True, + exclude_private: bool = True, +) -> type[Device]: + """Emit a :class:`Device` subclass from a device schema. + + The returned class exposes its registers through the ``REGISTER_MAP`` class + attribute (address -> register class) and carries ``__whoami__`` from the schema + (``0x0`` when absent). The device's registers are spread on top of the core common + map; on an address clash the device's register wins. ``exclude_private=True`` drops + registers whose DSL ``visibility`` is ``private``. A header-less register + fragment yields a device with no ``device`` name (falls back to ``"Device"``). + """ + device = source if isinstance(source, DeviceModel) else parse_device_schema(source) + registers = create_registers( + device, converters=converters, strict=strict, exclude_private=exclude_private + ) + by_address = {cls.address: cls for cls in registers.values()} + + namespace: dict[str, Any] = { + "__whoami__": int(device.whoAmI or 0), + "REGISTER_MAP": {**CORE_REGISTER_MAP, **by_address}, + } + return type(name or device.device or "Device", (Device,), namespace) diff --git a/src/packages/harp-device/src/harp/device/_schema/__init__.py b/src/packages/harp-device/src/harp/device/_schema/__init__.py new file mode 100644 index 0000000..cc41ac3 --- /dev/null +++ b/src/packages/harp-device/src/harp/device/_schema/__init__.py @@ -0,0 +1,45 @@ +from ._model import ( + Access, + BitMask, + Converter, + GroupMask, + InterfaceType, + MaskType, + MaskValue, + DeviceModel, + PayloadMember, + PayloadType, + Register, + Registers, + Visibility, +) +from ._emit import ( + ConverterContext, + ConverterFactory, + ConverterValue, + UnknownConverterError, + create_registers, + parse_device_schema, +) + +__all__ = [ + "parse_device_schema", + "create_registers", + "ConverterContext", + "ConverterFactory", + "ConverterValue", + "UnknownConverterError", + "DeviceModel", + "Registers", + "Register", + "PayloadMember", + "PayloadType", + "Access", + "Visibility", + "Converter", + "BitMask", + "GroupMask", + "MaskType", + "MaskValue", + "InterfaceType", +] diff --git a/src/packages/harp-device/src/harp/device/_schema/_emit.py b/src/packages/harp-device/src/harp/device/_schema/_emit.py new file mode 100644 index 0000000..966149f --- /dev/null +++ b/src/packages/harp-device/src/harp/device/_schema/_emit.py @@ -0,0 +1,431 @@ +import enum +import types +from dataclasses import dataclass +from typing import Any, Callable, Mapping, Optional, Union + +import numpy as np +from typing_extensions import Sentinel +from pydantic_yaml import parse_yaml_raw_as +from harp.protocol import ( + AnonymousPayload, + BitMask, + BoolConverter, + Converter, + Field, + GroupMask, + HarpVersionConverter, + IdentityConverter, + RegisterBase, + RegisterFloat, + RegisterFloatArray, + RegisterS8, + RegisterS8Array, + RegisterS16, + RegisterS16Array, + RegisterS32, + RegisterS32Array, + RegisterS64, + RegisterS64Array, + RegisterU8, + RegisterU8Array, + RegisterU16, + RegisterU16Array, + RegisterU32, + RegisterU32Array, + RegisterU64, + RegisterU64Array, + StringConverter, + StructPayload, +) +from harp.protocol import PayloadType as ProtoPayloadType + +from ._model import DeviceModel, PayloadMember, PayloadType, Register, Registers, Visibility + +# Register base element: schema PayloadType -> numpy scalar type (byte size via np.dtype). +_ELEMENT: dict[PayloadType, type[np.generic]] = { + PayloadType.U8: np.uint8, + PayloadType.S8: np.int8, + PayloadType.U16: np.uint16, + PayloadType.S16: np.int16, + PayloadType.U32: np.uint32, + PayloadType.S32: np.int32, + PayloadType.U64: np.uint64, + PayloadType.S64: np.int64, + PayloadType.Float: np.float32, +} +_SCALAR_REGISTER: dict[PayloadType, Any] = { + PayloadType.U8: RegisterU8, + PayloadType.S8: RegisterS8, + PayloadType.U16: RegisterU16, + PayloadType.S16: RegisterS16, + PayloadType.U32: RegisterU32, + PayloadType.S32: RegisterS32, + PayloadType.U64: RegisterU64, + PayloadType.S64: RegisterS64, + PayloadType.Float: RegisterFloat, +} +_ARRAY_REGISTER: dict[PayloadType, Any] = { + PayloadType.U8: RegisterU8Array, + PayloadType.S8: RegisterS8Array, + PayloadType.U16: RegisterU16Array, + PayloadType.S16: RegisterS16Array, + PayloadType.U32: RegisterU32Array, + PayloadType.S32: RegisterS32Array, + PayloadType.U64: RegisterU64Array, + PayloadType.S64: RegisterS64Array, + PayloadType.Float: RegisterFloatArray, +} + + +@dataclass(frozen=True) +class ConverterContext: + """A payload value's schema definition, resolved against its register context. + + Handed to every converter factory so it can construct the converter with the + right arguments — e.g. ``StringConverter(span)``, ``HarpVersionConverter(element)``, + or ``IdentityConverter(dtype)``. + """ + + name: str # yml field key ("__value__" for a whole-register value) + interface_type: Optional[str] # the DSL interfaceType (None = raw/native) + mask: Optional[int] # bit mask, when the value is bit-packed + length: int # element count this value spans (0 = unset -> scalar) + element: np.dtype # the register's base element dtype (from PayloadType) + element_size: int # the register's base element byte size + + @property + def span(self) -> int: + """Byte span of the value (element count * element size).""" + return max(1, self.length) * self.element_size + + @property + def member_dtype(self) -> np.dtype: + """The value's own numpy dtype — a native primitive interfaceType overrides the element.""" + if self.interface_type is not None: + entry = _INTERFACES.get(self.interface_type) + if entry is not None and entry.native_dtype is not None: + return np.dtype(entry.native_dtype) + return self.element + + @property + def raw_dtype(self) -> np.dtype: + """Native passthrough dtype — a sub-array when the value spans >1 element.""" + if self.length > 1: + return np.dtype((self.element.type, (self.length,))) + return self.element + + +# A converter factory builds a converter from a field's DSL context. +ConverterFactory = Callable[[ConverterContext], Converter[Any]] +# A user-supplied converter: a ready instance, or a factory that builds one from context. +ConverterValue = Union[Converter[Any], ConverterFactory] +# Internal built-in factory — may decline (return None) when the DSL type doesn't +# actually fit (e.g. a primitive whose declared byte span isn't its native size). +_InterfaceFactory = Callable[[ConverterContext], Optional[Converter[Any]]] +# Coerces a field's yml numeric default into its typed value (``_NO_DEFAULT`` = skip). +_DefaultCoercer = Callable[[float, ConverterContext], Any] + +_NO_DEFAULT = Sentinel("_NO_DEFAULT") # this interface has no numeric default representation + + +def _numpy_default(value: float, ctx: ConverterContext) -> Any: + literal = int(value) if value == np.floor(value) else value + return np.dtype(ctx.member_dtype).type(literal) + + +def _bool_default(value: float, ctx: ConverterContext) -> Any: + return bool(value != 0) + + +def _skip_default(value: float, ctx: ConverterContext) -> Any: + return _NO_DEFAULT + + +def _native(dtype: type[np.generic]) -> _InterfaceFactory: + # A native primitive decodes as an identity passthrough. Unmasked, it only fits + # when its declared byte span matches its width; otherwise it declines (returns + # None) and the field re-interprets the bytes via a custom ``{Field}Converter``. + # Masked, it is always a native slice of the element (member_dtype == this width). + d = np.dtype(dtype) + return lambda ctx: ( + IdentityConverter(d) if ctx.mask is not None or ctx.span == d.itemsize else None + ) + + +@dataclass(frozen=True) +class _Interface: + """A built-in interfaceType: how to build its converter, how to coerce its + default value, and its native numpy dtype (fixed-width primitives only).""" + + build: _InterfaceFactory + default: _DefaultCoercer + native_dtype: Optional[type[np.generic]] = None + + +# Every interfaceType the library handles natively, in one uniform table: the +# fixed-width primitives (identity passthrough, carrying their numpy scalar as +# ``native_dtype``) beside string/bool/HarpVersion. Custom interfaceTypes are +# supplied by the caller (see ``converters=``). +_INTERFACES: dict[str, _Interface] = { + "byte": _Interface(_native(np.uint8), _numpy_default, np.uint8), + "sbyte": _Interface(_native(np.int8), _numpy_default, np.int8), + "short": _Interface(_native(np.int16), _numpy_default, np.int16), + "ushort": _Interface(_native(np.uint16), _numpy_default, np.uint16), + "int": _Interface(_native(np.int32), _numpy_default, np.int32), + "uint": _Interface(_native(np.uint32), _numpy_default, np.uint32), + "long": _Interface(_native(np.int64), _numpy_default, np.int64), + "ulong": _Interface(_native(np.uint64), _numpy_default, np.uint64), + "float": _Interface(_native(np.float32), _numpy_default, np.float32), + "string": _Interface(lambda ctx: StringConverter(ctx.span), _skip_default), + "bool": _Interface(lambda ctx: BoolConverter(), _bool_default), + "HarpVersion": _Interface(lambda ctx: HarpVersionConverter(ctx.element), _skip_default), +} + + +def _materialize(value: ConverterValue, ctx: ConverterContext) -> Converter[Any]: + """A user converter value is either a ready instance or a ``(ctx) -> Converter`` factory.""" + return value if isinstance(value, Converter) else value(ctx) + + +class UnknownConverterError(ValueError): + """A custom ``interfaceType`` needs a converter not found in ``converters=``.""" + + +def _is_native(interface_type: Optional[str]) -> bool: + """True when a value decodes as a native numpy passthrough: no interfaceType, or a + fixed-width primitive one. Such a whole-register value needs no payload wrapper. + """ + if interface_type is None: + return True + entry = _INTERFACES.get(interface_type) + return entry is not None and entry.native_dtype is not None + + +def _new_class(name: str, bases: tuple, namespace: dict, kwds: Optional[dict] = None) -> type: + return types.new_class(name, bases, kwds or {}, lambda ns: ns.update(namespace)) + + +class _Emitter: + def __init__( + self, + device: Union[DeviceModel, Registers], + converters: Optional[Mapping[str, ConverterValue]], + strict: bool, + exclude_private: bool, + ) -> None: + self.device = device + self.converters = dict(converters or {}) + self.strict = strict + self.exclude_private = exclude_private + self.group_masks = device.groupMasks or {} + self.bit_masks = device.bitMasks or {} + self.enums = self._build_enums() + + # -- enums ------------------------------------------------------------ + def _build_enums(self) -> dict[str, Any]: + # Enum names and members are kept verbatim from the yml. + enums: dict[str, Any] = {} + for name, spec in self.bit_masks.items(): + # IntFlag has no zero-valued member; drop it if present. + members = {k: int(v) for k, v in spec.bits.items() if int(v) != 0} + enums[name] = enum.IntFlag(name, members) + for name, spec in self.group_masks.items(): + members = {k: int(v) for k, v in spec.values.items()} + enums[name] = enum.IntEnum(name, members) + return enums + + # -- converter resolution (one uniform factory pipeline) ------------- + def _resolve_converter(self, ctx: ConverterContext) -> Converter[Any]: + """Build the converter for a payload value from its schema context.""" + it = ctx.interface_type + entry = _INTERFACES.get(it) if it is not None else None + if entry is not None: + converter = entry.build(ctx) + if converter is not None: + return converter + if ctx.mask is not None: + return IdentityConverter(ctx.member_dtype) # bit-field: native slice of the element + if it is None: + return IdentityConverter(ctx.raw_dtype) # raw passthrough / sub-array + # A known primitive that didn't fit is re-interpreted per field (``{Name}Converter``); + # an unknown interfaceType is a domain type (``{InterfaceType}Converter``). + symbol = f"{ctx.name}Converter" if entry is not None else f"{it}Converter" + return self._extension(symbol, ctx) + + def _extension(self, symbol: str, ctx: ConverterContext) -> Converter[Any]: + value = self.converters.get(symbol) + if value is not None: + return _materialize(value, ctx) + if not self.strict: + return IdentityConverter(ctx.element) + raise UnknownConverterError( + f"no converter {symbol!r} in converters=; pass " + f"converters={{{symbol!r}: Converter>}} " + f"or strict=False to decode as the native type" + ) + + # -- defaults --------------------------------------------------------- + def _default(self, member: PayloadMember, type_name: str, ctx: ConverterContext) -> Any: + """The field's typed default value, or ``_NO_DEFAULT`` when it has none.""" + _default_value = member.defaultValue if member.defaultValue is not None else member.minValue + if _default_value is None or (member.length or 0) > 1: + return _NO_DEFAULT + value = float(_default_value.root) + if type_name in self.group_masks: + e = self.enums[type_name] + for mv in self.group_masks[type_name].values.values(): + if int(mv) == int(value): + return e(int(value)) + return int(value) + if member.converter is not None: + return _NO_DEFAULT # a custom converter owns its own decoding; no numeric default + it = ctx.interface_type + entry = _INTERFACES.get(it) if it is not None else None + if entry is not None: + return entry.default(value, ctx) + if it is None: + return _numpy_default(value, ctx) # raw native passthrough + return _NO_DEFAULT # custom domain interfaceType: no numeric default + + # -- fields ----------------------------------------------------------- + def _build_field(self, key: str, member: PayloadMember, reg: Register) -> tuple[str, Any]: + elem_np = _ELEMENT[reg.type] + elem_size = np.dtype(elem_np).itemsize + offset = member.offset or 0 + it = member.interfaceType.root if member.interfaceType else None + type_name = it or (member.maskType.root if member.maskType else "") + ctx = ConverterContext( + name=key, + interface_type=it, + mask=member.mask, + length=member.length or 0, + element=np.dtype(elem_np), + element_size=elem_size, + ) + default = self._default(member, type_name, ctx) + default_kwarg = {} if default is _NO_DEFAULT else {"default": default} + + # A group mask is an enum sub-field descriptor, not a Field(converter). + if type_name in self.group_masks: + full = (1 << (elem_size * 8)) - 1 + mask = member.mask if member.mask is not None else full + return key, GroupMask( + enum=self.enums[type_name], mask=mask, offset=offset, **default_kwarg + ) + + field_kwargs: dict[str, Any] = {"offset": offset, **default_kwarg} + if member.mask is not None: + field_kwargs["mask"] = member.mask + return key, Field(self._resolve_converter(ctx), **field_kwargs) + + # -- payloads --------------------------------------------------------- + def _build_payload(self, name: str, reg: Register) -> type: + elem_np = _ELEMENT[reg.type] + elem_size = np.dtype(elem_np).itemsize + length = reg.length or 1 + + if reg.payloadSpec is not None: + namespace = {} + for key, member in reg.payloadSpec.items(): + fname, descriptor = self._build_field(key, member, reg) + namespace[fname] = descriptor + kwds = {"length": length} if length > 1 else {} + return _new_class(f"{name}Payload", (StructPayload[elem_np],), namespace, kwds) + + # anonymous single-value payload + mt = reg.maskType.root if reg.maskType else None + it = reg.interfaceType.root if reg.interfaceType else None + if mt in self.group_masks: + full = (1 << (elem_size * 8)) - 1 + descriptor: Any = GroupMask(enum=self.enums[mt], mask=full) + elif mt in self.bit_masks: + descriptor = BitMask(enum=self.enums[mt]) + else: + assert it is not None, ( + f"{name}: register needs a payloadSpec, maskType, or interfaceType" + ) + ctx = ConverterContext( + name="__value__", + interface_type=it, + mask=None, + length=length, + element=np.dtype(elem_np), + element_size=elem_size, + ) + descriptor = Field(self._resolve_converter(ctx)) + return _new_class(f"{name}Payload", (AnonymousPayload[elem_np],), {"__value__": descriptor}) + + # -- registers -------------------------------------------------------- + def _build_register(self, name: str, reg: Register) -> type[RegisterBase[Any]]: + length = reg.length or 1 + it = reg.interfaceType.root if reg.interfaceType else None + + # A plain scalar/array register needs no payload wrapper: its whole value is a + # native passthrough (no payloadSpec, no maskType, no custom converter). + if ( + reg.payloadSpec is None + and reg.maskType is None + and reg.converter is None + and _is_native(it) + ): + if length > 1: # plain array register + cls = _ARRAY_REGISTER[reg.type](reg.address, length=length) + cls.__name__ = cls.__qualname__ = name + return cls + return _new_class(name, (_SCALAR_REGISTER[reg.type],), {"address": reg.address}) + + payload_cls = self._build_payload(name, reg) + return _new_class( + name, + (RegisterBase,), + { + "address": reg.address, + "payload_type": ProtoPayloadType[reg.type.name], + "payload_class": payload_cls, + }, + ) + + def emit(self) -> dict[str, type[RegisterBase[Any]]]: + return { + name: self._build_register(name, reg) + for name, reg in self.device.registers.items() + if not (self.exclude_private and reg.visibility is Visibility.private) + } + + +def parse_device_schema(text: str) -> DeviceModel: + """Parse a Harp ``device.yml`` (or a header-less fragment) into a :class:`DeviceModel`. + + A header-less fragment (just ``registers`` / ``bitMasks`` / ``groupMasks``) + parses fine — the identity fields (``device`` / ``whoAmI`` / ...) are simply + ``None``. Read files yourself, e.g. + ``parse_device_schema(Path("device.yml").read_text())``. + + Uses ``pydantic-yaml`` (ruamel-backed, YAML 1.2), so group-mask keys like + ``Off`` / ``On`` stay strings instead of being coerced to booleans. + """ + return parse_yaml_raw_as(DeviceModel, text) + + +def create_registers( + source: Union[str, DeviceModel, Registers], + *, + converters: Optional[Mapping[str, ConverterValue]] = None, + strict: bool = True, + exclude_private: bool = False, +) -> dict[str, type[RegisterBase[Any]]]: + """Emit runtime register classes from a device schema. + + ``source`` is yaml text or an already-parsed :class:`DeviceModel` / + :class:`Registers`. Identifiers (fields, enum members) are + kept verbatim from the yml. ``converters`` supplies custom converters keyed by + symbol name (e.g. ``{"DataConverter": ...}``); a value is either a ready + :class:`~harp.protocol.Converter` instance or a factory + ``(ctx: ConverterContext) -> Converter`` that builds one from the field's DSL + context. A custom type with no matching converter raises + ``UnknownConverterError`` when ``strict`` (the default); ``strict=False`` + decodes it as its native element type instead. ``exclude_private=True`` drops + registers whose DSL ``visibility`` is ``private``. + """ + device = source if isinstance(source, Registers) else parse_device_schema(source) + return _Emitter(device, converters, strict, exclude_private).emit() diff --git a/src/packages/harp-device/src/harp/device/_schema/_model.py b/src/packages/harp-device/src/harp/device/_schema/_model.py new file mode 100644 index 0000000..0d11f28 --- /dev/null +++ b/src/packages/harp-device/src/harp/device/_schema/_model.py @@ -0,0 +1,145 @@ +"""Pydantic object model of a Harp ``device.yml`` / ``registers`` schema. + +TODO: hand-maintained for now. Auto-generating it from the upstream +``harp-tech/protocol`` JSON schema is deferred until that schema stabilises. +""" + +from enum import Enum +from typing import Annotated, Dict, List, Optional, Union + +from pydantic import BaseModel, ConfigDict, Field, RootModel + + +class PayloadType(str, Enum): + """Register payload element type. Values match the schema enum (the yml names).""" + + U8 = "U8" + S8 = "S8" + U16 = "U16" + S16 = "S16" + U32 = "U32" + S32 = "S32" + U64 = "U64" + S64 = "S64" + Float = "Float" + + +class Access(Enum): + Read = "Read" + Write = "Write" + Event = "Event" + + +class Visibility(Enum): + public = "public" + private = "private" + + +class Converter(Enum): + None_ = "None" + Payload = "Payload" + RawPayload = "RawPayload" + + +class MaskValueItem(BaseModel): + model_config = ConfigDict(extra="forbid") + value: int = Field(..., description="Specifies the numerical mask value.") + description: Optional[str] = Field(None, description="Summary of the mask value function.") + + def __int__(self) -> int: + return self.value + + +class MaskValue(RootModel[Union[int, MaskValueItem]]): + root: Union[int, MaskValueItem] + + def __int__(self) -> int: + return int(self.root) + + +class BitMask(BaseModel): + description: Optional[str] = Field(None, description="Summary of the bit mask function.") + bits: Dict[str, MaskValue] + + +class GroupMask(BaseModel): + description: Optional[str] = Field(None, description="Summary of the group mask function.") + values: Dict[str, MaskValue] + + +class MaskType(RootModel[str]): + root: str + + +class InterfaceType(RootModel[str]): + root: str + + +class MinValue(RootModel[float]): + root: float + + +class MaxValue(RootModel[float]): + root: float + + +class DefaultValue(RootModel[float]): + root: float + + +class PayloadMember(BaseModel): + mask: Optional[int] = Field(None, description="Mask used to read/write this member.") + offset: Optional[int] = Field(None, description="Payload array offset of this member.") + length: Optional[int] = Field(None, description="Number of base elements this member spans.") + description: Optional[str] = Field(None, description="Summary of the payload member.") + minValue: Optional[MinValue] = None + maxValue: Optional[MaxValue] = None + defaultValue: Optional[DefaultValue] = None + maskType: Optional[MaskType] = None + interfaceType: Optional[InterfaceType] = None + converter: Optional[Converter] = None + + +class Register(BaseModel): + address: Annotated[int, Field(le=255, description="Unique 8-bit register address.")] + type: PayloadType + length: Annotated[Optional[int], Field(ge=1, default=1, description="Payload length.")] + access: Union[Access, List[Access]] = Field(..., description="Expected use of the register.") + description: Optional[str] = Field(None, description="Summary of the register function.") + minValue: Optional[MinValue] = None + maxValue: Optional[MaxValue] = None + defaultValue: Optional[DefaultValue] = None + maskType: Optional[MaskType] = None + visibility: Optional[Visibility] = Field( + None, description="Exposed in the high-level interface." + ) + volatile: Optional[bool] = Field(None, description="Value can be saved in non-volatile memory.") + payloadSpec: Optional[Dict[str, PayloadMember]] = None + interfaceType: Optional[InterfaceType] = None + converter: Optional[Converter] = None + + +class Registers(BaseModel): + """A bare register collection — a header-less ``device.yml`` fragment.""" + + registers: Dict[str, Register] = Field(..., description="The device's registers.") + bitMasks: Optional[Dict[str, BitMask]] = None + groupMasks: Optional[Dict[str, GroupMask]] = None + + +class DeviceModel(Registers): + """A device schema: a `Registers` collection plus (optional) device identity. + + Every identity field is optional, so a header-less fragment (just ``registers`` + / ``bitMasks`` / ``groupMasks``) is simply a ``DeviceModel`` with them all None + — parsing never needs to branch on "fragment vs full document". + """ + + device: Optional[str] = Field(None, description="The name of the device.") + whoAmI: Optional[int] = Field(None, description="Unique identifier for this device type.") + firmwareVersion: Optional[str] = Field( + None, description="Semantic version of the device firmware." + ) + hardwareTargets: Optional[str] = Field( + None, description="Semantic version of the device hardware." + ) diff --git a/src/packages/harp-protocol/src/harp/protocol/_payload.py b/src/packages/harp-protocol/src/harp/protocol/_payload.py index 22e5c74..896ff2a 100644 --- a/src/packages/harp-protocol/src/harp/protocol/_payload.py +++ b/src/packages/harp-protocol/src/harp/protocol/_payload.py @@ -180,7 +180,8 @@ def _columns( def _build_enum_lookup(enum_cls: type[enum.IntEnum]) -> "tuple[list[str], np.ndarray]": - """Helper for GroupMask to build the category list and code lookup table for a given enum.IntEnum class.""" + """Category list + a code table mapping each raw enum value (``0..max member``) to its + category index; a raw value with no member maps to -1.""" members = list(enum_cls) categories = [m.name for m in members] max_val = max(int(m) for m in members) @@ -234,10 +235,16 @@ def __init__( self._slot: str = "" self._dtype: np.dtype = _DEFAULT_ELEMENT self._categories, self._code_lookup = _build_enum_lookup(enum) + self._lookup_safe = (mask >> self._shift) < len(self._code_lookup) def _decode_raw(self, raw: Any) -> Any: - """Map an extracted (already masked + shifted) integer to its enum member.""" - return self._enum(int(raw)) + """Map an extracted (masked + shifted) integer to its enum member, preserving an + undefined code as its raw int (permissive, like C#'s unchecked enum cast).""" + value = int(raw) + try: + return self._enum(value) + except ValueError: + return value def _encode_value(self, value: Any) -> int: """Map a user value back to the integer to be masked + shifted into the slot.""" @@ -272,9 +279,24 @@ def _columns( ) -> "list[Column]": """One enum column: category codes + labels (``decode_enums``) or raw codes.""" raw = (arr[self._slot] & self._mask) >> self._shift - if decode_enums: - return [Column(name, self._code_lookup[raw], self._categories)] - return [Column(name, raw)] + if not decode_enums: + return [Column(name, raw)] + lookup = self._code_lookup + # ``_lookup_safe`` (the field's raw range fits the table) skips the bounds guard; + # an in-range gap still maps to -1, so the undefined branch below runs regardless. + if self._lookup_safe: + codes = lookup[raw] + else: + codes = np.where(raw < len(lookup), lookup.take(raw, mode="clip"), -1) + undefined = codes < 0 + if not undefined.any(): + return [Column(name, codes, self._categories)] + # An undefined code (an in-range gap, or a value past the enum's range) is kept + # as its raw integer — an extra category — matching the scalar decode and C#'s + codes = codes.astype(np.intp) + extras = np.unique(raw[undefined]) + codes[undefined] = len(self._categories) + np.searchsorted(extras, raw[undefined]) + return [Column(name, codes, list(self._categories) + extras.tolist())] class BitMask(Generic[F]): diff --git a/src/packages/harp-protocol/src/harp/protocol/_register.py b/src/packages/harp-protocol/src/harp/protocol/_register.py index 9f221dc..4e07864 100644 --- a/src/packages/harp-protocol/src/harp/protocol/_register.py +++ b/src/packages/harp-protocol/src/harp/protocol/_register.py @@ -2,7 +2,7 @@ from typing import Any, ClassVar, Generic, TypeVar, cast, final, overload import numpy as np -from numpy.typing import NDArray +from numpy.typing import ArrayLike, NDArray from typing_extensions import Sentinel from ._builder import build_message_frame @@ -15,7 +15,7 @@ _TS_MICROS_OFFSET, ) from ._message import HarpMessage -from ._message_type import MessageType +from ._message_type import MessageType, message_type_to_byte from ._payload import ( Batch, PayloadBase, @@ -38,10 +38,26 @@ PayloadU64, PayloadU64Array, ) -from ._payload_type import PayloadType +from ._payload_type import PayloadType, encode_payload_type _MISSING = Sentinel("_MISSING") + +def _encode_message_types(message_type: MessageType | ArrayLike, nrows: int) -> NDArray[np.uint8]: + """Resolve a scalar/array ``message_type`` argument to N message-type bytes. + + A single :class:`MessageType` (error-bit aware) fills all frames; a scalar int + is used verbatim; an array (e.g. the msgtype view from ``parse_bulk``, or a + list of ``MessageType``/ints) becomes the per-frame bytes. + """ + if isinstance(message_type, MessageType): + return np.full(nrows, message_type_to_byte(message_type), dtype=np.uint8) + values = np.asarray(message_type) + if values.ndim == 0: + return np.full(nrows, int(values.item()), dtype=np.uint8) + return values.astype(np.uint8) + + U = TypeVar("U") _R = TypeVar("_R") _AR = TypeVar("_AR", bound="RegisterBase[Any]") @@ -182,6 +198,76 @@ def parse_bulk( payload = payload_cls.from_array(payload_arr) return data, timestamps, msgtype_view, cast("Batch[Any]", payload) + @classmethod + def format_bulk( + cls, + values: PayloadBase | ArrayLike, + *, + timestamps: ArrayLike | None = None, + message_type: MessageType | ArrayLike = MessageType.Event, + port: int = _DEFAULT_PORT, + ) -> NDArray[np.uint8]: + """Build a flat buffer of N frames of this register type — the inverse of + :meth:`parse_bulk`. + + ``values`` is a payload (scalar or :class:`Batch`) or an ndarray of the + register's ``payload_class.dtype``. ``timestamps`` (a length-N array of + seconds) makes every frame timestamped. ``message_type`` is one + :class:`MessageType` for all frames, or a length-N array of message-type + bytes / values (e.g. the ``msgtype`` view returned by ``parse_bulk``). + """ + payload_cls = cls.payload_class + itemsize = payload_cls.dtype.itemsize + if isinstance(values, PayloadBase): + records = np.atleast_1d(np.asarray(values.raw_payload)) + else: + records = np.atleast_1d(np.asarray(values)) + # Coerce the element type only for plain scalar payloads (e.g. an int + # list for a scalar register). Struct/sub-array records already carry + # the right byte layout and must not be re-cast. + plain = ( + records.dtype.names is None + and records.dtype.subdtype is None + and payload_cls.dtype.names is None + and payload_cls.dtype.subdtype is None + ) + if plain and records.dtype != payload_cls.dtype: + records = records.astype(payload_cls.dtype) + nrows = len(records) + flat = np.ascontiguousarray(records).tobytes() + if len(flat) != nrows * itemsize: + raise ValueError( + f"{cls.__name__}.format_bulk: {len(flat)} payload bytes for {nrows} frames " + f"is not a multiple of itemsize {itemsize}; check the values shape/dtype" + ) + + is_timestamped = timestamps is not None + payload_offset = _TIMESTAMPED_PAYLOAD_OFFSET if is_timestamped else _HEADER_LEN + stride = payload_offset + itemsize + 1 # trailing checksum byte + + buf = np.zeros((nrows, stride), dtype=np.uint8) + buf[:, 0] = _encode_message_types(message_type, nrows) + buf[:, 1] = stride - 2 + buf[:, 2] = cls.address + buf[:, 3] = port + buf[:, 4] = encode_payload_type(cls.payload_type, has_timestamp=is_timestamped) + + if is_timestamped: + ts = np.atleast_1d(np.asarray(timestamps, dtype=np.float64)) + seconds = ts.astype(np.uint32) + micros = np.round((ts - seconds.astype(np.float64)) / _TICK_PERIOD_S).astype(np.uint16) + buf[:, _HEADER_LEN:_TS_MICROS_OFFSET] = np.frombuffer( + seconds.astype(" str: + """The generators test-metadata ``device.yml`` as text.""" + return (ASSETS / "device.yml").read_text() + + +@pytest.fixture(scope="session") +def common_yml() -> str: + """The Harp common (core) register set ``common.yml`` as text.""" + return (ASSETS / "common.yml").read_text() diff --git a/tests/device/__init__.py b/tests/device/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/device/converters.py b/tests/device/converters.py new file mode 100644 index 0000000..1957988 --- /dev/null +++ b/tests/device/converters.py @@ -0,0 +1,36 @@ +from typing import Any + +import numpy as np +from numpy.typing import NDArray + +from harp.protocol import Converter + + +class DataConverter(Converter[int]): + """Maps two raw little-endian signed bytes to and from a Python int. + + Models interfaceType: int over a two-byte sub-region of the CustomMemberConverter payload. + """ + + init_kwarg_type = int + + def __init__(self) -> None: + self._length = 2 + self.dtype = np.dtype((np.uint8, (self._length,))) + + def decode_scalar(self, view: np.generic) -> int: + return int.from_bytes(bytes(np.asarray(view).tolist()), "little", signed=True) + + def decode_batch(self, view: NDArray[np.generic]) -> Any: + return np.array( + [ + int.from_bytes(bytes(np.asarray(r).tolist()), "little", signed=True) + for r in np.atleast_2d(view) + ], + dtype=object, + ) + + def encode_into(self, view: NDArray[np.generic], value: int) -> None: + view[...] = np.frombuffer( + int(value).to_bytes(self._length, "little", signed=True), dtype=np.uint8 + ) diff --git a/tests/device/expected_core.py b/tests/device/expected_core.py new file mode 100644 index 0000000..6684278 --- /dev/null +++ b/tests/device/expected_core.py @@ -0,0 +1,229 @@ +# This file was automatically generated and should not be edited directly. +# To make changes, edit the device metadata and regenerate the interface. + +import enum +from typing import Any, ClassVar + +import numpy as np +from harp.protocol import ( + AnonymousPayload, + BitMask, + BoolConverter, + Field, + GroupMask, + PayloadType, + RegisterBase, + RegisterU16, + RegisterU32, + RegisterU8, + StringConverter, + StructPayload, +) + + +class ResetFlags(enum.IntFlag): + """Specifies the behavior of the non-volatile registers when resetting the device.""" + + RESTORE_DEFAULT = 0x1 + """The device will boot with all the registers reset to their default factory values.""" + RESTORE_EEPROM = 0x2 + """The device will boot and restore all the registers to the values stored in non-volatile memory.""" + SAVE = 0x4 + """The device will boot and save all the current register values to non-volatile memory.""" + RESTORE_NAME = 0x8 + """The device will boot with the default device name.""" + UPDATE_FIRMWARE = 0x20 + """The device will enter firmware update mode.""" + BOOT_FROM_DEFAULT = 0x40 + """Specifies that the device has booted from default factory values.""" + BOOT_FROM_EEPROM = 0x80 + """Specifies that the device has booted from non-volatile values stored in EEPROM.""" + + +class ClockConfigurationFlags(enum.IntFlag): + """Specifies configuration flags for the device synchronization clock.""" + + CLOCK_REPEATER = 0x1 + """The device will repeat the clock synchronization signal to the clock output connector, if available.""" + CLOCK_GENERATOR = 0x2 + """The device resets and generates the clock synchronization signal on the clock output connector, if available.""" + REPEATER_CAPABILITY = 0x8 + """Specifies the device has the capability to repeat the clock synchronization signal to the clock output connector.""" + GENERATOR_CAPABILITY = 0x10 + """Specifies the device has the capability to generate the clock synchronization signal to the clock output connector.""" + CLOCK_UNLOCK = 0x40 + """The device will unlock the timestamp register counter and will accept commands to set new timestamp values.""" + CLOCK_LOCK = 0x80 + """The device will lock the timestamp register counter and will not accept commands to set new timestamp values.""" + + +class OperationMode(enum.IntEnum): + """Specifies the operation mode of the device.""" + + STANDBY = 0 + """Disable all event reporting on the device.""" + ACTIVE = 1 + """Event detection is enabled. Only enabled events are reported by the device.""" + SPEED = 3 + """The device enters speed mode.""" + + +class EnableFlag(enum.IntEnum): + """Specifies whether a specific register flag is enabled or disabled.""" + + DISABLED = 0 + """Specifies that the flag is disabled.""" + ENABLED = 1 + """Specifies that the flag is enabled.""" + + +class OperationControlPayload(StructPayload[np.uint8]): + """Represents the payload of the OperationControl register.""" + + operation_mode: OperationMode = GroupMask(enum=OperationMode, mask=0x3) + """Specifies the operation mode of the device.""" + dump_registers: bool = Field(BoolConverter(), mask=0x8) + """Specifies whether the device should report the content of all registers on initialization.""" + mute_replies: bool = Field(BoolConverter(), mask=0x10) + """Specifies whether the replies to all commands will be muted, i.e. not sent by the device.""" + visual_indicators: EnableFlag = GroupMask(enum=EnableFlag, mask=0x20) + """Specifies the state of all visual indicators on the device.""" + operation_led: EnableFlag = GroupMask(enum=EnableFlag, mask=0x40) + """Specifies whether the device state LED should report the operation mode of the device.""" + heartbeat: EnableFlag = GroupMask(enum=EnableFlag, mask=0x80) + """Specifies whether the device should report the content of the seconds register each second.""" + + +class ResetDevicePayload(AnonymousPayload[np.uint8]): + """Represents the payload of the ResetDevice register.""" + + __value__: ResetFlags = BitMask(enum=ResetFlags) + + +class DeviceNamePayload(AnonymousPayload[np.uint8]): + """Represents the payload of the DeviceName register.""" + + __value__: str = Field(StringConverter(25)) + + +class ClockConfigurationPayload(AnonymousPayload[np.uint8]): + """Represents the payload of the ClockConfiguration register.""" + + __value__: ClockConfigurationFlags = BitMask(enum=ClockConfigurationFlags) + + +class WhoAmI(RegisterU16): + """Specifies the identity class of the device.""" + + address: ClassVar[int] = 0 + + +class HardwareVersionHigh(RegisterU8): + """Specifies the major hardware version of the device.""" + + address: ClassVar[int] = 1 + + +class HardwareVersionLow(RegisterU8): + """Specifies the minor hardware version of the device.""" + + address: ClassVar[int] = 2 + + +class AssemblyVersion(RegisterU8): + """Specifies the version of the assembled components in the device.""" + + address: ClassVar[int] = 3 + + +class CoreVersionHigh(RegisterU8): + """Specifies the major version of the Harp core implemented by the device.""" + + address: ClassVar[int] = 4 + + +class CoreVersionLow(RegisterU8): + """Specifies the minor version of the Harp core implemented by the device.""" + + address: ClassVar[int] = 5 + + +class FirmwareVersionHigh(RegisterU8): + """Specifies the major version of the Harp core implemented by the device.""" + + address: ClassVar[int] = 6 + + +class FirmwareVersionLow(RegisterU8): + """Specifies the minor version of the Harp core implemented by the device.""" + + address: ClassVar[int] = 7 + + +class TimestampSeconds(RegisterU32): + """Stores the integral part of the system timestamp, in seconds.""" + + address: ClassVar[int] = 8 + + +class TimestampMicroseconds(RegisterU16): + """Stores the fractional part of the system timestamp, in microseconds.""" + + address: ClassVar[int] = 9 + + +class OperationControl(RegisterBase[OperationControlPayload]): + """Stores the configuration mode of the device.""" + + address: ClassVar[int] = 10 + payload_type: ClassVar[PayloadType] = PayloadType.U8 + payload_class = OperationControlPayload + + +class ResetDevice(RegisterBase[ResetFlags]): + """Resets the device and saves non-volatile registers.""" + + address: ClassVar[int] = 11 + payload_type: ClassVar[PayloadType] = PayloadType.U8 + payload_class = ResetDevicePayload + + +class DeviceName(RegisterBase[str]): + """Stores the user-specified device name.""" + + address: ClassVar[int] = 12 + payload_type: ClassVar[PayloadType] = PayloadType.U8 + payload_class = DeviceNamePayload + + +class SerialNumber(RegisterU16): + """Specifies the unique serial number of the device.""" + + address: ClassVar[int] = 13 + + +class ClockConfiguration(RegisterBase[ClockConfigurationFlags]): + """Specifies the configuration for the device synchronization clock.""" + + address: ClassVar[int] = 14 + payload_type: ClassVar[PayloadType] = PayloadType.U8 + payload_class = ClockConfigurationPayload + + +REGISTER_MAP: dict[int, type[RegisterBase[Any]]] = { + 0: WhoAmI, + 1: HardwareVersionHigh, + 2: HardwareVersionLow, + 3: AssemblyVersion, + 4: CoreVersionHigh, + 5: CoreVersionLow, + 6: FirmwareVersionHigh, + 7: FirmwareVersionLow, + 8: TimestampSeconds, + 9: TimestampMicroseconds, + 10: OperationControl, + 11: ResetDevice, + 12: DeviceName, + 13: SerialNumber, + 14: ClockConfiguration, +} diff --git a/tests/device/expected_device.py b/tests/device/expected_device.py new file mode 100644 index 0000000..d2fbf4f --- /dev/null +++ b/tests/device/expected_device.py @@ -0,0 +1,252 @@ +# This file was automatically generated and should not be edited directly. +# To make changes, edit the device metadata and regenerate the interface. + +import enum +from typing import Any, ClassVar + +import numpy as np +from numpy.typing import NDArray +from harp.protocol import ( + AnonymousPayload, + BitMask, + BoolConverter, + Field, + GroupMask, + HarpVersion, + HarpVersionConverter, + IdentityConverter, + PayloadType, + RegisterBase, + RegisterS32, + RegisterU16, + RegisterU8, + StringConverter, + StructPayload, +) +from harp.device import REGISTER_MAP as _CORE_REGISTER_MAP + +from .converters import ( + DataConverter, +) + + +class PortDigitalIOS(enum.IntFlag): + DIO0 = 0x1 + DIO1 = 0x2 + DIO2 = 0x4 + DIO3 = 0x8 + DI_PORT0 = 0x100 + TEST_DI_PORT1 = 0x200 + SUPPLY_PORT0 = 0x400 + PORT_DIO1 = 0x800 + + +class PwmPort(enum.IntEnum): + PWM0 = 1 + PWM1 = 2 + PWM2 = 4 + PWM3 = 10 + + +class EncoderModeMask(enum.IntEnum): + """Specifies the type of encoder mode.""" + + POSITION = 0 + DISPLACEMENT = 1 + + +class AnalogDataPayload(StructPayload[np.float32], length=6): + """Represents the payload of the AnalogData register.""" + + analog0: np.float32 = Field(IdentityConverter(np.float32)) + analog1: np.float32 = Field(IdentityConverter(np.float32), offset=1) + analog2: np.float32 = Field(IdentityConverter(np.float32), offset=2) + accelerometer: NDArray[np.float32] = Field( + IdentityConverter(np.dtype((np.float32, (3,)))), offset=3 + ) + + +class ComplexConfigurationPayload(StructPayload[np.uint8], length=17): + """Represents the payload of the ComplexConfiguration register.""" + + pwm_port: PwmPort = GroupMask(enum=PwmPort, mask=0xFF) + duty_cycle: np.float32 = Field(IdentityConverter(np.float32), offset=4) + frequency: np.float32 = Field(IdentityConverter(np.float32), offset=8) + events_enabled: bool = Field(BoolConverter(), offset=12) + delta: np.uint32 = Field(IdentityConverter(np.uint32), offset=13) + + +class VersionPayload(StructPayload[np.uint8], length=32): + """Represents the payload of the Version register.""" + + protocol_version: HarpVersion = Field(HarpVersionConverter(np.uint8)) + firmware_version: HarpVersion = Field(HarpVersionConverter(np.uint8), offset=3) + hardware_version: HarpVersion = Field(HarpVersionConverter(np.uint8), offset=6) + core_id: str = Field(StringConverter(3), offset=9) + interface_hash: NDArray[np.uint8] = Field( + IdentityConverter(np.dtype((np.uint8, (20,)))), offset=12 + ) + + +class CustomPayloadPayload(AnonymousPayload[np.uint32]): + """Represents the payload of the CustomPayload register.""" + + __value__: HarpVersion = Field(HarpVersionConverter(np.uint32)) + + +class CustomRawPayloadPayload(AnonymousPayload[np.uint32]): + """Represents the payload of the CustomRawPayload register.""" + + __value__: HarpVersion = Field(HarpVersionConverter(np.uint32)) + + +class CustomMemberConverterPayload(StructPayload[np.uint8], length=3): + """Represents the payload of the CustomMemberConverter register.""" + + header: np.uint8 = Field(IdentityConverter(np.uint8)) + data: np.int32 = Field(DataConverter(), offset=1) + + +class BitmaskSplitterPayload(StructPayload[np.uint8]): + """Represents the payload of the BitmaskSplitter register.""" + + low: np.int32 = Field(IdentityConverter(np.int32), mask=0xF) + high: np.int32 = Field(IdentityConverter(np.int32), mask=0xF0) + + +class PortDIOSetPayload(AnonymousPayload[np.uint8]): + """Represents the payload of the PortDIOSet register.""" + + __value__: PortDigitalIOS = BitMask(enum=PortDigitalIOS) + + +class StartPulsePayload(StructPayload[np.uint16]): + """Represents the payload of the StartPulse register.""" + + digital_output: PwmPort = GroupMask(enum=PwmPort, mask=0xC00) + pulse_width: np.uint16 = Field(IdentityConverter(np.uint16), mask=0x3FF) + + +class StartPulseTrainPayload(StructPayload[np.uint16], length=2): + """Represents the payload of the StartPulseTrain register.""" + + digital_output: PwmPort = GroupMask(enum=PwmPort, mask=0xC00) + pulse_width: np.uint16 = Field(IdentityConverter(np.uint16), mask=0x3FF) + frequency: np.uint8 = Field( + IdentityConverter(np.uint8), mask=0xFF00, offset=1, default=np.uint8(1) + ) + pulse_count: np.uint8 = Field(IdentityConverter(np.uint8), mask=0xFF, offset=1) + + +class EncoderModePayload(AnonymousPayload[np.uint8]): + """Represents the payload of the EncoderMode register.""" + + __value__: EncoderModeMask = GroupMask(enum=EncoderModeMask, mask=0xFF) + + +class DigitalInputs(RegisterU8): + address: ClassVar[int] = 32 + + +class AnalogData(RegisterBase[AnalogDataPayload]): + address: ClassVar[int] = 33 + payload_type: ClassVar[PayloadType] = PayloadType.Float + payload_class = AnalogDataPayload + + +class ComplexConfiguration(RegisterBase[ComplexConfigurationPayload]): + address: ClassVar[int] = 34 + payload_type: ClassVar[PayloadType] = PayloadType.U8 + payload_class = ComplexConfigurationPayload + + +class Version(RegisterBase[VersionPayload]): + address: ClassVar[int] = 35 + payload_type: ClassVar[PayloadType] = PayloadType.U8 + payload_class = VersionPayload + + +class CustomPayload(RegisterBase[HarpVersion]): + address: ClassVar[int] = 36 + payload_type: ClassVar[PayloadType] = PayloadType.U32 + payload_class = CustomPayloadPayload + + +class CustomRawPayload(RegisterBase[HarpVersion]): + address: ClassVar[int] = 37 + payload_type: ClassVar[PayloadType] = PayloadType.U32 + payload_class = CustomRawPayloadPayload + + +class CustomMemberConverter(RegisterBase[CustomMemberConverterPayload]): + address: ClassVar[int] = 38 + payload_type: ClassVar[PayloadType] = PayloadType.U8 + payload_class = CustomMemberConverterPayload + + +class BitmaskSplitter(RegisterBase[BitmaskSplitterPayload]): + address: ClassVar[int] = 39 + payload_type: ClassVar[PayloadType] = PayloadType.U8 + payload_class = BitmaskSplitterPayload + + +class Counter0(RegisterS32): + address: ClassVar[int] = 40 + + +class PortDIOSet(RegisterBase[PortDigitalIOS]): + address: ClassVar[int] = 41 + payload_type: ClassVar[PayloadType] = PayloadType.U8 + payload_class = PortDIOSetPayload + + +class PulseDOPort0(RegisterU16): + address: ClassVar[int] = 42 + + +class PulseDO0(RegisterU16): + address: ClassVar[int] = 43 + + +class StartPulse(RegisterBase[StartPulsePayload]): + """Starts a PWM pulse.""" + + address: ClassVar[int] = 100 + payload_type: ClassVar[PayloadType] = PayloadType.U16 + payload_class = StartPulsePayload + + +class StartPulseTrain(RegisterBase[StartPulseTrainPayload]): + """Starts a PWM pulse train.""" + + address: ClassVar[int] = 101 + payload_type: ClassVar[PayloadType] = PayloadType.U16 + payload_class = StartPulseTrainPayload + + +class EncoderMode(RegisterBase[EncoderModeMask]): + """Configures the operation mode of the encoder.""" + + address: ClassVar[int] = 103 + payload_type: ClassVar[PayloadType] = PayloadType.U8 + payload_class = EncoderModePayload + + +REGISTER_MAP: dict[int, type[RegisterBase[Any]]] = { + **_CORE_REGISTER_MAP, + 32: DigitalInputs, + 33: AnalogData, + 34: ComplexConfiguration, + 35: Version, + 36: CustomPayload, + 37: CustomRawPayload, + 38: CustomMemberConverter, + 39: BitmaskSplitter, + 40: Counter0, + 41: PortDIOSet, + 42: PulseDOPort0, + 43: PulseDO0, + 100: StartPulse, + 101: StartPulseTrain, + 103: EncoderMode, +} diff --git a/tests/device/test_device_emit.py b/tests/device/test_device_emit.py new file mode 100644 index 0000000..c3e996c --- /dev/null +++ b/tests/device/test_device_emit.py @@ -0,0 +1,66 @@ +import pytest +from harp.device import Device, create_device + +from .converters import DataConverter + +CONVERTERS = {"DataConverter": DataConverter()} + + +@pytest.fixture +def test_device(device_yml): + return create_device(device_yml, converters=CONVERTERS) + + +def test_returns_device_subclass(test_device): + assert issubclass(test_device, Device) + assert test_device.__name__ == "Tests" + + +def test_whoami_defaults_to_zero_when_absent(test_device): + # device.yml (application-device metadata) omits whoAmI. + assert test_device.__whoami__ == 0 + + +def test_whoami_from_schema(): + Dev = create_device( + "device: D\nwhoAmI: 1216\nregisters:\n Foo: {address: 40, type: U16, access: Read}\n" + ) + assert Dev.__whoami__ == 1216 + + +def test_registers_are_reachable_by_address(test_device): + reg_map = test_device.REGISTER_MAP + assert reg_map[33].__name__ == "AnalogData" + assert reg_map[103].__name__ == "EncoderMode" + + +def test_register_map_spreads_core(test_device): + reg_map = test_device.REGISTER_MAP + assert reg_map[0].__name__ == "WhoAmI" # core register, always spread in + assert reg_map[33].__name__ == "AnalogData" # device-specific + assert reg_map[103].__name__ == "EncoderMode" + + +def test_device_register_overrides_core_on_clash(): + # A device register at a core address wins over the spread-in common one. + Dev = create_device( + "device: Clash\nregisters:\n Shadow: {address: 0, type: U32, access: Read}\n" + ) + assert Dev.REGISTER_MAP[0].__name__ == "Shadow" + + +def test_headerless_fragment_builds_default_device(): + # A register-only fragment is a valid (nameless) device; name falls back to "Device". + Dev = create_device("registers:\n Foo: {address: 40, type: U16, access: Read}\n") + assert Dev.__name__ == "Device" + assert Dev.__whoami__ == 0 + assert Dev.REGISTER_MAP[40].__name__ == "Foo" + + +def test_emitted_device_registers_are_usable(test_device): + reg = test_device.REGISTER_MAP[33] # AnalogData + # The emitted register class round-trips through the Device.read/write frame path. + frame = reg.format( + reg.payload_class(Analog0=1.0, Analog1=2.0, Analog2=3.0, Accelerometer=[4, 5, 6]) + ) + assert isinstance(frame, (bytes, bytearray)) diff --git a/tests/device/test_emit.py b/tests/device/test_emit.py new file mode 100644 index 0000000..4c61667 --- /dev/null +++ b/tests/device/test_emit.py @@ -0,0 +1,222 @@ +import zlib + +import numpy as np +import pytest +from harp.data import parse_to_dataframe +from harp.protocol import HarpMessage + +from harp.device._schema import UnknownConverterError, create_registers + +from . import expected_core, expected_device +from .converters import DataConverter + +CONVERTERS = {"DataConverter": DataConverter()} + + +@pytest.fixture +def device_registers(device_yml): + return create_registers(device_yml, converters=CONVERTERS) + + +def _device_registers(): + # expected_device.REGISTER_MAP spreads the core map; the device-specific + # registers (the ones the emitter builds from device.yml) are address >= 32. + return {cls.__name__: cls for addr, cls in expected_device.REGISTER_MAP.items() if addr >= 32} + + +def _layout(dt): + """Name-agnostic structural signature: element dtype + offset per field, and itemsize. + + Ignores field names (we keep the yml's verbatim names; the generator + snake_cases them) while still verifying the byte layout matches exactly. + """ + if dt.names is None: + return ("scalar", dt.str, dt.shape, dt.itemsize) + return ("struct", dt.itemsize, tuple((dt.fields[n][0], dt.fields[n][1]) for n in dt.names)) + + +# --------------------------------------------------------------------------- +# Device golden — layout/type parity with generator output +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("name", sorted(_device_registers())) +def test_device_register_matches_generator_layout(name, device_registers): + emitted = device_registers[name] + expected = _device_registers()[name] + assert emitted.address == expected.address + assert emitted.payload_type == expected.payload_type + assert _layout(emitted.payload_class.dtype) == _layout(expected.payload_class.dtype) + + +def test_device_emits_all_registers(device_registers): + assert set(device_registers) == set(_device_registers()) + + +# --------------------------------------------------------------------------- +# Verbatim naming — the yml is the single source of truth +# --------------------------------------------------------------------------- + + +def test_field_names_are_verbatim(device_registers): + fields = device_registers["AnalogData"].payload_class.dtype.names + assert fields == ("Analog0", "Analog1", "Analog2", "Accelerometer") + + +def test_enum_members_are_verbatim(device_registers): + flags = device_registers["PortDIOSet"].payload_class._mro_descriptor("__value__")._enum + # yml bit names are kept as-is (the generator would UPPER_SNAKE these). + assert {"DIO0", "DIPort0", "TestDIPort1", "PortDIO1"} <= set(flags.__members__) + + +# --------------------------------------------------------------------------- +# Core golden — from protocol common.yml +# --------------------------------------------------------------------------- + + +def _core_expected(): + return {cls.__name__: cls for cls in expected_core.REGISTER_MAP.values()} + + +@pytest.mark.parametrize("name", sorted(_core_expected())) +def test_core_register_structural(name, common_yml): + emitted = create_registers(common_yml)[name] + expected = _core_expected()[name] + assert emitted.address == expected.address + assert emitted.payload_type == expected.payload_type + assert emitted.payload_class.dtype.itemsize == expected.payload_class.dtype.itemsize + if name == "DeviceName": + # Generator enriches DeviceName to interfaceType: string; protocol's + # common.yml does not, so only the layout size matches here. + return + assert _layout(emitted.payload_class.dtype) == _layout(expected.payload_class.dtype) + + +# --------------------------------------------------------------------------- +# Behavioural round-trips +# --------------------------------------------------------------------------- + + +def _roundtrip(reg, value): + return reg.parse(HarpMessage.parse(reg.format(value))) + + +def test_whole_register_groupmask_unwraps_to_enum(device_registers): + reg = device_registers["EncoderMode"] + enum_cls = reg.payload_class._mro_descriptor("__value__")._enum + parsed = _roundtrip(reg, enum_cls["Displacement"]) + assert parsed == enum_cls["Displacement"] + assert isinstance(parsed, enum_cls) + + +def test_whole_register_bitmask_roundtrip(device_registers): + reg = device_registers["PortDIOSet"] + flags = reg.payload_class._mro_descriptor("__value__")._enum + value = flags["DIO0"] | flags["DIO3"] + assert _roundtrip(reg, value) == value + + +def test_struct_masked_members_roundtrip(device_registers): + reg = device_registers["StartPulse"] + payload_cls = reg.payload_class + pwm = payload_cls._mro_descriptor("DigitalOutput")._enum + # DigitalOutput is a 2-bit field (mask 0xC00); only Pwm0/Pwm1 fit it. This + # matches the generator's output verbatim (GroupMask(enum=PwmPort, mask=0xC00)). + payload = payload_cls(DigitalOutput=pwm["Pwm1"], PulseWidth=np.uint16(300)) + parsed = _roundtrip(reg, payload) + assert parsed.DigitalOutput == pwm["Pwm1"] + assert int(parsed.PulseWidth) == 300 + + +def test_custom_converter_roundtrip(device_registers): + reg = device_registers["CustomMemberConverter"] + payload_cls = reg.payload_class + parsed = _roundtrip(reg, payload_cls(Header=np.uint8(7), Data=-1234)) + assert int(parsed.Header) == 7 + assert int(parsed.Data) == -1234 + + +# --------------------------------------------------------------------------- +# Converter registry +# --------------------------------------------------------------------------- + + +def test_unknown_converter_raises(device_yml): + with pytest.raises(UnknownConverterError): + create_registers(device_yml) # CustomMemberConverter needs DataConverter + + +def test_non_strict_falls_back_to_native(device_yml): + regs = create_registers(device_yml, strict=False) + # Data decodes as the raw native element (u8[2]) rather than the custom int. + reg = regs["CustomMemberConverter"] + assert reg.payload_class.dtype.itemsize == 3 + + +def test_converter_factory_receives_dsl_context(device_yml): + seen = {} + + def factory(ctx): + seen["name"], seen["span"], seen["interface_type"] = ctx.name, ctx.span, ctx.interface_type + return DataConverter() + + regs = create_registers(device_yml, converters={"DataConverter": factory}) + parsed = _roundtrip( + regs["CustomMemberConverter"], + regs["CustomMemberConverter"].payload_class(Header=np.uint8(1), Data=42), + ) + assert int(parsed.Data) == 42 + # the factory was handed the Data field's resolved DSL context + assert seen == {"name": "Data", "span": 2, "interface_type": "int"} + + +# --------------------------------------------------------------------------- +# Visibility +# --------------------------------------------------------------------------- + + +def test_exclude_private_drops_private_registers(): + yml = ( + "registers:\n" + " Pub: {address: 40, type: U16, access: Read}\n" + " Priv: {address: 41, type: U16, access: Read, visibility: private}\n" + ) + assert set(create_registers(yml)) == {"Pub", "Priv"} # kept by default + assert set(create_registers(yml, exclude_private=True)) == {"Pub"} + + +# --------------------------------------------------------------------------- +# Golden bulk round-trip — the emitted register and the generator oracle are +# wire- and dataframe-compatible for the same payload bytes (cross read/write). +# --------------------------------------------------------------------------- + + +def _random_records(dtype, n, seed): + """``n`` deterministic records of ``dtype`` with random ASCII-range bytes. + + Bytes are held to 0..127 so every field varies while staying valid for any + ``StringConverter`` member and free of float NaN/inf (which would defeat the + value comparison); padding bytes are filled too but never read back. + """ + rng = np.random.default_rng(seed) + raw = rng.integers(0, 128, size=n * dtype.itemsize, dtype=np.uint8) + return raw.view(dtype).copy() + + +@pytest.mark.parametrize("name", sorted(_device_registers())) +def test_emitted_register_bulk_matches_oracle(name, device_registers): + emitted = device_registers[name] + oracle = _device_registers()[name] + records = _random_records(emitted.payload_class.dtype, 5, seed=zlib.crc32(name.encode())) + + # Cross-write: same address / payload_type / byte layout -> identical wire bytes. + buf = bytes(emitted.format_bulk(records)) + assert buf == bytes(oracle.format_bulk(records)) + + # Cross-read via harp.data: the shared bytes decode to equal frames through + # either class. Enum labels and field names diverge (verbatim yml vs generator + # snake_case), so compare raw codes by column position, not by name. + df_emitted = parse_to_dataframe(emitted, buf, timestamp=False, decode_enums=False) + df_oracle = parse_to_dataframe(oracle, buf, timestamp=False, decode_enums=False) + df_oracle.columns = df_emitted.columns + assert df_emitted.equals(df_oracle) diff --git a/tests/device/test_schema.py b/tests/device/test_schema.py new file mode 100644 index 0000000..7d01adc --- /dev/null +++ b/tests/device/test_schema.py @@ -0,0 +1,41 @@ +from harp.device._schema import DeviceModel, PayloadType, parse_device_schema + + +def test_parse_full_device(device_yml): + m = parse_device_schema(device_yml) + assert isinstance(m, DeviceModel) + assert m.device == "Tests" + assert m.whoAmI is None # this application-device metadata omits whoAmI + assert "AnalogData" in m.registers + ad = m.registers["AnalogData"] + assert ad.type is PayloadType.Float + assert ad.length == 6 + assert list(ad.payloadSpec) == ["Analog0", "Analog1", "Analog2", "Accelerometer"] + + +def test_parse_fragment_yields_null_device(): + m = parse_device_schema("registers:\n Foo: {address: 40, type: U16, access: Read}\n") + assert isinstance(m, DeviceModel) + assert m.device is None # header-less fragment -> identity fields are None + assert m.registers["Foo"].type is PayloadType.U16 + + +def test_parse_common_registers(common_yml): + c = parse_device_schema(common_yml) + assert c.device is None + assert "WhoAmI" in c.registers + # Off/On group-mask keys must stay strings (YAML 1.1 would coerce to bool). + assert {k: int(v) for k, v in c.groupMasks["LedState"].values.items()} == {"Off": 0, "On": 1} + # 'None' bit name stays a string, not YAML null. + assert "None" in c.bitMasks["ResetFlags"].bits + + +def test_bool_values_preserved(common_yml): + c = parse_device_schema(common_yml) + assert c.registers["TimestampSeconds"].volatile is True + + +def test_access_list_and_scalar(common_yml): + c = parse_device_schema(common_yml) + # TimestampSeconds has a list access [Read, Write, Event]; WhoAmI a scalar. + assert isinstance(c.registers["TimestampSeconds"].access, list) diff --git a/tests/protocol/test_payload.py b/tests/protocol/test_payload.py index 4444ec1..a7c8105 100644 --- a/tests/protocol/test_payload.py +++ b/tests/protocol/test_payload.py @@ -1,7 +1,9 @@ +import enum + import numpy as np import pytest from harp.data import payload_to_dataframe -from harp.protocol import Column +from harp.protocol import AnonymousPayload, Column, GroupMask from harp.protocol._payload import PayloadBase, Field, _IdentityConverter @@ -69,3 +71,30 @@ def test_from_buffer_zero_copy(): def test_payload_property(): p = SimplePayload.from_buffer(_make_simple_bytes(2)) assert p.raw_payload.dtype == SimplePayload.dtype + + +class _SparseMode(enum.IntEnum): + Low = 0 + High = 2 # gap at code 1; largest member is 2 + + +class _SparseModePayload(AnonymousPayload[np.uint8]): + # Whole-byte GroupMask over a sparse enum: raw can be 0..255, well past the + # largest member, so decode must not IndexError on undefined codes. + __value__ = GroupMask(enum=_SparseMode, mask=0xFF) + + +def test_groupmask_undefined_code_preserves_raw(): + # Codes: defined (0->Low, 2->High), an in-range gap (1), and out-of-range (90, 255). + # Every undefined code is preserved as its raw int (like C#'s unchecked cast) — + batch = _SparseModePayload.from_buffer(np.array([0, 2, 1, 90, 255], dtype=np.uint8).tobytes()) + assert list(payload_to_dataframe(batch)["value"]) == ["Low", "High", 1, 90, 255] + + +def test_groupmask_scalar_matches_batch_for_undefined(): + # Scalar decode is permissive the same way + defined = _SparseModePayload.from_buffer(np.array([2], dtype=np.uint8).tobytes()) + assert defined.__value__ is _SparseMode.High + undefined = _SparseModePayload.from_buffer(np.array([90], dtype=np.uint8).tobytes()) + assert undefined.__value__ == 90 + assert not isinstance(undefined.__value__, _SparseMode) diff --git a/tests/protocol/test_register.py b/tests/protocol/test_register.py index be165e4..b63c777 100644 --- a/tests/protocol/test_register.py +++ b/tests/protocol/test_register.py @@ -4,7 +4,7 @@ import numpy as np import pytest -from harp.data import payload_to_dataframe +from harp.data import parse_to_dataframe, payload_to_dataframe, to_buffer, to_file from harp.protocol._message import HarpMessage from harp.protocol._message_type import MessageType from harp.protocol._payload import ( @@ -514,3 +514,44 @@ class P(PayloadBase): high = Field(converter=_IdentityConverter("u1"), mask=0xF0, offset=0) assert P._repr_fields == ("low", "high") + + +# --------------------------------------------------------------------------- +# format_bulk (inverse of parse_bulk) + harp.data.to_buffer / to_file +# --------------------------------------------------------------------------- + + +def test_format_bulk_single_matches_format(): + reg = RegisterU16(0x20) + one = reg.format(np.uint16(42), message_type=MessageType.Event, timestamp=1.0) + bulk = reg.format_bulk( + np.array([42], dtype="