diff --git a/docs/doohickey-interface.md b/docs/doohickey-interface.md new file mode 100644 index 00000000..b53e9769 --- /dev/null +++ b/docs/doohickey-interface.md @@ -0,0 +1,32 @@ +# Doohickey Interface + +`src/doohickey_interface.py` provides a small registry and connection manager for +three supported device families: + +- `doohickey` +- `gizmo` +- `whatsit` + +The default adapter is in-memory so tests and dry runs do not need network or +hardware access. Production adapters can be registered per device type. + +```python +from doohickey_interface import DeviceSpec, DoohickeyInterface + +interface = DoohickeyInterface() +interface.register_device( + DeviceSpec( + name="field-gizmo", + device_type="gizmo", + endpoint="local://gizmos/field", + tags=("portable", "calibrated"), + ) +) + +connection = interface.connect("field-gizmo") +response = interface.send("field-gizmo", {"spin": 71}) +interface.disconnect("field-gizmo") +``` + +The manager validates supported device types, prevents duplicate names, and +raises explicit errors for unknown or disconnected devices. diff --git a/src/doohickey_interface.py b/src/doohickey_interface.py new file mode 100644 index 00000000..a656c671 --- /dev/null +++ b/src/doohickey_interface.py @@ -0,0 +1,231 @@ +"""Connection interface for doohickeys, gizmos, and whatsits.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from itertools import count +from types import MappingProxyType +from typing import Mapping, Protocol + + +SUPPORTED_DEVICE_TYPES = frozenset({"doohickey", "gizmo", "whatsit"}) + + +class DeviceInterfaceError(Exception): + """Base class for device interface errors.""" + + +class DuplicateDeviceError(DeviceInterfaceError): + def __init__(self, name: str) -> None: + super().__init__(f"Device already registered: {name!r}") + self.name = name + + +class UnknownDeviceError(DeviceInterfaceError): + def __init__(self, name: str) -> None: + super().__init__(f"Unknown device: {name!r}") + self.name = name + + +class UnsupportedDeviceTypeError(DeviceInterfaceError): + def __init__(self, device_type: str) -> None: + supported = ", ".join(sorted(SUPPORTED_DEVICE_TYPES)) + super().__init__(f"Unsupported device type {device_type!r}; expected {supported}") + self.device_type = device_type + + +class NotConnectedError(DeviceInterfaceError): + def __init__(self, name: str) -> None: + super().__init__(f"Device is not connected: {name!r}") + self.name = name + + +@dataclass(frozen=True) +class DeviceSpec: + """Static configuration for a doohickey, gizmo, or whatsit.""" + + name: str + device_type: str + endpoint: str + tags: tuple[str, ...] = () + metadata: Mapping[str, str] = field(default_factory=dict) + + def __post_init__(self) -> None: + normalized_type = self.device_type.strip().lower() + if normalized_type not in SUPPORTED_DEVICE_TYPES: + raise UnsupportedDeviceTypeError(self.device_type) + if not self.name.strip(): + raise ValueError("Device name is required") + if not self.endpoint.strip(): + raise ValueError("Device endpoint is required") + + object.__setattr__(self, "device_type", normalized_type) + object.__setattr__(self, "tags", tuple(sorted({tag.strip() for tag in self.tags if tag.strip()}))) + object.__setattr__(self, "metadata", MappingProxyType(dict(self.metadata))) + + +@dataclass(frozen=True) +class DeviceConnection: + """A live connection handle returned by an adapter.""" + + spec: DeviceSpec + session_id: str + adapter: str + status: str = "connected" + metadata: Mapping[str, str] = field(default_factory=dict) + + def __post_init__(self) -> None: + object.__setattr__(self, "metadata", MappingProxyType(dict(self.metadata))) + + +class DeviceAdapter(Protocol): + """Adapter contract used by the doohickey interface.""" + + name: str + + def connect(self, spec: DeviceSpec) -> DeviceConnection: + """Open a connection to a configured device.""" + + def send( + self, connection: DeviceConnection, payload: Mapping[str, object] + ) -> Mapping[str, object]: + """Send a payload through an open connection.""" + + def disconnect(self, connection: DeviceConnection) -> None: + """Close a connection opened by this adapter.""" + + +class InMemoryDeviceAdapter: + """Deterministic adapter for local tests and dry-run integrations.""" + + name = "in-memory" + + def __init__(self) -> None: + self._ids = count(1) + self.sent_payloads: list[tuple[str, Mapping[str, object]]] = [] + self.disconnected: list[str] = [] + + def connect(self, spec: DeviceSpec) -> DeviceConnection: + return DeviceConnection( + spec=spec, + session_id=f"{spec.device_type}-{next(self._ids)}", + adapter=self.name, + metadata={"endpoint": spec.endpoint}, + ) + + def send( + self, connection: DeviceConnection, payload: Mapping[str, object] + ) -> Mapping[str, object]: + snapshot = dict(payload) + self.sent_payloads.append((connection.session_id, snapshot)) + return { + "accepted": True, + "device": connection.spec.name, + "device_type": connection.spec.device_type, + "session_id": connection.session_id, + "payload": snapshot, + } + + def disconnect(self, connection: DeviceConnection) -> None: + self.disconnected.append(connection.session_id) + + +class DoohickeyInterface: + """Registry and connection manager for doohickeys, gizmos, and whatsits.""" + + def __init__(self, default_adapter: DeviceAdapter | None = None) -> None: + adapter = default_adapter or InMemoryDeviceAdapter() + self._adapters: dict[str, DeviceAdapter] = { + device_type: adapter for device_type in SUPPORTED_DEVICE_TYPES + } + self._devices: dict[str, DeviceSpec] = {} + self._connections: dict[str, DeviceConnection] = {} + + def register_adapter(self, device_type: str, adapter: DeviceAdapter) -> None: + normalized_type = device_type.strip().lower() + if normalized_type not in SUPPORTED_DEVICE_TYPES: + raise UnsupportedDeviceTypeError(device_type) + self._adapters[normalized_type] = adapter + + def register_device(self, spec: DeviceSpec) -> DeviceSpec: + if spec.name in self._devices: + raise DuplicateDeviceError(spec.name) + self._devices[spec.name] = spec + return spec + + def list_devices( + self, device_type: str | None = None, tag: str | None = None + ) -> list[DeviceSpec]: + normalized_type = device_type.strip().lower() if device_type else None + if normalized_type and normalized_type not in SUPPORTED_DEVICE_TYPES: + raise UnsupportedDeviceTypeError(device_type or "") + + desired_tag = tag.strip() if tag else None + devices = self._devices.values() + if normalized_type: + devices = [spec for spec in devices if spec.device_type == normalized_type] + if desired_tag: + devices = [spec for spec in devices if desired_tag in spec.tags] + return sorted(devices, key=lambda spec: spec.name) + + def connect(self, name: str) -> DeviceConnection: + spec = self._require_device(name) + if name in self._connections: + return self._connections[name] + connection = self._adapters[spec.device_type].connect(spec) + self._connections[name] = connection + return connection + + def send(self, name: str, payload: Mapping[str, object]) -> Mapping[str, object]: + connection = self._connections.get(name) + if connection is None: + raise NotConnectedError(name) + return self._adapters[connection.spec.device_type].send(connection, payload) + + def disconnect(self, name: str) -> None: + connection = self._connections.pop(name, None) + if connection is None: + raise NotConnectedError(name) + self._adapters[connection.spec.device_type].disconnect(connection) + + def connection_status(self, name: str) -> str: + self._require_device(name) + return "connected" if name in self._connections else "disconnected" + + def _require_device(self, name: str) -> DeviceSpec: + try: + return self._devices[name] + except KeyError as exc: + raise UnknownDeviceError(name) from exc + + +def default_device_specs() -> list[DeviceSpec]: + """Return a small starter catalog covering every supported device type.""" + return [ + DeviceSpec( + name="brass-doohickey", + device_type="doohickey", + endpoint="local://doohickeys/brass", + tags=("brass", "calibrated"), + ), + DeviceSpec( + name="field-gizmo", + device_type="gizmo", + endpoint="local://gizmos/field", + tags=("portable", "calibrated"), + ), + DeviceSpec( + name="bench-whatsit", + device_type="whatsit", + endpoint="local://whatsits/bench", + tags=("bench", "diagnostic"), + ), + ] + + +def build_default_interface() -> DoohickeyInterface: + """Create an interface preloaded with one doohickey, gizmo, and whatsit.""" + interface = DoohickeyInterface() + for spec in default_device_specs(): + interface.register_device(spec) + return interface diff --git a/src/tests/test_doohickey_interface.py b/src/tests/test_doohickey_interface.py new file mode 100644 index 00000000..df6419ed --- /dev/null +++ b/src/tests/test_doohickey_interface.py @@ -0,0 +1,115 @@ +"""Tests for the doohickey, gizmo, and whatsit interface.""" + +from __future__ import annotations + +import unittest +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from doohickey_interface import ( + DeviceConnection, + DeviceSpec, + DoohickeyInterface, + DuplicateDeviceError, + InMemoryDeviceAdapter, + NotConnectedError, + UnknownDeviceError, + UnsupportedDeviceTypeError, + build_default_interface, +) + + +class RecordingAdapter: + name = "recording" + + def __init__(self) -> None: + self.connected: list[str] = [] + self.disconnected: list[str] = [] + self.payloads: list[dict[str, object]] = [] + + def connect(self, spec: DeviceSpec) -> DeviceConnection: + self.connected.append(spec.name) + return DeviceConnection(spec=spec, session_id=f"session:{spec.name}", adapter=self.name) + + def send(self, connection: DeviceConnection, payload: dict[str, object]) -> dict[str, object]: + snapshot = dict(payload) + self.payloads.append(snapshot) + return {"device": connection.spec.name, "payload": snapshot} + + def disconnect(self, connection: DeviceConnection) -> None: + self.disconnected.append(connection.spec.name) + + +class DoohickeyInterfaceTests(unittest.TestCase): + def test_default_catalog_covers_all_supported_device_types(self) -> None: + interface = build_default_interface() + + self.assertEqual( + [spec.device_type for spec in interface.list_devices()], + ["whatsit", "doohickey", "gizmo"], + ) + self.assertEqual( + [spec.name for spec in interface.list_devices(tag="calibrated")], + ["brass-doohickey", "field-gizmo"], + ) + + def test_rejects_unknown_device_type(self) -> None: + with self.assertRaises(UnsupportedDeviceTypeError): + DeviceSpec(name="bad-widget", device_type="widget", endpoint="local://bad") + + def test_rejects_duplicate_device_names(self) -> None: + interface = DoohickeyInterface() + spec = DeviceSpec("brass-doohickey", "doohickey", "local://brass") + + interface.register_device(spec) + + with self.assertRaises(DuplicateDeviceError): + interface.register_device(spec) + + def test_connects_sends_and_disconnects_through_registered_adapter(self) -> None: + adapter = RecordingAdapter() + interface = DoohickeyInterface() + interface.register_adapter("gizmo", adapter) + interface.register_device(DeviceSpec("field-gizmo", "gizmo", "local://field")) + + connection = interface.connect("field-gizmo") + response = interface.send("field-gizmo", {"spin": 71}) + interface.disconnect("field-gizmo") + + self.assertEqual(connection.session_id, "session:field-gizmo") + self.assertEqual(response, {"device": "field-gizmo", "payload": {"spin": 71}}) + self.assertEqual(adapter.connected, ["field-gizmo"]) + self.assertEqual(adapter.payloads, [{"spin": 71}]) + self.assertEqual(adapter.disconnected, ["field-gizmo"]) + self.assertEqual(interface.connection_status("field-gizmo"), "disconnected") + + def test_default_adapter_echoes_payload_for_all_device_types(self) -> None: + adapter = InMemoryDeviceAdapter() + interface = DoohickeyInterface(default_adapter=adapter) + for spec in [ + DeviceSpec("d", "doohickey", "local://d"), + DeviceSpec("g", "gizmo", "local://g"), + DeviceSpec("w", "whatsit", "local://w"), + ]: + interface.register_device(spec) + + for name in ("d", "g", "w"): + connection = interface.connect(name) + response = interface.send(name, {"ping": name}) + self.assertEqual(response["accepted"], True) + self.assertEqual(response["session_id"], connection.session_id) + + def test_send_requires_connection_and_known_device(self) -> None: + interface = DoohickeyInterface() + interface.register_device(DeviceSpec("bench-whatsit", "whatsit", "local://bench")) + + with self.assertRaises(NotConnectedError): + interface.send("bench-whatsit", {"ping": True}) + with self.assertRaises(UnknownDeviceError): + interface.connect("missing") + + +if __name__ == "__main__": + unittest.main()