Add optional ReactiveX adapter for device events

pyproject.toml

@@ -87,6 +87,7 @@ dev = [
     "harp-data",
     "harp-benchmarks",
     "typing-extensions>=4.15.0",
+    "reactivex>=4",  # for harp-device[rx] adapter + its tests
 ]
 
 [tool.codespell]

scripts/event_monitor.py

@@ -0,0 +1,72 @@
+"""Example: subscribe to all events from a Harp device and print them.
+
+Connects to a device on COM3, registers a single catch-all handler for *every*
+event, and prints each one to the console until you press Ctrl+C.
+
+Run with:
+    uv run python scripts/event_monitor.py
+"""
+
+import time
+
+from harp.device import (
+    REGISTER_MAP,
+    Device,
+    OperationControl,
+    OperationControlPayload,
+    OperationMode,
+    TimestampSeconds,
+)
+from harp.device.rx import observe
+from harp.protocol import HarpMessage, MessageType
+from harp.serial import open_serial_device
+from reactivex import operators as ops
+
+PORT = "COM3"
+
+
+def print_event(msg: HarpMessage) -> None:
+    if msg.address == 44:  # too damn noisy
+        return
+    register = REGISTER_MAP.get(msg.address, None)
+    if register is not None:
+        value = register.parse(msg)
+    else:
+        value = msg.payload.hex()
+    print(f"NonRx: {msg.timestamp:.6f}  {msg.address}  {msg.message_type.name:<5s}  {value}")
+
+
+if __name__ == "__main__":
+    with open_serial_device(Device, port=PORT) as dev:
+        dev.write(
+            OperationControl,
+            OperationControlPayload(
+                operation_mode=OperationMode.ACTIVE,
+                dump_registers=True,
+                heartbeat=True,
+                mute_replies=False,
+                operation_led=True,
+                visual_indicators=True,
+            ),
+        )
+        disposable = (
+            observe(dev, TimestampSeconds)
+            .pipe(
+                ops.filter(lambda msg: msg.message_type == MessageType.Event),
+                ops.filter(lambda msg: msg.parsed % 2 == 0),
+            )
+            .subscribe(
+                lambda msg: print(
+                    f"Rx:{msg.timestamp:.6f}  {msg.address}  {msg.message_type.name:<5s}  {msg.parsed}"
+                )
+            )
+        )
+
+        dev.subscribe_all(print_event)
+        print(f"Listening for events on {PORT}. Press Ctrl+C to stop.\n")
+        try:
+            while True:
+                time.sleep(1.0)
+        except KeyboardInterrupt:
+            print("\nStopping.")
+            disposable.dispose()

src/packages/harp-device/pyproject.toml

@@ -7,6 +7,9 @@ dependencies = [
     "harp-protocol",
 ]
 
+[project.optional-dependencies]
+rx = ["reactivex>=4"]
+
 [build-system]
 requires = ["uv_build>=0.9.5"]
 build-backend = "uv_build"

src/packages/harp-device/src/harp/device/rx.py

@@ -0,0 +1,69 @@
+from typing import TYPE_CHECKING, Any, TypeVar
+
+try:
+    import reactivex as rx
+    import reactivex.operators as ops
+    from reactivex.disposable import Disposable
+except ModuleNotFoundError as exc:
+    raise ImportError(
+        "harp.device.rx requires the 'reactivex' package, which is not installed. "
+        "Install the optional extra with:  pip install 'harp-device[rx]'."
+    ) from exc
+
+from harp.protocol import HarpMessage, MessageType
+from harp.protocol._message import ParsedHarpMessage
+from harp.protocol._register import RegisterBase
+
+from ._device import MessageTypeFilter
+
+if TYPE_CHECKING:
+    from ._device import Device
+
+P = TypeVar("P")
+
+__all__ = ["observe", "observe_all"]
+
+
+def observe(
+    device: "Device",
+    register: type[RegisterBase[P]],
+    *,
+    message_types: MessageTypeFilter = MessageType.Event,
+) -> "rx.Observable[ParsedHarpMessage[P]]":
+    """Return a hot, multicast Observable of parsed messages for ``register``.
+
+    Mirrors :meth:`Device.subscribe`: by default only ``Event`` messages are
+    emitted; pass ``message_types`` for read/write replies too.
+
+    The stream is ref-counted (``ops.share``): the *first* subscriber creates a
+    single underlying :meth:`Device.subscribe`, additional subscribers share it,
+    and the device subscription is disposed once the last one unsubscribes.
+
+    Note: emissions run on the device's shared event thread. Use
+    ``ops.observe_on(scheduler)`` to move heavy downstream work off it.
+    """
+
+    def on_subscribe(observer: "rx.abc.ObserverBase[Any]", _: Any = None) -> Disposable:
+        sub = device.subscribe(register, observer.on_next, message_types=message_types)
+        return Disposable(sub.unsubscribe)
+
+    return rx.create(on_subscribe).pipe(ops.share())
+
+
+def observe_all(
+    device: "Device",
+    *,
+    message_types: MessageTypeFilter = MessageType.Event,
+) -> "rx.Observable[HarpMessage]":
+    """Return a hot, multicast Observable of raw messages for *every* address.
+
+    Mirrors :meth:`Device.subscribe_all` (a full-traffic firehose when passed
+    more message types). Same ref-counted, hot, single-underlying-subscription
+    lifecycle and threading semantics as :func:`observe`.
+    """
+
+    def on_subscribe(observer: "rx.abc.ObserverBase[Any]", _: Any = None) -> Disposable:
+        sub = device.subscribe_all(observer.on_next, message_types=message_types)
+        return Disposable(sub.unsubscribe)
+
+    return rx.create(on_subscribe).pipe(ops.share())