From e9e4faff2e33ff0d72ae5fe267bf01e4fd1a6bc1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Sun, 1 Mar 2026 16:15:35 -0500 Subject: [PATCH 01/17] Fix DebugPort bugs and add multi-endpoint support Fix flush() writing to nonexistent attribute, fix endPointIndex unbound when endPoint is not None, add numberOfEndPoints parameter for USB devices, add defaultTimeout attribute, remove broken DebugDataCommand. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/communication/debugport.py | 63 +++++++--------------- 1 file changed, 20 insertions(+), 43 deletions(-) diff --git a/hardwarelibrary/communication/debugport.py b/hardwarelibrary/communication/debugport.py index b7ebe45f..e13545e1 100644 --- a/hardwarelibrary/communication/debugport.py +++ b/hardwarelibrary/communication/debugport.py @@ -4,79 +4,57 @@ from hardwarelibrary.communication import * from threading import Thread, Lock -class DebugDataCommand(Command): - def __init__(self, name, dataHexRegex = None, unpackingMask = None, endPoints = (None, None)): - Command.__init__(self, name, endPoints=endPoints) - self.data : bytearray = data - self.dataHexRegex: str = dataHexRegex - self.unpackingMask:str = unpackingMask - - def send(self, port) -> bool: - try: - self.isSent = True - nBytes = port.writeData(data=self.data, endPoint=self.endPoints[0]) - if self.replyDataLength > 0: - self.reply = port.readData(length=self.replyDataLength) - elif self.replyHexRegex is not None: - raise NotImplementedError("DataCommand reply pattern not implemented") - # self.reply = port.readData(length=self.replyDataLength) - self.isSentSuccessfully = True - except Exception as err: - - self.exceptions.append(err) - self.isSentSuccessfully = False - raise(err) - - return False - class DebugPort(CommunicationPort): - def __init__(self, delay=0): - self.inputBuffers = [bytearray()] - self.outputBuffers = [bytearray()] + def __init__(self, delay=0, numberOfEndPoints=1): + self.inputBuffers = [bytearray() for _ in range(numberOfEndPoints)] + self.outputBuffers = [bytearray() for _ in range(numberOfEndPoints)] self.delay = delay + self.defaultTimeout = 500 self._isOpen = False super(DebugPort, self).__init__() @property def isOpen(self): - return self._isOpen + return self._isOpen def open(self): if self._isOpen: - raise Exception() + raise Exception("Port already open") self._isOpen = True - return def close(self): self._isOpen = False - return def bytesAvailable(self, endPoint=0): - return len(self.outputBuffers[endPoint]) + endPointIndex = endPoint if endPoint is not None else 0 + return len(self.outputBuffers[endPointIndex]) def flush(self): - self.buffers = [bytearray(),bytearray()] + for i in range(len(self.inputBuffers)): + self.inputBuffers[i] = bytearray() + for i in range(len(self.outputBuffers)): + self.outputBuffers[i] = bytearray() def readData(self, length, endPoint=None): - if endPoint is None: - endPointIndex = 0 + endPointIndex = endPoint if endPoint is not None else 0 with self.portLock: - time.sleep(self.delay*random.random()) + if self.delay > 0: + time.sleep(self.delay * random.random()) + data = bytearray() - for i in range(0, length): + for i in range(length): if len(self.outputBuffers[endPointIndex]) > 0: byte = self.outputBuffers[endPointIndex].pop(0) data.append(byte) else: - raise CommunicationReadTimeout("Unable to read data") + raise CommunicationReadTimeout("Unable to read {0} bytes, only {1} available".format(length, len(data))) return data def writeData(self, data, endPoint=None): - if endPoint is None: - endPointIndex = 0 + endPointIndex = endPoint if endPoint is not None else 0 with self.portLock: self.inputBuffers[endPointIndex].extend(data) @@ -85,12 +63,11 @@ def writeData(self, data, endPoint=None): return len(data) - def writeToOutputBuffer(self, data, endPointIndex): + def writeToOutputBuffer(self, data, endPointIndex=0): self.outputBuffers[endPointIndex].extend(data) def processInputBuffers(self, endPointIndex): # We default to ECHO for simplicity - inputBytes = self.inputBuffers[endPointIndex] # Do something, here we do an Echo From e6e88adb1fa6a526cb3beddbbd8f2b0c2b7c1545 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Sun, 1 Mar 2026 16:40:00 -0500 Subject: [PATCH 02/17] Add TableDrivenDebugPort for declarative mock device creation Replace manual if/elif byte-parsing in debug ports with a table-driven approach: declare command entries (binary or text) and implement only device logic in process_command(). Migrate Sutter's DebugSerialPort, create Intellidrive's missing DebugSerialPort, and add 12 new tests. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/communication/__init__.py | 2 +- hardwarelibrary/communication/debugport.py | 85 +++++++- hardwarelibrary/motion/intellidrivedevice.py | 28 ++- hardwarelibrary/motion/sutterdevice.py | 41 ++-- .../tests/testTableDrivenDebugPort.py | 198 ++++++++++++++++++ 5 files changed, 327 insertions(+), 27 deletions(-) create mode 100644 hardwarelibrary/tests/testTableDrivenDebugPort.py diff --git a/hardwarelibrary/communication/__init__.py b/hardwarelibrary/communication/__init__.py index bc0034e1..72a9f894 100644 --- a/hardwarelibrary/communication/__init__.py +++ b/hardwarelibrary/communication/__init__.py @@ -2,7 +2,7 @@ from .serialport import SerialPort from .usbport import USBPort from .diagnostics import USBParameters, DeviceCommand, USBDeviceDescription -from .debugport import DebugPort +from .debugport import DebugPort, TableDrivenDebugPort, BinaryCommandEntry, TextCommandEntry from .echoport import DebugEchoPort import usb.backend.libusb1 import platform diff --git a/hardwarelibrary/communication/debugport.py b/hardwarelibrary/communication/debugport.py index e13545e1..9391645d 100644 --- a/hardwarelibrary/communication/debugport.py +++ b/hardwarelibrary/communication/debugport.py @@ -1,9 +1,28 @@ import time import random +import struct +import re +from dataclasses import dataclass from hardwarelibrary.communication import * from threading import Thread, Lock + +@dataclass +class BinaryCommandEntry: + name: str + prefix: bytes + request_format: str = None + request_length: int = None + response_format: str = None + + +@dataclass +class TextCommandEntry: + name: str + pattern: str + response: str = None + class DebugPort(CommunicationPort): def __init__(self, delay=0, numberOfEndPoints=1): self.inputBuffers = [bytearray() for _ in range(numberOfEndPoints)] @@ -72,4 +91,68 @@ def processInputBuffers(self, endPointIndex): # Do something, here we do an Echo self.writeToOutputBuffer(inputBytes, endPointIndex) - self.inputBuffers[endPointIndex] = bytearray() \ No newline at end of file + self.inputBuffers[endPointIndex] = bytearray() + + +class TableDrivenDebugPort(DebugPort): + def __init__(self, delay=0, numberOfEndPoints=1): + super().__init__(delay=delay, numberOfEndPoints=numberOfEndPoints) + self.binary_commands = [] + self.text_commands = [] + + def processInputBuffers(self, endPointIndex): + inputBytes = self.inputBuffers[endPointIndex] + if len(inputBytes) == 0: + return + + for cmd in self.binary_commands: + prefix_len = len(cmd.prefix) + if inputBytes[:prefix_len].upper() == cmd.prefix.upper(): + if cmd.request_format is not None: + unpack_len = struct.calcsize(cmd.request_format) + params = struct.unpack(cmd.request_format, bytes(inputBytes[:unpack_len])) + else: + params = () + + result = self.process_command(cmd.name, params, endPointIndex) + self._write_binary_response(result, cmd, endPointIndex) + self.inputBuffers[endPointIndex] = bytearray() + return + + inputStr = inputBytes.decode('utf-8', errors='replace') + for cmd in self.text_commands: + match = re.match(cmd.pattern, inputStr) + if match: + params = match.groups() + result = self.process_command(cmd.name, params, endPointIndex) + self._write_text_response(result, cmd, endPointIndex) + self.inputBuffers[endPointIndex] = bytearray() + return + + print("Unrecognized command: {0}".format(inputBytes)) + self.inputBuffers[endPointIndex] = bytearray() + + def _write_binary_response(self, result, cmd, endPointIndex): + if result is None: + return + elif isinstance(result, (bytes, bytearray)): + self.writeToOutputBuffer(bytearray(result), endPointIndex) + elif isinstance(result, str): + self.writeToOutputBuffer(bytearray(result.encode('utf-8')), endPointIndex) + elif isinstance(result, tuple) and cmd.response_format is not None: + data = struct.pack(cmd.response_format, *result) + self.writeToOutputBuffer(bytearray(data), endPointIndex) + + def _write_text_response(self, result, cmd, endPointIndex): + if result is None: + return + elif isinstance(result, (bytes, bytearray)): + self.writeToOutputBuffer(bytearray(result), endPointIndex) + elif isinstance(result, str): + self.writeToOutputBuffer(bytearray(result.encode('utf-8')), endPointIndex) + elif isinstance(result, tuple) and cmd.response is not None: + formatted = cmd.response.format(*result) + self.writeToOutputBuffer(bytearray(formatted.encode('utf-8')), endPointIndex) + + def process_command(self, name, params, endPointIndex): + raise NotImplementedError("Subclasses must implement process_command") \ No newline at end of file diff --git a/hardwarelibrary/motion/intellidrivedevice.py b/hardwarelibrary/motion/intellidrivedevice.py index d94a5b91..aa76f72c 100644 --- a/hardwarelibrary/motion/intellidrivedevice.py +++ b/hardwarelibrary/motion/intellidrivedevice.py @@ -1,7 +1,7 @@ from hardwarelibrary.motion.rotationdevice import * from hardwarelibrary.communication.serialport import SerialPort from hardwarelibrary.communication.commands import DataCommand -from hardwarelibrary.communication.debugport import DebugPort +from hardwarelibrary.communication.debugport import TableDrivenDebugPort, TextCommandEntry import time from struct import * @@ -117,3 +117,29 @@ def doHome(self): if not self.isReferenced(): raise Exception("Homing failed") + + class DebugSerialPort(TableDrivenDebugPort): + def __init__(self): + super().__init__() + self.registers = {'0xc9': 0} + self.text_commands = [ + TextCommandEntry(name='set_register', pattern=r's r(0x[0-9a-fA-F]+) (-?\d+)[\r\n]'), + TextCommandEntry(name='get_register', pattern=r'g r(0x[0-9a-fA-F]+)[\r\n]'), + TextCommandEntry(name='trajectory', pattern=r't (\d+)[\r\n]'), + ] + + def process_command(self, name, params, endPointIndex): + if name == 'set_register': + register, value = params + self.registers[register] = int(value) + return "ok\r" + elif name == 'get_register': + register = params[0] + value = self.registers.get(register, 0) + return "v {}\r".format(value) + elif name == 'trajectory': + mode = int(params[0]) + if mode == 2: # home + self.registers['0xca'] = 0 + self.registers['0xc9'] = (1 << 12) # referenced bit + return "ok\r" diff --git a/hardwarelibrary/motion/sutterdevice.py b/hardwarelibrary/motion/sutterdevice.py index 5625bf13..556ad12d 100644 --- a/hardwarelibrary/motion/sutterdevice.py +++ b/hardwarelibrary/motion/sutterdevice.py @@ -4,7 +4,7 @@ from hardwarelibrary.communication.usbport import USBPort from hardwarelibrary.communication.serialport import SerialPort from hardwarelibrary.communication.commands import DataCommand -from hardwarelibrary.communication.debugport import DebugPort +from hardwarelibrary.communication.debugport import TableDrivenDebugPort, BinaryCommandEntry import re import time @@ -143,31 +143,24 @@ def work(self): raise Exception(f"Expected carriage return, but got {replyBytes} instead.") - class DebugSerialPort(DebugPort): + class DebugSerialPort(TableDrivenDebugPort): def __init__(self): super().__init__() self.xSteps = 0 self.ySteps = 0 self.zSteps = 0 - - def processInputBuffers(self, endPointIndex): - inputBytes = self.inputBuffers[endPointIndex] - - if inputBytes[0] == b'm'[0] or inputBytes[0] == b'M'[0]: - x,y,z = unpack(" Date: Sun, 1 Mar 2026 17:18:01 -0500 Subject: [PATCH 03/17] Unify command recognition: Command objects now serve both send and mock sides Add matches(), extractParams(), formatResponse() to Command/TextCommand/DataCommand so the same objects define both the client protocol and debug port recognition. Remove BinaryCommandEntry/TextCommandEntry dataclasses from debugport.py, replacing them with a single polymorphic loop over Command objects in TableDrivenDebugPort. Update SutterDevice and IntellidriveDevice to use class-level commands dicts. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/communication/__init__.py | 2 +- hardwarelibrary/communication/commands.py | 99 ++++++++++++- hardwarelibrary/communication/debugport.py | 70 +-------- hardwarelibrary/motion/intellidrivedevice.py | 31 ++-- hardwarelibrary/motion/sutterdevice.py | 25 ++-- .../tests/testCommandRecognition.py | 139 ++++++++++++++++++ .../tests/testTableDrivenDebugPort.py | 46 +++--- 7 files changed, 297 insertions(+), 115 deletions(-) create mode 100644 hardwarelibrary/tests/testCommandRecognition.py diff --git a/hardwarelibrary/communication/__init__.py b/hardwarelibrary/communication/__init__.py index 72a9f894..69ef7128 100644 --- a/hardwarelibrary/communication/__init__.py +++ b/hardwarelibrary/communication/__init__.py @@ -2,7 +2,7 @@ from .serialport import SerialPort from .usbport import USBPort from .diagnostics import USBParameters, DeviceCommand, USBDeviceDescription -from .debugport import DebugPort, TableDrivenDebugPort, BinaryCommandEntry, TextCommandEntry +from .debugport import DebugPort, TableDrivenDebugPort from .echoport import DebugEchoPort import usb.backend.libusb1 import platform diff --git a/hardwarelibrary/communication/commands.py b/hardwarelibrary/communication/commands.py index 99cd7290..3d6d257f 100644 --- a/hardwarelibrary/communication/commands.py +++ b/hardwarelibrary/communication/commands.py @@ -1,4 +1,5 @@ import re +import struct class Command: def __init__(self, name:str, endPoints = (None, None)): @@ -26,7 +27,7 @@ def matchAsFloat(self, index=0): if self.matchGroups is not None: return float(self.matchGroups[index]) return None - + @property def hasError(self): return len(self.exceptions) != 0 @@ -34,14 +35,33 @@ def hasError(self): def send(self, port) -> bool: raise NotImplementedError("Subclasses must implement the send() command") + def matches(self, inputBytes): + return False + + def extractParams(self, inputBytes): + return () + + def formatResponse(self, result): + if result is None: + return None + elif isinstance(result, (bytes, bytearray)): + return bytearray(result) + elif isinstance(result, str): + return bytearray(result.encode('utf-8')) + return None + class TextCommand(Command): - def __init__(self, name, text, replyPattern = None, - alternatePattern = None, - endPoints = (None, None)): + def __init__(self, name, text, replyPattern = None, + alternatePattern = None, + endPoints = (None, None), + matchPattern = None, + responseTemplate = None): Command.__init__(self, name, endPoints=endPoints) self.text : str = text self.replyPattern: str = replyPattern self.alternatePattern: str = alternatePattern + self.matchPattern: str = matchPattern + self.responseTemplate: str = responseTemplate @property def payload(self): @@ -54,6 +74,41 @@ def numberOfArguments(self): return 0 return len(match.groups()) + @property + def _autoMatchPattern(self): + parts = re.split(r'\{[^}]*\}', self.text) + escaped = [re.escape(p) for p in parts] + return '(.+?)'.join(escaped) + + @property + def effectiveMatchPattern(self): + if self.matchPattern is not None: + return self.matchPattern + return self._autoMatchPattern + + def matches(self, inputBytes): + try: + inputStr = inputBytes.decode('utf-8', errors='replace') + except Exception: + return False + return re.match(self.effectiveMatchPattern, inputStr) is not None + + def extractParams(self, inputBytes): + try: + inputStr = inputBytes.decode('utf-8', errors='replace') + except Exception: + return () + match = re.match(self.effectiveMatchPattern, inputStr) + if match: + return match.groups() + return () + + def formatResponse(self, result): + if isinstance(result, tuple) and self.responseTemplate is not None: + formatted = self.responseTemplate.format(*result) + return bytearray(formatted.encode('utf-8')) + return super().formatResponse(result) + def send(self, port, params=None) -> bool: try: if params is not None: @@ -154,17 +209,51 @@ def send(self, port, params=None) -> bool: class DataCommand(Command): - def __init__(self, name, data, replyHexRegex = None, replyDataLength = 0, unpackingMask = None, endPoints = (None, None)): + def __init__(self, name, data=None, replyHexRegex = None, replyDataLength = 0, unpackingMask = None, endPoints = (None, None), + prefix = None, requestFormat = None, responseFormat = None): Command.__init__(self, name, endPoints=endPoints) self.data : bytearray = data self.replyHexRegex: str = replyHexRegex self.replyDataLength: int = replyDataLength self.unpackingMask:str = unpackingMask + self._prefix: bytes = prefix + self.requestFormat: str = requestFormat + self.responseFormat: str = responseFormat @property def payload(self): return self.data + @property + def effectivePrefix(self): + if self._prefix is not None: + return self._prefix + if self.data is not None and len(self.data) > 0: + return self.data[0:1] + return None + + def matches(self, inputBytes): + prefix = self.effectivePrefix + if prefix is None: + return False + prefixLen = len(prefix) + if len(inputBytes) < prefixLen: + return False + return inputBytes[:prefixLen].upper() == prefix.upper() + + def extractParams(self, inputBytes): + if self.requestFormat is not None: + unpackLen = struct.calcsize(self.requestFormat) + if len(inputBytes) >= unpackLen: + return struct.unpack(self.requestFormat, bytes(inputBytes[:unpackLen])) + return () + + def formatResponse(self, result): + if isinstance(result, tuple) and self.responseFormat is not None: + data = struct.pack(self.responseFormat, *result) + return bytearray(data) + return super().formatResponse(result) + def send(self, port) -> bool: try: nBytes = port.writeData(data=self.data, endPoint=self.endPoints[0]) diff --git a/hardwarelibrary/communication/debugport.py b/hardwarelibrary/communication/debugport.py index 9391645d..60361329 100644 --- a/hardwarelibrary/communication/debugport.py +++ b/hardwarelibrary/communication/debugport.py @@ -2,27 +2,10 @@ import random import struct import re -from dataclasses import dataclass from hardwarelibrary.communication import * from threading import Thread, Lock - -@dataclass -class BinaryCommandEntry: - name: str - prefix: bytes - request_format: str = None - request_length: int = None - response_format: str = None - - -@dataclass -class TextCommandEntry: - name: str - pattern: str - response: str = None - class DebugPort(CommunicationPort): def __init__(self, delay=0, numberOfEndPoints=1): self.inputBuffers = [bytearray() for _ in range(numberOfEndPoints)] @@ -95,64 +78,27 @@ def processInputBuffers(self, endPointIndex): class TableDrivenDebugPort(DebugPort): - def __init__(self, delay=0, numberOfEndPoints=1): + def __init__(self, delay=0, numberOfEndPoints=1, commands=None): super().__init__(delay=delay, numberOfEndPoints=numberOfEndPoints) - self.binary_commands = [] - self.text_commands = [] + self.commands = commands if commands is not None else {} def processInputBuffers(self, endPointIndex): inputBytes = self.inputBuffers[endPointIndex] if len(inputBytes) == 0: return - for cmd in self.binary_commands: - prefix_len = len(cmd.prefix) - if inputBytes[:prefix_len].upper() == cmd.prefix.upper(): - if cmd.request_format is not None: - unpack_len = struct.calcsize(cmd.request_format) - params = struct.unpack(cmd.request_format, bytes(inputBytes[:unpack_len])) - else: - params = () - + for cmd in self.commands.values(): + if cmd.matches(inputBytes): + params = cmd.extractParams(inputBytes) result = self.process_command(cmd.name, params, endPointIndex) - self._write_binary_response(result, cmd, endPointIndex) - self.inputBuffers[endPointIndex] = bytearray() - return - - inputStr = inputBytes.decode('utf-8', errors='replace') - for cmd in self.text_commands: - match = re.match(cmd.pattern, inputStr) - if match: - params = match.groups() - result = self.process_command(cmd.name, params, endPointIndex) - self._write_text_response(result, cmd, endPointIndex) + response = cmd.formatResponse(result) + if response is not None: + self.writeToOutputBuffer(response, endPointIndex) self.inputBuffers[endPointIndex] = bytearray() return print("Unrecognized command: {0}".format(inputBytes)) self.inputBuffers[endPointIndex] = bytearray() - def _write_binary_response(self, result, cmd, endPointIndex): - if result is None: - return - elif isinstance(result, (bytes, bytearray)): - self.writeToOutputBuffer(bytearray(result), endPointIndex) - elif isinstance(result, str): - self.writeToOutputBuffer(bytearray(result.encode('utf-8')), endPointIndex) - elif isinstance(result, tuple) and cmd.response_format is not None: - data = struct.pack(cmd.response_format, *result) - self.writeToOutputBuffer(bytearray(data), endPointIndex) - - def _write_text_response(self, result, cmd, endPointIndex): - if result is None: - return - elif isinstance(result, (bytes, bytearray)): - self.writeToOutputBuffer(bytearray(result), endPointIndex) - elif isinstance(result, str): - self.writeToOutputBuffer(bytearray(result.encode('utf-8')), endPointIndex) - elif isinstance(result, tuple) and cmd.response is not None: - formatted = cmd.response.format(*result) - self.writeToOutputBuffer(bytearray(formatted.encode('utf-8')), endPointIndex) - def process_command(self, name, params, endPointIndex): raise NotImplementedError("Subclasses must implement process_command") \ No newline at end of file diff --git a/hardwarelibrary/motion/intellidrivedevice.py b/hardwarelibrary/motion/intellidrivedevice.py index aa76f72c..328bf3de 100644 --- a/hardwarelibrary/motion/intellidrivedevice.py +++ b/hardwarelibrary/motion/intellidrivedevice.py @@ -1,7 +1,7 @@ from hardwarelibrary.motion.rotationdevice import * from hardwarelibrary.communication.serialport import SerialPort -from hardwarelibrary.communication.commands import DataCommand -from hardwarelibrary.communication.debugport import TableDrivenDebugPort, TextCommandEntry +from hardwarelibrary.communication.commands import TextCommand +from hardwarelibrary.communication.debugport import TableDrivenDebugPort import time from struct import * @@ -21,7 +21,19 @@ class State(Enum): class IntellidriveDevice(RotationDevice): classIdVendor = 0x0403 - classIdProduct = 0x6001 + classIdProduct = 0x6001 + + commands = { + "SET_REGISTER": TextCommand(name="SET_REGISTER", text="s r{0} {1}\r", + matchPattern=r's r(0x[0-9a-fA-F]+) (-?\d+)[\r\n]', + replyPattern="ok", responseTemplate="ok\r"), + "GET_REGISTER": TextCommand(name="GET_REGISTER", text="g r{0}\n", + matchPattern=r'g r(0x[0-9a-fA-F]+)[\r\n]', + replyPattern=r'v\s(-?\d+)', responseTemplate="v {0}\r"), + "TRAJECTORY": TextCommand(name="TRAJECTORY", text="t {0}\n", + matchPattern=r't (\d+)[\r\n]', + replyPattern="ok", responseTemplate="ok\r"), + } def __init__(self, serialNumber): super().__init__(serialNumber=serialNumber, idVendor=self.classIdVendor, idProduct=self.classIdProduct) @@ -120,24 +132,19 @@ def doHome(self): class DebugSerialPort(TableDrivenDebugPort): def __init__(self): - super().__init__() + super().__init__(commands=IntellidriveDevice.commands) self.registers = {'0xc9': 0} - self.text_commands = [ - TextCommandEntry(name='set_register', pattern=r's r(0x[0-9a-fA-F]+) (-?\d+)[\r\n]'), - TextCommandEntry(name='get_register', pattern=r'g r(0x[0-9a-fA-F]+)[\r\n]'), - TextCommandEntry(name='trajectory', pattern=r't (\d+)[\r\n]'), - ] def process_command(self, name, params, endPointIndex): - if name == 'set_register': + if name == 'SET_REGISTER': register, value = params self.registers[register] = int(value) return "ok\r" - elif name == 'get_register': + elif name == 'GET_REGISTER': register = params[0] value = self.registers.get(register, 0) return "v {}\r".format(value) - elif name == 'trajectory': + elif name == 'TRAJECTORY': mode = int(params[0]) if mode == 2: # home self.registers['0xca'] = 0 diff --git a/hardwarelibrary/motion/sutterdevice.py b/hardwarelibrary/motion/sutterdevice.py index 556ad12d..0ae312ea 100644 --- a/hardwarelibrary/motion/sutterdevice.py +++ b/hardwarelibrary/motion/sutterdevice.py @@ -4,7 +4,7 @@ from hardwarelibrary.communication.usbport import USBPort from hardwarelibrary.communication.serialport import SerialPort from hardwarelibrary.communication.commands import DataCommand -from hardwarelibrary.communication.debugport import TableDrivenDebugPort, BinaryCommandEntry +from hardwarelibrary.communication.debugport import TableDrivenDebugPort import re import time @@ -16,6 +16,16 @@ class SutterDevice(LinearMotionDevice): classIdVendor = 4930 classIdProduct = 1 + commands = { + "MOVE": DataCommand(name="MOVE", prefix=b'M', requestFormat=' Date: Sun, 1 Mar 2026 17:37:16 -0500 Subject: [PATCH 04/17] Add debug-mode tests for IntellidriveDevice and fix terminator Add TestIntellidriveDebugDevice with 9 tests exercising init, home, move, orientation, register setup, and status flags without hardware connected. Set terminator to b'\r' on DebugSerialPort so readString stops correctly on carriage-return-terminated responses. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/motion/intellidrivedevice.py | 1 + hardwarelibrary/tests/testIntellidrive.py | 50 ++++++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/hardwarelibrary/motion/intellidrivedevice.py b/hardwarelibrary/motion/intellidrivedevice.py index 328bf3de..2cf1cde1 100644 --- a/hardwarelibrary/motion/intellidrivedevice.py +++ b/hardwarelibrary/motion/intellidrivedevice.py @@ -133,6 +133,7 @@ def doHome(self): class DebugSerialPort(TableDrivenDebugPort): def __init__(self): super().__init__(commands=IntellidriveDevice.commands) + self.terminator = b'\r' self.registers = {'0xc9': 0} def process_command(self, name, params, endPointIndex): diff --git a/hardwarelibrary/tests/testIntellidrive.py b/hardwarelibrary/tests/testIntellidrive.py index 00eeeb4f..d06c101e 100644 --- a/hardwarelibrary/tests/testIntellidrive.py +++ b/hardwarelibrary/tests/testIntellidrive.py @@ -129,6 +129,56 @@ def testHomeWithCommands(self): self.assertFalse( status & (1 << 14) ) port.close() +class TestIntellidriveDebugDevice(unittest.TestCase): + def setUp(self): + self.device = IntellidriveDevice(serialNumber="debug") + self.device.initializeDevice() + + def tearDown(self): + self.device.shutdownDevice() + self.device = None + + def testDeviceInitializes(self): + self.assertIsNotNone(self.device) + self.assertEqual(self.device.internalState.name, 'ready') + + def testDeviceHome(self): + self.device.home() + orientation = self.device.orientation() + self.assertAlmostEqual(orientation, 0, places=2) + + def testDeviceMove(self): + angle = 90 + self.device.moveTo(angle) + orientation = self.device.orientation() + self.assertAlmostEqual(orientation, angle, places=0) + + def testDeviceMoveToZero(self): + self.device.moveTo(180) + self.device.moveTo(0) + orientation = self.device.orientation() + self.assertAlmostEqual(orientation, 0, places=2) + + def testDeviceOrientation(self): + orientation = self.device.orientation() + self.assertIsNotNone(orientation) + self.assertAlmostEqual(orientation, 0, places=2) + + def testRegistersSetDuringInit(self): + port = self.device.port + self.assertEqual(port.registers.get('0x24'), 31) + self.assertEqual(port.registers.get('0xc2'), 514) + + def testIsReferenced(self): + self.assertTrue(self.device.isReferenced()) + + def testIsNotMoving(self): + self.assertFalse(self.device.isMoving()) + + def testIsNotHoming(self): + self.assertFalse(self.device.isHoming()) + + class TestIntellidrivePhysicalDevice(unittest.TestCase): def setUp(self): try: From 4ca4aa17c1915ef5be64bb6c028662b94b3e955c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Sun, 1 Mar 2026 17:57:12 -0500 Subject: [PATCH 05/17] Support named regex groups and add documentation to TableDrivenDebugPort TextCommand.extractParams() now returns a dict when the matchPattern uses named groups (?P...), and formatResponse() accepts dicts for named template parameters. Anonymous groups and positional tuples still work. Add class and method docstrings to TableDrivenDebugPort with examples showing named parameters for self-documenting command definitions. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/communication/commands.py | 5 ++ hardwarelibrary/communication/debugport.py | 59 +++++++++++++++++++ .../tests/testCommandRecognition.py | 15 +++++ 3 files changed, 79 insertions(+) diff --git a/hardwarelibrary/communication/commands.py b/hardwarelibrary/communication/commands.py index 3d6d257f..d7091a14 100644 --- a/hardwarelibrary/communication/commands.py +++ b/hardwarelibrary/communication/commands.py @@ -100,10 +100,15 @@ def extractParams(self, inputBytes): return () match = re.match(self.effectiveMatchPattern, inputStr) if match: + if match.groupdict(): + return match.groupdict() return match.groups() return () def formatResponse(self, result): + if isinstance(result, dict) and self.responseTemplate is not None: + formatted = self.responseTemplate.format(**result) + return bytearray(formatted.encode('utf-8')) if isinstance(result, tuple) and self.responseTemplate is not None: formatted = self.responseTemplate.format(*result) return bytearray(formatted.encode('utf-8')) diff --git a/hardwarelibrary/communication/debugport.py b/hardwarelibrary/communication/debugport.py index 60361329..401fd7ce 100644 --- a/hardwarelibrary/communication/debugport.py +++ b/hardwarelibrary/communication/debugport.py @@ -78,6 +78,47 @@ def processInputBuffers(self, endPointIndex): class TableDrivenDebugPort(DebugPort): + """A mock port that recognizes incoming commands and produces responses, + driven by a dict of Command objects (TextCommand or DataCommand). + + Each Command object knows how to recognize itself in raw input bytes + (matches/extractParams) and how to format a response (formatResponse). + This means the same Command objects used by a real device to *send* + commands can also be reused here to *recognize* them, eliminating + protocol duplication between device code and mock code. + + To create a mock port for a device: + + 1. Pass the device's commands dict to __init__: + + commands = { + "GET": TextCommand(name="GET", text="GET {key}\\r", + matchPattern=r'GET (?P\\w+)\\r', + responseTemplate="VAL {value}\\r"), + "SET": DataCommand(name="SET", prefix=b'S', + requestFormat='...) + patterns, a positional tuple for anonymous groups or struct.unpack. + The return value is passed to formatResponse: dicts use named template + parameters, tuples use positional parameters, raw bytes/strings are + returned directly, None sends nothing. + """ + def __init__(self, delay=0, numberOfEndPoints=1, commands=None): super().__init__(delay=delay, numberOfEndPoints=numberOfEndPoints) self.commands = commands if commands is not None else {} @@ -101,4 +142,22 @@ def processInputBuffers(self, endPointIndex): self.inputBuffers[endPointIndex] = bytearray() def process_command(self, name, params, endPointIndex): + """Process a recognized command and return a result. + + Subclasses must override this to implement device behavior. + + Args: + name: the Command's name (matches a key in self.commands) + params: extracted parameters — a dict when using named regex + groups (?P...), a tuple for anonymous groups or + struct.unpack fields + endPointIndex: the endpoint the command arrived on + + Returns: + The return value is passed to the command's formatResponse(): + - dict: formatted using named responseTemplate parameters + - tuple: formatted using positional responseTemplate/responseFormat + - bytes/str: written to the output buffer as-is + - None: no response is sent + """ raise NotImplementedError("Subclasses must implement process_command") \ No newline at end of file diff --git a/hardwarelibrary/tests/testCommandRecognition.py b/hardwarelibrary/tests/testCommandRecognition.py index b37786fc..237e5d0a 100644 --- a/hardwarelibrary/tests/testCommandRecognition.py +++ b/hardwarelibrary/tests/testCommandRecognition.py @@ -68,6 +68,21 @@ def testFormatResponseNone(self): cmd = TextCommand(name="test", text="cmd\r") self.assertIsNone(cmd.formatResponse(None)) + def testNamedGroupExtractParams(self): + cmd = TextCommand(name="test", text="SET {key} {value}\r", + matchPattern=r'SET (?P\w+) (?P-?\d+)\r') + params = cmd.extractParams(b'SET foo 42\r') + self.assertIsInstance(params, dict) + self.assertEqual(params["key"], "foo") + self.assertEqual(params["value"], "42") + + def testNamedGroupFormatResponse(self): + cmd = TextCommand(name="test", text="GET {key}\r", + matchPattern=r'GET (?P\w+)\r', + responseTemplate="VAL {value}\r") + result = cmd.formatResponse({"value": "99"}) + self.assertEqual(result, bytearray(b'VAL 99\r')) + def testEffectiveMatchPatternUsesExplicit(self): cmd = TextCommand(name="test", text="SET {0}\r", matchPattern=r'SET (\w+)\r') From d68f613eef24cd4a79c61bacb0de5d2c353d883e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Sun, 1 Mar 2026 17:58:00 -0500 Subject: [PATCH 06/17] Convert Intellidrive commands to named regex parameters Use (?P...), (?P...), (?P...) in matchPatterns and {register}, {value}, {mode} in text/responseTemplate fields. process_command now uses params["register"] instead of params[0]. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/motion/intellidrivedevice.py | 24 +++++++++----------- 1 file changed, 11 insertions(+), 13 deletions(-) diff --git a/hardwarelibrary/motion/intellidrivedevice.py b/hardwarelibrary/motion/intellidrivedevice.py index 2cf1cde1..c276e001 100644 --- a/hardwarelibrary/motion/intellidrivedevice.py +++ b/hardwarelibrary/motion/intellidrivedevice.py @@ -24,14 +24,14 @@ class IntellidriveDevice(RotationDevice): classIdProduct = 0x6001 commands = { - "SET_REGISTER": TextCommand(name="SET_REGISTER", text="s r{0} {1}\r", - matchPattern=r's r(0x[0-9a-fA-F]+) (-?\d+)[\r\n]', + "SET_REGISTER": TextCommand(name="SET_REGISTER", text="s r{register} {value}\r", + matchPattern=r's r(?P0x[0-9a-fA-F]+) (?P-?\d+)[\r\n]', replyPattern="ok", responseTemplate="ok\r"), - "GET_REGISTER": TextCommand(name="GET_REGISTER", text="g r{0}\n", - matchPattern=r'g r(0x[0-9a-fA-F]+)[\r\n]', - replyPattern=r'v\s(-?\d+)', responseTemplate="v {0}\r"), - "TRAJECTORY": TextCommand(name="TRAJECTORY", text="t {0}\n", - matchPattern=r't (\d+)[\r\n]', + "GET_REGISTER": TextCommand(name="GET_REGISTER", text="g r{register}\n", + matchPattern=r'g r(?P0x[0-9a-fA-F]+)[\r\n]', + replyPattern=r'v\s(-?\d+)', responseTemplate="v {value}\r"), + "TRAJECTORY": TextCommand(name="TRAJECTORY", text="t {mode}\n", + matchPattern=r't (?P\d+)[\r\n]', replyPattern="ok", responseTemplate="ok\r"), } @@ -138,15 +138,13 @@ def __init__(self): def process_command(self, name, params, endPointIndex): if name == 'SET_REGISTER': - register, value = params - self.registers[register] = int(value) + self.registers[params["register"]] = int(params["value"]) return "ok\r" elif name == 'GET_REGISTER': - register = params[0] - value = self.registers.get(register, 0) - return "v {}\r".format(value) + value = self.registers.get(params["register"], 0) + return {"value": str(value)} elif name == 'TRAJECTORY': - mode = int(params[0]) + mode = int(params["mode"]) if mode == 2: # home self.registers['0xca'] = 0 self.registers['0xc9'] = (1 << 12) # referenced bit From b021358794cca12683f48970d689fcc037cbb344 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Sun, 1 Mar 2026 18:06:55 -0500 Subject: [PATCH 07/17] Convert Sutter commands to named parameters Add requestFields and responseFields to DataCommand, mapping struct fields to names so extractParams returns a dict and formatResponse accepts a dict. Sutter process_command now uses params['x'], params['y'], params['z'] and returns {'header': ..., 'x': ..., 'y': ..., 'z': ..., 'terminator': ...}. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/communication/commands.py | 14 ++++++++++++-- hardwarelibrary/motion/sutterdevice.py | 12 +++++++++--- hardwarelibrary/tests/testCommandRecognition.py | 17 +++++++++++++++++ 3 files changed, 38 insertions(+), 5 deletions(-) diff --git a/hardwarelibrary/communication/commands.py b/hardwarelibrary/communication/commands.py index d7091a14..47e26c4a 100644 --- a/hardwarelibrary/communication/commands.py +++ b/hardwarelibrary/communication/commands.py @@ -215,7 +215,8 @@ def send(self, port, params=None) -> bool: class DataCommand(Command): def __init__(self, name, data=None, replyHexRegex = None, replyDataLength = 0, unpackingMask = None, endPoints = (None, None), - prefix = None, requestFormat = None, responseFormat = None): + prefix = None, requestFormat = None, responseFormat = None, + requestFields = None, responseFields = None): Command.__init__(self, name, endPoints=endPoints) self.data : bytearray = data self.replyHexRegex: str = replyHexRegex @@ -224,6 +225,8 @@ def __init__(self, name, data=None, replyHexRegex = None, replyDataLength = 0, u self._prefix: bytes = prefix self.requestFormat: str = requestFormat self.responseFormat: str = responseFormat + self.requestFields: tuple = requestFields + self.responseFields: tuple = responseFields @property def payload(self): @@ -250,10 +253,17 @@ def extractParams(self, inputBytes): if self.requestFormat is not None: unpackLen = struct.calcsize(self.requestFormat) if len(inputBytes) >= unpackLen: - return struct.unpack(self.requestFormat, bytes(inputBytes[:unpackLen])) + values = struct.unpack(self.requestFormat, bytes(inputBytes[:unpackLen])) + if self.requestFields is not None: + return dict(zip(self.requestFields, values)) + return values return () def formatResponse(self, result): + if isinstance(result, dict) and self.responseFormat is not None and self.responseFields is not None: + values = tuple(result[f] for f in self.responseFields) + data = struct.pack(self.responseFormat, *values) + return bytearray(data) if isinstance(result, tuple) and self.responseFormat is not None: data = struct.pack(self.responseFormat, *result) return bytearray(data) diff --git a/hardwarelibrary/motion/sutterdevice.py b/hardwarelibrary/motion/sutterdevice.py index 0ae312ea..254cee8b 100644 --- a/hardwarelibrary/motion/sutterdevice.py +++ b/hardwarelibrary/motion/sutterdevice.py @@ -18,10 +18,12 @@ class SutterDevice(LinearMotionDevice): commands = { "MOVE": DataCommand(name="MOVE", prefix=b'M', requestFormat=' Date: Sun, 1 Mar 2026 18:15:35 -0500 Subject: [PATCH 08/17] Add docstrings to commands.py Document the module, all four classes (Command, TextCommand, MultilineTextCommand, DataCommand), and their key methods. Covers both the send side and recognition side roles, with parameter descriptions and examples. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/communication/commands.py | 148 ++++++++++++++++++++++ 1 file changed, 148 insertions(+) diff --git a/hardwarelibrary/communication/commands.py b/hardwarelibrary/communication/commands.py index 47e26c4a..a90ccf95 100644 --- a/hardwarelibrary/communication/commands.py +++ b/hardwarelibrary/communication/commands.py @@ -1,7 +1,41 @@ +"""Command classes that describe a device's communication protocol. + +Each Command object serves two roles: + +1. **Send side** (real device): build a payload, send it through a port, + and parse the reply. This is done via the `send()` method. + +2. **Recognition side** (mock/debug port): examine raw incoming bytes, + decide if they match this command, extract parameters, and format + a response. This is done via `matches()`, `extractParams()`, and + `formatResponse()`. + +By defining the protocol once in Command objects, the same definitions +drive both real communication and mock ports (TableDrivenDebugPort), +eliminating protocol duplication. + +Subclasses: + TextCommand — text-based protocols (e.g. "s r0x24 31\\r" → "ok\\r") + DataCommand — binary protocols (e.g. struct-packed position commands) + MultilineTextCommand — text protocols that return multiple lines +""" + import re import struct class Command: + """Base class for all device commands. + + A Command has a name and tracks send/reply state. Subclasses override + `send()` for the client side, and `matches()`/`extractParams()`/ + `formatResponse()` for the mock port side. + + Args: + name: identifier for this command (e.g. "GET_POSITION", "SET_REGISTER") + endPoints: tuple of (writeEndPoint, readEndPoint) for USB devices + that use separate endpoints for sending and receiving + """ + def __init__(self, name:str, endPoints = (None, None)): self.name = name self.reply = None @@ -33,15 +67,33 @@ def hasError(self): return len(self.exceptions) != 0 def send(self, port) -> bool: + """Send this command through a port. Subclasses must override.""" raise NotImplementedError("Subclasses must implement the send() command") def matches(self, inputBytes): + """Return True if inputBytes represents this command. + Base implementation always returns False.""" return False def extractParams(self, inputBytes): + """Extract parameters from recognized input bytes. + Returns a dict (named params) or tuple (positional params). + Base implementation returns an empty tuple.""" return () def formatResponse(self, result): + """Convert a process_command() return value into bytes for the output buffer. + + Args: + result: the value returned by process_command(): + - None → returns None (no response sent) + - bytes/bytearray → returned as-is + - str → encoded to UTF-8 + - tuple or dict → handled by subclass (template/struct formatting) + + Returns: + bytearray to write to the output buffer, or None. + """ if result is None: return None elif isinstance(result, (bytes, bytearray)): @@ -51,6 +103,31 @@ def formatResponse(self, result): return None class TextCommand(Command): + """A command that communicates via text strings (UTF-8). + + Used for devices with text-based protocols like "s r0x24 31\\r" → "ok\\r". + + Send-side parameters (used by send()): + text: format string sent to the device, e.g. "g r{register}\\n" + replyPattern: regex to validate the device's reply + alternatePattern: secondary regex if replyPattern doesn't match + + Recognition-side parameters (used by matches/extractParams/formatResponse): + matchPattern: regex to recognize incoming bytes in a mock port. + Use named groups for readable code: + r'g r(?P0x[0-9a-fA-F]+)[\\r\\n]' + If None, an auto-derived pattern from `text` is used. + responseTemplate: format string for the mock response, e.g. "v {value}\\r". + Filled with named (dict) or positional (tuple) params + returned by process_command(). + + Example: + TextCommand(name="GET_REGISTER", text="g r{register}\\n", + matchPattern=r'g r(?P0x[0-9a-fA-F]+)[\\r\\n]', + replyPattern=r'v\\s(-?\\d+)', + responseTemplate="v {value}\\r") + """ + def __init__(self, name, text, replyPattern = None, alternatePattern = None, endPoints = (None, None), @@ -76,17 +153,21 @@ def numberOfArguments(self): @property def _autoMatchPattern(self): + """Derive a regex from the text format string by replacing {…} + placeholders with (.+?) capture groups.""" parts = re.split(r'\{[^}]*\}', self.text) escaped = [re.escape(p) for p in parts] return '(.+?)'.join(escaped) @property def effectiveMatchPattern(self): + """Return the explicit matchPattern if set, otherwise the auto-derived one.""" if self.matchPattern is not None: return self.matchPattern return self._autoMatchPattern def matches(self, inputBytes): + """Return True if inputBytes matches this text command's pattern.""" try: inputStr = inputBytes.decode('utf-8', errors='replace') except Exception: @@ -94,6 +175,10 @@ def matches(self, inputBytes): return re.match(self.effectiveMatchPattern, inputStr) is not None def extractParams(self, inputBytes): + """Extract parameters from the input bytes using the match pattern. + + Returns a dict if the pattern uses named groups (?P...), + otherwise a tuple of positional groups.""" try: inputStr = inputBytes.decode('utf-8', errors='replace') except Exception: @@ -106,6 +191,12 @@ def extractParams(self, inputBytes): return () def formatResponse(self, result): + """Format a mock response using the responseTemplate. + + - dict result → template.format(**result), e.g. {"value": "42"} → "v 42\\r" + - tuple result → template.format(*result), e.g. ("42",) → "v 42\\r" + - other types → delegated to base class (bytes/str as-is, None → None) + """ if isinstance(result, dict) and self.responseTemplate is not None: formatted = self.responseTemplate.format(**result) return bytearray(formatted.encode('utf-8')) @@ -145,6 +236,17 @@ def send(self, port, params=None) -> bool: class MultilineTextCommand(Command): + """A text command that expects a multi-line reply. + + The reply is read either a fixed number of times (lineCount > 1) or + until a line matches lastLinePattern. Each line is matched against + replyPattern and the results are collected into lists. + + Args: + lineCount: number of reply lines to read (if > 1) + lastLinePattern: regex that signals the final line (alternative to lineCount) + """ + def __init__(self, name, text, replyPattern=None, alternatePattern=None, @@ -214,6 +316,40 @@ def send(self, port, params=None) -> bool: class DataCommand(Command): + """A command that communicates via binary data (struct-packed bytes). + + Used for devices with binary protocols like the Sutter micromanipulator + where commands are single-byte prefixes followed by packed integers. + + Send-side parameters (used by send()): + data: raw bytes to send to the device + replyDataLength: number of bytes to read back + unpackingMask: struct format to unpack the reply + + Recognition-side parameters (used by matches/extractParams/formatResponse): + prefix: byte(s) that identify this command in incoming data. + Matched case-insensitively. If None, derived from data[0:1]. + requestFormat: struct format to unpack incoming request parameters. + Padding bytes (x) are skipped automatically by struct. + requestFields: tuple of field names for the unpacked values. + When set, extractParams() returns a dict: + e.g. requestFields=('x','y','z') → {'x': 10, 'y': 20, 'z': 30} + When None, extractParams() returns a positional tuple. + responseFormat: struct format to pack the mock response. + responseFields: tuple of field names expected in the response dict. + Defines the order for struct.pack when process_command() + returns a dict. + + Example: + DataCommand(name="MOVE", prefix=b'M', + requestFormat=' 0: @@ -241,6 +378,7 @@ def effectivePrefix(self): return None def matches(self, inputBytes): + """Return True if inputBytes starts with this command's prefix (case-insensitive).""" prefix = self.effectivePrefix if prefix is None: return False @@ -250,6 +388,10 @@ def matches(self, inputBytes): return inputBytes[:prefixLen].upper() == prefix.upper() def extractParams(self, inputBytes): + """Unpack parameters from the input bytes using requestFormat. + + Returns a dict if requestFields is set (e.g. {'x': 10, 'y': 20}), + otherwise a positional tuple from struct.unpack.""" if self.requestFormat is not None: unpackLen = struct.calcsize(self.requestFormat) if len(inputBytes) >= unpackLen: @@ -260,6 +402,12 @@ def extractParams(self, inputBytes): return () def formatResponse(self, result): + """Pack a mock response using responseFormat. + + - dict result + responseFields → values ordered by responseFields, then packed + - tuple result → packed directly with struct.pack + - other types → delegated to base class (bytes/str as-is, None → None) + """ if isinstance(result, dict) and self.responseFormat is not None and self.responseFields is not None: values = tuple(result[f] for f in self.responseFields) data = struct.pack(self.responseFormat, *values) From 91e7a8f444e79ef3da2c0c4a11e6c67feec93a4a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Sun, 1 Mar 2026 18:19:21 -0500 Subject: [PATCH 09/17] Rewrite module docstring to explain the two-way role of commands dicts Every PhysicalDevice defines a commands dict that serves both sides: sending commands to real hardware and creating mock debug ports that receive those same commands and reply in the correct format. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/communication/commands.py | 35 ++++++++++++++--------- 1 file changed, 22 insertions(+), 13 deletions(-) diff --git a/hardwarelibrary/communication/commands.py b/hardwarelibrary/communication/commands.py index a90ccf95..f39071d0 100644 --- a/hardwarelibrary/communication/commands.py +++ b/hardwarelibrary/communication/commands.py @@ -1,18 +1,27 @@ """Command classes that describe a device's communication protocol. -Each Command object serves two roles: - -1. **Send side** (real device): build a payload, send it through a port, - and parse the reply. This is done via the `send()` method. - -2. **Recognition side** (mock/debug port): examine raw incoming bytes, - decide if they match this command, extract parameters, and format - a response. This is done via `matches()`, `extractParams()`, and - `formatResponse()`. - -By defining the protocol once in Command objects, the same definitions -drive both real communication and mock ports (TableDrivenDebugPort), -eliminating protocol duplication. +Every PhysicalDevice can define a ``commands`` dictionary — a dict of +Command objects that fully describes its protocol. For example, a +Sutter micromanipulator defines MOVE, GET_POSITION, and HOME commands, +while an Intellidrive rotation stage defines SET_REGISTER, GET_REGISTER, +and TRAJECTORY commands. + +This dictionary is useful in two complementary ways: + +1. **Talking to a real device**: each Command knows how to build a + payload, send it through a port, and parse the reply. The device + code calls ``send()`` and reads back ``reply`` / ``matchGroups``. + +2. **Creating a mock debug port**: the same Command objects can be + passed to a ``TableDrivenDebugPort``, which reverses the roles — + it *receives* those commands instead of sending them, extracts the + parameters from the incoming bytes, and replies in the correct + format. This makes it trivial to write a mock port for any device: + just pass its ``commands`` dict and implement ``process_command()`` + with the device-specific logic. + +Because both sides share the same Command definitions, the protocol is +defined once and cannot drift between real and mock implementations. Subclasses: TextCommand — text-based protocols (e.g. "s r0x24 31\\r" → "ok\\r") From 812bcee065f21f32d400aa18c5cd44847ef2ac21 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Sun, 1 Mar 2026 18:59:11 -0500 Subject: [PATCH 10/17] Migrate SutterDevice send-side to use the commands dict Add buildSendData()/unpackReply() to DataCommand so the same commands dict that drives the mock debug port also builds outgoing packets. Rewrite doGetPosition, doMoveTo, doHome, and work to use the new sendCommand() helper, eliminating hand-built pack() calls. Add WORK command to the commands dict and DebugSerialPort. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/communication/commands.py | 23 +++++++- hardwarelibrary/motion/sutterdevice.py | 59 ++++++++++--------- .../tests/testCommandRecognition.py | 31 ++++++++++ 3 files changed, 85 insertions(+), 28 deletions(-) diff --git a/hardwarelibrary/communication/commands.py b/hardwarelibrary/communication/commands.py index f39071d0..4d90fa15 100644 --- a/hardwarelibrary/communication/commands.py +++ b/hardwarelibrary/communication/commands.py @@ -361,7 +361,8 @@ class DataCommand(Command): def __init__(self, name, data=None, replyHexRegex = None, replyDataLength = 0, unpackingMask = None, endPoints = (None, None), prefix = None, requestFormat = None, responseFormat = None, - requestFields = None, responseFields = None): + requestFields = None, responseFields = None, + sendFormat = None, sendFields = None, sendDefaults = None): Command.__init__(self, name, endPoints=endPoints) self.data : bytearray = data self.replyHexRegex: str = replyHexRegex @@ -372,6 +373,9 @@ def __init__(self, name, data=None, replyHexRegex = None, replyDataLength = 0, u self.responseFormat: str = responseFormat self.requestFields: tuple = requestFields self.responseFields: tuple = responseFields + self.sendFormat: str = sendFormat + self.sendFields: tuple = sendFields + self.sendDefaults: dict = sendDefaults @property def payload(self): @@ -426,6 +430,23 @@ def formatResponse(self, result): return bytearray(data) return super().formatResponse(result) + def buildSendData(self, **params): + """Build send bytes from named params merged with defaults. + Falls back to self.data if sendFormat is not set.""" + if self.sendFormat is None: + return self.data + merged = dict(self.sendDefaults) if self.sendDefaults else {} + merged.update(params) + values = tuple(merged[f] for f in self.sendFields) + return struct.pack(self.sendFormat, *values) + + def unpackReply(self, replyBytes): + """Unpack reply bytes using unpackingMask. + Returns raw bytes if unpackingMask is not set.""" + if self.unpackingMask is None: + return replyBytes + return struct.unpack(self.unpackingMask, replyBytes) + def send(self, port) -> bool: try: nBytes = port.writeData(data=self.data, endPoint=self.endPoints[0]) diff --git a/hardwarelibrary/motion/sutterdevice.py b/hardwarelibrary/motion/sutterdevice.py index 085f2ffd..3e6500ad 100644 --- a/hardwarelibrary/motion/sutterdevice.py +++ b/hardwarelibrary/motion/sutterdevice.py @@ -17,14 +17,22 @@ class SutterDevice(LinearMotionDevice): classIdProduct = 1 commands = { - "MOVE": DataCommand(name="MOVE", prefix=b'M', requestFormat=' tuple: - """ The function to read a reply from the endpoint. It will initialize the device - if it is not already initialized. On failure, it will warn and shutdown. + """ The function to read a reply from the endpoint. It will initialize the device + if it is not already initialized. On failure, it will warn and shutdown. It will unpack the reply into a tuple. """ if format is None: @@ -106,24 +114,27 @@ def readReply(self, size, format) -> tuple: # print(unpack(format, replyBytes)) return unpack(format, replyBytes) + def sendCommand(self, name, **params): + """Send a command from the commands dict and return the unpacked reply.""" + cmd = self.commands[name] + data = cmd.buildSendData(**params) + self.sendCommandBytes(data) + if cmd.replyDataLength > 0: + return self.readReply(cmd.replyDataLength, cmd.unpackingMask) + return None + def positionInMicrosteps(self) -> (int, int, int): # for compatibility return self.doGetPosition() def doGetPosition(self) -> (int, int, int): """ Returns the position in microsteps """ - commandBytes = pack(' Date: Sun, 1 Mar 2026 19:26:09 -0500 Subject: [PATCH 11/17] Rename TextCommand text parameter to text_format The parameter is a format string, not plain text. Rename it everywhere: TextCommand/MultilineTextCommand __init__, all self.text references, all call sites across device classes, debug ports, and tests. Co-Authored-By: Claude Opus 4.6 --- hardwarelibrary/communication/commands.py | 30 +++++++++---------- hardwarelibrary/communication/debugport.py | 2 +- hardwarelibrary/devicemanager.py | 2 +- hardwarelibrary/echodevice.py | 4 +-- hardwarelibrary/motion/intellidrivedevice.py | 6 ++-- hardwarelibrary/powermeters/integradevice.py | 10 +++---- .../tests/testCommandRecognition.py | 24 +++++++-------- .../tests/testCommunicationPorts.py | 4 +-- hardwarelibrary/tests/testIntegraDevice.py | 22 +++++++------- .../tests/testTableDrivenDebugPort.py | 6 ++-- 10 files changed, 55 insertions(+), 55 deletions(-) diff --git a/hardwarelibrary/communication/commands.py b/hardwarelibrary/communication/commands.py index 4d90fa15..eb9d3802 100644 --- a/hardwarelibrary/communication/commands.py +++ b/hardwarelibrary/communication/commands.py @@ -117,7 +117,7 @@ class TextCommand(Command): Used for devices with text-based protocols like "s r0x24 31\\r" → "ok\\r". Send-side parameters (used by send()): - text: format string sent to the device, e.g. "g r{register}\\n" + text_format: format string sent to the device, e.g. "g r{register}\\n" replyPattern: regex to validate the device's reply alternatePattern: secondary regex if replyPattern doesn't match @@ -125,25 +125,25 @@ class TextCommand(Command): matchPattern: regex to recognize incoming bytes in a mock port. Use named groups for readable code: r'g r(?P0x[0-9a-fA-F]+)[\\r\\n]' - If None, an auto-derived pattern from `text` is used. + If None, an auto-derived pattern from `text_format` is used. responseTemplate: format string for the mock response, e.g. "v {value}\\r". Filled with named (dict) or positional (tuple) params returned by process_command(). Example: - TextCommand(name="GET_REGISTER", text="g r{register}\\n", + TextCommand(name="GET_REGISTER", text_format="g r{register}\\n", matchPattern=r'g r(?P0x[0-9a-fA-F]+)[\\r\\n]', replyPattern=r'v\\s(-?\\d+)', responseTemplate="v {value}\\r") """ - def __init__(self, name, text, replyPattern = None, + def __init__(self, name, text_format, replyPattern = None, alternatePattern = None, endPoints = (None, None), matchPattern = None, responseTemplate = None): Command.__init__(self, name, endPoints=endPoints) - self.text : str = text + self.text_format : str = text_format self.replyPattern: str = replyPattern self.alternatePattern: str = alternatePattern self.matchPattern: str = matchPattern @@ -151,11 +151,11 @@ def __init__(self, name, text, replyPattern = None, @property def payload(self): - return self.text + return self.text_format @property def numberOfArguments(self): - match = re.search(r"\{(.*?)\}", self.text) + match = re.search(r"\{(.*?)\}", self.text_format) if match is None: return 0 return len(match.groups()) @@ -164,7 +164,7 @@ def numberOfArguments(self): def _autoMatchPattern(self): """Derive a regex from the text format string by replacing {…} placeholders with (.+?) capture groups.""" - parts = re.split(r'\{[^}]*\}', self.text) + parts = re.split(r'\{[^}]*\}', self.text_format) escaped = [re.escape(p) for p in parts] return '(.+?)'.join(escaped) @@ -217,9 +217,9 @@ def formatResponse(self, result): def send(self, port, params=None) -> bool: try: if params is not None: - textCommand = self.text.format(params) + textCommand = self.text_format.format(params) else: - textCommand = self.text + textCommand = self.text_format if port is None: raise RuntimeError("port cannot be None") @@ -256,14 +256,14 @@ class MultilineTextCommand(Command): lastLinePattern: regex that signals the final line (alternative to lineCount) """ - def __init__(self, name, text, + def __init__(self, name, text_format, replyPattern=None, alternatePattern=None, lineCount=1, lastLinePattern=None, endPoints=(None, None)): Command.__init__(self, name=name, endPoints=endPoints) - self.text: str = text + self.text_format: str = text_format self.replyPattern: str = replyPattern self.alternatePattern: str = alternatePattern self.lineCount = lineCount @@ -271,14 +271,14 @@ def __init__(self, name, text, @property def payload(self): - return self.text + return self.text_format def send(self, port, params=None) -> bool: try: if params is not None: - textCommand = self.text.format(params) + textCommand = self.text_format.format(params) else: - textCommand = self.text + textCommand = self.text_format if port is None: raise RuntimeError("port cannot be None") diff --git a/hardwarelibrary/communication/debugport.py b/hardwarelibrary/communication/debugport.py index 1316ae29..f2b54088 100644 --- a/hardwarelibrary/communication/debugport.py +++ b/hardwarelibrary/communication/debugport.py @@ -92,7 +92,7 @@ class TableDrivenDebugPort(DebugPort): 1. Pass the device's commands dict to __init__: commands = { - "GET": TextCommand(name="GET", text="GET {key}\\r", + "GET": TextCommand(name="GET", text_format="GET {key}\\r", matchPattern=r'GET (?P\\w+)\\r', responseTemplate="VAL {value}\\r"), "SET": DataCommand(name="SET", prefix=b'S', diff --git a/hardwarelibrary/devicemanager.py b/hardwarelibrary/devicemanager.py index 46036ae6..4cab51d3 100644 --- a/hardwarelibrary/devicemanager.py +++ b/hardwarelibrary/devicemanager.py @@ -314,7 +314,7 @@ def sendCommand(self, commandName, deviceIdentifier=0): if device.state == DeviceState.Ready: command = device.commands[commandName] command.send(port=device.port) - return (commandName, command.text, command.matchGroups) + return (commandName, command.text_format, command.matchGroups) else: print("Device {0} is not Ready: call initializeDevice()".format(device)) diff --git a/hardwarelibrary/echodevice.py b/hardwarelibrary/echodevice.py index b262f3d6..57ef738d 100644 --- a/hardwarelibrary/echodevice.py +++ b/hardwarelibrary/echodevice.py @@ -14,8 +14,8 @@ class EchoDevice(PhysicalDevice): classIdProduct = 0x6001 classIdVendor = 0x0403 - commands = {"ECHO1": TextCommand(name="ECHO1", text="someText", replyPattern="someText"), - "ECHO2": TextCommand(name="ECHO2", text="someOtherText", replyPattern="someOtherText"), + commands = {"ECHO1": TextCommand(name="ECHO1", text_format="someText", replyPattern="someText"), + "ECHO2": TextCommand(name="ECHO2", text_format="someOtherText", replyPattern="someOtherText"), "ECHO3": DataCommand(name="ECHO3", data=b"someData", replyDataLength=len(b"someData"))} def __init__(self, serialNumber='ftDXIKC4', idProduct=classIdProduct, idVendor=classIdVendor): diff --git a/hardwarelibrary/motion/intellidrivedevice.py b/hardwarelibrary/motion/intellidrivedevice.py index c276e001..8fd3fefb 100644 --- a/hardwarelibrary/motion/intellidrivedevice.py +++ b/hardwarelibrary/motion/intellidrivedevice.py @@ -24,13 +24,13 @@ class IntellidriveDevice(RotationDevice): classIdProduct = 0x6001 commands = { - "SET_REGISTER": TextCommand(name="SET_REGISTER", text="s r{register} {value}\r", + "SET_REGISTER": TextCommand(name="SET_REGISTER", text_format="s r{register} {value}\r", matchPattern=r's r(?P0x[0-9a-fA-F]+) (?P-?\d+)[\r\n]', replyPattern="ok", responseTemplate="ok\r"), - "GET_REGISTER": TextCommand(name="GET_REGISTER", text="g r{register}\n", + "GET_REGISTER": TextCommand(name="GET_REGISTER", text_format="g r{register}\n", matchPattern=r'g r(?P0x[0-9a-fA-F]+)[\r\n]', replyPattern=r'v\s(-?\d+)', responseTemplate="v {value}\r"), - "TRAJECTORY": TextCommand(name="TRAJECTORY", text="t {mode}\n", + "TRAJECTORY": TextCommand(name="TRAJECTORY", text_format="t {mode}\n", matchPattern=r't (?P\d+)[\r\n]', replyPattern="ok", responseTemplate="ok\r"), } diff --git a/hardwarelibrary/powermeters/integradevice.py b/hardwarelibrary/powermeters/integradevice.py index 640801d8..d9c95aff 100644 --- a/hardwarelibrary/powermeters/integradevice.py +++ b/hardwarelibrary/powermeters/integradevice.py @@ -8,11 +8,11 @@ class IntegraDevice(PowerMeterDevice): classIdProduct = 0x0300 classIdVendor = 0x1ad5 commands = { - "GETPOWER": TextCommand(name="GETPOWER", text="*CVU", replyPattern=r"(.+?)\r\n"), - "VERSION": TextCommand(name="VERSION", text="*VER", replyPattern=r"(.+?)\r\n"), - "STATUS": MultilineTextCommand(name="STATUS", text="*STS", replyPattern=r"(.+?)\r\n", lastLinePattern=":100000000"), - "GETWAVELENGTH": TextCommand(name="GETWAVELENGTH", text="*GWL", replyPattern=r"PWC\s*:\s*(.+?)\r\n"), - "SETWAVELENGTH": TextCommand(name="SETWAVELENGTH", text="*PWC{0:05d}") + "GETPOWER": TextCommand(name="GETPOWER", text_format="*CVU", replyPattern=r"(.+?)\r\n"), + "VERSION": TextCommand(name="VERSION", text_format="*VER", replyPattern=r"(.+?)\r\n"), + "STATUS": MultilineTextCommand(name="STATUS", text_format="*STS", replyPattern=r"(.+?)\r\n", lastLinePattern=":100000000"), + "GETWAVELENGTH": TextCommand(name="GETWAVELENGTH", text_format="*GWL", replyPattern=r"PWC\s*:\s*(.+?)\r\n"), + "SETWAVELENGTH": TextCommand(name="SETWAVELENGTH", text_format="*PWC{0:05d}") } def __init__(self, serialNumber:str = None, idProduct:int = 0x0300, idVendor:int = 0x1ad5): diff --git a/hardwarelibrary/tests/testCommandRecognition.py b/hardwarelibrary/tests/testCommandRecognition.py index bf105029..f23ea836 100644 --- a/hardwarelibrary/tests/testCommandRecognition.py +++ b/hardwarelibrary/tests/testCommandRecognition.py @@ -29,47 +29,47 @@ def testFormatResponseString(self): class TestTextCommandRecognition(unittest.TestCase): def testAutoMatchPattern(self): - cmd = TextCommand(name="test", text="SET {0} {1}\r") + cmd = TextCommand(name="test", text_format="SET {0} {1}\r") self.assertTrue(cmd.matches(b'SET foo 42\r')) def testAutoMatchPatternNoMatch(self): - cmd = TextCommand(name="test", text="SET {0} {1}\r") + cmd = TextCommand(name="test", text_format="SET {0} {1}\r") self.assertFalse(cmd.matches(b'GET foo\r')) def testExplicitMatchPattern(self): - cmd = TextCommand(name="test", text="s r{0}\r", + cmd = TextCommand(name="test", text_format="s r{0}\r", matchPattern=r's r(0x[0-9a-fA-F]+)[\r\n]') self.assertTrue(cmd.matches(b's r0xFF\r')) self.assertTrue(cmd.matches(b's r0xFF\n')) def testExtractParams(self): - cmd = TextCommand(name="test", text="SET {0} {1}\r") + cmd = TextCommand(name="test", text_format="SET {0} {1}\r") params = cmd.extractParams(b'SET foo 42\r') self.assertEqual(params, ('foo', '42')) def testExtractParamsExplicitPattern(self): - cmd = TextCommand(name="test", text="g r{0}\n", + cmd = TextCommand(name="test", text_format="g r{0}\n", matchPattern=r'g r(0x[0-9a-fA-F]+)[\r\n]') params = cmd.extractParams(b'g r0xc9\n') self.assertEqual(params, ('0xc9',)) def testFormatResponseWithTemplate(self): - cmd = TextCommand(name="test", text="g r{0}\n", + cmd = TextCommand(name="test", text_format="g r{0}\n", responseTemplate="v {0}\r") result = cmd.formatResponse(("42",)) self.assertEqual(result, bytearray(b'v 42\r')) def testFormatResponseStringFallback(self): - cmd = TextCommand(name="test", text="cmd\r") + cmd = TextCommand(name="test", text_format="cmd\r") result = cmd.formatResponse("ok\r") self.assertEqual(result, bytearray(b'ok\r')) def testFormatResponseNone(self): - cmd = TextCommand(name="test", text="cmd\r") + cmd = TextCommand(name="test", text_format="cmd\r") self.assertIsNone(cmd.formatResponse(None)) def testNamedGroupExtractParams(self): - cmd = TextCommand(name="test", text="SET {key} {value}\r", + cmd = TextCommand(name="test", text_format="SET {key} {value}\r", matchPattern=r'SET (?P\w+) (?P-?\d+)\r') params = cmd.extractParams(b'SET foo 42\r') self.assertIsInstance(params, dict) @@ -77,19 +77,19 @@ def testNamedGroupExtractParams(self): self.assertEqual(params["value"], "42") def testNamedGroupFormatResponse(self): - cmd = TextCommand(name="test", text="GET {key}\r", + cmd = TextCommand(name="test", text_format="GET {key}\r", matchPattern=r'GET (?P\w+)\r', responseTemplate="VAL {value}\r") result = cmd.formatResponse({"value": "99"}) self.assertEqual(result, bytearray(b'VAL 99\r')) def testEffectiveMatchPatternUsesExplicit(self): - cmd = TextCommand(name="test", text="SET {0}\r", + cmd = TextCommand(name="test", text_format="SET {0}\r", matchPattern=r'SET (\w+)\r') self.assertEqual(cmd.effectiveMatchPattern, r'SET (\w+)\r') def testEffectiveMatchPatternUsesAuto(self): - cmd = TextCommand(name="test", text="SET {0}\r") + cmd = TextCommand(name="test", text_format="SET {0}\r") self.assertEqual(cmd.effectiveMatchPattern, cmd._autoMatchPattern) diff --git a/hardwarelibrary/tests/testCommunicationPorts.py b/hardwarelibrary/tests/testCommunicationPorts.py index 78a2635f..0fcb740c 100644 --- a/hardwarelibrary/tests/testCommunicationPorts.py +++ b/hardwarelibrary/tests/testCommunicationPorts.py @@ -189,7 +189,7 @@ def testThreadSafety(self): self.fail("Thread {0}?".format(threadFailed)) def testTextCommandNoReply(self): - command = TextCommand("Test", text="1234\n") + command = TextCommand("Test", text_format="1234\n") self.assertIsNotNone(command) self.assertFalse(command.send(self.port)) self.assertTrue(command.isSentSuccessfully) @@ -197,7 +197,7 @@ def testTextCommandNoReply(self): self.assertFalse(command.hasError) def testTextCommand(self): - command = TextCommand("Test", text="1234\n", replyPattern="1234") + command = TextCommand("Test", text_format="1234\n", replyPattern="1234") self.assertIsNotNone(command) self.assertFalse(command.send(self.port)) self.assertTrue(command.isSentSuccessfully) diff --git a/hardwarelibrary/tests/testIntegraDevice.py b/hardwarelibrary/tests/testIntegraDevice.py index 9246de32..d991169a 100644 --- a/hardwarelibrary/tests/testIntegraDevice.py +++ b/hardwarelibrary/tests/testIntegraDevice.py @@ -78,10 +78,10 @@ def testValidateCommands(self): def testTextCommandsNoParameter(self): commands = [ - TextCommand(name="GETPOWER", text="*CVU", replyPattern = r"(.+?)\r\n"), - TextCommand(name="VERSION", text="*VER", replyPattern = r"(.+?)\r\n"), - MultilineTextCommand(name="STATUS", text="*STS", replyPattern = r"(.+?)\r\n", lastLinePattern=":100000000"), - TextCommand(name="GETWAVELENGTH", text="*GWL", replyPattern = r"PWC\s*:\s*(.+?)\r\n") + TextCommand(name="GETPOWER", text_format="*CVU", replyPattern = r"(.+?)\r\n"), + TextCommand(name="VERSION", text_format="*VER", replyPattern = r"(.+?)\r\n"), + MultilineTextCommand(name="STATUS", text_format="*STS", replyPattern = r"(.+?)\r\n", lastLinePattern=":100000000"), + TextCommand(name="GETWAVELENGTH", text_format="*GWL", replyPattern = r"PWC\s*:\s*(.+?)\r\n") ] for command in commands: @@ -92,7 +92,7 @@ def testTextCommandsNoParameter(self): def testStatusCommand(self): - #TextCommand(name="STATUS", text="*STS", replyPattern=r"(.+)\r\n", finalReplyPattern=":100000000"), + #TextCommand(name="STATUS", text_format="*STS", replyPattern=r"(.+)\r\n", finalReplyPattern=":100000000"), self.port.writeString("*STS") for i in range(47): @@ -112,8 +112,8 @@ def testIntegraBugUnresponsiveAfterSetWavelength(self): It is not sufficient to set a longer timeout as demonstrated with testIntegraBugUnresponsiveAfterSetWavelengthMustSleep() """ - setCommand = TextCommand(name="SETWAVELENGTH", text="*PWC{0:05d}") - getCommand = TextCommand(name="GETWAVELENGTH", text="*GWL", replyPattern = r"PWC\s*:\s*(.+?)\r\n") + setCommand = TextCommand(name="SETWAVELENGTH", text_format="*PWC{0:05d}") + getCommand = TextCommand(name="GETWAVELENGTH", text_format="*GWL", replyPattern = r"PWC\s*:\s*(.+?)\r\n") # This should succeed setCommand.send(port=self.port, params=(800)) @@ -129,8 +129,8 @@ def testIntegraBugUnresponsiveAfterSetWavelength(self): def testIntegraBugUnresponsiveAfterSetWavelengthMustSleep(self): """ See above. """ - setCommand = TextCommand(name="SETWAVELENGTH", text="*PWC{0:05d}") - getCommand = TextCommand(name="GETWAVELENGTH", text="*GWL", replyPattern = r"PWC\s*:\s*(.+?)\r\n") + setCommand = TextCommand(name="SETWAVELENGTH", text_format="*PWC{0:05d}") + getCommand = TextCommand(name="GETWAVELENGTH", text_format="*GWL", replyPattern = r"PWC\s*:\s*(.+?)\r\n") self.port.defaultTimeout = 2000 setCommand.send(port=self.port, params=(800)) @@ -141,8 +141,8 @@ def testIntegraBugUnresponsiveAfterSetWavelengthMustSleep(self): def testIntegraBugUnresponsiveOnlyAfterSetWavelength(self): """ See above. """ - setCommand = TextCommand(name="SETAUTOSCALE", text="*SAS1") - getCommand = TextCommand(name="GETWAVELENGTH", text="*GWL", replyPattern = r"PWC\s*:\s*(.+?)\r\n") + setCommand = TextCommand(name="SETAUTOSCALE", text_format="*SAS1") + getCommand = TextCommand(name="GETWAVELENGTH", text_format="*GWL", replyPattern = r"PWC\s*:\s*(.+?)\r\n") # This will succeed, so it's only Setwavelength that is the problem setCommand.send(port=self.port, params=(800)) time.sleep(0.001) diff --git a/hardwarelibrary/tests/testTableDrivenDebugPort.py b/hardwarelibrary/tests/testTableDrivenDebugPort.py index 546569ea..3ec23890 100644 --- a/hardwarelibrary/tests/testTableDrivenDebugPort.py +++ b/hardwarelibrary/tests/testTableDrivenDebugPort.py @@ -25,9 +25,9 @@ def process_command(self, name, params, endPointIndex): class TextFixture(TableDrivenDebugPort): def __init__(self): super().__init__(commands={ - 'set': TextCommand(name='set', text='SET {0} {1}\r', + 'set': TextCommand(name='set', text_format='SET {0} {1}\r', matchPattern=r'SET (\w+) (-?\d+)\r'), - 'get': TextCommand(name='get', text='GET {0}\r', + 'get': TextCommand(name='get', text_format='GET {0}\r', matchPattern=r'GET (\w+)\r', responseTemplate='VAL {0}\r'), }) @@ -47,7 +47,7 @@ class MixedFixture(TableDrivenDebugPort): def __init__(self): super().__init__(commands={ 'bin_set': DataCommand(name='bin_set', prefix=b'\x01', requestFormat=' Date: Mon, 18 May 2026 12:29:43 -0400 Subject: [PATCH 12/17] Configure pytest to collect testFoo.py files from the tests directory The project's test files follow a 'testFooBar.py' naming convention (no underscore separator). Pytest's default 'python_files' pattern is 'test_*.py', so 'pytest hardwarelibrary/tests/' returned zero collected tests -- tests could only be run by naming files explicitly. That had become a real friction now that the suite crosses 500 tests. Added a [tool.pytest.ini_options] section to pyproject.toml: - testpaths = ["hardwarelibrary/tests"] -- so plain 'pytest' works - python_files = ["test*.py"] -- match the project's naming - norecursedirs = ["deprecated"] -- skip unmaintained legacy tests that error on import After this, 'pytest' from the repo root collects 476 tests and reports 263 passed, 213 skipped (hardware-gated), 0 failures. Co-Authored-By: Claude Opus 4.7 (1M context) --- pyproject.toml | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 66e861de..b41f639e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,7 +48,11 @@ Homepage = "https://github.com/DCC-Lab/PyHardwareLibrary" [tool.setuptools] include-package-data = true - [tool.setuptools_scm] version_scheme = "post-release" -local_scheme = "no-local-version" \ No newline at end of file +local_scheme = "no-local-version" + +[tool.pytest.ini_options] +testpaths = ["hardwarelibrary/tests"] +python_files = ["test*.py"] +norecursedirs = ["deprecated"] \ No newline at end of file From cc173590963c9ab9d482e2de8e7986966dc7cf3d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Mon, 18 May 2026 12:42:38 -0400 Subject: [PATCH 13/17] Migrate EchoDevice to the Sutter-style debug pattern Restructures EchoDevice to follow the single-class-with-sentinel pattern already used by SutterDevice and IntellidriveDevice rather than maintaining a separate top-level DebugEchoDevice class. EchoDevice changes: - doInitializeDevice now branches on serialNumber == "debug" to pick between SerialPort (real FTDI) and DebugSerialPort (mock). - Adds a nested DebugSerialPort(TableDrivenDebugPort) class that reuses the device's commands dict and implements process_command for ECHO1/ECHO2/ECHO3. - DebugEchoDevice class deleted -- callers now use EchoDevice(serialNumber="debug") instead. Test changes: - TestDebugEchoPhysicalDevice now inherits BaseTestCases.TestPhysicalDeviceBase directly (matching TestSutterPhysicalDevice) rather than from TestEchoPhysicalDevice. The previous inheritance ran TestEchoPhysicalDevice.setUp via super().setUp(), which always raised SkipTest because no real FTDI hardware was attached. Debug tests therefore never actually ran. They run now, with 11 passing and 2 USB-enumeration-dependent tests still skipping. - testEchoCommands duplicated onto TestDebugEchoPhysicalDevice since it no longer inherits from TestEchoPhysicalDevice. Co-Authored-By: Claude Opus 4.7 (1M context) --- hardwarelibrary/echodevice.py | 33 +++++++++++---------- hardwarelibrary/tests/testPhysicalDevice.py | 19 +++++++----- 2 files changed, 29 insertions(+), 23 deletions(-) diff --git a/hardwarelibrary/echodevice.py b/hardwarelibrary/echodevice.py index 57ef738d..56349d3f 100644 --- a/hardwarelibrary/echodevice.py +++ b/hardwarelibrary/echodevice.py @@ -1,8 +1,8 @@ from hardwarelibrary.physicaldevice import * from hardwarelibrary.communication.communicationport import * from hardwarelibrary.communication.serialport import * -from hardwarelibrary.communication.commands import DataCommand -from hardwarelibrary.communication.debugport import DebugPort +from hardwarelibrary.communication.commands import DataCommand, TextCommand +from hardwarelibrary.communication.debugport import TableDrivenDebugPort import re import time @@ -22,22 +22,23 @@ def __init__(self, serialNumber='ftDXIKC4', idProduct=classIdProduct, idVendor=c PhysicalDevice.__init__(self, serialNumber=serialNumber, idProduct=idProduct, idVendor=idVendor) def doInitializeDevice(self): - self.port = SerialPort(idVendor=self.idVendor, idProduct=self.idProduct) + if self.serialNumber == "debug": + self.port = self.DebugSerialPort() + else: + self.port = SerialPort(idVendor=self.idVendor, idProduct=self.idProduct) self.port.open() def doShutdownDevice(self): self.port.close() -class DebugEchoDevice(EchoDevice): - classIdProduct = 0xfffa - classIdVendor = debugClassIdVendor - - def __init__(self, serialNumber='debug', idProduct=classIdProduct, idVendor=classIdVendor): - PhysicalDevice.__init__(self, serialNumber=serialNumber, idProduct=idProduct, idVendor=idVendor) - - def doInitializeDevice(self): - self.port = DebugPort() - self.port.open() - - def doShutdownDevice(self): - self.port.close() + class DebugSerialPort(TableDrivenDebugPort): + def __init__(self): + super().__init__(commands=EchoDevice.commands) + + def process_command(self, name, params, endPointIndex): + if name == 'ECHO1': + return 'someText' + elif name == 'ECHO2': + return 'someOtherText' + elif name == 'ECHO3': + return b'someData' diff --git a/hardwarelibrary/tests/testPhysicalDevice.py b/hardwarelibrary/tests/testPhysicalDevice.py index 85bd1083..6cf1b29c 100644 --- a/hardwarelibrary/tests/testPhysicalDevice.py +++ b/hardwarelibrary/tests/testPhysicalDevice.py @@ -8,7 +8,7 @@ from hardwarelibrary.powermeters import IntegraDevice from hardwarelibrary.spectrometers import USB2000, USB4000, USB2000Plus, StellarNet # from hardwarelibrary.cameras import OpenCVCamera -from hardwarelibrary.echodevice import EchoDevice, DebugEchoDevice +from hardwarelibrary.echodevice import EchoDevice from hardwarelibrary.utils import * @@ -242,14 +242,19 @@ def testEchoCommands(self): self.fail("Unable to send command {0} to device {1}: {2}".format(command.name, self.device, err)) self.device.shutdownDevice() -class TestDebugEchoPhysicalDevice(TestEchoPhysicalDevice): +class TestDebugEchoPhysicalDevice(BaseTestCases.TestPhysicalDeviceBase): def setUp(self): super().setUp() - try: - self.device = DebugEchoDevice() - self.assertIsNotNone(self.device) - except Exception as err: - self.skipTest("No ECHO connected") + self.device = EchoDevice(serialNumber="debug") + + def testEchoCommands(self): + self.device.initializeDevice() + for name, command in self.device.commands.items(): + try: + self.device.sendCommand(command) + except Exception as err: + self.fail("Unable to send command {0} to device {1}: {2}".format(command.name, self.device, err)) + self.device.shutdownDevice() class TestCameraPhysicalDevice(BaseTestCases.TestPhysicalDeviceBase): From 450f327e34f068b524d7c57998210745f9f928aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Mon, 18 May 2026 12:58:33 -0400 Subject: [PATCH 14/17] Migrate CoboltDevice to the TableDrivenDebugPort pattern Replaces the ad-hoc CoboltDebugSerial class with a nested DebugSerialPort(TableDrivenDebugPort) inside CoboltDevice, and moves the protocol definition into a class-level 'commands' dict. Eleven commands captured: GET_POWER, GET_REQUESTED_POWER, SET_POWER, GET_ON_OFF, TURN_ON, TURN_OFF, GET_SERIAL_NUMBER, TURN_AUTOSTART_ON, TURN_AUTOSTART_OFF, GET_AUTOSTART, GET_INTERLOCK. Each is declared once with text_format/replyPattern (for the device send-side) plus matchPattern/responseTemplate (for the debug port to recognize and respond). The doX methods now use commands["NAME"].send(port=self.port) and read .matchGroups for the parsed values, eliminating the parallel hand-rolled regex dispatch that lived in CoboltDebugSerial.write (~120 lines of inline pattern matching). The DebugSerialPort holds the mock device state (power, isOn, autostart, serialNumber, interlock) and process_command returns either a dict (for templated responses) or a raw "OK\r\n" / error string. The background power-ramp thread (increasePowerSlowlyInBackground) is preserved. Manual sanity check: end-to-end roundtrip works via CoboltDevice(bsdPath='debug') -- initializeDevice succeeds, all queries return sensible values, doSetPower(0.5) converges through the background ramp. Co-Authored-By: Claude Opus 4.7 (1M context) --- hardwarelibrary/sources/cobolt.py | 300 ++++++++++++++---------------- 1 file changed, 138 insertions(+), 162 deletions(-) diff --git a/hardwarelibrary/sources/cobolt.py b/hardwarelibrary/sources/cobolt.py index 7b6e2255..8bb91986 100644 --- a/hardwarelibrary/sources/cobolt.py +++ b/hardwarelibrary/sources/cobolt.py @@ -1,5 +1,7 @@ from hardwarelibrary.physicaldevice import * from hardwarelibrary.communication import * +from hardwarelibrary.communication.commands import TextCommand +from hardwarelibrary.communication.debugport import TableDrivenDebugPort from .lasersourcedevice import LaserSourceDevice import re @@ -12,6 +14,69 @@ class CoboltCantTurnOnWithAutostartOn(Exception): pass class CoboltDevice(PhysicalDevice, LaserSourceDevice): + commands = { + "GET_POWER": TextCommand( + name="GET_POWER", + text_format="pa?\r", + replyPattern=r"(\d+\.\d+)", + matchPattern=r"pa\?\r", + responseTemplate="{power:0.4f}\r\n"), + "GET_REQUESTED_POWER": TextCommand( + name="GET_REQUESTED_POWER", + text_format="p?\r", + replyPattern=r"(\d+\.\d+)", + matchPattern=r"p\?\r", + responseTemplate="{requestedPower:0.4f}\r\n"), + "SET_POWER": TextCommand( + name="SET_POWER", + text_format="p {0:0.3f}\r", + replyPattern=r"OK", + matchPattern=r"p (?P\d+\.?\d+)\r"), + "GET_ON_OFF": TextCommand( + name="GET_ON_OFF", + text_format="l?\r", + replyPattern=r"(0|1)", + matchPattern=r"l\?\r", + responseTemplate="{isOn}\r\n"), + "TURN_ON": TextCommand( + name="TURN_ON", + text_format="l1\r", + replyPattern=r"OK", + matchPattern=r"l1\r"), + "TURN_OFF": TextCommand( + name="TURN_OFF", + text_format="l0\r", + replyPattern=r"OK", + matchPattern=r"l0\r"), + "GET_SERIAL_NUMBER": TextCommand( + name="GET_SERIAL_NUMBER", + text_format="sn?\r", + replyPattern=r"(\d+)", + matchPattern=r"sn\?\r", + responseTemplate="{serialNumber}\r\n"), + "TURN_AUTOSTART_ON": TextCommand( + name="TURN_AUTOSTART_ON", + text_format="@cobas 1\r", + replyPattern=r"OK", + matchPattern=r"@cobas 1\r"), + "TURN_AUTOSTART_OFF": TextCommand( + name="TURN_AUTOSTART_OFF", + text_format="@cobas 0\r", + replyPattern=r"OK", + matchPattern=r"@cobas 0\r"), + "GET_AUTOSTART": TextCommand( + name="GET_AUTOSTART", + text_format="@cobas?\r", + replyPattern=r"(0|1)", + matchPattern=r"@cobas\?\r", + responseTemplate="{autostart}\r\n"), + "GET_INTERLOCK": TextCommand( + name="GET_INTERLOCK", + text_format="ilk?\r", + replyPattern=r"(0|1)", + matchPattern=r"ilk\?\r", + responseTemplate="{interlock}\r\n"), + } def __init__(self, bsdPath=None, portPath=None, serialNumber: str = None, idProduct: int = None, idVendor: int = None): @@ -40,6 +105,7 @@ def __del__(self): except: # ignore if already closed return + def autostartIsOn(self) -> bool: self.doGetAutostart() return self.autostart @@ -50,15 +116,15 @@ def turnAutostartOn(self): def turnAutostartOff(self): self.doTurnAutostartOff() - def doInitializeDevice(self): + def doInitializeDevice(self): try: if self.portPath == "debug": - self.port = CommunicationPort(port=CoboltDebugSerial()) + self.port = self.DebugSerialPort() else: self.port = SerialPort(portPath=self.portPath) - + if self.port is None: - raise PhysicalDevice.UnableToInitialize("Cannot allocate port {0}".format(self.portPath)) + raise PhysicalDevice.UnableToInitialize("Cannot allocate port for serial '{0}'".format(self.portPath)) self.port.open() self.doGetLaserSerialNumber() @@ -74,7 +140,6 @@ def doInitializeDevice(self): if self.port is not None and self.port.isOpen: self.port.close() raise PhysicalDevice.UnableToInitialize() - def doShutdownDevice(self): self.port.close() @@ -82,43 +147,47 @@ def doShutdownDevice(self): return def doGetInterlockState(self) -> bool: - value = self.port.writeStringExpectMatchingString('ilk?\r', replyPattern='(0|1)') - self.interlockState = bool(value) + cmd = CoboltDevice.commands["GET_INTERLOCK"] + cmd.send(port=self.port) + self.interlockState = bool(int(cmd.matchGroups[0])) return self.interlockState def doGetLaserSerialNumber(self) -> str: - self.laserSerialNumber = self.port.writeStringReadFirstMatchingGroup('sn?\r', replyPattern='(\\d+)') + cmd = CoboltDevice.commands["GET_SERIAL_NUMBER"] + cmd.send(port=self.port) + self.laserSerialNumber = cmd.matchGroups[0] def doGetOnOffState(self) -> bool: - value = self.port.writeStringExpectMatchingString('l?\r', replyPattern='(0|1)') - self.isOn = (int(value) == 1) - return self.isOn + cmd = CoboltDevice.commands["GET_ON_OFF"] + cmd.send(port=self.port) + self.isOn = (int(cmd.matchGroups[0]) == 1) + return self.isOn def doTurnOn(self): if not self.doGetAutostart(): - self.port.writeStringExpectMatchingString('l1\r', replyPattern='OK') + CoboltDevice.commands["TURN_ON"].send(port=self.port) else: raise CoboltCantTurnOnWithAutostartOn() def doTurnOff(self): - self.port.writeStringExpectMatchingString('l0\r', replyPattern='OK') + CoboltDevice.commands["TURN_OFF"].send(port=self.port) def doGetAutostart(self) -> bool: - value = self.port.writeStringReadFirstMatchingGroup('@cobas?\r', '([1|0])') - self.autostart = (int(value) == 1) + cmd = CoboltDevice.commands["GET_AUTOSTART"] + cmd.send(port=self.port) + self.autostart = (int(cmd.matchGroups[0]) == 1) return self.autostart def doTurnAutostartOn(self): - self.port.writeStringExpectMatchingString('@cobas 1\r', 'OK') + CoboltDevice.commands["TURN_AUTOSTART_ON"].send(port=self.port) self.autostart = True def doTurnAutostartOff(self): - self.port.writeStringExpectMatchingString('@cobas 0\r', 'OK') + CoboltDevice.commands["TURN_AUTOSTART_OFF"].send(port=self.port) self.autostart = False def doSetPower(self, powerInWatts) -> float: - command = 'p {0:0.3f}\r'.format(powerInWatts) - self.port.writeStringExpectMatchingString(command, replyPattern='OK') + CoboltDevice.commands["SET_POWER"].send(port=self.port, params=powerInWatts) actualPower = 0 acceptableDifference = 0.1 * powerInWatts for i in range(10): # It is not an error if we don't converge @@ -130,150 +199,58 @@ def doSetPower(self, powerInWatts) -> float: return actualPower def doGetPower(self) -> float: - value = self.port.writeStringReadFirstMatchingGroup('pa?\r', replyPattern='(\\d.\\d+)') - return float(value) - -class CoboltDebugSerial: - def __init__(self): - self.outputBuffer = bytearray() - self.lineEnding = b'\r' - self.power = 0.1 - self.isOn = 0 - self.requestedPower = 0 - self.autostart = 1 - self._isOpen = True - - @property - def is_open(self): - return self._isOpen - - def open(self): - with globalLock: - if self._isOpen: - raise IOError("port is already open") - else: - self._isOpen = True - - return - - def close(self): - with globalLock: - self._isOpen = False - - return - - def flush(self): - return - - def read(self, length) -> bytearray: - with globalLock: - data = bytearray() - for i in range(0, length): - if len(self.outputBuffer) > 0: - byte = self.outputBuffer.pop(0) - data.append(byte) - else: - raise CommunicationReadTimeout("Unable to read data") - - return data - - - def write(self, data:bytearray) -> int : - with globalLock: - string = data.decode('utf-8') - - match = re.search("pa\\?", string) - if match is not None: - replyData = bytearray("{0:0.4f}\r\n".format(self.power), encoding='utf-8') - self.outputBuffer.extend(replyData) - return len(data) - - match = re.search("p\\?", string) - if match is not None: - replyData = bytearray("{0:0.4f}\r\n".format(self.requestedPower), encoding='utf-8') - self.outputBuffer.extend(replyData) - return len(data) - - match = re.search("p (\\d+\\.?\\d+)\r", string) - if match is not None: - requestedPower = float(match.groups()[0]) - process = Thread(target=increasePowerSlowlyInBackground, - kwargs=dict(port=self, - endPower=requestedPower, - duration=1.0)) - process.start() # will complete in the background - replyData = bytearray("OK\r\n", encoding='utf-8') - self.outputBuffer.extend(replyData) - return len(data) - - match = re.search("l\\?\r", string) - if match is not None: - replyData = bytearray("{0}\r\n".format(self.isOn), encoding='utf-8') - self.outputBuffer.extend(replyData) - return len(data) - - match = re.search("l1\r", string) - if match is not None: - replyData = bytearray() - if self.autostart == 1: - replyData = bytearray("Syntax error: not allowed in autostart mode\r\n", encoding='utf-8') - else: + cmd = CoboltDevice.commands["GET_POWER"] + cmd.send(port=self.port) + return float(cmd.matchGroups[0]) + + class DebugSerialPort(TableDrivenDebugPort): + def __init__(self): + super().__init__(commands=CoboltDevice.commands) + self.power = 0.1 + self.requestedPower = 0 + self.isOn = 0 + self.autostart = 1 + self.serialNumber = "123456" + self.interlock = 1 + + def process_command(self, name, params, endPointIndex): + with globalLock: + if name == "GET_POWER": + return {"power": self.power} + elif name == "GET_REQUESTED_POWER": + return {"requestedPower": self.requestedPower} + elif name == "SET_POWER": + requestedPower = float(params["value"]) + process = Thread(target=increasePowerSlowlyInBackground, + kwargs=dict(port=self, + endPower=requestedPower, + duration=1.0)) + process.start() + return "OK\r\n" + elif name == "GET_ON_OFF": + return {"isOn": self.isOn} + elif name == "TURN_ON": + if self.autostart == 1: + return "Syntax error: not allowed in autostart mode\r\n" self.isOn = 1 - replyData = bytearray("OK\r\n", encoding='utf-8') - self.outputBuffer.extend(replyData) - return len(data) - - match = re.search("l0\r", string) - if match is not None: - self.isOn = 0 - replyData = bytearray("OK\r\n", encoding='utf-8') - self.outputBuffer.extend(replyData) - return len(data) - - - match = re.search("sn\\?\r", string) - if match is not None: - replyData = bytearray("123456\r\n", encoding='utf-8') - self.outputBuffer.extend(replyData) - return len(data) - - match = re.search("@cobas 1\r", string) - if match is not None: - if self.autostart == 0: + return "OK\r\n" + elif name == "TURN_OFF": self.isOn = 0 - else: - pass - self.autostart = 1 - replyData = bytearray("OK\r\n", encoding='utf-8') - self.outputBuffer.extend(replyData) - return len(data) - - match = re.search("@cobas 0\r", string) - if match is not None: - self.autostart = 0 - replyData = bytearray("OK\r\n", encoding='utf-8') - self.outputBuffer.extend(replyData) - return len(data) - - match = re.search("@cobas\\?\r", string) - if match is not None: - replyData = bytearray("{0}\r\n".format(self.autostart), encoding='utf-8') - self.outputBuffer.extend(replyData) - return len(data) - - - match = re.search("ilk\\?\r", string) - if match is not None: - replyData = bytearray("1\r\n", encoding='utf-8') - self.outputBuffer.extend(replyData) - - return len(data) - - # Error (string already includes \n) - replyData = bytearray("Syntax error: {0}\n".format(string), encoding='utf-8') - self.outputBuffer.extend(replyData) - - return len(data) + return "OK\r\n" + elif name == "GET_SERIAL_NUMBER": + return {"serialNumber": self.serialNumber} + elif name == "TURN_AUTOSTART_ON": + if self.autostart == 0: + self.isOn = 0 + self.autostart = 1 + return "OK\r\n" + elif name == "TURN_AUTOSTART_OFF": + self.autostart = 0 + return "OK\r\n" + elif name == "GET_AUTOSTART": + return {"autostart": self.autostart} + elif name == "GET_INTERLOCK": + return {"interlock": self.interlock} def increasePowerSlowlyInBackground(port, endPower, duration): @@ -287,4 +264,3 @@ def increasePowerSlowlyInBackground(port, endPower, duration): print("Unable to set power") time.sleep(0.05) port.power = endPower - From 60edefe8248a419930450d8d6b45f045cabc785f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Mon, 18 May 2026 12:58:42 -0400 Subject: [PATCH 15/17] Remove redundant DebugEchoPort alias DebugEchoPort was a no-op subclass of DebugPort with an empty __init__ that just called super(). It existed as a named alias for "DebugPort used for echo testing", but added no behavior -- the echo logic lives in DebugPort.processInputBuffers. Removed: - hardwarelibrary/communication/echoport.py (the alias) - DebugEchoPort export from hardwarelibrary/communication/__init__.py - TestDebugEchoPort class in testCommunicationPorts.py (was a duplicate of TestDebugPortDefaultsToECho, with the only difference being which alias it instantiated) Updated: - TestSlowDebugEchoPort now instantiates DebugPort() directly instead of DebugEchoPort(). Same behavior; one less indirection. Tests still pass (246 passed, 202 skipped on the full suite). Co-Authored-By: Claude Opus 4.7 (1M context) --- hardwarelibrary/communication/__init__.py | 1 - hardwarelibrary/communication/echoport.py | 10 ---------- hardwarelibrary/tests/testCommunicationPorts.py | 16 +--------------- 3 files changed, 1 insertion(+), 26 deletions(-) delete mode 100644 hardwarelibrary/communication/echoport.py diff --git a/hardwarelibrary/communication/__init__.py b/hardwarelibrary/communication/__init__.py index 69ef7128..5cca2db0 100644 --- a/hardwarelibrary/communication/__init__.py +++ b/hardwarelibrary/communication/__init__.py @@ -3,7 +3,6 @@ from .usbport import USBPort from .diagnostics import USBParameters, DeviceCommand, USBDeviceDescription from .debugport import DebugPort, TableDrivenDebugPort -from .echoport import DebugEchoPort import usb.backend.libusb1 import platform from pathlib import * diff --git a/hardwarelibrary/communication/echoport.py b/hardwarelibrary/communication/echoport.py deleted file mode 100644 index c0f77c45..00000000 --- a/hardwarelibrary/communication/echoport.py +++ /dev/null @@ -1,10 +0,0 @@ -import time -import random - -from hardwarelibrary.communication import * -from threading import Thread, Lock - -class DebugEchoPort(DebugPort): - def __init__(self, delay=0): - super(DebugEchoPort, self).__init__(delay=delay) - diff --git a/hardwarelibrary/tests/testCommunicationPorts.py b/hardwarelibrary/tests/testCommunicationPorts.py index 0fcb740c..880adf63 100644 --- a/hardwarelibrary/tests/testCommunicationPorts.py +++ b/hardwarelibrary/tests/testCommunicationPorts.py @@ -240,20 +240,6 @@ def threadReadWrite(port, index): threadFailed = index -class TestDebugEchoPort(BaseTestCases.TestEchoPort): - - def setUp(self): - self.port = DebugEchoPort() - self.assertIsNotNone(self.port) - self.port.open() - self.assertTrue(self.port.isOpen) - self.port.flush() - - def tearDown(self): - self.port.close() - self.assertFalse(self.port.isOpen) - - class TestDebugPortDefaultsToECho(BaseTestCases.TestEchoPort): def setUp(self): @@ -322,7 +308,7 @@ def testFindConfigureInterfaceEndpointsDevice(self): class TestSlowDebugEchoPort(BaseTestCases.TestEchoPort): def setUp(self): - self.port = DebugEchoPort() + self.port = DebugPort() self.assertIsNotNone(self.port) self.port.open() self.assertTrue(self.port.isOpen) From 55052f60eb7066f5156d6dec3745ac322c451868 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Mon, 18 May 2026 13:14:02 -0400 Subject: [PATCH 16/17] Document DebugPort / TableDrivenDebugPort strategy Adds README-8-DebugPort.md (~340 lines) explaining the debug-port infrastructure that PR #73 introduces: - TL;DR and when to pick DebugPort vs TableDrivenDebugPort. - Class hierarchy diagram (Mermaid) showing CommunicationPort -> DebugPort -> TableDrivenDebugPort and the four device-specific DebugSerialPort subclasses that adopt the pattern. - Reference table of Command attributes by role (send / recognize / respond) for both TextCommand and DataCommand. - Sequence diagram (Mermaid) walking through SutterDevice.doMoveTo end-to-end -- buildSendData, matches, extractParams, process_command, formatResponse, unpackReply -- showing the same Command object consulted on both sides of the wire. - A 7-step recipe for adopting the pattern in a new device, with text-protocol and binary-protocol examples and pointers to the reference implementations. - Edge cases / gotchas section (unmatched input, terminator requirements, replyDataLength, auto-derived match patterns). - "Where it's used" table listing the current adopters plus the remaining candidates (IntegraDevice, OISpectrometer). Also adds a pointer from README-5-Communication-ports.md so the existing README chain leads readers here. Co-Authored-By: Claude Opus 4.7 (1M context) --- README-5-Communication-ports.md | 1 + README-8-DebugPort.md | 344 ++++++++++++++++++++++++++++++++ 2 files changed, 345 insertions(+) create mode 100644 README-8-DebugPort.md diff --git a/README-5-Communication-ports.md b/README-5-Communication-ports.md index 7686022c..9a1adf04 100644 --- a/README-5-Communication-ports.md +++ b/README-5-Communication-ports.md @@ -399,3 +399,4 @@ Important points to notice: This will be discussed later, but `Command`, `TextCommand` and `DataCommand` classes are defined to manage everything in a single entity. 95% of the time, we send a command, read a reply and extract a value from the reply, then do something with this value. This can be encapsulated in a class that will manage all the details for us. +Take a look at [README-8-DebugPort.md]() diff --git a/README-8-DebugPort.md b/README-8-DebugPort.md new file mode 100644 index 00000000..33259744 --- /dev/null +++ b/README-8-DebugPort.md @@ -0,0 +1,344 @@ +# DebugPort, TableDrivenDebugPort, and the dual-role commands dict + +This document explains the debug-port infrastructure used to write **hardware-free mocks** of physical devices, and the recipe for migrating a new device onto it. + +If you've ever wanted to: + +- run device tests on a CI box that has no hardware attached; +- demo the API without plugging in a real laser/stage/spectrometer; +- iterate on protocol code without the slow round-trip to a real serial port; + +…this is the abstraction you want. + +## TL;DR + +- **`DebugPort`** is a fake port that echoes everything back. Useful when you want to test *port-level behavior* without caring what the bytes mean. +- **`TableDrivenDebugPort`** is a `DebugPort` that dispatches incoming bytes against a dict of `Command` objects and lets you implement device-specific reply logic. Useful when you want to test a *device* without hardware. +- The **same `commands` dict** on a `PhysicalDevice` describes both how it sends commands to the real hardware *and* how its `TableDrivenDebugPort` recognizes those same commands when used as a mock. One declaration, two roles, no drift. + +## Class hierarchy + +```mermaid +classDiagram + direction LR + class CommunicationPort { + <> + +open() + +close() + +writeData(data, endPoint) + +readData(length, endPoint) + +writeString(string, endPoint) + +readString(endPoint) + } + + class DebugPort { + +inputBuffers + +outputBuffers + +delay + +processInputBuffers(endPointIndex) + Note: default impl echoes + } + + class TableDrivenDebugPort { + +commands + +processInputBuffers(endPointIndex) + +process_command(name, params, endPointIndex) + Note: dispatches via commands + } + + class SutterDevice_DebugSerialPort { + +xSteps, ySteps, zSteps + +process_command(name, params, _) + } + + class IntellidriveDevice_DebugSerialPort { + +registers + +process_command(name, params, _) + } + + class EchoDevice_DebugSerialPort { + +process_command(name, params, _) + } + + class CoboltDevice_DebugSerialPort { + +power, isOn, autostart + +process_command(name, params, _) + } + + CommunicationPort <|-- DebugPort + DebugPort <|-- TableDrivenDebugPort + TableDrivenDebugPort <|-- SutterDevice_DebugSerialPort + TableDrivenDebugPort <|-- IntellidriveDevice_DebugSerialPort + TableDrivenDebugPort <|-- EchoDevice_DebugSerialPort + TableDrivenDebugPort <|-- CoboltDevice_DebugSerialPort +``` + +Each device that wants a debug mode declares a nested class: + +```python +class SutterDevice(LinearMotionDevice): + commands = { "MOVE": DataCommand(...), ... } + + class DebugSerialPort(TableDrivenDebugPort): + def __init__(self): + super().__init__(commands=SutterDevice.commands) + self.xSteps = 0 + ... + + def process_command(self, name, params, endPointIndex): + if name == "MOVE": + ... +``` + +## When to pick which base class + +| You want to… | Use | +|---|---| +| Verify that a `CommunicationPort` writes/reads bytes correctly | `DebugPort` | +| Test echo-style behavior with arbitrary payloads | `DebugPort` | +| Mock a real device's protocol so tests can exercise its `do*` methods | `TableDrivenDebugPort` subclass | +| Stub *only* the device's high-level methods (no wire protocol) | A `DebugXxxDevice` subclass that overrides `doX` directly — see `DebugLinearMotionDevice`, `DebugRotationDevice` | + +## The dual-role `Command` object + +Every `Command` (`TextCommand`, `MultilineTextCommand`, `DataCommand`) carries two parallel sets of attributes: + +| Role | TextCommand attributes | DataCommand attributes | +|---|---|---| +| **Send** (real device → wire) | `text_format`, `replyPattern`, `alternatePattern` | `data` or `sendFormat`+`sendFields`+`sendDefaults`, `replyDataLength`, `unpackingMask` | +| **Recognize** (mock receives wire bytes) | `matchPattern` (or auto-derived from `text_format`) | `prefix` (or auto-derived from `data[0:1]`), `requestFormat`, `requestFields` | +| **Respond** (mock writes reply) | `responseTemplate` | `responseFormat`, `responseFields` | + +The send side and the recognize side describe the **same protocol** from opposite ends — one packs, the other unpacks. Putting them on a single object means the protocol can never drift between the real driver and its mock. + +## How a roundtrip works + +When a device with `serialNumber="debug"` calls `sendCommand(...)`, the bytes travel through `inputBuffers`, get dispatched by `processInputBuffers`, and the reply comes back through `outputBuffers`. Here's a `SutterDevice.doMoveTo((10, 20, 30))` end-to-end: + +```mermaid +sequenceDiagram + participant DM as SutterDevice (debug mode) + participant CMD as commands["MOVE"] + participant PORT as DebugSerialPort + + DM->>CMD: buildSendData(x=10, y=20, z=30) + Note over CMD: pack via sendFormat='+ sendFields + sendDefaults + CMD-->>DM: b'M\x0a..\x14..\x1e..\r' + + DM->>PORT: writeData(bytes) + Note over PORT: inputBuffers[0] += bytes
processInputBuffers() fires + + PORT->>CMD: matches(inputBytes) + Note over CMD: check prefix b'M' + CMD-->>PORT: True + + PORT->>CMD: extractParams(inputBytes) + Note over CMD: unpack via requestFormat='+ requestFields → dict + CMD-->>PORT: {'x': 10, 'y': 20, 'z': 30} + + PORT->>PORT: process_command('MOVE', {x,y,z}, 0) + Note over PORT: device-specific:
self.xSteps = params['x'] etc.
return b'\r' + + PORT->>CMD: formatResponse(b'\r') + Note over CMD: bytes pass-through
(no responseFormat needed here) + CMD-->>PORT: bytearray(b'\r') + + Note over PORT: outputBuffers[0] += b'\r' + + DM->>PORT: readData(1) + PORT-->>DM: b'\r' + + DM->>CMD: unpackReply(b'\r') + Note over CMD: struct.unpack('>DM: (b'\r',) + Note over DM: doMoveTo verifies and returns +``` + +Notice that the **same `commands["MOVE"]` object** is consulted four times — twice during send (`buildSendData`, `unpackReply`) and twice during recognize (`matches`/`extractParams` for parsing, `formatResponse` for replying). Add one `DataCommand` to the dict and you've described both halves of the protocol. + +## Recipe — getting a new device working with TableDrivenDebugPort + +Follow this 7-step checklist. Reference implementation: `motion/sutterdevice.py` (binary protocol) or `motion/intellidrivedevice.py` (text protocol with named groups). + +### 1. Subclass the family base **and** `PhysicalDevice` + +```python +from hardwarelibrary.physicaldevice import PhysicalDevice +from hardwarelibrary.motion.linearmotiondevice import LinearMotionDevice + +class MyDevice(LinearMotionDevice): + classIdVendor = 0xXXXX + classIdProduct = 0xYYYY +``` + +### 2. Declare the `commands` dict as a class attribute + +Each entry is a `TextCommand`, `MultilineTextCommand`, or `DataCommand` with **both** the send-side and recognize-side attributes filled in. + +For a text protocol: + +```python +commands = { + "GET_REGISTER": TextCommand( + name="GET_REGISTER", + # send side: + text_format="g r{register}\n", + replyPattern=r"v\s(-?\d+)", + # recognize side: + matchPattern=r'g r(?P0x[0-9a-fA-F]+)[\r\n]', + responseTemplate="v {value}\r"), +} +``` + +For a binary protocol: + +```python +commands = { + "MOVE": DataCommand( + name="MOVE", prefix=b'M', + # send side: + sendFormat='...)`) in `matchPattern` and matching `{name}` placeholders in `responseTemplate`. `extractParams` returns a dict — far more readable than positional tuples. +- For `DataCommand`, the `x` characters in struct formats are pad bytes — use them to skip the prefix and terminator when unpacking. +- The send-side and recognize-side formats are usually mirror images: ` float: + cmd = MyDevice.commands["GET_POWER"] + cmd.send(port=self.port) + return float(cmd.matchGroups[0]) +``` + +### 5. Add a nested `DebugSerialPort(TableDrivenDebugPort)` + +Inside the device class, **after** the methods: + +```python +class DebugSerialPort(TableDrivenDebugPort): + def __init__(self): + super().__init__(commands=MyDevice.commands) + # initialize any mock state: + self.registers = {} + + def process_command(self, name, params, endPointIndex): + if name == "GET_REGISTER": + register = params["register"] + return {"value": str(self.registers.get(register, 0))} + elif name == "MOVE": + self.x, self.y, self.z = params["x"], params["y"], params["z"] + return b'\r' + ... +``` + +`process_command` receives: +- `name`: the matched command's name (string) +- `params`: the parsed parameters — a **dict** if you used named groups / `requestFields`, a **tuple** otherwise +- `endPointIndex`: which endpoint (relevant for multi-endpoint USB devices) + +It must return one of: +- `dict` → formatted via `responseTemplate` / `responseFormat`+`responseFields` +- `tuple` → formatted via `responseTemplate` / `responseFormat` positionally +- `bytes` / `bytearray` → returned as-is +- `str` → encoded to UTF-8 +- `None` → no reply sent (input is still consumed) + +### 6. Export from the family's `__init__.py` + +```python +from .mydevice import MyDevice +``` + +### 7. Add tests + +Create `hardwarelibrary/tests/testMyDevice.py`: + +```python +import env +import unittest +from hardwarelibrary.mymodule import MyDevice + +class TestMyDevice(unittest.TestCase): + def setUp(self): + try: + self.device = MyDevice() # real hardware + self.device.initializeDevice() + except Exception: + self.skipTest("No MyDevice connected") + ... + +class TestDebugMyDevice(unittest.TestCase): + def setUp(self): + self.device = MyDevice(serialNumber="debug") + self.device.initializeDevice() + + def testDoMoveTo(self): + self.device.doMoveTo((10, 20, 30)) + self.assertEqual(self.device.doGetPosition(), (10, 20, 30)) +``` + +Follow the established convention: the real-hardware test class **skips** when no hardware is attached; the debug test class **always runs**. + +## Edge cases and gotchas + +- **Unmatched input is discarded silently** (with a `print("Unrecognized command: ...")`). If a test sends something that doesn't match any command, the input buffer is cleared and no reply is generated — `readData`/`readString` on the device side will time out. That's usually the right signal that your `matchPattern` is wrong. + +- **`re.match` anchors at the start.** Patterns like `r'p\?\r'` won't accidentally match `'pa?\r'`. Order of commands in the dict doesn't matter for unambiguous patterns. + +- **Auto-derived match patterns.** For `TextCommand`, if you omit `matchPattern`, one is derived from `text_format` by escaping it and replacing `{…}` placeholders with `(.+?)`. For `DataCommand`, if you omit `prefix`, the first byte of `data` is used. The explicit forms exist for when the auto-derivation isn't expressive enough (e.g., you need a regex character class). + +- **Reply terminator matters for `readString`.** `CommunicationPort.readString()` reads byte-by-byte until `self.terminator` (default `b'\n'`). If your `responseTemplate` doesn't end with `\n`, the read will time out. `\r\n` works because `\n` is the terminator. + +- **`replyDataLength` for `DataCommand`.** Binary replies are read by **fixed length**, not by terminator. Set `replyDataLength` to the exact byte count or you'll either time out (too short) or block (too long). + +## Where it's used in this repo + +| Device | File | Protocol type | +|---|---|---| +| `SutterDevice` | `motion/sutterdevice.py` | Binary (`DataCommand`) | +| `IntellidriveDevice` | `motion/intellidrivedevice.py` | Text with named groups | +| `EchoDevice` | `echodevice.py` | Text + binary (mixed) | +| `CoboltDevice` | `sources/cobolt.py` | Text | + +`IntegraDevice` (`powermeters/integradevice.py`) and `OISpectrometer` (`spectrometers/oceaninsight.py`) use the `commands` dict **without** the table-driven debug port — they still talk to real hardware only. They're candidates for the same migration. + +## See also + +- `hardwarelibrary/communication/commands.py` — `Command` / `TextCommand` / `DataCommand` source with full docstrings +- `hardwarelibrary/communication/debugport.py` — `DebugPort` / `TableDrivenDebugPort` source +- `hardwarelibrary/tests/testCommandRecognition.py` — unit tests for the `Command` machinery (good for understanding semantics) +- `hardwarelibrary/tests/testTableDrivenDebugPort.py` — three minimal fixture classes that show the pattern stripped of device specifics +- `README-4-New-device-coding-example.md` — the older, human-oriented walkthrough of writing a new driver From c961f08123706548d7865a975c86e148fa02755e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Mon, 18 May 2026 13:15:54 -0400 Subject: [PATCH 17/17] Renumber DebugPort README from 8 to 6 and fix link Co-Authored-By: Claude Opus 4.7 (1M context) --- README-5-Communication-ports.md | 2 +- README-8-DebugPort.md => README-6-DebugPort.md | 0 2 files changed, 1 insertion(+), 1 deletion(-) rename README-8-DebugPort.md => README-6-DebugPort.md (100%) diff --git a/README-5-Communication-ports.md b/README-5-Communication-ports.md index 9a1adf04..19a1c6aa 100644 --- a/README-5-Communication-ports.md +++ b/README-5-Communication-ports.md @@ -399,4 +399,4 @@ Important points to notice: This will be discussed later, but `Command`, `TextCommand` and `DataCommand` classes are defined to manage everything in a single entity. 95% of the time, we send a command, read a reply and extract a value from the reply, then do something with this value. This can be encapsulated in a class that will manage all the details for us. -Take a look at [README-8-DebugPort.md]() +Take a look at [README-6-DebugPort.md](README-6-DebugPort.md) diff --git a/README-8-DebugPort.md b/README-6-DebugPort.md similarity index 100% rename from README-8-DebugPort.md rename to README-6-DebugPort.md