From 8467f9260d2384897cafab869942b767b2b62a2d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Sat, 27 Jun 2026 17:32:44 -0400 Subject: [PATCH 1/3] Add DeviceController: headless threaded wrapper for any PhysicalDevice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Building an interactive app on a PhysicalDevice means re-solving the same problems each time: device calls block (the Millennia confirms ON/OFF over ~seconds) so they can't run on a UI thread; status polling and user actions share one port and must be serialized; connections drop and want to recover; and none of this should know about the UI toolkit. DeviceController wraps any PhysicalDevice and runs ALL device access — submitted commands and the periodic status poll — through a single worker thread, so there are no races and ordering is preserved. It reports through the existing NotificationCenter (didConnect / didDisconnect / connectionLost / connectionFailed / status / commandFailed), so any front-end (mytk, Qt, CLI, tests) just observes; marshalling to a UI thread is the front-end's job. It imports only the standard library + NotificationCenter — no toolkit dependency. Features: submit(action) async commands; connect/disconnect with intent; auto-reconnect with configurable interval; status polling via doGetStatusUserInfo() that disables itself for devices without a snapshot; and a toolkit-agnostic connection_error_reason() that classifies busy/missing/ permission off the exception chain. Adds a full unittest suite (9 tests) driven by the Debug Millennia device. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01K6LVhfM3d9CcJdkuSMMqag --- hardwarelibrary/__init__.py | 1 + hardwarelibrary/devicecontroller.py | 314 ++++++++++++++++++ hardwarelibrary/tests/testDeviceController.py | 218 ++++++++++++ 3 files changed, 533 insertions(+) create mode 100644 hardwarelibrary/devicecontroller.py create mode 100644 hardwarelibrary/tests/testDeviceController.py diff --git a/hardwarelibrary/__init__.py b/hardwarelibrary/__init__.py index 8cfbd4e..d5e67b1 100644 --- a/hardwarelibrary/__init__.py +++ b/hardwarelibrary/__init__.py @@ -26,6 +26,7 @@ from hardwarelibrary.physicaldevice import * from hardwarelibrary.notificationcenter import * +from hardwarelibrary.devicecontroller import * from hardwarelibrary.communication import * # We want these modules in their namespace diff --git a/hardwarelibrary/devicecontroller.py b/hardwarelibrary/devicecontroller.py new file mode 100644 index 0000000..96667cd --- /dev/null +++ b/hardwarelibrary/devicecontroller.py @@ -0,0 +1,314 @@ +"""A headless, GUI-agnostic controller around any PhysicalDevice. + +Building an interactive app (e.g. a GUI) on top of a PhysicalDevice means +solving the same problems every time: + + * device calls block (the Millennia eV confirms ON/OFF by writing, settling, + and reading back -- up to several seconds), so they cannot run on a UI + thread; + * status polling and user actions both touch the same port, so they must be + serialized to avoid interleaving bytes (writeCommand is not under the + port's transactionLock); + * connections drop -- the cable is pulled, the port disappears -- and the app + wants to recover automatically; + * none of this should know anything about the UI toolkit. + +DeviceController owns a single worker thread through which *all* device access +flows -- submitted commands and the periodic status poll alike -- so there are +no races and ordering is preserved. It reports everything through the existing +NotificationCenter, so any front-end (Tk/mytk, Qt, a CLI, a test) just observes +notifications; marshalling them onto a UI thread is the front-end's job. + +It is deliberately decoupled from any toolkit: this module imports only the +standard library and NotificationCenter. + +Example:: + + from hardwarelibrary.devicecontroller import ( + DeviceController, DeviceControllerNotification as N) + from hardwarelibrary.notificationcenter import NotificationCenter + from hardwarelibrary.sources.millennia import MillenniaDevice + + controller = DeviceController(MillenniaDevice(portPath="/dev/cu.usbmodem1")) + + def on_status(notification): + print(notification.userInfo) # {'power': .., 'isLaserOn': .., ...} + + NotificationCenter().addObserver(self, on_status, N.status) + controller.start() + controller.connect() + controller.submit(lambda device: device.turnOn()) + ... + controller.stop() +""" + +import queue +import time +from enum import Enum +from threading import Event, RLock, Thread + +from hardwarelibrary.notificationcenter import NotificationCenter + +__all__ = [ + "DeviceController", + "DeviceControllerNotification", + "connection_error_reason", +] + + +class DeviceControllerNotification(Enum): + """Notifications posted by a DeviceController (the controller is the object). + + userInfo payloads: + didConnect -> the device + didDisconnect -> None + connectionLost -> the exception that revealed the drop + connectionFailed -> the exception from the failed (re)connect attempt + status -> the dict from device.doGetStatusUserInfo() + commandFailed -> the exception raised by the submitted command + """ + didConnect = "didConnect" + didDisconnect = "didDisconnect" + connectionLost = "connectionLost" + connectionFailed = "connectionFailed" + status = "status" + commandFailed = "commandFailed" + + +def connection_error_reason(error): + """Classify a connection exception as 'busy'/'permission'/'missing'/None. + + Walks the exception's __cause__/__context__ chain (PhysicalDevice drivers + are encouraged to preserve the underlying transport error) and matches it + against common serial-port failure strings. Toolkit-agnostic and free of + any pyserial import, so a front-end can turn a connectionFailed/lost + notification into a human message without re-probing the port. + """ + parts, seen, current = [], set(), error + while current is not None and id(current) not in seen: + seen.add(id(current)) + text = str(current).strip() + if text: + parts.append(text) + current = current.__cause__ or current.__context__ + text = " | ".join(parts).lower() + if "resource busy" in text or "errno 16" in text or "busy" in text: + return "busy" + if "permission" in text or "errno 13" in text or "access is denied" in text: + return "permission" + if ("no such file" in text or "errno 2" in text or "could not open" in text + or "filenotfounderror" in text): + return "missing" + return None + + +class DeviceController: + """Drive a PhysicalDevice from a single worker thread, reporting via notifications. + + Args: + device: any PhysicalDevice instance. + poll_interval: seconds between status polls while connected. The poll + calls device.doGetStatusUserInfo(); if the device has none (returns + None), polling is disabled automatically. + reconnect_interval: seconds between reconnection attempts while + disconnected-but-wanted. + auto_reconnect: when True, an unexpected drop (or a connect that fails + because the device is not yet present) is retried until it succeeds + or disconnect() is called. When False, a single failure gives up. + """ + + def __init__(self, device, poll_interval=1.0, reconnect_interval=2.0, + auto_reconnect=True): + self.device = device + self.poll_interval = poll_interval + self.reconnect_interval = reconnect_interval + self.auto_reconnect = auto_reconnect + + self._queue = queue.Queue() + self._stop = Event() + self._lock = RLock() + self._thread = Thread(target=self._run, name="DeviceController", + daemon=True) + + self._connected = False + self._want_connected = False + self._supports_status = True # set False if doGetStatusUserInfo()->None + self._next_poll = 0.0 + self._next_reconnect = 0.0 + self._last_failure_signature = None + + # -- public API (any thread) -- + + def start(self): + """Start the worker thread.""" + self._thread.start() + + def connect(self): + """Request connection (and, with auto_reconnect, keep it connected).""" + self._queue.put(("connect", None)) + + def disconnect(self): + """Request disconnection; clears the reconnect intent.""" + self._queue.put(("disconnect", None)) + + def submit(self, action, name=None): + """Run ``action(device)`` on the worker thread. + + Returns immediately. On failure a commandFailed notification is posted + with the exception. A status poll is forced right after a successful + command so observers see the new state promptly. + """ + self._queue.put(("command", (action, name))) + + def stop(self): + """Stop the worker thread and shut the device down.""" + self._stop.set() + self._queue.put(("disconnect", None)) + if self._thread.is_alive(): + self._thread.join(timeout=5.0) + + @property + def is_connected(self): + with self._lock: + return self._connected + + @property + def is_running(self): + return self._thread.is_alive() + + # -- worker thread -- + + def _post(self, name, userInfo=None): + NotificationCenter().postNotification(name, notifyingObject=self, + userInfo=userInfo) + + def _run(self): + while not self._stop.is_set(): + try: + task = self._queue.get(timeout=self._seconds_until_due()) + except queue.Empty: + task = None + + if task is not None: + self._handle(task) + + now = time.monotonic() + if self._connected and self._supports_status and now >= self._next_poll: + self._next_poll = now + self.poll_interval + self._poll() + elif (not self._connected and self._want_connected + and now >= self._next_reconnect): + self._next_reconnect = now + self.reconnect_interval + self._attempt_connect() + + self._teardown() + + def _seconds_until_due(self): + now = time.monotonic() + if self._connected: + target = self._next_poll if self._supports_status else now + 0.5 + elif self._want_connected: + target = self._next_reconnect + else: + return None # nothing scheduled; block until a task arrives + return max(0.0, target - now) + + def _handle(self, task): + kind, payload = task + if kind == "connect": + self._want_connected = True + self._next_reconnect = 0.0 + self._attempt_connect() + elif kind == "disconnect": + self._want_connected = False + self._do_disconnect() + elif kind == "command": + self._do_command(*payload) + + def _attempt_connect(self): + with self._lock: + if self._connected: + return + try: + self.device.initializeDevice() + except Exception as error: + self._connected = False + self._post_failure_if_new( + DeviceControllerNotification.connectionFailed, error) + if not self.auto_reconnect: + self._want_connected = False + return + + with self._lock: + self._connected = True + self._supports_status = True + self._next_poll = time.monotonic() # poll immediately + self._last_failure_signature = None + self._post(DeviceControllerNotification.didConnect, userInfo=self.device) + + def _poll(self): + try: + info = self.device.doGetStatusUserInfo() + except Exception as error: + self._handle_lost(error) + return + if info is None: + # Device exposes no status snapshot; stop polling for it. + self._supports_status = False + return + self._post(DeviceControllerNotification.status, userInfo=info) + + def _do_command(self, action, name): + if not self.is_connected: + self._post(DeviceControllerNotification.commandFailed, + userInfo=RuntimeError( + "Cannot run command {0!r}: not connected".format(name))) + return + try: + action(self.device) + except Exception as error: + # A failed command may mean the link dropped; verify with a poll. + self._post(DeviceControllerNotification.commandFailed, userInfo=error) + self._poll() + return + # Reflect the new state promptly. + if self._supports_status: + self._next_poll = time.monotonic() + self._poll() + + def _handle_lost(self, error): + was_connected = self._connected + with self._lock: + self._connected = False + self._safe_shutdown() + self._next_reconnect = time.monotonic() + self.reconnect_interval + if not self.auto_reconnect: + self._want_connected = False + if was_connected: + self._post(DeviceControllerNotification.connectionLost, userInfo=error) + + def _do_disconnect(self): + with self._lock: + self._connected = False + self._safe_shutdown() + self._post(DeviceControllerNotification.didDisconnect) + + def _safe_shutdown(self): + try: + self.device.shutdownDevice() + except Exception: + pass + + def _post_failure_if_new(self, name, error): + # Avoid flooding identical connectionFailed notifications every interval + # while a device stays unavailable; re-post only when the reason changes. + signature = (type(error).__name__, connection_error_reason(error) or str(error)) + if signature != self._last_failure_signature: + self._last_failure_signature = signature + self._post(name, userInfo=error) + + def _teardown(self): + if self._connected: + with self._lock: + self._connected = False + self._safe_shutdown() diff --git a/hardwarelibrary/tests/testDeviceController.py b/hardwarelibrary/tests/testDeviceController.py new file mode 100644 index 0000000..69d3362 --- /dev/null +++ b/hardwarelibrary/tests/testDeviceController.py @@ -0,0 +1,218 @@ +import env # noqa: F401 (sets sys.path) +import time +import unittest +from threading import Lock + +from hardwarelibrary.devicecontroller import ( + DeviceController, DeviceControllerNotification, connection_error_reason) +from hardwarelibrary.notificationcenter import NotificationCenter +from hardwarelibrary.sources.millennia import DebugMillenniaDevice + + +class FlakyMillennia(DebugMillenniaDevice): + """A debug Millennia that can be taken 'offline' to simulate a dropped link.""" + + def __init__(self): + super().__init__() + self.online = True + + def doInitializeDevice(self): + if not self.online: + raise IOError("port gone") + return super().doInitializeDevice() + + def doGetStatusUserInfo(self): + if not self.online: + raise IOError("[Errno 16] Resource busy: link down") + return super().doGetStatusUserInfo() + + +class NoStatusDevice(DebugMillenniaDevice): + """A debug device that exposes no status snapshot.""" + + def doGetStatusUserInfo(self): + return None + + +class Recorder: + """Collects a controller's notifications for assertions.""" + + def __init__(self, controller): + self.events = [] + self._lock = Lock() + self._nc = NotificationCenter() + for name in DeviceControllerNotification: + self._nc.addObserver(self, self._handle, name, observedObject=controller) + + def _handle(self, notification): + with self._lock: + self.events.append((notification.name, notification.userInfo)) + + def names(self): + with self._lock: + return [name for name, _ in self.events] + + def wait_for(self, name, timeout=3.0): + """Wait for a notification and return its userInfo (which may be None).""" + deadline = time.time() + timeout + while time.time() < deadline: + with self._lock: + for nm, info in self.events: + if nm == name: + return info + time.sleep(0.02) + return None + + def wait_seen(self, name, timeout=3.0): + """Wait for a notification to arrive; return True/False (payload-agnostic).""" + deadline = time.time() + timeout + while time.time() < deadline: + if name in self.names(): + return True + time.sleep(0.02) + return False + + def last_status(self): + with self._lock: + for nm, info in reversed(self.events): + if nm == DeviceControllerNotification.status: + return info + return None + + def remove(self): + self._nc.removeObserver(self) + + +class TestDeviceController(unittest.TestCase): + def setUp(self): + self.controllers = [] + self.recorders = [] + + def tearDown(self): + for controller in self.controllers: + controller.stop() + for recorder in self.recorders: + recorder.remove() + + def make(self, device=None, **kwargs): + device = device or DebugMillenniaDevice() + kwargs.setdefault("poll_interval", 0.2) + kwargs.setdefault("reconnect_interval", 0.3) + controller = DeviceController(device, **kwargs) + recorder = Recorder(controller) + self.controllers.append(controller) + self.recorders.append(recorder) + controller.start() + return controller, recorder + + # -- connection + status -- + + def testConnectPostsDidConnectAndStatus(self): + controller, rec = self.make() + controller.connect() + self.assertIsNotNone(rec.wait_for(DeviceControllerNotification.didConnect)) + status = rec.wait_for(DeviceControllerNotification.status) + self.assertIsNotNone(status) + self.assertEqual(set(status), {"power", "isLaserOn", "isShutterOpen"}) + self.assertTrue(controller.is_connected) + + def testSubmittedCommandRunsAndStatusReflectsIt(self): + controller, rec = self.make() + controller.connect() + rec.wait_for(DeviceControllerNotification.didConnect) + controller.submit(lambda device: device.turnOn()) + controller.submit(lambda device: device.openShutter()) + deadline = time.time() + 3.0 + while time.time() < deadline: + status = rec.last_status() + if status and status["isLaserOn"] and status["isShutterOpen"]: + break + time.sleep(0.05) + self.assertTrue(status["isLaserOn"]) + self.assertTrue(status["isShutterOpen"]) + + def testFailedCommandPostsCommandFailed(self): + controller, rec = self.make() + controller.connect() + rec.wait_for(DeviceControllerNotification.didConnect) + + def boom(device): + raise ValueError("nope") + + controller.submit(boom) + info = rec.wait_for(DeviceControllerNotification.commandFailed) + self.assertIsInstance(info, ValueError) + + def testCommandWhileDisconnectedFails(self): + controller, rec = self.make() + controller.submit(lambda device: device.turnOn()) + info = rec.wait_for(DeviceControllerNotification.commandFailed) + self.assertIsInstance(info, RuntimeError) + + # -- drop / reconnect -- + + def testConnectionLostThenAutoReconnect(self): + device = FlakyMillennia() + controller, rec = self.make(device=device) + controller.connect() + rec.wait_for(DeviceControllerNotification.didConnect) + + device.online = False + self.assertIsNotNone( + rec.wait_for(DeviceControllerNotification.connectionLost), + "should report the drop") + self.assertFalse(controller.is_connected) + + # Count reconnects so we can confirm a *new* one after recovery. + before = rec.names().count(DeviceControllerNotification.didConnect) + device.online = True + deadline = time.time() + 3.0 + while time.time() < deadline: + if (rec.names().count(DeviceControllerNotification.didConnect) > before + and controller.is_connected): + break + time.sleep(0.05) + self.assertTrue(controller.is_connected, "should have auto-reconnected") + + def testExplicitDisconnectDoesNotReconnect(self): + controller, rec = self.make() + controller.connect() + rec.wait_for(DeviceControllerNotification.didConnect) + controller.disconnect() + self.assertTrue(rec.wait_seen(DeviceControllerNotification.didDisconnect)) + time.sleep(0.8) # longer than a couple of reconnect intervals + self.assertFalse(controller.is_connected) + + def testNoAutoReconnectGivesUpAfterDrop(self): + device = FlakyMillennia() + controller, rec = self.make(device=device, auto_reconnect=False) + controller.connect() + rec.wait_for(DeviceControllerNotification.didConnect) + device.online = False + rec.wait_for(DeviceControllerNotification.connectionLost) + before = rec.names().count(DeviceControllerNotification.didConnect) + device.online = True + time.sleep(1.0) + self.assertEqual( + rec.names().count(DeviceControllerNotification.didConnect), before, + "must not reconnect when auto_reconnect is False") + + # -- misc -- + + def testDeviceWithoutStatusStillConnects(self): + controller, rec = self.make(device=NoStatusDevice()) + controller.connect() + self.assertIsNotNone(rec.wait_for(DeviceControllerNotification.didConnect)) + time.sleep(0.6) + self.assertNotIn(DeviceControllerNotification.status, rec.names()) + + def testConnectionErrorReasonClassifiesChain(self): + cause = OSError("[Errno 16] Resource busy: '/dev/cu.x'") + wrapper = RuntimeError("could not initialize") + wrapper.__cause__ = cause + self.assertEqual(connection_error_reason(wrapper), "busy") + self.assertIsNone(connection_error_reason(RuntimeError("weird"))) + + +if __name__ == "__main__": + unittest.main() From f9f22909e45f30558fa104e313df5422f6a1b6bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Sat, 27 Jun 2026 17:36:13 -0400 Subject: [PATCH 2/3] DeviceController.submit() returns a Future with the action's result MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Generality fix: submit() was fire-and-forget, which suits move/turnOn but is useless for query-style devices (powermeter/oscilloscope reads return data). It now returns a concurrent.futures.Future that resolves with the action's return value or its exception, so any PhysicalDevice — not just actuators — is fully usable: reading = controller.submit(lambda d: d.power()).result() commandFailed is still posted for observers that don't hold the Future. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01K6LVhfM3d9CcJdkuSMMqag --- hardwarelibrary/devicecontroller.py | 36 ++++++++++++------- hardwarelibrary/tests/testDeviceController.py | 14 ++++++-- 2 files changed, 36 insertions(+), 14 deletions(-) diff --git a/hardwarelibrary/devicecontroller.py b/hardwarelibrary/devicecontroller.py index 96667cd..0999fe3 100644 --- a/hardwarelibrary/devicecontroller.py +++ b/hardwarelibrary/devicecontroller.py @@ -37,13 +37,15 @@ def on_status(notification): NotificationCenter().addObserver(self, on_status, N.status) controller.start() controller.connect() - controller.submit(lambda device: device.turnOn()) + controller.submit(lambda device: device.turnOn()) # fire-and-forget + reading = controller.submit(lambda device: device.power()).result() # query ... controller.stop() """ import queue import time +from concurrent.futures import Future from enum import Enum from threading import Event, RLock, Thread @@ -152,13 +154,18 @@ def disconnect(self): self._queue.put(("disconnect", None)) def submit(self, action, name=None): - """Run ``action(device)`` on the worker thread. - - Returns immediately. On failure a commandFailed notification is posted - with the exception. A status poll is forced right after a successful - command so observers see the new state promptly. + """Run ``action(device)`` on the worker thread; return a Future. + + The Future resolves with the action's return value (so query-style + devices can read a measurement back: ``controller.submit(lambda d: + d.power()).result()``) or with its exception. On failure a commandFailed + notification is also posted, for observers that don't hold the Future. + A status poll is forced after a successful command so observers see the + new state promptly. Do not block on ``.result()`` from a UI thread. """ - self._queue.put(("command", (action, name))) + future = Future() + self._queue.put(("command", (action, name, future))) + return future def stop(self): """Stop the worker thread and shut the device down.""" @@ -258,19 +265,24 @@ def _poll(self): return self._post(DeviceControllerNotification.status, userInfo=info) - def _do_command(self, action, name): + def _do_command(self, action, name, future): + if not future.set_running_or_notify_cancel(): + return # the caller cancelled the Future before it ran if not self.is_connected: - self._post(DeviceControllerNotification.commandFailed, - userInfo=RuntimeError( - "Cannot run command {0!r}: not connected".format(name))) + error = RuntimeError( + "Cannot run command {0!r}: not connected".format(name)) + self._post(DeviceControllerNotification.commandFailed, userInfo=error) + future.set_exception(error) return try: - action(self.device) + result = action(self.device) except Exception as error: # A failed command may mean the link dropped; verify with a poll. self._post(DeviceControllerNotification.commandFailed, userInfo=error) + future.set_exception(error) self._poll() return + future.set_result(result) # Reflect the new state promptly. if self._supports_status: self._next_poll = time.monotonic() diff --git a/hardwarelibrary/tests/testDeviceController.py b/hardwarelibrary/tests/testDeviceController.py index 69d3362..0271880 100644 --- a/hardwarelibrary/tests/testDeviceController.py +++ b/hardwarelibrary/tests/testDeviceController.py @@ -131,7 +131,15 @@ def testSubmittedCommandRunsAndStatusReflectsIt(self): self.assertTrue(status["isLaserOn"]) self.assertTrue(status["isShutterOpen"]) - def testFailedCommandPostsCommandFailed(self): + def testSubmitReturnsResultViaFuture(self): + controller, rec = self.make() + controller.connect() + rec.wait_for(DeviceControllerNotification.didConnect) + controller.submit(lambda device: device.setPower(8.0)).result(timeout=3.0) + reading = controller.submit(lambda device: device.power()).result(timeout=3.0) + self.assertEqual(reading, 8.0) + + def testFailedCommandPostsCommandFailedAndRaisesInFuture(self): controller, rec = self.make() controller.connect() rec.wait_for(DeviceControllerNotification.didConnect) @@ -139,7 +147,9 @@ def testFailedCommandPostsCommandFailed(self): def boom(device): raise ValueError("nope") - controller.submit(boom) + future = controller.submit(boom) + with self.assertRaises(ValueError): + future.result(timeout=3.0) info = rec.wait_for(DeviceControllerNotification.commandFailed) self.assertIsInstance(info, ValueError) From 7949c55b182bb6905104d93f1ac54955a2e3f9e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Sat, 27 Jun 2026 17:52:55 -0400 Subject: [PATCH 3/3] Match library house style: camelCase, no underscore-prefixed members MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DeviceController was written in PEP 8 style (leading-underscore "privates", snake_case), which clashes with the rest of PyHardwareLibrary (camelCase, no underscore prefixing — cf. PhysicalDevice.startBackgroundStatusUpdates, self.refreshInterval, NotificationCenter.postNotification). Renamed throughout: isConnected/isRunning, pollInterval/reconnectInterval/autoReconnect, commandQueue/stopEvent/lock/thread/connected/wantConnected/supportsStatus/ nextPoll/nextReconnect/lastFailureSignature, and methods runLoop/handleTask/ attemptConnect/poll/doCommand/handleConnectionLost/doDisconnect/safeShutdown/ postFailureIfNew/teardown/post. Module helper connection_error_reason -> connectionErrorReason. No behaviour change; tests updated, 10/10 pass. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01K6LVhfM3d9CcJdkuSMMqag --- hardwarelibrary/devicecontroller.py | 238 +++++++++--------- hardwarelibrary/tests/testDeviceController.py | 24 +- 2 files changed, 131 insertions(+), 131 deletions(-) diff --git a/hardwarelibrary/devicecontroller.py b/hardwarelibrary/devicecontroller.py index 0999fe3..ec588b3 100644 --- a/hardwarelibrary/devicecontroller.py +++ b/hardwarelibrary/devicecontroller.py @@ -31,10 +31,10 @@ controller = DeviceController(MillenniaDevice(portPath="/dev/cu.usbmodem1")) - def on_status(notification): + def onStatus(notification): print(notification.userInfo) # {'power': .., 'isLaserOn': .., ...} - NotificationCenter().addObserver(self, on_status, N.status) + NotificationCenter().addObserver(self, onStatus, N.status) controller.start() controller.connect() controller.submit(lambda device: device.turnOn()) # fire-and-forget @@ -54,7 +54,7 @@ def on_status(notification): __all__ = [ "DeviceController", "DeviceControllerNotification", - "connection_error_reason", + "connectionErrorReason", ] @@ -77,7 +77,7 @@ class DeviceControllerNotification(Enum): commandFailed = "commandFailed" -def connection_error_reason(error): +def connectionErrorReason(error): """Classify a connection exception as 'busy'/'permission'/'missing'/None. Walks the exception's __cause__/__context__ chain (PhysicalDevice drivers @@ -109,49 +109,49 @@ class DeviceController: Args: device: any PhysicalDevice instance. - poll_interval: seconds between status polls while connected. The poll + pollInterval: seconds between status polls while connected. The poll calls device.doGetStatusUserInfo(); if the device has none (returns None), polling is disabled automatically. - reconnect_interval: seconds between reconnection attempts while + reconnectInterval: seconds between reconnection attempts while disconnected-but-wanted. - auto_reconnect: when True, an unexpected drop (or a connect that fails + autoReconnect: when True, an unexpected drop (or a connect that fails because the device is not yet present) is retried until it succeeds or disconnect() is called. When False, a single failure gives up. """ - def __init__(self, device, poll_interval=1.0, reconnect_interval=2.0, - auto_reconnect=True): + def __init__(self, device, pollInterval=1.0, reconnectInterval=2.0, + autoReconnect=True): self.device = device - self.poll_interval = poll_interval - self.reconnect_interval = reconnect_interval - self.auto_reconnect = auto_reconnect - - self._queue = queue.Queue() - self._stop = Event() - self._lock = RLock() - self._thread = Thread(target=self._run, name="DeviceController", - daemon=True) - - self._connected = False - self._want_connected = False - self._supports_status = True # set False if doGetStatusUserInfo()->None - self._next_poll = 0.0 - self._next_reconnect = 0.0 - self._last_failure_signature = None + self.pollInterval = pollInterval + self.reconnectInterval = reconnectInterval + self.autoReconnect = autoReconnect + + self.commandQueue = queue.Queue() + self.stopEvent = Event() + self.lock = RLock() + self.thread = Thread(target=self.runLoop, name="DeviceController", + daemon=True) + + self.connected = False + self.wantConnected = False + self.supportsStatus = True # set False if doGetStatusUserInfo()->None + self.nextPoll = 0.0 + self.nextReconnect = 0.0 + self.lastFailureSignature = None # -- public API (any thread) -- def start(self): """Start the worker thread.""" - self._thread.start() + self.thread.start() def connect(self): - """Request connection (and, with auto_reconnect, keep it connected).""" - self._queue.put(("connect", None)) + """Request connection (and, with autoReconnect, keep it connected).""" + self.commandQueue.put(("connect", None)) def disconnect(self): """Request disconnection; clears the reconnect intent.""" - self._queue.put(("disconnect", None)) + self.commandQueue.put(("disconnect", None)) def submit(self, action, name=None): """Run ``action(device)`` on the worker thread; return a Future. @@ -164,163 +164,163 @@ def submit(self, action, name=None): new state promptly. Do not block on ``.result()`` from a UI thread. """ future = Future() - self._queue.put(("command", (action, name, future))) + self.commandQueue.put(("command", (action, name, future))) return future def stop(self): """Stop the worker thread and shut the device down.""" - self._stop.set() - self._queue.put(("disconnect", None)) - if self._thread.is_alive(): - self._thread.join(timeout=5.0) + self.stopEvent.set() + self.commandQueue.put(("disconnect", None)) + if self.thread.is_alive(): + self.thread.join(timeout=5.0) @property - def is_connected(self): - with self._lock: - return self._connected + def isConnected(self): + with self.lock: + return self.connected @property - def is_running(self): - return self._thread.is_alive() + def isRunning(self): + return self.thread.is_alive() # -- worker thread -- - def _post(self, name, userInfo=None): + def post(self, name, userInfo=None): NotificationCenter().postNotification(name, notifyingObject=self, userInfo=userInfo) - def _run(self): - while not self._stop.is_set(): + def runLoop(self): + while not self.stopEvent.is_set(): try: - task = self._queue.get(timeout=self._seconds_until_due()) + task = self.commandQueue.get(timeout=self.secondsUntilDue()) except queue.Empty: task = None if task is not None: - self._handle(task) + self.handleTask(task) now = time.monotonic() - if self._connected and self._supports_status and now >= self._next_poll: - self._next_poll = now + self.poll_interval - self._poll() - elif (not self._connected and self._want_connected - and now >= self._next_reconnect): - self._next_reconnect = now + self.reconnect_interval - self._attempt_connect() + if self.connected and self.supportsStatus and now >= self.nextPoll: + self.nextPoll = now + self.pollInterval + self.poll() + elif (not self.connected and self.wantConnected + and now >= self.nextReconnect): + self.nextReconnect = now + self.reconnectInterval + self.attemptConnect() - self._teardown() + self.teardown() - def _seconds_until_due(self): + def secondsUntilDue(self): now = time.monotonic() - if self._connected: - target = self._next_poll if self._supports_status else now + 0.5 - elif self._want_connected: - target = self._next_reconnect + if self.connected: + target = self.nextPoll if self.supportsStatus else now + 0.5 + elif self.wantConnected: + target = self.nextReconnect else: return None # nothing scheduled; block until a task arrives return max(0.0, target - now) - def _handle(self, task): + def handleTask(self, task): kind, payload = task if kind == "connect": - self._want_connected = True - self._next_reconnect = 0.0 - self._attempt_connect() + self.wantConnected = True + self.nextReconnect = 0.0 + self.attemptConnect() elif kind == "disconnect": - self._want_connected = False - self._do_disconnect() + self.wantConnected = False + self.doDisconnect() elif kind == "command": - self._do_command(*payload) + self.doCommand(*payload) - def _attempt_connect(self): - with self._lock: - if self._connected: + def attemptConnect(self): + with self.lock: + if self.connected: return try: self.device.initializeDevice() except Exception as error: - self._connected = False - self._post_failure_if_new( + self.connected = False + self.postFailureIfNew( DeviceControllerNotification.connectionFailed, error) - if not self.auto_reconnect: - self._want_connected = False + if not self.autoReconnect: + self.wantConnected = False return - with self._lock: - self._connected = True - self._supports_status = True - self._next_poll = time.monotonic() # poll immediately - self._last_failure_signature = None - self._post(DeviceControllerNotification.didConnect, userInfo=self.device) + with self.lock: + self.connected = True + self.supportsStatus = True + self.nextPoll = time.monotonic() # poll immediately + self.lastFailureSignature = None + self.post(DeviceControllerNotification.didConnect, userInfo=self.device) - def _poll(self): + def poll(self): try: info = self.device.doGetStatusUserInfo() except Exception as error: - self._handle_lost(error) + self.handleConnectionLost(error) return if info is None: # Device exposes no status snapshot; stop polling for it. - self._supports_status = False + self.supportsStatus = False return - self._post(DeviceControllerNotification.status, userInfo=info) + self.post(DeviceControllerNotification.status, userInfo=info) - def _do_command(self, action, name, future): + def doCommand(self, action, name, future): if not future.set_running_or_notify_cancel(): return # the caller cancelled the Future before it ran - if not self.is_connected: + if not self.isConnected: error = RuntimeError( "Cannot run command {0!r}: not connected".format(name)) - self._post(DeviceControllerNotification.commandFailed, userInfo=error) + self.post(DeviceControllerNotification.commandFailed, userInfo=error) future.set_exception(error) return try: result = action(self.device) except Exception as error: # A failed command may mean the link dropped; verify with a poll. - self._post(DeviceControllerNotification.commandFailed, userInfo=error) + self.post(DeviceControllerNotification.commandFailed, userInfo=error) future.set_exception(error) - self._poll() + self.poll() return future.set_result(result) # Reflect the new state promptly. - if self._supports_status: - self._next_poll = time.monotonic() - self._poll() - - def _handle_lost(self, error): - was_connected = self._connected - with self._lock: - self._connected = False - self._safe_shutdown() - self._next_reconnect = time.monotonic() + self.reconnect_interval - if not self.auto_reconnect: - self._want_connected = False - if was_connected: - self._post(DeviceControllerNotification.connectionLost, userInfo=error) - - def _do_disconnect(self): - with self._lock: - self._connected = False - self._safe_shutdown() - self._post(DeviceControllerNotification.didDisconnect) - - def _safe_shutdown(self): + if self.supportsStatus: + self.nextPoll = time.monotonic() + self.poll() + + def handleConnectionLost(self, error): + wasConnected = self.connected + with self.lock: + self.connected = False + self.safeShutdown() + self.nextReconnect = time.monotonic() + self.reconnectInterval + if not self.autoReconnect: + self.wantConnected = False + if wasConnected: + self.post(DeviceControllerNotification.connectionLost, userInfo=error) + + def doDisconnect(self): + with self.lock: + self.connected = False + self.safeShutdown() + self.post(DeviceControllerNotification.didDisconnect) + + def safeShutdown(self): try: self.device.shutdownDevice() except Exception: pass - def _post_failure_if_new(self, name, error): + def postFailureIfNew(self, name, error): # Avoid flooding identical connectionFailed notifications every interval # while a device stays unavailable; re-post only when the reason changes. - signature = (type(error).__name__, connection_error_reason(error) or str(error)) - if signature != self._last_failure_signature: - self._last_failure_signature = signature - self._post(name, userInfo=error) - - def _teardown(self): - if self._connected: - with self._lock: - self._connected = False - self._safe_shutdown() + signature = (type(error).__name__, connectionErrorReason(error) or str(error)) + if signature != self.lastFailureSignature: + self.lastFailureSignature = signature + self.post(name, userInfo=error) + + def teardown(self): + if self.connected: + with self.lock: + self.connected = False + self.safeShutdown() diff --git a/hardwarelibrary/tests/testDeviceController.py b/hardwarelibrary/tests/testDeviceController.py index 0271880..ab5874f 100644 --- a/hardwarelibrary/tests/testDeviceController.py +++ b/hardwarelibrary/tests/testDeviceController.py @@ -4,7 +4,7 @@ from threading import Lock from hardwarelibrary.devicecontroller import ( - DeviceController, DeviceControllerNotification, connection_error_reason) + DeviceController, DeviceControllerNotification, connectionErrorReason) from hardwarelibrary.notificationcenter import NotificationCenter from hardwarelibrary.sources.millennia import DebugMillenniaDevice @@ -96,8 +96,8 @@ def tearDown(self): def make(self, device=None, **kwargs): device = device or DebugMillenniaDevice() - kwargs.setdefault("poll_interval", 0.2) - kwargs.setdefault("reconnect_interval", 0.3) + kwargs.setdefault("pollInterval", 0.2) + kwargs.setdefault("reconnectInterval", 0.3) controller = DeviceController(device, **kwargs) recorder = Recorder(controller) self.controllers.append(controller) @@ -114,7 +114,7 @@ def testConnectPostsDidConnectAndStatus(self): status = rec.wait_for(DeviceControllerNotification.status) self.assertIsNotNone(status) self.assertEqual(set(status), {"power", "isLaserOn", "isShutterOpen"}) - self.assertTrue(controller.is_connected) + self.assertTrue(controller.isConnected) def testSubmittedCommandRunsAndStatusReflectsIt(self): controller, rec = self.make() @@ -171,7 +171,7 @@ def testConnectionLostThenAutoReconnect(self): self.assertIsNotNone( rec.wait_for(DeviceControllerNotification.connectionLost), "should report the drop") - self.assertFalse(controller.is_connected) + self.assertFalse(controller.isConnected) # Count reconnects so we can confirm a *new* one after recovery. before = rec.names().count(DeviceControllerNotification.didConnect) @@ -179,10 +179,10 @@ def testConnectionLostThenAutoReconnect(self): deadline = time.time() + 3.0 while time.time() < deadline: if (rec.names().count(DeviceControllerNotification.didConnect) > before - and controller.is_connected): + and controller.isConnected): break time.sleep(0.05) - self.assertTrue(controller.is_connected, "should have auto-reconnected") + self.assertTrue(controller.isConnected, "should have auto-reconnected") def testExplicitDisconnectDoesNotReconnect(self): controller, rec = self.make() @@ -191,11 +191,11 @@ def testExplicitDisconnectDoesNotReconnect(self): controller.disconnect() self.assertTrue(rec.wait_seen(DeviceControllerNotification.didDisconnect)) time.sleep(0.8) # longer than a couple of reconnect intervals - self.assertFalse(controller.is_connected) + self.assertFalse(controller.isConnected) def testNoAutoReconnectGivesUpAfterDrop(self): device = FlakyMillennia() - controller, rec = self.make(device=device, auto_reconnect=False) + controller, rec = self.make(device=device, autoReconnect=False) controller.connect() rec.wait_for(DeviceControllerNotification.didConnect) device.online = False @@ -205,7 +205,7 @@ def testNoAutoReconnectGivesUpAfterDrop(self): time.sleep(1.0) self.assertEqual( rec.names().count(DeviceControllerNotification.didConnect), before, - "must not reconnect when auto_reconnect is False") + "must not reconnect when autoReconnect is False") # -- misc -- @@ -220,8 +220,8 @@ def testConnectionErrorReasonClassifiesChain(self): cause = OSError("[Errno 16] Resource busy: '/dev/cu.x'") wrapper = RuntimeError("could not initialize") wrapper.__cause__ = cause - self.assertEqual(connection_error_reason(wrapper), "busy") - self.assertIsNone(connection_error_reason(RuntimeError("weird"))) + self.assertEqual(connectionErrorReason(wrapper), "busy") + self.assertIsNone(connectionErrorReason(RuntimeError("weird"))) if __name__ == "__main__":