Add dependency injection for the device constructor

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 --------------------------------------------------

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`).

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:

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()

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)

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