diff --git a/README.md b/README.md index c7b3645..9368bfd 100644 --- a/README.md +++ b/README.md @@ -80,16 +80,17 @@ everything = reader.read_all() # {register_name: DataFrame} ``` Both paths are driven by a device schema. If you have only a `device.yml` and no -pre-generated package, `create_device` compiles it into a typed `Device` at runtime — -no code-generation step — which is exactly what `create_dataset_reader` does under -the hood: +pre-generated package, `create_device_module` compiles it into a module of register classes +at runtime — no code-generation step — which is exactly what `create_dataset_reader` +does under the hood: ```python from pathlib import Path -from harp.device import create_device +from harp.device import create_device_module -Behavior = create_device(Path("device.yml").read_text()) -AnalogData = Behavior.REGISTER_MAP[44] # registers are reached by address +behavior = create_device_module(Path("device.yml").read_text()) +AnalogData = behavior.AnalogData # registers are reached by name... +assert behavior.REGISTER_MAP[44] is AnalogData # ...or by address ``` See the [Examples](https://harp-tech.org/pyharp/examples/) for the full walkthroughs, diff --git a/docs/api/device.md b/docs/api/device.md index 652e4c7..f1ea7a1 100644 --- a/docs/api/device.md +++ b/docs/api/device.md @@ -3,7 +3,7 @@ --- ::: harp.device.Device -::: harp.device.create_device +::: harp.device.create_device_module ::: harp.device.parse_device_schema ::: harp.device.ConverterContext ::: harp.device.HarpFramer diff --git a/docs/examples/create_device/create_device.md b/docs/examples/create_device_module/create_device_module.md similarity index 54% rename from docs/examples/create_device/create_device.md rename to docs/examples/create_device_module/create_device_module.md index 15907d9..5eb61e7 100644 --- a/docs/examples/create_device/create_device.md +++ b/docs/examples/create_device_module/create_device_module.md @@ -1,24 +1,27 @@ -# Generating a Device from a Schema +# Generating Registers 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. +This example demonstrates how to turn a Harp `device.yml` into a module of register +classes at runtime with `create_device_module`, 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. +A generated device package is a module: register classes at module level, with a +`REGISTER_MAP` beside them keyed by address. `create_device_module` builds that same shape +from a schema, so registers are reached the same way either way — by name +(`behavior.AnalogData`) or by address (`behavior.REGISTER_MAP[44]`). From there they +work exactly like a pre-generated package's: drive them over a transport with +[`Device`](../../api/device.md) to talk to hardware, or use them to decode recorded +data. ## When to use runtime generation -`create_device` trades statically generated device packages for schema-driven +`create_device_module` 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, + becomes a working module 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. @@ -27,9 +30,10 @@ convenience. It's worth understanding what that buys you and what it costs. **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. +- **Static typing and autocomplete.** The names exist only once the module is built, + so an editor can't offer them and a type checker can't verify them. A generated + package is a real module on disk, so both work. The module also isn't in + `sys.modules`, so you bind it yourself rather than `import`-ing it. - **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 @@ -40,7 +44,7 @@ convenience. It's worth understanding what that buys you and what it costs. 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 +`create_device_module` when you want to go from a schema to working code with no generation step. !!! warning @@ -48,6 +52,6 @@ generation step. ```python -[](./create_device.py) +[](./create_device_module.py) ``` diff --git a/docs/examples/create_device/create_device.py b/docs/examples/create_device_module/create_device_module.py similarity index 50% rename from docs/examples/create_device/create_device.py rename to docs/examples/create_device_module/create_device_module.py index 11ec1b8..4bca277 100644 --- a/docs/examples/create_device/create_device.py +++ b/docs/examples/create_device_module/create_device_module.py @@ -1,23 +1,26 @@ from pathlib import Path from harp.data import parse_to_dataframe -from harp.device import create_device +from harp.device import Device, create_device_module 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 +# `create_device_module` compiles a Harp `device.yml` into a module of register classes 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: +# get the same shape a generated package has, registers at module level beside a +# `REGISTER_MAP`. +behavior = create_device_module(Path("device.yml").read_text()) + +print("WhoAmI:", behavior.WHO_AM_I) # device identity, taken from the schema +AnalogData = behavior.AnalogData # registers are reached by name... +assert behavior.REGISTER_MAP[44] is AnalogData # ...or by address + +# Registers are ordinary register classes, so they drive `read`/`write` on any +# `Device` over a transport. The schema carries no Python code, so there is no +# generated device class here: use `Device` itself. +with open_serial_device(Device, 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 @@ -25,14 +28,20 @@ df = parse_to_dataframe(AnalogData, "Behavior_44.bin") print(df.head()) +# To have the identity checked on connect, subclass `Device` with the schema's +# WhoAmI — the same one-liner a generated package ships: +# +# class Behavior(Device): +# __whoami__ = behavior.WHO_AM_I + # --- 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()}) +# behavior = create_device_module(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. +# than a module, `parse_device_schema(yml_text)` returns that directly. diff --git a/docs/examples/index.md b/docs/examples/index.md index ff83850..0feb522 100644 --- a/docs/examples/index.md +++ b/docs/examples/index.md @@ -4,7 +4,7 @@ This section contains some examples to help you get started with `harp`. 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`. +- [Generating Registers from a Schema](./create_device_module/create_device_module.md) - compile a `device.yml` into a module of register classes at runtime with `create_device_module`. Talking to a device: diff --git a/docs/examples/read_and_write_from_registers/read_and_write_from_registers.py b/docs/examples/read_and_write_from_registers/read_and_write_from_registers.py index 10371d3..8b735ae 100755 --- a/docs/examples/read_and_write_from_registers/read_and_write_from_registers.py +++ b/docs/examples/read_and_write_from_registers/read_and_write_from_registers.py @@ -1,4 +1,11 @@ -from harp.device import Device, OperationControl, OperationControlPayload, OperationMode, WhoAmI +from harp.device import ( + Device, + EnableFlag, + OperationControl, + OperationControlPayload, + OperationMode, + WhoAmI, +) from harp.serial import open_serial_device SERIAL_PORT = "/dev/ttyUSB0" # or "COMx" in Windows ("x" is the number of the serial port) @@ -11,7 +18,18 @@ control = device.read(OperationControl).parsed print("operation_mode before:", control.operation_mode) - # Write the register, then read it back to confirm the change. - device.write(OperationControl, OperationControlPayload(operation_mode=OperationMode.ACTIVE)) + # Write the register, then read it back to confirm the change. A struct payload + # is built whole, so every field is given a value. + device.write( + OperationControl, + OperationControlPayload( + operation_mode=OperationMode.ACTIVE, + dump_registers=False, + mute_replies=False, + visual_indicators=EnableFlag.ENABLED, + operation_led=EnableFlag.ENABLED, + heartbeat=EnableFlag.DISABLED, + ), + ) control = device.read(OperationControl).parsed print("operation_mode after:", control.operation_mode) diff --git a/docs/examples/read_dataset/read_dataset.py b/docs/examples/read_dataset/read_dataset.py index 1e9b834..b2186ae 100644 --- a/docs/examples/read_dataset/read_dataset.py +++ b/docs/examples/read_dataset/read_dataset.py @@ -12,8 +12,8 @@ # ┗ 📜 device.yml # # `create_dataset_reader` does the right thing: it finds `device.yml` inside the -# folder, builds the device that knows how to decode each register, and hands back -# a reader ready to go. +# folder, builds the module of register classes that knows how to decode each +# register, and hands back a reader ready to go. reader = create_dataset_reader("session.harp") # Read one register into a DataFrame — by register class (any register in the @@ -34,13 +34,13 @@ absolute = reader.read(44, epoch=REFERENCE_EPOCH) print(absolute.index[:3]) -# --- Already have a device class? ------------------------------------------- -# A pre-generated device package, or one you built yourself with `create_device`, -# can drive the reader directly — construct `DatasetReader(Device, folder)`: +# --- Already have a device module? ------------------------------------------- +# A pre-generated device package, or one you built yourself with `create_device_module`, +# can drive the reader directly — construct `DatasetReader(module, folder)`: # # from harp.data import DatasetReader -# from harp.device import create_device +# from harp.device import create_device_module # from pathlib import Path # -# Behavior = create_device((Path("session.harp") / "device.yml").read_text()) -# reader = DatasetReader(Behavior, "session.harp") +# behavior = create_device_module((Path("session.harp") / "device.yml").read_text()) +# reader = DatasetReader(behavior, "session.harp") diff --git a/docs/examples/subscribing_to_events/subscribing_to_events.py b/docs/examples/subscribing_to_events/subscribing_to_events.py index c649514..42ccca6 100644 --- a/docs/examples/subscribing_to_events/subscribing_to_events.py +++ b/docs/examples/subscribing_to_events/subscribing_to_events.py @@ -1,6 +1,8 @@ +import numpy as np from harp.device import ( REGISTER_MAP, Device, + EnableFlag, OperationControl, OperationControlPayload, OperationMode, @@ -12,7 +14,7 @@ SERIAL_PORT = "/dev/ttyUSB0" # or "COMx" in Windows ("x" is the number of the serial port) -def print_timestamp(msg: ParsedHarpMessage[float]) -> None: +def print_timestamp(msg: ParsedHarpMessage[np.uint32]) -> None: print(f"[timestamp] {msg.timestamp:.6f} {msg.parsed}") @@ -34,10 +36,10 @@ def print_any_event(msg: HarpMessage) -> None: OperationControlPayload( operation_mode=OperationMode.ACTIVE, dump_registers=True, - heartbeat=True, + heartbeat=EnableFlag.ENABLED, mute_replies=False, - operation_led=True, - visual_indicators=True, + operation_led=EnableFlag.ENABLED, + visual_indicators=EnableFlag.ENABLED, ), ) diff --git a/mkdocs.yml b/mkdocs.yml index c4a63b4..23d980e 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -73,7 +73,7 @@ nav: - Home: index.md - Examples: - examples/index.md - - Generating a Device from a Schema: examples/create_device/create_device.md + - Generating Registers from a Schema: examples/create_device_module/create_device_module.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 - Subscribing to Events: examples/subscribing_to_events/subscribing_to_events.md diff --git a/pyproject.toml b/pyproject.toml index c7d9169..1f3ce03 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -118,13 +118,14 @@ include = [ "src/packages/harp-device/src", "src/packages/harp-serial/src", "src/packages/harp-data/src", + "docs/examples", + "tests/conformance.py", ] exclude = [ "**/node_modules", "**/__pycache__", "**/.*", ".venv", - "tests", ] venvPath = "." venv = ".venv" diff --git a/src/packages/harp-data/README.md b/src/packages/harp-data/README.md index 80045b8..ebc9aba 100644 --- a/src/packages/harp-data/README.md +++ b/src/packages/harp-data/README.md @@ -37,16 +37,16 @@ df = reader.read(44) # by address everything = reader.read_all() # {register_name: DataFrame} ``` -Already have a device class (e.g. a pre-generated package, or one built with -`create_device`)? Drive `DatasetReader` with it directly: +Already have a device module (e.g. a pre-generated package, or one built with +`create_device_module`)? Drive `DatasetReader` with it directly: ```python from pathlib import Path from harp.data import DatasetReader -from harp.device import create_device +from harp.device import create_device_module -Behavior = create_device((Path("session.harp") / "device.yml").read_text()) -reader = DatasetReader(Behavior, "session.harp") +behavior = create_device_module((Path("session.harp") / "device.yml").read_text()) +reader = DatasetReader(behavior, "session.harp") ``` Timestamps are auto-detected per register and placed on the DataFrame index diff --git a/src/packages/harp-data/src/harp/data/_dataset.py b/src/packages/harp-data/src/harp/data/_dataset.py index bad29a6..6047769 100644 --- a/src/packages/harp-data/src/harp/data/_dataset.py +++ b/src/packages/harp-data/src/harp/data/_dataset.py @@ -6,7 +6,7 @@ from typing import Any import pandas as pd -from harp.device import Device, create_device +from harp.device import DeviceModuleLike, create_device_module from harp.protocol import RegisterBase from harp.protocol._constants import _TIMESTAMP_FLAG @@ -34,17 +34,18 @@ def default_file_resolver(root: Path, name: str) -> dict[int, list[Path]]: class DatasetReader: """Reader over a de-multiplexed Harp dataset folder. - Construct from a generated device and a dataset folder, then read a register's + Construct from a device module and a dataset folder, then read a register's frames into a DataFrame by register class or by address:: - reader = DatasetReader(Behavior, "session.harp") - df = reader.read(AnalogData) # by register class - df = reader.read(44) # by address - everything = reader.read_all() # {register_name: DataFrame} + reader = DatasetReader(behavior, "session.harp") + df = reader.read(behavior.AnalogData) # by register class + df = reader.read(44) # by address + everything = reader.read_all() # {register_name: DataFrame} - ``device`` is a generated :class:`~harp.device.Device` subclass; its - ``REGISTER_MAP`` and class name are read on demand. ``name`` overrides the - ```` file prefix, which defaults to the device class name. + ``device_module`` is a device module -- a generated device package, or one built from a + schema with :func:`~harp.device.create_device_module`. Its ``REGISTER_MAP`` and + ``__name__`` are read on demand. ``name`` overrides the ```` file + prefix, which defaults to the module name. File resolution defaults to the Harp file format: ``_
.bin`` and, when a register was logged as several ``_
_.bin`` chunks, @@ -54,13 +55,13 @@ class DatasetReader: def __init__( self, - device: type[Device], + device_module: DeviceModuleLike, root: str | PathLike[str], *, name: str | None = None, resolver: FileNameResolver = default_file_resolver, ) -> None: - self._device = device + self._device_module = device_module self._root = Path(root) self._name_override = name self._resolver = resolver @@ -72,19 +73,19 @@ def root(self) -> Path: return self._root @property - def device(self) -> type[Device]: - """The generated device this reader parses against.""" - return self._device + def device_module(self) -> DeviceModuleLike: + """The device module this reader parses against.""" + return self._device_module @property def name(self) -> str: """The ```` prefix used to match binary files.""" - return self._name_override or self._device.__name__ + return self._name_override or self._device_module.__name__ @property def registers(self) -> Mapping[int, type[RegisterBase[Any]]]: - """The device's address -> register-class map.""" - return self._device.REGISTER_MAP + """The address -> register-class map the module carries as ``REGISTER_MAP``.""" + return self._device_module.REGISTER_MAP @property def files(self) -> Mapping[int, list[Path]]: @@ -196,24 +197,26 @@ def create_dataset_reader( """Build a :class:`DatasetReader` for a dataset folder, device and all. Convenience wrapper that finds the device schema inside ``root`` (``device.yml`` - by default), generates a device from it with :func:`~harp.device.create_device`, - and returns a reader ready to :meth:`~DatasetReader.read`:: + by default), builds its module with :func:`~harp.device.create_device_module`, and + returns a reader ready to :meth:`~DatasetReader.read`:: reader = create_dataset_reader("session.harp") df = reader.read(44) ``schema`` points at the schema file explicitly when it isn't ``root/device.yml``. - ``converters`` and ``strict`` are forwarded to :func:`~harp.device.create_device` + ``converters`` and ``strict`` are forwarded to :func:`~harp.device.create_device_module` for custom ``interfaceType`` decoding; ``name`` and ``resolver`` are forwarded to - :class:`DatasetReader`. Use ``DatasetReader(device, root)`` directly when you - already have a (e.g. pre-generated) device class. + :class:`DatasetReader`. Use ``DatasetReader(device_module, root)`` directly when you + already have a (e.g. pre-generated) device module. """ root_path = Path(root) schema_path = Path(schema) if schema is not None else root_path / DEVICE_SCHEMA_FILENAME if not schema_path.is_file(): raise FileNotFoundError( f"No device schema at '{schema_path}'. Pass schema= to point at a device.yml, " - f"or build the device yourself and use DatasetReader(device, root)." + f"or build the device module yourself and use DatasetReader(device_module, root)." ) - device = create_device(schema_path.read_text(), converters=converters, strict=strict) - return DatasetReader(device, root_path, name=name, resolver=resolver) + device_module = create_device_module( + schema_path.read_text(), converters=converters, strict=strict + ) + return DatasetReader(device_module, root_path, name=name, resolver=resolver) diff --git a/src/packages/harp-device/README.md b/src/packages/harp-device/README.md index 3221dd8..fe70936 100644 --- a/src/packages/harp-device/README.md +++ b/src/packages/harp-device/README.md @@ -19,42 +19,75 @@ device.write(OperationControl, payload) # write a register ## Extending for a specific device -Downstream (often generated) packages add their registers and spread the core -`REGISTER_MAP`, and may set `__whoami__` for identity validation on connect: +A device is described by a module. Downstream, often generated, packages record the +device identity as `WHO_AM_I`, declare the register classes at module level, and expand +the core `REGISTER_MAP` beside them: ```python -from harp.device import Device, REGISTER_MAP as _CORE_REGISTER_MAP +from harp.device import REGISTER_MAP as _CORE_REGISTER_MAP +WHO_AM_I: int = 1216 +REGISTER_MAP = {**_CORE_REGISTER_MAP, 32: DigitalInputState, ...} +``` + +This is the same structure `create_device_module` builds from a schema, so a device +reads the same way whether it was generated ahead of time or compiled at runtime. A +`WHO_AM_I` of `0` marks an unregistered device, used while a device is in development +or outside the official registry, and identity checks are skipped for it. + +A device module names only the registers its schema declares, so `REGISTER_MAP` is the +device address space while the module namespace is what the device adds to it. The +common registers have a single definition, currently exported from `harp.device`, and +are not a device, so the core register set carries no `WHO_AM_I`. + +`Device` itself holds no register collection. `read`, `write` and `subscribe` take a +register class, so the module is the only place registers need to live: + +```python +from harp import behavior + +# `device` is a Device opened over some transport (see harp-serial) +device.read(behavior.DigitalInputState) +``` + +Identity is not yet read from the module. To validate `WhoAmI` on connect, subclass +`Device` with the same value, which is what `open` checks against today: + +```python class MyDevice(Device): __whoami__ = 1216 - -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` +## Generating registers 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. +Without a pre-generated device package, `create_device_module` builds the same +structure at runtime from Harp `device.yml` text: register classes at module level, a +`REGISTER_MAP` beside them, and the identity declared by the schema as `WHO_AM_I`. +Field and enum names come from the yml verbatim. ```python from pathlib import Path -from harp.device import create_device +from harp.device import create_device_module -Behavior = create_device(Path("device.yml").read_text()) -reg = Behavior.REGISTER_MAP[44] +behavior = create_device_module(Path("device.yml").read_text()) +reg = behavior.AnalogData # by name +reg = behavior.REGISTER_MAP[44] # or by address ``` +The module is not registered in `sys.modules`, so it has to be bound rather than +imported. Names come from the schema at runtime, so they don't autocomplete and +aren't statically checked. A generated package on disk gives both. + 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()}) +create_device_module(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`. +`parse_device_schema(yml_text)` is also public, returning the parsed schema model +without a module: registers, masks, and optional device identity. diff --git a/src/packages/harp-device/src/harp/device/__init__.py b/src/packages/harp-device/src/harp/device/__init__.py index d352a19..3bed88a 100644 --- a/src/packages/harp-device/src/harp/device/__init__.py +++ b/src/packages/harp-device/src/harp/device/__init__.py @@ -1,5 +1,5 @@ from ._device import Device, EventHandler, Subscription -from ._emit_device import create_device +from ._emit_module import DeviceModule, DeviceModuleLike, create_device_module from ._framer import HarpFramer from ._registers import ( AssemblyVersion, @@ -34,7 +34,9 @@ "Device", "EventHandler", "Subscription", - "create_device", + "create_device_module", + "DeviceModule", + "DeviceModuleLike", "parse_device_schema", "ConverterContext", "HarpFramer", diff --git a/src/packages/harp-device/src/harp/device/_device.py b/src/packages/harp-device/src/harp/device/_device.py index d765da3..71098e9 100644 --- a/src/packages/harp-device/src/harp/device/_device.py +++ b/src/packages/harp-device/src/harp/device/_device.py @@ -71,9 +71,17 @@ class Device: """Harp device protocol logic (framing, request/reply, register access) over an :class:`~harp.device.ITransport`. - Must be opened before use, via ``with`` or :meth:`open`. Subclasses add - register class attributes and set :attr:`__whoami__` to validate device - identity on open (``0x0`` skips the check). + Must be opened before use, via ``with`` or :meth:`open`. :meth:`read`, + :meth:`write` and :meth:`subscribe` take a register class, so the device holds + no register collection of its own: a device's registers live in its module, + beside a ``REGISTER_MAP`` (see :func:`~harp.device.create_device_module`, or the + ``harp-device`` README for the statically generated equivalent). + + A subclass sets :attr:`__whoami__` to validate device identity on open + (``0x0`` skips the check):: + + class Behavior(Device): + __whoami__ = 1216 """ REPLY_TIMEOUT: ClassVar[float] = 5.0 # seconds @@ -81,9 +89,6 @@ 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 deleted file mode 100644 index 82f43b4..0000000 --- a/src/packages/harp-device/src/harp/device/_emit_device.py +++ /dev/null @@ -1,37 +0,0 @@ -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/_emit_module.py b/src/packages/harp-device/src/harp/device/_emit_module.py new file mode 100644 index 0000000..19e23db --- /dev/null +++ b/src/packages/harp-device/src/harp/device/_emit_module.py @@ -0,0 +1,111 @@ +"""Emit a Python module of register classes from a device schema. + +A generated device package is already a module: register classes at module level +and a ``REGISTER_MAP`` beside them (see the ``harp-device`` README). +:func:`create_device_module` builds that same shape at runtime from a ``device.yml``, so a +schema-driven device and a generated one are reached the same way, by name from the +module or by address through ``REGISTER_MAP``. +""" + +import types +from typing import Any, Mapping, Optional, Protocol, runtime_checkable + +from harp.protocol import RegisterBase + +from ._register_map import REGISTER_MAP as CORE_REGISTER_MAP +from ._schema import create_registers, parse_device_schema +from ._schema._emit import ConverterValue + +#: Module name used when the schema carries no ``device`` header. +_DEFAULT_NAME = "Device" + + +@runtime_checkable +class DeviceModuleLike(Protocol): + """Any module describing a device, however it was produced. + + A generated device package is a plain module, so it cannot be named by a class; + what identifies it is describing a device. Matching structurally accepts both it + and :class:`DeviceModule`, and rejects the common register set, which carries + registers but is not a device. + """ + + __name__: str + REGISTER_MAP: dict[int, type[RegisterBase[Any]]] + WHO_AM_I: int + + +class DeviceModule(types.ModuleType): + """The module :func:`create_device_module` returns, describing what a device module holds. + + Declaring the members is what lets a linter resolve them. Register names come + from the schema, so they can only be described collectively, through + :meth:`__getattr__`; ``REGISTER_MAP`` and ``WHO_AM_I`` are named and keep their + own types. A statically generated device package is a plain module and needs + none of this, since its registers are written out. + """ + + #: Address -> register class, the common Harp registers merged with the schema's. + REGISTER_MAP: dict[int, type[RegisterBase[Any]]] + #: The device identity declared by the schema; ``0`` when absent. + WHO_AM_I: int + + def __getattr__(self, name: str) -> type[RegisterBase[Any]]: + raise AttributeError(f"module {self.__name__!r} has no register named {name!r}") + + +def create_device_module( + text: str, + *, + name: Optional[str] = None, + converters: Optional[Mapping[str, ConverterValue]] = None, + strict: bool = True, + exclude_private: bool = True, +) -> DeviceModule: + """Emit a module of register classes from ``device.yml`` text. + + The module names the registers the schema declares, so ``behavior.AnalogData`` + resolves while a common register such as ``WhoAmI`` is imported from + :mod:`harp.device`, keeping one definition of each. Beside them it holds: + + * ``REGISTER_MAP``, the device address space, so the common registers are + present here even though the module does not name them; + * ``WHO_AM_I``, the schema's identity (``0`` for an unregistered device); + * ``__name__``, the schema's ``device`` name, or ``name`` when given + (``"Device"`` for a header-less register fragment). + + Because the names come from the schema at runtime they don't autocomplete, and + each resolves as ``type[RegisterBase[Any]]`` rather than its own register type; + a generated device package is a real module on disk and gives both. On an + address clash the device's register replaces the common one in ``REGISTER_MAP``. + ``exclude_private=True`` drops registers whose DSL ``visibility`` is ``private``. + + ``text`` is the schema itself rather than a path to it, matching + :func:`parse_device_schema`, so read the file first. The module is **not** + registered in :data:`sys.modules`, so it cannot be reached by ``import`` and two + schemas may share a name without clashing. Bind it yourself:: + + behavior = create_device_module(Path("device.yml").read_text()) + behavior.AnalogData + """ + device = parse_device_schema(text) + registers = create_registers( + device, converters=converters, strict=strict, exclude_private=exclude_private + ) + module_name = name or device.device or _DEFAULT_NAME + + contents: dict[str, type[RegisterBase[Any]]] = dict(registers) + register_map = {cls.address: cls for cls in CORE_REGISTER_MAP.values()} + register_map.update({cls.address: cls for cls in registers.values()}) + + for register in registers.values(): + register.__module__ = module_name + + module = DeviceModule(module_name, f"Harp registers for {module_name}, from a schema.") + vars(module).update( + contents, + REGISTER_MAP=register_map, + WHO_AM_I=int(device.whoAmI or 0), + __all__=[*sorted(contents), "REGISTER_MAP", "WHO_AM_I"], + ) + return module diff --git a/src/packages/harp-protocol/README.md b/src/packages/harp-protocol/README.md index 1afee4d..8f7b044 100644 --- a/src/packages/harp-protocol/README.md +++ b/src/packages/harp-protocol/README.md @@ -19,4 +19,15 @@ frame = WhoAmI.format(np.uint16(1216)) # build a Write frame value = WhoAmI.parse(HarpMessage.parse(frame)) # -> np.uint16(1216) ``` +## Register value types + +`parse` returns numpy scalars rather than plain `int` or `float`, so a value carries the width its register declares. A Python `int` has no width and no upper bound, so it cannot distinguish a `U8` from a `U32`, nor detect a value leaving the register range. + +```python +np.uint16(65535) + 1 # RuntimeWarning: overflow encountered in scalar add +65535 + 1 # 65536, wider than the register can hold +``` + +Numpy scalars behave like plain Python numbers in arithmetic, comparison and formatting. Use `int()` or `float()` where a built-in type is required. + It carries no transport or device logic — see [`harp-device`](../harp-device) for the device layer. diff --git a/tests/conformance.py b/tests/conformance.py new file mode 100644 index 0000000..89316c7 --- /dev/null +++ b/tests/conformance.py @@ -0,0 +1,52 @@ +"""Static conformance checks for the documented public API. + +Nothing here runs. Every function is a type-checker fixture, asserting the type a +documented expression resolves to, so a change that silently degrades an inferred +type fails the build rather than being noticed downstream. +""" + +from typing import Any, assert_type + +import numpy as np +from harp.data import DatasetReader +from harp.device import ( + Device, + DeviceModule, + DeviceModuleLike, + OperationControl, + OperationControlPayload, + WhoAmI, + create_device_module, +) +from harp.protocol import ParsedHarpMessage, RegisterBase + + +def schema_built_registers(yml: str) -> None: + """A module built from a schema types its registers collectively.""" + behavior = create_device_module(yml) + assert_type(behavior, DeviceModule) + assert_type(behavior.AnalogData, type[RegisterBase[Any]]) + assert_type(behavior.REGISTER_MAP, dict[int, type[RegisterBase[Any]]]) + assert_type(behavior.WHO_AM_I, int) + + +def statically_declared_registers(device: Device) -> None: + """A register written out in a module carries its payload type through read.""" + assert_type(device.read(WhoAmI), ParsedHarpMessage[np.uint16]) + assert_type(device.read(WhoAmI).parsed, np.uint16) + assert_type(device.read(OperationControl).parsed, OperationControlPayload) + + +def register_writes(device: Device, payload: OperationControlPayload) -> None: + """Write accepts the payload type its register parses to.""" + assert_type(device.write(OperationControl, payload).parsed, OperationControlPayload) + + +def dataset_reader_accepts_either_module( + schema_built: DeviceModule, generated: DeviceModuleLike +) -> None: + """The reader takes a schema-built module and a generated package alike.""" + DatasetReader(schema_built, "session.harp") + reader = DatasetReader(generated, "session.harp") + reader.read(WhoAmI) + reader.read(44) diff --git a/tests/data/test_dataset.py b/tests/data/test_dataset.py index 581f415..ba30ee1 100644 --- a/tests/data/test_dataset.py +++ b/tests/data/test_dataset.py @@ -9,7 +9,7 @@ create_dataset_reader, parse_to_dataframe, ) -from harp.device import create_device +from harp.device import TimestampSeconds, WhoAmI, create_device_module def _records(cls, n, seed): @@ -20,42 +20,68 @@ def _records(cls, n, seed): @pytest.fixture -def emitted_device(device_yml): +def emitted_module(device_yml): # strict=False: the test device.yml uses a custom DataConverter we don't inject # here; native decoding is enough to exercise file resolution and parsing. - return create_device(device_yml, strict=False) + return create_device_module(device_yml, strict=False) @pytest.fixture -def dataset(emitted_device, tmp_path): +def dataset(emitted_module, tmp_path): """A dataset folder with three app registers; the first is timestamped.""" - dev = emitted_device - name = dev.__name__ - addresses = [a for a in sorted(dev.REGISTER_MAP) if a >= 32][:3] + mod = emitted_module + name = mod.__name__ + addresses = [a for a in sorted(mod.REGISTER_MAP) if a >= 32][:3] specs = {} for i, address in enumerate(addresses): - cls = dev.REGISTER_MAP[address] + cls = mod.REGISTER_MAP[address] records = _records(cls, 5, seed=address) timestamped = i == 0 timestamps = np.arange(5, dtype=np.float64) if timestamped else None buf = bytes(cls.format_bulk(records, timestamps=timestamps)) (tmp_path / f"{name}_{address}.bin").write_bytes(buf) specs[address] = (cls, timestamped, buf) - return dev, name, tmp_path, specs + return mod, name, tmp_path, specs def test_read_by_class_and_by_address(dataset): - dev, _name, root, specs = dataset - reader = DatasetReader(dev, root) + mod, _name, root, specs = dataset + reader = DatasetReader(mod, root) for address, (cls, timestamped, buf) in specs.items(): expected = parse_to_dataframe(cls, buf, timestamp=timestamped) assert reader.read(cls).equals(expected) assert reader.read(address).equals(expected) +def test_read_by_name_from_module(dataset): + mod, _name, root, specs = dataset + reader = DatasetReader(mod, root) + for address, (cls, _timestamped, _buf) in specs.items(): + # The register reached by name off the module is the one at that address. + assert reader.read(getattr(mod, cls.__name__)).equals(reader.read(address)) + + +def test_reads_common_registers_not_named_by_module(emitted_module, tmp_path): + """A device module names only its own registers, but a session folder also holds + files for the common ones, so the reader must still decode those.""" + mod = emitted_module + assert not hasattr(mod, "WhoAmI") # imported from harp.device, not re-exported + + for cls in (WhoAmI, TimestampSeconds): + records = _records(cls, 4, seed=cls.address) + buf = bytes(cls.format_bulk(records)) + (tmp_path / f"{mod.__name__}_{cls.address}.bin").write_bytes(buf) + + reader = DatasetReader(mod, tmp_path) + # By address, and by the class imported from harp.device, and in read_all. + assert len(reader.read(WhoAmI.address)) == 4 + assert reader.read(WhoAmI).equals(reader.read(WhoAmI.address)) + assert set(reader.read_all()) == {"WhoAmI", "TimestampSeconds"} + + def test_timestamp_is_auto_detected(dataset): - dev, _name, root, specs = dataset - reader = DatasetReader(dev, root) + mod, _name, root, specs = dataset + reader = DatasetReader(mod, root) for address, (_cls, timestamped, _buf) in specs.items(): df = reader.read(address) # Timestamped frames get a "Time" index; untimestamped keep a plain RangeIndex. @@ -63,8 +89,8 @@ def test_timestamp_is_auto_detected(dataset): def test_time_index_is_float_seconds_without_epoch(dataset): - dev, _name, root, specs = dataset - reader = DatasetReader(dev, root) + mod, _name, root, specs = dataset + reader = DatasetReader(mod, root) address = next(a for a, (_c, ts, _b) in specs.items() if ts) # the timestamped register df = reader.read(address) assert df.index.name == "Time" @@ -72,8 +98,8 @@ def test_time_index_is_float_seconds_without_epoch(dataset): def test_epoch_gives_absolute_datetime_index(dataset): - dev, _name, root, specs = dataset - reader = DatasetReader(dev, root) + mod, _name, root, specs = dataset + reader = DatasetReader(mod, root) address = next(a for a, (_c, ts, _b) in specs.items() if ts) df = reader.read(address, epoch=REFERENCE_EPOCH) assert isinstance(df.index, pd.DatetimeIndex) @@ -84,25 +110,25 @@ def test_epoch_gives_absolute_datetime_index(dataset): def test_read_all_keyed_by_register_name(dataset): - dev, _name, root, specs = dataset - reader = DatasetReader(dev, root) + mod, _name, root, specs = dataset + reader = DatasetReader(mod, root) frames = reader.read_all() assert set(frames) == {cls.__name__ for cls, _ts, _buf in specs.values()} for cls, _timestamped, _buf in specs.values(): assert frames[cls.__name__].equals(reader.read(cls.address)) -def test_suffix_chunks_are_concatenated(emitted_device, tmp_path): - dev = emitted_device - name = dev.__name__ - address = next(a for a in sorted(dev.REGISTER_MAP) if a >= 32) - cls = dev.REGISTER_MAP[address] +def test_suffix_chunks_are_concatenated(emitted_module, tmp_path): + mod = emitted_module + name = mod.__name__ + address = next(a for a in sorted(mod.REGISTER_MAP) if a >= 32) + cls = mod.REGISTER_MAP[address] chunk0 = bytes(cls.format_bulk(_records(cls, 3, seed=1))) chunk1 = bytes(cls.format_bulk(_records(cls, 2, seed=2))) (tmp_path / f"{name}_{address}_0.bin").write_bytes(chunk0) (tmp_path / f"{name}_{address}_1.bin").write_bytes(chunk1) - reader = DatasetReader(dev, tmp_path) + reader = DatasetReader(mod, tmp_path) combined = parse_to_dataframe(cls, chunk0 + chunk1, timestamp=False) assert reader.read(cls).reset_index(drop=True).equals(combined) # A specific chunk can still be selected by suffix. @@ -110,42 +136,42 @@ def test_suffix_chunks_are_concatenated(emitted_device, tmp_path): assert reader.read(cls, suffix="0").equals(only0) -def test_non_device_raises_on_register_access(dataset): - _dev, _name, root, _specs = dataset - # Registers are derived lazily; a non-Device fails when they are accessed. +def test_non_module_raises_on_register_access(dataset): + _mod, _name, root, _specs = dataset + # Registers are derived lazily; anything without a REGISTER_MAP fails on access. reader = DatasetReader(object, root) with pytest.raises(AttributeError, match="REGISTER_MAP"): _ = reader.registers def test_explicit_name_overrides(dataset): - dev, name, root, _specs = dataset - reader = DatasetReader(dev, root, name=name) + mod, name, root, _specs = dataset + reader = DatasetReader(mod, root, name=name) assert isinstance(reader, DatasetReader) assert reader.name == name def test_missing_register_file_raises(dataset): - dev, _name, root, _specs = dataset - reader = DatasetReader(dev, root) + mod, _name, root, _specs = dataset + reader = DatasetReader(mod, root) # WhoAmI (address 0) is in the map but has no file in this dataset. with pytest.raises(FileNotFoundError): reader.read(0) def test_unknown_address_raises(dataset): - dev, _name, root, _specs = dataset - reader = DatasetReader(dev, root) + mod, _name, root, _specs = dataset + reader = DatasetReader(mod, root) with pytest.raises(KeyError): reader.read(9999) -def test_custom_file_resolver_supports_alternative_layout(emitted_device, tmp_path): - dev = emitted_device - addresses = [a for a in sorted(dev.REGISTER_MAP) if a >= 32][:2] +def test_custom_file_resolver_supports_alternative_layout(emitted_module, tmp_path): + mod = emitted_module + addresses = [a for a in sorted(mod.REGISTER_MAP) if a >= 32][:2] expected = {} for address in addresses: - cls = dev.REGISTER_MAP[address] + cls = mod.REGISTER_MAP[address] buf = bytes(cls.format_bulk(_records(cls, 3, seed=address))) (tmp_path / f"reg{address}.bin").write_bytes(buf) # not the Harp layout expected[cls.__name__] = parse_to_dataframe(cls, buf, timestamp=False) @@ -158,7 +184,7 @@ def resolver(root, _name): found.setdefault(int(match.group(1)), []).append(path) return found - reader = DatasetReader(dev, tmp_path, resolver=resolver) + reader = DatasetReader(mod, tmp_path, resolver=resolver) assert set(reader.files) == set(addresses) frames = reader.read_all() assert set(frames) == set(expected) @@ -167,17 +193,17 @@ def resolver(root, _name): def test_files_property_lists_discovered_bins(dataset): - dev, _name, root, specs = dataset - reader = DatasetReader(dev, root) + mod, _name, root, specs = dataset + reader = DatasetReader(mod, root) assert set(reader.files) == set(specs) -def test_read_all_registers_of_mock_device(emitted_device, tmp_path): +def test_read_all_registers_of_mock_device(emitted_module, tmp_path): """Write one .bin per register of the device.yml device, then read them all back.""" - dev = emitted_device - name = dev.__name__ + mod = emitted_module + name = mod.__name__ expected = {} - for address, cls in dev.REGISTER_MAP.items(): + for address, cls in mod.REGISTER_MAP.items(): records = _records(cls, 4, seed=address) # Alternate timestamped/untimestamped to exercise both parse paths. timestamped = address % 2 == 0 @@ -186,39 +212,39 @@ def test_read_all_registers_of_mock_device(emitted_device, tmp_path): (tmp_path / f"{name}_{address}.bin").write_bytes(buf) expected[cls.__name__] = parse_to_dataframe(cls, buf, timestamp=timestamped) - reader = DatasetReader(dev, tmp_path) + reader = DatasetReader(mod, tmp_path) frames = reader.read_all() - assert set(reader.files) == set(dev.REGISTER_MAP) + assert set(reader.files) == set(mod.REGISTER_MAP) assert set(frames) == set(expected) - assert len(frames) == len(dev.REGISTER_MAP) + assert len(frames) == len(mod.REGISTER_MAP) for register_name, df in frames.items(): assert len(df) == 4 assert df.equals(expected[register_name]) -def test_reader_derives_name_and_registers_from_device(dataset): - dev, name, root, _specs = dataset - reader = DatasetReader(dev, root) - assert reader.device is dev +def test_reader_derives_name_and_registers_from_module(dataset): + mod, name, root, _specs = dataset + reader = DatasetReader(mod, root) + assert reader.device_module is mod assert reader.name == name - assert reader.registers == dev.REGISTER_MAP + assert reader.registers == mod.REGISTER_MAP -def test_create_dataset_reader_builds_device_from_device_yml(dataset, device_yml): - dev, _name, root, specs = dataset +def test_create_dataset_reader_builds_module_from_device_yml(dataset, device_yml): + mod, _name, root, specs = dataset (root / "device.yml").write_text(device_yml) - # strict=False mirrors the emitted_device fixture (custom DataConverter not injected). + # strict=False mirrors the emitted_module fixture (custom DataConverter not injected). reader = create_dataset_reader(root, strict=False) assert isinstance(reader, DatasetReader) - # Reads match a reader built from an explicitly-generated device. - reference = DatasetReader(dev, root) + # Reads match a reader built from an explicitly-generated module. + reference = DatasetReader(mod, root) for address, (cls, _timestamped, _buf) in specs.items(): assert reader.read(address).equals(reference.read(cls)) def test_create_dataset_reader_accepts_explicit_schema_path(dataset, device_yml, tmp_path): - _dev, _name, root, specs = dataset + _mod, _name, root, specs = dataset schema_path = tmp_path / "elsewhere.yml" # not inside the dataset folder schema_path.write_text(device_yml) reader = create_dataset_reader(root, schema=schema_path, strict=False) @@ -227,6 +253,6 @@ def test_create_dataset_reader_accepts_explicit_schema_path(dataset, device_yml, def test_create_dataset_reader_missing_schema_raises(dataset): - _dev, _name, root, _specs = dataset # no device.yml written into the folder + _mod, _name, root, _specs = dataset # no device.yml written into the folder with pytest.raises(FileNotFoundError, match="device.yml"): create_dataset_reader(root) diff --git a/tests/device/expected_device.py b/tests/device/expected_device.py index d2fbf4f..304fc07 100644 --- a/tests/device/expected_device.py +++ b/tests/device/expected_device.py @@ -30,6 +30,9 @@ ) +WHO_AM_I: int = 0 + + class PortDigitalIOS(enum.IntFlag): DIO0 = 0x1 DIO1 = 0x2 diff --git a/tests/device/test_create_device_module.py b/tests/device/test_create_device_module.py new file mode 100644 index 0000000..a56fb43 --- /dev/null +++ b/tests/device/test_create_device_module.py @@ -0,0 +1,156 @@ +import sys +import types + +import harp.device +import pytest +from harp.device import REGISTER_MAP as CORE_REGISTER_MAP +from harp.device import DeviceModule, DeviceModuleLike, WhoAmI, create_device_module + +from . import expected_device +from .converters import DataConverter + +CONVERTERS = {"DataConverter": DataConverter()} +MODULE_CONSTANTS = {"REGISTER_MAP", "WHO_AM_I"} + + +@pytest.fixture +def test_module(device_yml): + return create_device_module(device_yml, converters=CONVERTERS) + + +def test_returns_module_named_after_schema(test_module): + assert isinstance(test_module, types.ModuleType) + assert test_module.__name__ == "Tests" + + +def test_returns_device_module(test_module): + # The subclass is what declares REGISTER_MAP, WHO_AM_I and the register names, + # so a linter can resolve them on a module built at runtime. + assert isinstance(test_module, DeviceModule) + + +def test_whoami_defaults_to_zero_when_absent(test_module): + # device.yml (application-device metadata) omits whoAmI. + assert test_module.WHO_AM_I == 0 + + +def test_whoami_from_schema(): + mod = create_device_module( + "device: D\nwhoAmI: 1216\nregisters:\n Foo: {address: 40, type: U16, access: Read}\n" + ) + assert mod.WHO_AM_I == 1216 + + +def test_registers_are_reachable_by_name(test_module): + assert test_module.AnalogData.address == 33 + assert test_module.EncoderMode.address == 103 + + +def test_registers_are_reachable_by_address(test_module): + reg_map = test_module.REGISTER_MAP + assert reg_map[33].__name__ == "AnalogData" + assert reg_map[103].__name__ == "EncoderMode" + + +def test_register_map_spreads_core(test_module): + reg_map = test_module.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_core_registers_are_not_named_by_module(test_module): + # A common register has one definition, in harp.device, so a device module does + # not re-export it. It is still in the address space the device can send from. + assert not hasattr(test_module, "WhoAmI") + assert test_module.REGISTER_MAP[0] is WhoAmI + + +def test_module_names_exactly_schema_registers(test_module): + named = {n for n in vars(test_module) if not n.startswith("_") and n not in MODULE_CONSTANTS} + addresses = {cls.address for cls in test_module.REGISTER_MAP.values()} + # Everything the module names is in the address space, and the map carries the + # common registers on top, which is the whole difference between the two. + assert all(getattr(test_module, n).address in addresses for n in named) + assert {c.__name__ for c in CORE_REGISTER_MAP.values()}.isdisjoint(named) + assert len(test_module.REGISTER_MAP) > len(named) + + +def test_emitted_module_matches_device_protocol(test_module): + assert isinstance(test_module, DeviceModuleLike) + + +def test_generated_package_matches_device_protocol(): + # expected_device is a sample of generator output, so this pins that what the + # generator emits is accepted wherever a device module is required. + assert isinstance(expected_device, DeviceModuleLike) + + +def test_common_registers_are_not_device_module(): + # They carry REGISTER_MAP but describe no device, so they cannot be passed + # where a device module is required, such as to a DatasetReader. + assert hasattr(harp.device, "REGISTER_MAP") + assert not hasattr(harp.device, "WHO_AM_I") + assert not isinstance(harp.device, DeviceModuleLike) + + +def test_unknown_name_raises_attribute_error(test_module): + # The message names the module and the register, since a schema-built module + # cannot offer the name in an editor. + with pytest.raises(AttributeError, match="'Tests' has no register named 'Nonexistent'"): + _ = test_module.Nonexistent + + +def test_named_registers_are_subset_of_address_space(test_module): + # The two are deliberately different sets: the module names what the schema + # declares, the map is everything the device can send. + for cls in test_module.REGISTER_MAP.values(): + if cls.address >= 32: + assert getattr(test_module, cls.__name__) is cls + assert 0 in test_module.REGISTER_MAP + + +def test_device_register_overrides_core_on_clash(): + # A device register at a common address replaces it in the address space. + mod = create_device_module( + "device: Clash\nregisters:\n Shadow: {address: 0, type: U32, access: Read}\n" + ) + assert mod.REGISTER_MAP[0].__name__ == "Shadow" + assert mod.Shadow.address == 0 + + +def test_headerless_fragment_builds_default_module(): + # A register-only fragment is a valid (nameless) device; name falls back to "Device". + mod = create_device_module("registers:\n Foo: {address: 40, type: U16, access: Read}\n") + assert mod.__name__ == "Device" + assert mod.WHO_AM_I == 0 + assert mod.REGISTER_MAP[40].__name__ == "Foo" + assert mod.Foo.address == 40 + + +def test_all_covers_registers_and_module_constants(test_module): + exported = set(test_module.__all__) + assert {"REGISTER_MAP", "WHO_AM_I"} <= exported + assert {"AnalogData", "EncoderMode"} <= exported + assert "WhoAmI" not in exported # a common register is not re-exported + assert exported - MODULE_CONSTANTS == { + cls.__name__ for cls in test_module.REGISTER_MAP.values() if cls.address >= 32 + } + + +def test_module_is_not_registered_in_sys_modules(test_module): + # Two schemas may share a device name, so the module is handed back unbound. + assert sys.modules.get(test_module.__name__) is not test_module + + +def test_emitted_registers_carry_module_name(test_module): + assert test_module.AnalogData.__module__ == "Tests" + + +def test_emitted_registers_are_usable(test_module): + reg = test_module.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_device_emit.py b/tests/device/test_device_emit.py deleted file mode 100644 index c3e996c..0000000 --- a/tests/device/test_device_emit.py +++ /dev/null @@ -1,66 +0,0 @@ -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))