diff --git a/docs/examples/create_device_module/create_device_module.py b/docs/examples/create_device_module/create_device_module.py index 1d55e3e..9746aa2 100644 --- a/docs/examples/create_device_module/create_device_module.py +++ b/docs/examples/create_device_module/create_device_module.py @@ -29,11 +29,11 @@ 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: +# To have the identity checked on connect, pass the module itself. The check is +# driven by its `WHO_AM_I`, and `0` skips it: # -# class Behavior(Device): -# __whoami__ = behavior.WHO_AM_I +# with open_serial_device(behavior, port=SERIAL_PORT) as device: +# print("AnalogData:", device.read(AnalogData).parsed) # --- Custom interface types -------------------------------------------------- diff --git a/src/packages/harp-device/README.md b/src/packages/harp-device/README.md index b73ad68..4f16f99 100644 --- a/src/packages/harp-device/README.md +++ b/src/packages/harp-device/README.md @@ -38,26 +38,24 @@ 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 +common registers have a single definition, in `harp.device.core`, 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: +Pass the module to `Device` (or `open_serial_device`) to validate identity on open: ```python -from harp import behavior +from harp.device import behavior, client, core -# `device` is a Device opened over some transport (see harp-serial) -device.read(behavior.DigitalInputState) +with client.Device(transport, behavior) as device: + device.read(core.WhoAmI) # a common register + device.read(behavior.DigitalInputState) # declared by the schema ``` -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 -``` +`WHO_AM_I` in the module drives the check; `0` skips it. Omitting the module skips +validation. The module is not otherwise consulted: registers reach `read`, `write` and +`subscribe` as arguments either way, and only a subscribed register is parsed on +arrival. Common registers such as `WhoAmI` and `OperationControl` come from +`harp.device.core` and are read the same way. A new transport is just an object implementing the `ITransport` protocol (`open`/`write`/`read`/`close`). diff --git a/src/packages/harp-device/src/harp/device/client/_device.py b/src/packages/harp-device/src/harp/device/client/_device.py index 0d0e556..3f778f9 100644 --- a/src/packages/harp-device/src/harp/device/client/_device.py +++ b/src/packages/harp-device/src/harp/device/client/_device.py @@ -1,7 +1,7 @@ """Transport-agnostic Harp device base class.""" from collections.abc import Callable, Iterable -from typing import Any, ClassVar, Self, TypeVar +from typing import Any, ClassVar, Generic, Self, TypeVar, overload import logging import queue @@ -11,12 +11,14 @@ from harp.protocol._message import ParsedHarpMessage from harp.protocol._register import RegisterBase +from harp.device.schema import DeviceModuleLike from ._framer import HarpFramer from ._transport import ITransport, TransportError from harp.device.core import ( WhoAmI, ) +M = TypeVar("M", bound="DeviceModuleLike | None") P = TypeVar("P") _logger = logging.getLogger(__name__) @@ -67,30 +69,50 @@ def __exit__(self, *args: object) -> None: self.unsubscribe() -class Device: +class Device(Generic[M]): """Harp device protocol logic (framing, request/reply, register access) over an :class:`~harp.device.client.ITransport`. 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.schema.create_device_module`, or the - ``harp-device`` README for the statically generated equivalent). + :meth:`write` and :meth:`subscribe` take a register class directly. - A subclass sets :attr:`__whoami__` to validate device identity on open - (``0x0`` skips the check):: + Pass a ``device_module`` (from :func:`~harp.device.schema.create_device_module` or a + statically generated device package) to validate device identity on open:: - class Behavior(Device): - __whoami__ = 1216 + behavior = create_device_module(schema_text) + with Device(transport, behavior) as dev: + dev.read(behavior.OperationControl) + + Omitting ``device_module`` skips that check. The module is not otherwise + consulted: registers reach :meth:`read`, :meth:`write` and :meth:`subscribe` + as arguments either way, and only a subscribed register is parsed on arrival. """ REPLY_TIMEOUT: ClassVar[float] = 5.0 # seconds - #: Expected ``WhoAmI`` of the device this class models; ``0x0`` skips the check. - __whoami__: ClassVar[int] = 0x0 + @overload + def __init__( + self, transport: ITransport, device_module: M, *, raise_on_error: bool = ... + ) -> None: ... + + @overload + def __init__( + self: "Device[None]", + transport: ITransport, + device_module: None = ..., + *, + raise_on_error: bool = ..., + ) -> None: ... - def __init__(self, transport: ITransport, *, raise_on_error: bool = True) -> None: + def __init__( + self, + transport: ITransport, + device_module: M | None = None, + *, + raise_on_error: bool = True, + ) -> None: self._transport = transport + self._device_module = device_module self.raise_on_error = raise_on_error self._framer = HarpFramer() self._pending: dict[int, queue.SimpleQueue] = {} @@ -106,6 +128,11 @@ def __init__(self, transport: ITransport, *, raise_on_error: bool = True) -> Non self._event_queue: queue.SimpleQueue[HarpMessage | None] = queue.SimpleQueue() self._event_thread: threading.Thread | None = None + @property + def module(self) -> M: + """The device module injected at construction, or ``None`` if not set.""" + return self._device_module # type: ignore[return-value] + # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ @@ -130,15 +157,18 @@ def open(self) -> Self: return self def _validate_whoami(self) -> None: - """Check the device's ``WhoAmI`` against :attr:`__whoami__` (``0x0`` skips).""" - expected = self.__whoami__ + """Check the device's ``WhoAmI`` against the module (skipped if no module or ``WHO_AM_I == 0x0``).""" + module = self._device_module + if module is None: + return + expected = module.WHO_AM_I if expected == 0x0: return actual = int(self.read(WhoAmI).parsed) if actual != expected: raise RuntimeError( - f"WhoAmI mismatch: {type(self).__name__} expected 0x{expected:04x} " - f"but device reported 0x{actual:04x}." + f"WhoAmI mismatch: {module.__name__} expects 0x{expected:04x} " + f"but the device reported 0x{actual:04x}." ) def close(self) -> None: diff --git a/src/packages/harp-device/src/harp/device/py.typed b/src/packages/harp-device/src/harp/device/client/py.typed similarity index 100% rename from src/packages/harp-device/src/harp/device/py.typed rename to src/packages/harp-device/src/harp/device/client/py.typed diff --git a/src/packages/harp-device/src/harp/device/core/py.typed b/src/packages/harp-device/src/harp/device/core/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/packages/harp-device/src/harp/device/schema/py.typed b/src/packages/harp-device/src/harp/device/schema/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/packages/harp-serial/src/harp/serial/_serial.py b/src/packages/harp-serial/src/harp/serial/_serial.py index f7d4e5b..deb9ab6 100644 --- a/src/packages/harp-serial/src/harp/serial/_serial.py +++ b/src/packages/harp-serial/src/harp/serial/_serial.py @@ -1,15 +1,17 @@ """Serial transport and factory for Harp devices.""" -from typing import TypeVar +from typing import TypeVar, overload import serial from harp.device.client import Device, TransportError - -D = TypeVar("D", bound=Device) +from harp.device.schema import DeviceModuleLike DEFAULT_BAUDRATE: int = 1_000_000 +D = TypeVar("D", bound=Device) +M = TypeVar("M", bound=DeviceModuleLike) + class SerialTransport: """A serial-port :class:`~harp.device.client.ITransport` (structural conformance).""" @@ -51,20 +53,66 @@ def close(self) -> None: self._serial.close() +@overload def open_serial_device( - device: type[D], + device_or_module: type[D], + *, + port: str, + baudrate: int = ..., + raise_on_error: bool = ..., +) -> D: ... + + +@overload +def open_serial_device( + device_or_module: M, + *, + port: str, + baudrate: int = ..., + raise_on_error: bool = ..., +) -> Device[M]: ... + + +@overload +def open_serial_device( + device_or_module: None = ..., + *, + port: str, + baudrate: int = ..., + raise_on_error: bool = ..., +) -> Device[None]: ... + + +def open_serial_device( + device_or_module: DeviceModuleLike | type[D] | None = None, *, port: str, baudrate: int = DEFAULT_BAUDRATE, raise_on_error: bool = True, -) -> D: - """Build ``device`` over a serial transport and open it. +) -> Device: + """Build a :class:`~harp.device.client.Device` over a serial transport and open it. + + Accepts either a device module or a :class:`~harp.device.client.Device` subclass: + + - **Module** (preferred): validates identity on open:: + + from harp.device import behavior, core + + with open_serial_device(behavior, port="COM3") as dev: + dev.read(core.WhoAmI) # a common register + dev.read(behavior.AnalogData) # declared by the schema + + - **Device subclass**: instantiates the subclass directly, preserving its type:: + + with open_serial_device(MyBehavior, port="COM3") as dev: + dev.arm() # method defined on MyBehavior - Like the builtin :func:`open`, the returned device is already connected; - use it directly or in a ``with`` block for guaranteed close:: + Omit the first argument for schema-free access, which skips the identity check. - with open_serial_device(behavior.Device, port="COM3") as dev: - dev.read(behavior.WhoAmI) + Like the builtin :func:`open`, the returned device is already connected; use it + directly or in a ``with`` block for guaranteed close. """ transport = SerialTransport(port, baudrate) - return device(transport, raise_on_error=raise_on_error).open() + if isinstance(device_or_module, type): + return device_or_module(transport, raise_on_error=raise_on_error).open() + return Device(transport, device_or_module, raise_on_error=raise_on_error).open() diff --git a/tests/conformance.py b/tests/conformance.py index 0e11319..c2df806 100644 --- a/tests/conformance.py +++ b/tests/conformance.py @@ -5,14 +5,15 @@ type fails the build rather than being noticed downstream. """ -from typing import Any, assert_type +from typing import Any, ClassVar, assert_type import numpy as np from harp.data import DatasetReader +from harp.device.client import Device, ITransport from harp.device.core import OperationControl, OperationControlPayload, WhoAmI -from harp.device.client import Device from harp.device.schema import DeviceModule, DeviceModuleLike, create_device_module from harp.protocol import ParsedHarpMessage, RegisterBase +from harp.serial import open_serial_device def schema_built_registers(yml: str) -> None: @@ -36,6 +37,44 @@ def register_writes(device: Device, payload: OperationControlPayload) -> None: assert_type(device.write(OperationControl, payload).parsed, OperationControlPayload) +def device_with_module(transport: ITransport, module: DeviceModule) -> None: + """Device constructed with a module is typed on that module.""" + device = Device(transport, module) + assert_type(device, Device[DeviceModule]) + assert_type(device.module, DeviceModule) + + +def device_without_module(transport: ITransport) -> None: + """Device constructed without a module is Device[None].""" + device = Device(transport) + assert_type(device, Device[None]) + assert_type(device.module, None) + + +def open_serial_device_with_module(module: DeviceModule) -> None: + """open_serial_device with a module returns Device[M].""" + device = open_serial_device(module, port="COM3") + assert_type(device, Device[DeviceModule]) + assert_type(device.module, DeviceModule) + + +def open_serial_device_without_module() -> None: + """open_serial_device without a module returns Device[None].""" + device = open_serial_device(port="COM3") + assert_type(device, Device[None]) + assert_type(device.module, None) + + +def open_serial_device_with_subclass() -> None: + """open_serial_device with a Device subclass preserves its type.""" + + class MyDevice(Device[DeviceModule]): + def arm(self) -> None: ... + + device = open_serial_device(MyDevice, port="COM3") + assert_type(device, MyDevice) + + def dataset_reader_accepts_either_module( schema_built: DeviceModule, generated: DeviceModuleLike ) -> None: @@ -44,3 +83,19 @@ def dataset_reader_accepts_either_module( reader = DatasetReader(generated, "session.harp") reader.read(WhoAmI) reader.read(44) + + +def open_serial_device_prefers_the_subclass_overload() -> None: + """A Device subclass is matched as a subclass even when it looks like a module. + + type[D] is narrower than the structural module overload, so it has to come first: + a class carrying REGISTER_MAP and WHO_AM_I satisfies DeviceModuleLike too, and the + module overload would otherwise win and return Device[type[Hybrid]]. + """ + + class Hybrid(Device[None]): + REGISTER_MAP: ClassVar[dict[int, type[RegisterBase[Any]]]] = {} + WHO_AM_I: ClassVar[int] = 1216 + + device = open_serial_device(Hybrid, port="COM3") + assert_type(device, Hybrid) diff --git a/tests/device/test_device.py b/tests/device/test_device.py new file mode 100644 index 0000000..570be9c --- /dev/null +++ b/tests/device/test_device.py @@ -0,0 +1,34 @@ +import types + +from harp.device.client import Device + + +class _NullTransport: + def open(self) -> None: ... + + def close(self) -> None: ... + + def write(self, data: bytes) -> None: ... + + def read(self) -> bytes: + return b"" + + +def _module(name: str, **attrs: object) -> types.ModuleType: + mod = types.ModuleType(name) + for key, value in attrs.items(): + setattr(mod, key, value) + return mod + + +def test_whoami_of_zero_skips_the_check(): + # 0 marks an unregistered device, so opening must not read WhoAmI at all. + device = Device(_NullTransport(), _module("Unregistered", WHO_AM_I=0, REGISTER_MAP={})) + with device: + assert device.module.WHO_AM_I == 0 + + +def test_module_is_returned_by_the_property(): + module = _module("Behavior", WHO_AM_I=0, REGISTER_MAP={}) + assert Device(_NullTransport(), module).module is module + assert Device(_NullTransport()).module is None