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..ec588b3 --- /dev/null +++ b/hardwarelibrary/devicecontroller.py @@ -0,0 +1,326 @@ +"""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 onStatus(notification): + print(notification.userInfo) # {'power': .., 'isLaserOn': .., ...} + + NotificationCenter().addObserver(self, onStatus, N.status) + controller.start() + controller.connect() + 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 + +from hardwarelibrary.notificationcenter import NotificationCenter + +__all__ = [ + "DeviceController", + "DeviceControllerNotification", + "connectionErrorReason", +] + + +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 connectionErrorReason(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. + pollInterval: seconds between status polls while connected. The poll + calls device.doGetStatusUserInfo(); if the device has none (returns + None), polling is disabled automatically. + reconnectInterval: seconds between reconnection attempts while + disconnected-but-wanted. + 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, pollInterval=1.0, reconnectInterval=2.0, + autoReconnect=True): + self.device = device + 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() + + def connect(self): + """Request connection (and, with autoReconnect, keep it connected).""" + self.commandQueue.put(("connect", None)) + + def disconnect(self): + """Request disconnection; clears the reconnect intent.""" + self.commandQueue.put(("disconnect", None)) + + def submit(self, action, name=None): + """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. + """ + future = Future() + self.commandQueue.put(("command", (action, name, future))) + return future + + def stop(self): + """Stop the worker thread and shut the device down.""" + self.stopEvent.set() + self.commandQueue.put(("disconnect", None)) + if self.thread.is_alive(): + self.thread.join(timeout=5.0) + + @property + def isConnected(self): + with self.lock: + return self.connected + + @property + def isRunning(self): + return self.thread.is_alive() + + # -- worker thread -- + + def post(self, name, userInfo=None): + NotificationCenter().postNotification(name, notifyingObject=self, + userInfo=userInfo) + + def runLoop(self): + while not self.stopEvent.is_set(): + try: + task = self.commandQueue.get(timeout=self.secondsUntilDue()) + except queue.Empty: + task = None + + if task is not None: + self.handleTask(task) + + now = time.monotonic() + 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() + + def secondsUntilDue(self): + now = time.monotonic() + 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 handleTask(self, task): + kind, payload = task + if kind == "connect": + self.wantConnected = True + self.nextReconnect = 0.0 + self.attemptConnect() + elif kind == "disconnect": + self.wantConnected = False + self.doDisconnect() + elif kind == "command": + self.doCommand(*payload) + + def attemptConnect(self): + with self.lock: + if self.connected: + return + try: + self.device.initializeDevice() + except Exception as error: + self.connected = False + self.postFailureIfNew( + DeviceControllerNotification.connectionFailed, error) + if not self.autoReconnect: + self.wantConnected = False + return + + 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): + try: + info = self.device.doGetStatusUserInfo() + except Exception as error: + self.handleConnectionLost(error) + return + if info is None: + # Device exposes no status snapshot; stop polling for it. + self.supportsStatus = False + return + self.post(DeviceControllerNotification.status, userInfo=info) + + 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.isConnected: + error = RuntimeError( + "Cannot run command {0!r}: not connected".format(name)) + 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) + future.set_exception(error) + self.poll() + return + future.set_result(result) + # Reflect the new state promptly. + 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 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__, 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 new file mode 100644 index 0000000..ab5874f --- /dev/null +++ b/hardwarelibrary/tests/testDeviceController.py @@ -0,0 +1,228 @@ +import env # noqa: F401 (sets sys.path) +import time +import unittest +from threading import Lock + +from hardwarelibrary.devicecontroller import ( + DeviceController, DeviceControllerNotification, connectionErrorReason) +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("pollInterval", 0.2) + kwargs.setdefault("reconnectInterval", 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.isConnected) + + 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 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) + + def boom(device): + raise ValueError("nope") + + future = controller.submit(boom) + with self.assertRaises(ValueError): + future.result(timeout=3.0) + 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.isConnected) + + # 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.isConnected): + break + time.sleep(0.05) + self.assertTrue(controller.isConnected, "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.isConnected) + + def testNoAutoReconnectGivesUpAfterDrop(self): + device = FlakyMillennia() + controller, rec = self.make(device=device, autoReconnect=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 autoReconnect 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(connectionErrorReason(wrapper), "busy") + self.assertIsNone(connectionErrorReason(RuntimeError("weird"))) + + +if __name__ == "__main__": + unittest.main()