Yes chef. Right away chef. Add doohickey interface

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.

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

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