diff --git a/README-5-Communication-ports.md b/README-5-Communication-ports.md index 7686022c..19a1c6aa 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-6-DebugPort.md](README-6-DebugPort.md) diff --git a/README-6-DebugPort.md b/README-6-DebugPort.md new file mode 100644 index 00000000..33259744 --- /dev/null +++ b/README-6-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 diff --git a/hardwarelibrary/communication/__init__.py b/hardwarelibrary/communication/__init__.py index bc0034e1..5cca2db0 100644 --- a/hardwarelibrary/communication/__init__.py +++ b/hardwarelibrary/communication/__init__.py @@ -2,8 +2,7 @@ from .serialport import SerialPort from .usbport import USBPort from .diagnostics import USBParameters, DeviceCommand, USBDeviceDescription -from .debugport import DebugPort -from .echoport import DebugEchoPort +from .debugport import DebugPort, TableDrivenDebugPort import usb.backend.libusb1 import platform from pathlib import * diff --git a/hardwarelibrary/communication/commands.py b/hardwarelibrary/communication/commands.py index 99cd7290..eb9d3802 100644 --- a/hardwarelibrary/communication/commands.py +++ b/hardwarelibrary/communication/commands.py @@ -1,6 +1,50 @@ +"""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. + +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") + 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 @@ -26,40 +70,156 @@ 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 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)): + 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)): + """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(). + + 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") + """ + + 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 + self.responseTemplate: str = responseTemplate @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()) + @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) + 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: + return False + 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: + return () + match = re.match(self.effectiveMatchPattern, inputStr) + if match: + if match.groupdict(): + return match.groupdict() + return match.groups() + 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')) + 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: - 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") @@ -85,14 +245,25 @@ def send(self, port, params=None) -> bool: class MultilineTextCommand(Command): - def __init__(self, name, text, + """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_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 @@ -100,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") @@ -154,17 +325,128 @@ def send(self, port, params=None) -> bool: class DataCommand(Command): - def __init__(self, name, data, replyHexRegex = None, replyDataLength = 0, unpackingMask = None, endPoints = (None, None)): + """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: + 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 + prefixLen = len(prefix) + if len(inputBytes) < prefixLen: + return False + 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 () + + 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) + return bytearray(data) + if isinstance(result, tuple) and self.responseFormat is not None: + data = struct.pack(self.responseFormat, *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/communication/debugport.py b/hardwarelibrary/communication/debugport.py index 8839134b..f2b54088 100644 --- a/hardwarelibrary/communication/debugport.py +++ b/hardwarelibrary/communication/debugport.py @@ -1,75 +1,57 @@ import time import random +import struct +import re 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): endPointIndex = 0 if endPoint is None else endPoint 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 @@ -83,14 +65,99 @@ 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 self.writeToOutputBuffer(inputBytes, endPointIndex) - self.inputBuffers[endPointIndex] = bytearray() \ No newline at end of file + self.inputBuffers[endPointIndex] = bytearray() + + +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_format="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 {} + + def processInputBuffers(self, endPointIndex): + inputBytes = self.inputBuffers[endPointIndex] + if len(inputBytes) == 0: + return + + for cmd in self.commands.values(): + if cmd.matches(inputBytes): + params = cmd.extractParams(inputBytes) + result = self.process_command(cmd.name, params, 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 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/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/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..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 @@ -14,30 +14,31 @@ 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): 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/motion/intellidrivedevice.py b/hardwarelibrary/motion/intellidrivedevice.py index d94a5b91..8fd3fefb 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.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_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"), + } def __init__(self, serialNumber): super().__init__(serialNumber=serialNumber, idVendor=self.classIdVendor, idProduct=self.classIdProduct) @@ -117,3 +129,23 @@ def doHome(self): if not self.isReferenced(): raise Exception("Homing failed") + + 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): + if name == 'SET_REGISTER': + self.registers[params["register"]] = int(params["value"]) + return "ok\r" + elif name == 'GET_REGISTER': + value = self.registers.get(params["register"], 0) + return {"value": str(value)} + elif name == 'TRAJECTORY': + mode = int(params["mode"]) + 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 8cb0a47c..3e6500ad 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 import re import time @@ -16,6 +16,26 @@ class SutterDevice(LinearMotionDevice): classIdVendor = 4930 classIdProduct = 1 + commands = { + "MOVE": DataCommand(name="MOVE", prefix=b'M', + sendFormat=' 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: @@ -94,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('\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 - diff --git a/hardwarelibrary/tests/testCommandRecognition.py b/hardwarelibrary/tests/testCommandRecognition.py new file mode 100644 index 00000000..f23ea836 --- /dev/null +++ b/hardwarelibrary/tests/testCommandRecognition.py @@ -0,0 +1,202 @@ +import env +import unittest +from struct import pack + +from hardwarelibrary.communication.commands import Command, TextCommand, DataCommand + + +class TestCommandBase(unittest.TestCase): + def testMatchesReturnsFalse(self): + cmd = Command(name="base") + self.assertFalse(cmd.matches(b'anything')) + + def testExtractParamsReturnsEmpty(self): + cmd = Command(name="base") + self.assertEqual(cmd.extractParams(b'anything'), ()) + + def testFormatResponseNone(self): + cmd = Command(name="base") + self.assertIsNone(cmd.formatResponse(None)) + + def testFormatResponseBytes(self): + cmd = Command(name="base") + self.assertEqual(cmd.formatResponse(b'\x01\x02'), bytearray(b'\x01\x02')) + + def testFormatResponseString(self): + cmd = Command(name="base") + self.assertEqual(cmd.formatResponse("ok\r"), bytearray(b'ok\r')) + + +class TestTextCommandRecognition(unittest.TestCase): + def testAutoMatchPattern(self): + 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_format="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]') + 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") + 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]') + 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") + result = cmd.formatResponse(("42",)) + self.assertEqual(result, bytearray(b'v 42\r')) + + def testFormatResponseStringFallback(self): + 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_format="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') + 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") + 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 testEffectiveMatchPatternUsesAuto(self): + cmd = TextCommand(name="test", text_format="SET {0}\r") + self.assertEqual(cmd.effectiveMatchPattern, cmd._autoMatchPattern) + + +class TestDataCommandRecognition(unittest.TestCase): + def testMatchesByPrefix(self): + cmd = DataCommand(name="test", prefix=b'M') + self.assertTrue(cmd.matches(b'M\x01\x02')) + + def testMatchesCaseInsensitive(self): + cmd = DataCommand(name="test", prefix=b'M') + self.assertTrue(cmd.matches(b'm\x01\x02')) + + def testPrefixDerivedFromData(self): + cmd = DataCommand(name="test", data=b'C\r') + self.assertTrue(cmd.matches(b'C')) + self.assertTrue(cmd.matches(b'c')) + + def testNoMatchWrongPrefix(self): + cmd = DataCommand(name="test", prefix=b'M') + self.assertFalse(cmd.matches(b'G\x01')) + + def testExtractParamsWithFormat(self): + cmd = DataCommand(name="test", prefix=b'S', requestFormat='