From 02005af117e802cfac5a80a177cbf8cfcdba1c8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20C=C3=B4t=C3=A9?= Date: Mon, 18 May 2026 21:57:28 -0400 Subject: [PATCH] Rename Command class attributes around the four conceptual operations Every Command (TextCommand, MultilineTextCommand, DataCommand) carries the same four conceptual operations: requestEncoder encode an outgoing request payload (device -> wire) requestDecoder decode an incoming request payload (wire -> mock) replyEncoder encode an outgoing reply payload (mock -> wire) replyDecoder decode an incoming reply payload (wire -> device) Before this commit, those operations were named ad-hoc per subclass -- TextCommand used text_format/matchPattern/replyPattern/responseTemplate while DataCommand used sendFormat+sendFields+sendDefaults / requestFormat+ requestFields+prefix / unpackingMask+replyDataLength / responseFormat+ responseFields. Twenty-two attribute slots describing four conceptual roles, no two sharing a name across subclasses. This commit gives every Command subclass the same four attribute names for the same four operations. For TextCommand they hold strings (format strings for encoders, regexes for decoders). For DataCommand they hold small frozen dataclasses (DataEncoder bundles struct format + fields + defaults; DataDecoder bundles struct format + fields + length + prefix). Deleted attributes: - alternatePattern: TextCommand + MultilineTextCommand kwarg that was plumbed through readMatchingGroups but never actually set by any Command in the codebase. Fold into replyDecoder regex if needed. - replyHexRegex: DataCommand kwarg whose only code path raised NotImplementedError. Dead. Migration of every Command in the codebase to the new constructor: - SutterDevice (4 commands): each DataCommand now wraps the struct format + fields + defaults inside DataEncoder/DataDecoder. The MOVE command goes from 7 keyword args + a prefix=b'M' to one tidy requestEncoder/requestDecoder/replyDecoder triple. - IntellidriveDevice (3 text commands), IntegraDevice (5 text/multiline commands), EchoDevice (3 mixed commands), CoboltDevice (11 text commands): same pattern, attribute names renamed. - testCommandRecognition.py (38 tests), testTableDrivenDebugPort.py (3 fixtures + 12 tests), testCommunicationPorts.py (5 tests): all rewritten to use the new constructor signatures. Internal property renames: - TextCommand.effectiveMatchPattern -> effectiveDecoder - TextCommand._autoMatchPattern (was a property) -> _autoMatchPattern() (now a method, since it's called rather than read) These are private/dunder-ish; the only tests that referenced them were updated. Verified: - 261 passed, 203 skipped, 0 failures on the full test suite. The skipped tests are all hardware-gated (no spectrometer / no Sutter / no LabJack attached) and were also skipping before this commit. Closes the first item of the redesign discussed in PR #85's review: "TextCommand and DataCommand don't share a shape". The remaining items (separate Command schema from per-call instance, unify error model, mock hook contract) are left for follow-up work. Co-Authored-By: Claude Opus 4.7 (1M context) --- hardwarelibrary/communication/commands.py | 447 +++++++++--------- hardwarelibrary/echodevice.py | 11 +- hardwarelibrary/motion/intellidrivedevice.py | 24 +- hardwarelibrary/motion/sutterdevice.py | 43 +- hardwarelibrary/powermeters/integradevice.py | 10 +- hardwarelibrary/sources/cobolt.py | 111 ++--- .../tests/testCommandRecognition.py | 96 ++-- .../tests/testCommunicationPorts.py | 10 +- .../tests/testTableDrivenDebugPort.py | 34 +- 9 files changed, 417 insertions(+), 369 deletions(-) diff --git a/hardwarelibrary/communication/commands.py b/hardwarelibrary/communication/commands.py index eb9d3802..2e542404 100644 --- a/hardwarelibrary/communication/commands.py +++ b/hardwarelibrary/communication/commands.py @@ -1,10 +1,10 @@ """Command classes that describe a device's communication protocol. -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. +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: @@ -13,39 +13,78 @@ 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 — + 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. + format. -Because both sides share the same Command definitions, the protocol is -defined once and cannot drift between real and mock implementations. +Every Command has four conceptual operations, named consistently across +``TextCommand`` and ``DataCommand``: -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 + requestEncoder encode an outgoing request payload (device -> wire) + requestDecoder decode an incoming request payload (wire -> mock) + replyEncoder encode an outgoing reply payload (mock -> wire) + replyDecoder decode an incoming reply payload (wire -> device) + +For ``TextCommand`` each is a string -- a Python format string for the +encoders, a regex for the decoders. + +For ``DataCommand`` each is a small dataclass (``DataEncoder`` / +``DataDecoder``) that bundles a struct format with its field names. """ import re import struct +from dataclasses import dataclass, field +from typing import Optional + + +@dataclass(frozen=True) +class DataEncoder: + """Schema for serializing structured params into bytes (DataCommand). + + Attributes: + format: struct format string, e.g. ' positional tuple unpack. + length: bytes to read from the port. Used by replyDecoder; ignored by requestDecoder. + prefix: leading bytes that identify the command for mock dispatch. + Used by requestDecoder; ignored by replyDecoder. + """ + format: str = '' + fields: tuple = () + length: int = 0 + prefix: Optional[bytes] = None + 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. + A Command has a name and tracks per-call state (reply, matchGroups, + exceptions). 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 + name: identifier (key in the device's ``commands`` dict) + endPoints: tuple of (writeEndPoint, readEndPoint) for USB + devices that use distinct endpoints """ - def __init__(self, name:str, endPoints = (None, None)): + def __init__(self, name: str, endPoints=(None, None)): self.name = name self.reply = None self.matchGroups = None @@ -76,32 +115,19 @@ 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") + raise NotImplementedError("Subclasses must implement send()") 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. + Handles the common cases (None / bytes / str). Subclasses extend + for typed encoding (dict + responseTemplate, dict + responseFormat). """ if result is None: return None @@ -111,88 +137,102 @@ def formatResponse(self, result): return bytearray(result.encode('utf-8')) 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: 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_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(). + Each of the four operations is a string -- a format string for encoders, + a regex for decoders. + + Attributes: + requestEncoder: format string for the outgoing request, + e.g. "g r{register}\\n" + requestDecoder: regex with named groups for the mock to recognize + an incoming request, e.g. + r'g r(?P0x[0-9a-fA-F]+)[\\r\\n]'. + If None, derived from requestEncoder by replacing + each {...} placeholder with (.+?). + replyDecoder: regex to parse the device's reply, + e.g. r'v\\s(-?\\d+)'. + replyEncoder: format string for the mock's reply, + e.g. "v {value}\\r". Example: - 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") + TextCommand( + name="GET_REGISTER", + requestEncoder="g r{register}\\n", + requestDecoder=r'g r(?P0x[0-9a-fA-F]+)[\\r\\n]', + replyDecoder=r'v\\s(-?\\d+)', + replyEncoder="v {value}\\r", + ) """ - def __init__(self, name, text_format, replyPattern = None, - alternatePattern = None, - endPoints = (None, None), - matchPattern = None, - responseTemplate = None): + def __init__(self, name, + requestEncoder=None, + requestDecoder=None, + replyDecoder=None, + replyEncoder=None, + endPoints=(None, None)): Command.__init__(self, name, endPoints=endPoints) - self.text_format : str = text_format - self.replyPattern: str = replyPattern - self.alternatePattern: str = alternatePattern - self.matchPattern: str = matchPattern - self.responseTemplate: str = responseTemplate + self.requestEncoder = requestEncoder + self.requestDecoder = requestDecoder + self.replyDecoder = replyDecoder + self.replyEncoder = replyEncoder @property def payload(self): - return self.text_format + return self.requestEncoder @property def numberOfArguments(self): - match = re.search(r"\{(.*?)\}", self.text_format) + if self.requestEncoder is None: + return 0 + match = re.search(r"\{(.*?)\}", self.requestEncoder) if match is None: return 0 return len(match.groups()) - @property def _autoMatchPattern(self): - """Derive a regex from the text format string by replacing {…} - placeholders with (.+?) capture groups.""" - parts = re.split(r'\{[^}]*\}', self.text_format) + """Derive a regex from requestEncoder by escaping it and replacing + every {...} placeholder with a (.+?) capture group.""" + if self.requestEncoder is None: + return None + parts = re.split(r'\{[^}]*\}', self.requestEncoder) 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 effectiveDecoder(self): + """The regex used by matches/extractParams: requestDecoder if set, + otherwise the auto-derived pattern from requestEncoder.""" + if self.requestDecoder is not None: + return self.requestDecoder + return self._autoMatchPattern() def matches(self, inputBytes): - """Return True if inputBytes matches this text command's pattern.""" + pattern = self.effectiveDecoder + if pattern is None: + return False try: inputStr = inputBytes.decode('utf-8', errors='replace') except Exception: return False - return re.match(self.effectiveMatchPattern, inputStr) is not None + return re.match(pattern, inputStr) is not None def extractParams(self, inputBytes): - """Extract parameters from the input bytes using the match pattern. + """Extract params from inputBytes using effectiveDecoder. Returns a dict if the pattern uses named groups (?P...), - otherwise a tuple of positional groups.""" + otherwise a tuple of positional capture groups. + """ + pattern = self.effectiveDecoder + if pattern is None: + return () try: inputStr = inputBytes.decode('utf-8', errors='replace') except Exception: return () - match = re.match(self.effectiveMatchPattern, inputStr) + match = re.match(pattern, inputStr) if match: if match.groupdict(): return match.groupdict() @@ -200,26 +240,26 @@ def extractParams(self, inputBytes): return () def formatResponse(self, result): - """Format a mock response using the responseTemplate. + """Format a mock response using replyEncoder. - - 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) + - dict result + replyEncoder -> replyEncoder.format(**result) + - tuple result + replyEncoder -> replyEncoder.format(*result) + - other types -> delegated to base (bytes/str/None) """ - if isinstance(result, dict) and self.responseTemplate is not None: - formatted = self.responseTemplate.format(**result) + if isinstance(result, dict) and self.replyEncoder is not None: + formatted = self.replyEncoder.format(**result) return bytearray(formatted.encode('utf-8')) - if isinstance(result, tuple) and self.responseTemplate is not None: - formatted = self.responseTemplate.format(*result) + if isinstance(result, tuple) and self.replyEncoder is not None: + formatted = self.replyEncoder.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: - textCommand = self.text_format.format(params) + textCommand = self.requestEncoder.format(params) else: - textCommand = self.text_format + textCommand = self.requestEncoder if port is None: raise RuntimeError("port cannot be None") @@ -227,14 +267,11 @@ def send(self, port, params=None) -> bool: port.writeString(string=textCommand, endPoint=self.endPoints[0]) self.isSent = True - if self.replyPattern is not None: + if self.replyDecoder is not None: self.reply, self.matchGroups = port.readMatchingGroups( - replyPattern=self.replyPattern, - alternatePattern=self.alternatePattern, - endPoint=self.endPoints[1]) - else: - pass - + replyPattern=self.replyDecoder, + endPoint=self.endPoints[1]) + self.isSentSuccessfully = True except Exception as err: self.exceptions.append(err) @@ -245,40 +282,42 @@ def send(self, port, params=None) -> bool: class MultilineTextCommand(Command): - """A text command that expects a multi-line reply. + """A TextCommand variant whose reply spans multiple lines. 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. + replyDecoder; the collected results land in self.reply / self.matchGroups + as lists. - Args: - lineCount: number of reply lines to read (if > 1) + Attributes: + requestEncoder: format string for the outgoing request + replyDecoder: regex to parse each reply line + lineCount: number of reply lines to read (if > 1) lastLinePattern: regex that signals the final line (alternative to lineCount) """ - def __init__(self, name, text_format, - replyPattern=None, - alternatePattern=None, + def __init__(self, name, + requestEncoder=None, + replyDecoder=None, lineCount=1, lastLinePattern=None, endPoints=(None, None)): Command.__init__(self, name=name, endPoints=endPoints) - self.text_format: str = text_format - self.replyPattern: str = replyPattern - self.alternatePattern: str = alternatePattern + self.requestEncoder = requestEncoder + self.replyDecoder = replyDecoder self.lineCount = lineCount self.lastLinePattern = lastLinePattern @property def payload(self): - return self.text_format + return self.requestEncoder def send(self, port, params=None) -> bool: try: if params is not None: - textCommand = self.text_format.format(params) + textCommand = self.requestEncoder.format(params) else: - textCommand = self.text_format + textCommand = self.requestEncoder if port is None: raise RuntimeError("port cannot be None") @@ -289,27 +328,21 @@ def send(self, port, params=None) -> bool: if self.lineCount > 1: self.reply = [] self.matchGroups = [] - for i in range(self.lineCount): reply, matchGroups = port.readMatchingGroups( - replyPattern=self.replyPattern, - alternatePattern=self.alternatePattern, + replyPattern=self.replyDecoder, endPoint=self.endPoints[1]) self.reply.append(reply) self.matchGroups.append(matchGroups) - elif self.lastLinePattern is not None: self.reply = [] self.matchGroups = [] - while True: reply, matchGroups = port.readMatchingGroups( - replyPattern=self.replyPattern, - alternatePattern=self.alternatePattern, + replyPattern=self.replyDecoder, endPoint=self.endPoints[1]) self.reply.append(reply) self.matchGroups.append(matchGroups) - if re.search(self.lastLinePattern, reply) is not None: break else: @@ -325,57 +358,46 @@ 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. + """A command that communicates via binary struct-packed bytes. + + Used for devices with binary protocols like the Sutter MP-285 where + commands are single-byte prefixes followed by packed integers. + + Attributes: + requestEncoder: DataEncoder for the outgoing request bytes + requestDecoder: DataDecoder used by a mock to parse the incoming request + replyDecoder: DataDecoder used by the device to parse the reply + (its ``length`` field tells how many bytes to read) + replyEncoder: DataEncoder used by a mock to pack its reply + data: precomputed request bytes. Used when requestEncoder is + None (e.g. fixed commands with no parameters), and also + provides a prefix fallback: ``data[0:1]`` is the + command prefix when ``requestDecoder.prefix`` is unset. Example: - DataCommand(name="MOVE", prefix=b'M', - requestFormat=' 0: return self.data[0:1] 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 @@ -401,65 +423,68 @@ 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: - values = struct.unpack(self.requestFormat, bytes(inputBytes[:unpackLen])) - if self.requestFields is not None: - return dict(zip(self.requestFields, values)) - return values - return () + """Unpack params from inputBytes via requestDecoder. + + Returns a dict if requestDecoder.fields is set (named unpack), + otherwise a positional tuple from struct.unpack. + """ + if self.requestDecoder is None: + return () + fmt = self.requestDecoder.format + if not fmt: + return () + unpackLen = struct.calcsize(fmt) + if len(inputBytes) < unpackLen: + return () + values = struct.unpack(fmt, bytes(inputBytes[:unpackLen])) + if self.requestDecoder.fields: + return dict(zip(self.requestDecoder.fields, values)) + return values def formatResponse(self, result): - """Pack a mock response using responseFormat. + """Pack a mock response using replyEncoder. - - 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) + - dict result + replyEncoder.fields -> values ordered by fields, then packed + - tuple result + replyEncoder -> packed directly via struct.pack + - other types -> delegated to base """ - 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) + if self.replyEncoder is not None: + fmt = self.replyEncoder.format + fields = self.replyEncoder.fields + if isinstance(result, dict) and fields: + values = tuple(result[f] for f in fields) + return bytearray(struct.pack(fmt, *values)) + if isinstance(result, tuple): + return bytearray(struct.pack(fmt, *result)) 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: + """Build outgoing bytes from requestEncoder + named params + defaults. + Falls back to self.data when requestEncoder is not set.""" + if self.requestEncoder is None: return self.data - merged = dict(self.sendDefaults) if self.sendDefaults else {} + merged = dict(self.requestEncoder.defaults) merged.update(params) - values = tuple(merged[f] for f in self.sendFields) - return struct.pack(self.sendFormat, *values) + values = tuple(merged[f] for f in self.requestEncoder.fields) + return struct.pack(self.requestEncoder.format, *values) def unpackReply(self, replyBytes): - """Unpack reply bytes using unpackingMask. - Returns raw bytes if unpackingMask is not set.""" - if self.unpackingMask is None: + """Unpack reply bytes via replyDecoder.format. + Returns raw bytes if replyDecoder is not set.""" + if self.replyDecoder is None or not self.replyDecoder.format: return replyBytes - return struct.unpack(self.unpackingMask, replyBytes) + return struct.unpack(self.replyDecoder.format, replyBytes) def send(self, port) -> bool: try: - nBytes = port.writeData(data=self.data, endPoint=self.endPoints[0]) + port.writeData(data=self.data, endPoint=self.endPoints[0]) self.isSent = True - 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) + if self.replyDecoder is not None and self.replyDecoder.length > 0: + self.reply = port.readData(length=self.replyDecoder.length) self.isSentSuccessfully = True except Exception as err: self.exceptions.append(err) self.isSentSuccessfully = False - raise(err) + raise return False diff --git a/hardwarelibrary/echodevice.py b/hardwarelibrary/echodevice.py index 56349d3f..6705924d 100644 --- a/hardwarelibrary/echodevice.py +++ b/hardwarelibrary/echodevice.py @@ -1,7 +1,7 @@ from hardwarelibrary.physicaldevice import * from hardwarelibrary.communication.communicationport import * from hardwarelibrary.communication.serialport import * -from hardwarelibrary.communication.commands import DataCommand, TextCommand +from hardwarelibrary.communication.commands import DataCommand, DataDecoder, TextCommand from hardwarelibrary.communication.debugport import TableDrivenDebugPort import re @@ -14,9 +14,12 @@ class EchoDevice(PhysicalDevice): classIdProduct = 0x6001 classIdVendor = 0x0403 - 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"))} + commands = { + "ECHO1": TextCommand(name="ECHO1", requestEncoder="someText", replyDecoder="someText"), + "ECHO2": TextCommand(name="ECHO2", requestEncoder="someOtherText", replyDecoder="someOtherText"), + "ECHO3": DataCommand(name="ECHO3", data=b"someData", + replyDecoder=DataDecoder(length=len(b"someData"))), + } def __init__(self, serialNumber='ftDXIKC4', idProduct=classIdProduct, idVendor=classIdVendor): PhysicalDevice.__init__(self, serialNumber=serialNumber, idProduct=idProduct, idVendor=idVendor) diff --git a/hardwarelibrary/motion/intellidrivedevice.py b/hardwarelibrary/motion/intellidrivedevice.py index 8fd3fefb..e54430e7 100644 --- a/hardwarelibrary/motion/intellidrivedevice.py +++ b/hardwarelibrary/motion/intellidrivedevice.py @@ -24,15 +24,21 @@ class IntellidriveDevice(RotationDevice): classIdProduct = 0x6001 commands = { - "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_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_format="t {mode}\n", - matchPattern=r't (?P\d+)[\r\n]', - replyPattern="ok", responseTemplate="ok\r"), + "SET_REGISTER": TextCommand(name="SET_REGISTER", + requestEncoder="s r{register} {value}\r", + requestDecoder=r's r(?P0x[0-9a-fA-F]+) (?P-?\d+)[\r\n]', + replyDecoder="ok", + replyEncoder="ok\r"), + "GET_REGISTER": TextCommand(name="GET_REGISTER", + requestEncoder="g r{register}\n", + requestDecoder=r'g r(?P0x[0-9a-fA-F]+)[\r\n]', + replyDecoder=r'v\s(-?\d+)', + replyEncoder="v {value}\r"), + "TRAJECTORY": TextCommand(name="TRAJECTORY", + requestEncoder="t {mode}\n", + requestDecoder=r't (?P\d+)[\r\n]', + replyDecoder="ok", + replyEncoder="ok\r"), } def __init__(self, serialNumber): diff --git a/hardwarelibrary/motion/sutterdevice.py b/hardwarelibrary/motion/sutterdevice.py index c7abc73a..a2eb3da1 100644 --- a/hardwarelibrary/motion/sutterdevice.py +++ b/hardwarelibrary/motion/sutterdevice.py @@ -3,7 +3,7 @@ from hardwarelibrary.communication.communicationport import * from hardwarelibrary.communication.usbport import USBPort from hardwarelibrary.communication.serialport import SerialPort -from hardwarelibrary.communication.commands import DataCommand +from hardwarelibrary.communication.commands import DataCommand, DataEncoder, DataDecoder from hardwarelibrary.communication.debugport import TableDrivenDebugPort import re @@ -17,23 +17,26 @@ class SutterDevice(LinearMotionDevice): classIdProduct = 1 commands = { - "MOVE": DataCommand(name="MOVE", prefix=b'M', - sendFormat=' 0: - return self.readReply(cmd.replyDataLength, cmd.unpackingMask) + if cmd.replyDecoder is not None and cmd.replyDecoder.length > 0: + return self.readReply(cmd.replyDecoder.length, cmd.replyDecoder.format) return None def positionInMicrosteps(self) -> (int, int, int): # for compatibility diff --git a/hardwarelibrary/powermeters/integradevice.py b/hardwarelibrary/powermeters/integradevice.py index d9c95aff..c2a08cb4 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_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}") + "GETPOWER": TextCommand(name="GETPOWER", requestEncoder="*CVU", replyDecoder=r"(.+?)\r\n"), + "VERSION": TextCommand(name="VERSION", requestEncoder="*VER", replyDecoder=r"(.+?)\r\n"), + "STATUS": MultilineTextCommand(name="STATUS", requestEncoder="*STS", replyDecoder=r"(.+?)\r\n", lastLinePattern=":100000000"), + "GETWAVELENGTH": TextCommand(name="GETWAVELENGTH", requestEncoder="*GWL", replyDecoder=r"PWC\s*:\s*(.+?)\r\n"), + "SETWAVELENGTH": TextCommand(name="SETWAVELENGTH", requestEncoder="*PWC{0:05d}"), } def __init__(self, serialNumber:str = None, idProduct:int = 0x0300, idVendor:int = 0x1ad5): diff --git a/hardwarelibrary/sources/cobolt.py b/hardwarelibrary/sources/cobolt.py index 4b6e6468..8d5e2189 100644 --- a/hardwarelibrary/sources/cobolt.py +++ b/hardwarelibrary/sources/cobolt.py @@ -15,67 +15,56 @@ class CoboltCantTurnOnWithAutostartOn(Exception): class CoboltDevice(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"), + "GET_POWER": TextCommand(name="GET_POWER", + requestEncoder="pa?\r", + requestDecoder=r"pa\?\r", + replyDecoder=r"(\d+\.\d+)", + replyEncoder="{power:0.4f}\r\n"), + "GET_REQUESTED_POWER": TextCommand(name="GET_REQUESTED_POWER", + requestEncoder="p?\r", + requestDecoder=r"p\?\r", + replyDecoder=r"(\d+\.\d+)", + replyEncoder="{requestedPower:0.4f}\r\n"), + "SET_POWER": TextCommand(name="SET_POWER", + requestEncoder="p {0:0.3f}\r", + requestDecoder=r"p (?P\d+\.?\d+)\r", + replyDecoder=r"OK"), + "GET_ON_OFF": TextCommand(name="GET_ON_OFF", + requestEncoder="l?\r", + requestDecoder=r"l\?\r", + replyDecoder=r"(0|1)", + replyEncoder="{isOn}\r\n"), + "TURN_ON": TextCommand(name="TURN_ON", + requestEncoder="l1\r", + requestDecoder=r"l1\r", + replyDecoder=r"OK"), + "TURN_OFF": TextCommand(name="TURN_OFF", + requestEncoder="l0\r", + requestDecoder=r"l0\r", + replyDecoder=r"OK"), + "GET_SERIAL_NUMBER": TextCommand(name="GET_SERIAL_NUMBER", + requestEncoder="sn?\r", + requestDecoder=r"sn\?\r", + replyDecoder=r"(\d+)", + replyEncoder="{serialNumber}\r\n"), + "TURN_AUTOSTART_ON": TextCommand(name="TURN_AUTOSTART_ON", + requestEncoder="@cobas 1\r", + requestDecoder=r"@cobas 1\r", + replyDecoder=r"OK"), + "TURN_AUTOSTART_OFF": TextCommand(name="TURN_AUTOSTART_OFF", + requestEncoder="@cobas 0\r", + requestDecoder=r"@cobas 0\r", + replyDecoder=r"OK"), + "GET_AUTOSTART": TextCommand(name="GET_AUTOSTART", + requestEncoder="@cobas?\r", + requestDecoder=r"@cobas\?\r", + replyDecoder=r"(0|1)", + replyEncoder="{autostart}\r\n"), + "GET_INTERLOCK": TextCommand(name="GET_INTERLOCK", + requestEncoder="ilk?\r", + requestDecoder=r"ilk\?\r", + replyDecoder=r"(0|1)", + replyEncoder="{interlock}\r\n"), } def __init__(self, bsdPath=None, portPath=None, serialNumber: str = None, diff --git a/hardwarelibrary/tests/testCommandRecognition.py b/hardwarelibrary/tests/testCommandRecognition.py index f23ea836..491119d2 100644 --- a/hardwarelibrary/tests/testCommandRecognition.py +++ b/hardwarelibrary/tests/testCommandRecognition.py @@ -2,13 +2,17 @@ import unittest from struct import pack -from hardwarelibrary.communication.commands import Command, TextCommand, DataCommand +from hardwarelibrary.communication.commands import ( + Command, TextCommand, DataCommand, DataEncoder, DataDecoder, +) class TestCommandBase(unittest.TestCase): - def testMatchesReturnsFalse(self): + def testMatchesReturnsAlwaysFalse(self): cmd = Command(name="base") self.assertFalse(cmd.matches(b'anything')) + cmd = Command(name="base") + self.assertFalse(cmd.matches(b'base')) def testExtractParamsReturnsEmpty(self): cmd = Command(name="base") @@ -29,77 +33,77 @@ def testFormatResponseString(self): class TestTextCommandRecognition(unittest.TestCase): def testAutoMatchPattern(self): - cmd = TextCommand(name="test", text_format="SET {0} {1}\r") + cmd = TextCommand(name="test", requestEncoder="SET {0} {1}\r") self.assertTrue(cmd.matches(b'SET foo 42\r')) def testAutoMatchPatternNoMatch(self): - cmd = TextCommand(name="test", text_format="SET {0} {1}\r") + cmd = TextCommand(name="test", requestEncoder="SET {0} {1}\r") self.assertFalse(cmd.matches(b'GET foo\r')) def testExplicitMatchPattern(self): - cmd = TextCommand(name="test", text_format="s r{0}\r", - matchPattern=r's r(0x[0-9a-fA-F]+)[\r\n]') + cmd = TextCommand(name="test", requestEncoder="s r{0}\r", + requestDecoder=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_format="SET {0} {1}\r") + cmd = TextCommand(name="test", requestEncoder="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_format="g r{0}\n", - matchPattern=r'g r(0x[0-9a-fA-F]+)[\r\n]') + cmd = TextCommand(name="test", requestEncoder="g r{0}\n", + requestDecoder=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_format="g r{0}\n", - responseTemplate="v {0}\r") + cmd = TextCommand(name="test", requestEncoder="g r{0}\n", + replyEncoder="v {0}\r") result = cmd.formatResponse(("42",)) self.assertEqual(result, bytearray(b'v 42\r')) def testFormatResponseStringFallback(self): - cmd = TextCommand(name="test", text_format="cmd\r") + cmd = TextCommand(name="test", requestEncoder="cmd\r") result = cmd.formatResponse("ok\r") self.assertEqual(result, bytearray(b'ok\r')) def testFormatResponseNone(self): - cmd = TextCommand(name="test", text_format="cmd\r") + cmd = TextCommand(name="test", requestEncoder="cmd\r") self.assertIsNone(cmd.formatResponse(None)) def testNamedGroupExtractParams(self): - cmd = TextCommand(name="test", text_format="SET {key} {value}\r", - matchPattern=r'SET (?P\w+) (?P-?\d+)\r') + cmd = TextCommand(name="test", requestEncoder="SET {key} {value}\r", + requestDecoder=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_format="GET {key}\r", - matchPattern=r'GET (?P\w+)\r', - responseTemplate="VAL {value}\r") + cmd = TextCommand(name="test", requestEncoder="GET {key}\r", + requestDecoder=r'GET (?P\w+)\r', + replyEncoder="VAL {value}\r") result = cmd.formatResponse({"value": "99"}) self.assertEqual(result, bytearray(b'VAL 99\r')) - def testEffectiveMatchPatternUsesExplicit(self): - cmd = TextCommand(name="test", text_format="SET {0}\r", - matchPattern=r'SET (\w+)\r') - self.assertEqual(cmd.effectiveMatchPattern, r'SET (\w+)\r') + def testEffectiveDecoderUsesExplicit(self): + cmd = TextCommand(name="test", requestEncoder="SET {0}\r", + requestDecoder=r'SET (\w+)\r') + self.assertEqual(cmd.effectiveDecoder, r'SET (\w+)\r') - def testEffectiveMatchPatternUsesAuto(self): - cmd = TextCommand(name="test", text_format="SET {0}\r") - self.assertEqual(cmd.effectiveMatchPattern, cmd._autoMatchPattern) + def testEffectiveDecoderUsesAuto(self): + cmd = TextCommand(name="test", requestEncoder="SET {0}\r") + self.assertEqual(cmd.effectiveDecoder, cmd._autoMatchPattern()) class TestDataCommandRecognition(unittest.TestCase): def testMatchesByPrefix(self): - cmd = DataCommand(name="test", prefix=b'M') + cmd = DataCommand(name="test", requestDecoder=DataDecoder(prefix=b'M')) self.assertTrue(cmd.matches(b'M\x01\x02')) def testMatchesCaseInsensitive(self): - cmd = DataCommand(name="test", prefix=b'M') + cmd = DataCommand(name="test", requestDecoder=DataDecoder(prefix=b'M')) self.assertTrue(cmd.matches(b'm\x01\x02')) def testPrefixDerivedFromData(self): @@ -108,36 +112,40 @@ def testPrefixDerivedFromData(self): self.assertTrue(cmd.matches(b'c')) def testNoMatchWrongPrefix(self): - cmd = DataCommand(name="test", prefix=b'M') + cmd = DataCommand(name="test", requestDecoder=DataDecoder(prefix=b'M')) self.assertFalse(cmd.matches(b'G\x01')) def testExtractParamsWithFormat(self): - cmd = DataCommand(name="test", prefix=b'S', requestFormat='