diff --git a/pylabrobot/bmg_labtech/clariostar/driver/__init__.py b/pylabrobot/bmg_labtech/clariostar/driver/__init__.py index c8e2f6842a5..25678c9af99 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/__init__.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/__init__.py @@ -1,6 +1,8 @@ +from pylabrobot.bmg_labtech.protocol.errors import ChecksumError, FrameError + from .configuration import DeviceConfiguration from .confirmed_firmware_versions import CONFIRMED_FIRMWARE_VERSIONS -from .errors import ChecksumError, CommandRefusedError, FrameError +from .errors import CommandRefusedError from .features import ( Absorbance, AbsorbanceConfiguration, diff --git a/pylabrobot/bmg_labtech/clariostar/driver/configuration_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/configuration_tests.py index 080b31df97b..934d619c3b7 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/configuration_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/configuration_tests.py @@ -13,7 +13,7 @@ DeviceConfiguration, _parse_monochromator_slides, ) -from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload, ftdi_wrap_payload +from pylabrobot.bmg_labtech.clariostar.driver.framing import FTDI_HEADER from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( FIRMWARE_INFO_REPLY, INIT_REPLY, @@ -30,6 +30,7 @@ SERIAL_NUMBER_ADDRESSES, SIGNED_PARAMETERS, ) +from pylabrobot.bmg_labtech.protocol.framing import ftdi_extract_payload, ftdi_wrap_payload INIT_PAYLOAD = ftdi_extract_payload(INIT_REPLY) @@ -202,7 +203,7 @@ def reply_with_serial(first: int, second: int) -> bytes: payload = bytearray(ftdi_extract_payload(PARAMETERS_REPLY)) for address, value in zip(SERIAL_NUMBER_ADDRESSES, (first, second)): payload[address + REPLY_OFFSET : address + REPLY_OFFSET + 2] = value.to_bytes(2, "little") - return ftdi_wrap_payload(bytes(payload)) + return ftdi_wrap_payload(bytes(payload), header=FTDI_HEADER) cases: List[Tuple[Tuple[int, int], Dict[str, Optional[str]], str, bool]] = [ ((123, 4567), {"device_id": None}, "123-4567", False), diff --git a/pylabrobot/bmg_labtech/clariostar/driver/errors.py b/pylabrobot/bmg_labtech/clariostar/driver/errors.py index 09f251283fa..cd302db03e7 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/errors.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/errors.py @@ -2,15 +2,6 @@ from typing import Dict, Optional, Tuple - -class FrameError(Exception): - """A reply frame is malformed, truncated, or missing.""" - - -class ChecksumError(FrameError): - """An FTDI reply frame's checksum does not match its bytes.""" - - # Reply byte 1 bits that put quit codes in bytes 6-10 of a type 0x01 reply. REFUSED_BIT = 1 << 4 WARNING_BIT = 1 << 7 diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance.py b/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance.py index 8b199766db0..ecf457534a0 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance.py @@ -6,10 +6,10 @@ from datetime import datetime from typing import TYPE_CHECKING, Dict, List, Literal, Optional, Sequence, Tuple +from pylabrobot.bmg_labtech.protocol.errors import FrameError from pylabrobot.lib.absorbance.results import AbsorbanceResult, AbsorbanceSpectrumResult from pylabrobot.resources import Plate, Well -from ..errors import FrameError from .measurement import ( Measurement, MeasurementConfiguration, diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py index 83d53ba7f6f..302d1454468 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py @@ -13,7 +13,6 @@ from pylabrobot.bmg_labtech.clariostar.device import CLARIOstarPlus from pylabrobot.bmg_labtech.clariostar.driver.configuration_tests import decoded -from pylabrobot.bmg_labtech.clariostar.driver.errors import FrameError from pylabrobot.bmg_labtech.clariostar.driver.features import measurement as measurement_module from pylabrobot.bmg_labtech.clariostar.driver.features.absorbance import ( Absorbance, @@ -27,7 +26,6 @@ MeasurementSettings, ShakeSettings, ) -from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( STATUS_IDLE, STATUS_QUERY_FRAME, @@ -36,6 +34,8 @@ make_driver, ) from pylabrobot.bmg_labtech.clariostar.driver.simulator import CLARIOstarSimulationAbsorbance +from pylabrobot.bmg_labtech.protocol.errors import FrameError +from pylabrobot.bmg_labtech.protocol.framing import ftdi_extract_payload from pylabrobot.lib.absorbance.results import AbsorbanceResult from pylabrobot.resources import Plate, Well, cor_96_wellplate_360uL_Fb, create_ordered_items_2d from pylabrobot.resources.biorad.plates import biorad_96_wellplate_200uL_Vb diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/drawer_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/drawer_tests.py index d26afc11e95..6b6dbd1e83f 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/drawer_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/drawer_tests.py @@ -2,7 +2,7 @@ import unittest -from pylabrobot.bmg_labtech.clariostar.driver.errors import CommandRefusedError, FrameError +from pylabrobot.bmg_labtech.clariostar.driver.errors import CommandRefusedError from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( STATUS_BUSY, STATUS_DRAWER_OPEN, @@ -12,6 +12,7 @@ FakeFTDI, make_driver, ) +from pylabrobot.bmg_labtech.protocol.errors import FrameError from pylabrobot.resources import Coordinate, PlateHolder from pylabrobot.resources.corning.plates import cor_96_wellplate_2mL_Vb, cor_96_wellplate_360uL_Fb diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py index c915c298ab4..cfbaabd2be6 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py @@ -24,6 +24,7 @@ overload, ) +from pylabrobot.bmg_labtech.protocol.errors import FrameError from pylabrobot.lib.fluorescence.channels import FluorescenceChannel from pylabrobot.lib.fluorescence.results import FluorescenceResult, FluorescenceSpectrumResult from pylabrobot.lib.optics.filters import DichroicFilter, OpticalFilter @@ -36,7 +37,6 @@ ) from pylabrobot.resources import Coordinate, Plate, Well -from ..errors import FrameError from .measurement import ( Measurement, MeasurementConfiguration, diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py index c73fe8ac780..2364f5bba2c 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py @@ -38,7 +38,7 @@ MeasurementSettings, ShakeSettings, ) -from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload, ftdi_wrap_payload +from pylabrobot.bmg_labtech.clariostar.driver.framing import FTDI_HEADER from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( STATUS_IDLE, STATUS_QUERY_FRAME, @@ -47,6 +47,7 @@ make_driver, ) from pylabrobot.bmg_labtech.clariostar.driver.simulator import CLARIOstarSimulationFluorescence +from pylabrobot.bmg_labtech.protocol.framing import ftdi_extract_payload, ftdi_wrap_payload from pylabrobot.lib.fluorescence.channels import FluorescenceChannel from pylabrobot.lib.fluorescence.results import FluorescenceSpectrumResult from pylabrobot.lib.optics.filters import DichroicFilter, OpticalFilter @@ -1786,7 +1787,7 @@ def with_nominal_scan_edges(ground_truth: bytes) -> bytes: for offset, edges in ((21, _compute_band_edges(488, 14)), (27, _compute_band_edges(535, 30))): for index, value in enumerate(edges): payload[offset + 2 * index : offset + 2 * index + 2] = value.to_bytes(2, "big") - return ftdi_wrap_payload(bytes(payload)) + return ftdi_wrap_payload(bytes(payload), header=FTDI_HEADER) async def scan_ground_truth(**arguments): diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py index ad4324bf46d..5f45a2074e2 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/luminescence_tests.py @@ -25,13 +25,13 @@ Measurement, MeasurementSettings, ) -from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( STATUS_IDLE, FakeFTDI, make_driver, ) from pylabrobot.bmg_labtech.clariostar.driver.simulator import CLARIOstarSimulationLuminescence +from pylabrobot.bmg_labtech.protocol.framing import ftdi_extract_payload from pylabrobot.resources import Plate, cor_96_wellplate_360uL_Fb # The ground truth: every well, 580-80 nm, gain 3500, focal height 8.5 mm, settling 0.0 s, diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py b/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py index fe262622d11..a6b294aa1c4 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py @@ -13,9 +13,10 @@ from datetime import datetime from typing import TYPE_CHECKING, Dict, List, Literal, Optional, Sequence, Set, Tuple +from pylabrobot.bmg_labtech.protocol.errors import FrameError from pylabrobot.resources import Plate, Well -from ..errors import CommandRefusedError, FrameError +from ..errors import CommandRefusedError if TYPE_CHECKING: from ..master import CLARIOstarDriver diff --git a/pylabrobot/bmg_labtech/clariostar/driver/framing.py b/pylabrobot/bmg_labtech/clariostar/driver/framing.py index 2a600b0171e..23665f36576 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/framing.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/framing.py @@ -3,11 +3,8 @@ Both interfaces carry the same logical payloads (command family, command, parameters); only the framing around them differs. -FTDI (VID 0x0403 / PID 0xBB68), one frame per message, 8 bytes of overhead:: - - STX (1) | size (2 BE, whole frame) | 0x0C (1) | payload (n) | checksum (3 BE) | CR (1) - -The checksum is the 24-bit sum of every byte from STX through the end of the payload. +FTDI (VID 0x0403 / PID 0xBB68): BMG's FTDI frame (`pylabrobot.bmg_labtech.protocol.framing`) +with the CLARIOstar's header byte, 0x0C. USB (VID 0x0483 / PID 0xA29A), 64-byte packets, each with a 7-byte header:: @@ -21,55 +18,10 @@ from typing import List -from .errors import ChecksumError, FrameError - -# -- FTDI -------------------------------------------------------------------- +from pylabrobot.bmg_labtech.protocol.errors import FrameError -FTDI_STX = 0x02 +# The CLARIOstar's header byte, in its FTDI frames and its USB packets. FTDI_HEADER = 0x0C -FTDI_CR = 0x0D -FTDI_OVERHEAD = 8 # STX + size(2) + header + checksum(3) + CR - - -def ftdi_wrap_payload(payload: bytes) -> bytes: - """Frame a logical payload for the FTDI interface.""" - frame = bytearray([FTDI_STX]) - frame.extend((len(payload) + FTDI_OVERHEAD).to_bytes(2, "big")) - frame.append(FTDI_HEADER) - frame.extend(payload) - frame.extend((sum(frame) & 0xFFFFFF).to_bytes(3, "big")) - frame.append(FTDI_CR) - return bytes(frame) - - -def ftdi_validate_frame(frame: bytes) -> None: - """Check an FTDI frame's delimiters, size field and checksum. - - Raises: - FrameError: on a wrong STX, CR, header or size field. - ChecksumError: on a checksum mismatch. - """ - if len(frame) < FTDI_OVERHEAD: - raise FrameError(f"Reply too short ({len(frame)} bytes): {frame.hex()}") - if frame[0] != FTDI_STX: - raise FrameError(f"Expected STX 0x{FTDI_STX:02x}, got 0x{frame[0]:02x}") - if frame[-1] != FTDI_CR: - raise FrameError(f"Expected CR 0x{FTDI_CR:02x}, got 0x{frame[-1]:02x}") - if frame[3] != FTDI_HEADER: - raise FrameError(f"Expected header 0x{FTDI_HEADER:02x} at byte 3, got 0x{frame[3]:02x}") - size = int.from_bytes(frame[1:3], "big") - if size != len(frame): - raise FrameError(f"Size field says {size} bytes, got {len(frame)}") - received = int.from_bytes(frame[-4:-1], "big") - computed = sum(frame[:-4]) & 0xFFFFFF - if received != computed: - raise ChecksumError(f"Checksum 0x{received:06x}, computed 0x{computed:06x}") - - -def ftdi_extract_payload(frame: bytes) -> bytes: - """Return the logical payload of a validated FTDI frame.""" - return frame[4:-4] - # -- USB --------------------------------------------------------------------- diff --git a/pylabrobot/bmg_labtech/clariostar/driver/framing_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/framing_tests.py index eaf5e835d5d..2cf36c0b51e 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/framing_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/framing_tests.py @@ -2,10 +2,6 @@ import unittest -from pylabrobot.bmg_labtech.clariostar.driver.errors import ( - ChecksumError, - FrameError, -) from pylabrobot.bmg_labtech.clariostar.driver.features.absorbance_tests import ( VERIFICATION_DATA_REPLY, ) @@ -16,9 +12,7 @@ OPEN_DRAWER_REPLY, ) from pylabrobot.bmg_labtech.clariostar.driver.framing import ( - ftdi_extract_payload, - ftdi_validate_frame, - ftdi_wrap_payload, + FTDI_HEADER, usb_packet_type, usb_total_length, usb_wrap_payload, @@ -42,6 +36,12 @@ STATUS_TRUNCATED, STATUS_WARNING, ) +from pylabrobot.bmg_labtech.protocol.errors import ChecksumError, FrameError +from pylabrobot.bmg_labtech.protocol.framing import ( + ftdi_extract_payload, + ftdi_validate_frame, + ftdi_wrap_payload, +) # Every recorded exchange whose reply arrived whole. REAL_EXCHANGES = [ @@ -70,37 +70,37 @@ class FTDIFramingTests(unittest.TestCase): def test_wrap_reproduces_every_real_sent_frame(self): for sent, _ in REAL_EXCHANGES: - self.assertEqual(ftdi_wrap_payload(ftdi_extract_payload(sent)), sent) - self.assertEqual(ftdi_wrap_payload(b"\x80"), STATUS_QUERY_FRAME) + self.assertEqual(ftdi_wrap_payload(ftdi_extract_payload(sent), header=FTDI_HEADER), sent) + self.assertEqual(ftdi_wrap_payload(b"\x80", header=FTDI_HEADER), STATUS_QUERY_FRAME) def test_every_real_frame_validates(self): for sent, received in REAL_EXCHANGES: - ftdi_validate_frame(sent) - ftdi_validate_frame(received) + ftdi_validate_frame(sent, header=FTDI_HEADER) + ftdi_validate_frame(received, header=FTDI_HEADER) self.assertEqual(len(ftdi_extract_payload(received)), len(received) - 8) def test_real_truncated_reply_fails(self): with self.assertRaises(FrameError): - ftdi_validate_frame(STATUS_TRUNCATED) + ftdi_validate_frame(STATUS_TRUNCATED, header=FTDI_HEADER) def test_a_checksum_past_16_bits_validates(self): # The 96-well verification read's data reply: its checksum is 0x01c462. self.assertGreater(int.from_bytes(VERIFICATION_DATA_REPLY[-4:-1], "big"), 0xFFFF) - ftdi_validate_frame(VERIFICATION_DATA_REPLY) + ftdi_validate_frame(VERIFICATION_DATA_REPLY, header=FTDI_HEADER) def test_corrupted_byte_fails_checksum(self): # Not a reader's reply: the recorded frame with one bit flipped. frame = bytearray(STATUS_REPLY_FRAME) frame[6] ^= 0x01 with self.assertRaises(ChecksumError): - ftdi_validate_frame(bytes(frame)) + ftdi_validate_frame(bytes(frame), header=FTDI_HEADER) def test_truncated_frame_fails(self): # Not a reader's reply: the recorded frame cut short. with self.assertRaises(FrameError): - ftdi_validate_frame(STATUS_REPLY_FRAME[:-1]) + ftdi_validate_frame(STATUS_REPLY_FRAME[:-1], header=FTDI_HEADER) with self.assertRaises(FrameError): - ftdi_validate_frame(STATUS_REPLY_FRAME[:5]) + ftdi_validate_frame(STATUS_REPLY_FRAME[:5], header=FTDI_HEADER) class USBFramingTests(unittest.TestCase): diff --git a/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads_tests.py index 0ee953a428e..bccbcbdfd27 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads_tests.py @@ -17,7 +17,6 @@ from pylabrobot.bmg_labtech.clariostar.driver import ftdi_queued_reads from pylabrobot.bmg_labtech.clariostar.driver import master as master_module -from pylabrobot.bmg_labtech.clariostar.driver.errors import FrameError from pylabrobot.bmg_labtech.clariostar.driver.features.absorbance_tests import ( RUN_REPLY, STATUS_MEASURED, @@ -26,7 +25,6 @@ ground_truth_96_well_plate, make_absorbance, ) -from pylabrobot.bmg_labtech.clariostar.driver.framing import ftdi_extract_payload from pylabrobot.bmg_labtech.clariostar.driver.ftdi_queued_reads import QueuedReads from pylabrobot.bmg_labtech.clariostar.driver.master import CLARIOstarDriver from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( @@ -36,6 +34,8 @@ FakeFTDI, make_driver, ) +from pylabrobot.bmg_labtech.protocol.errors import FrameError +from pylabrobot.bmg_labtech.protocol.framing import ftdi_extract_payload from pylabrobot.io.validation_utils import LOG_LEVEL_IO DATA_REQUEST = bytes.fromhex("05020000000000") diff --git a/pylabrobot/bmg_labtech/clariostar/driver/master.py b/pylabrobot/bmg_labtech/clariostar/driver/master.py index ffe5de56236..10762c6f7c8 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/master.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/master.py @@ -11,6 +11,14 @@ from dataclasses import dataclass from typing import Any, Callable, Dict, Literal, Optional, Tuple, Union, cast +from pylabrobot.bmg_labtech.protocol.errors import FrameError +from pylabrobot.bmg_labtech.protocol.framing import ( + FTDI_CR, + FTDI_STX, + ftdi_extract_payload, + ftdi_validate_frame, + ftdi_wrap_payload, +) from pylabrobot.io.capture import get_capture_or_validation_active from pylabrobot.io.ftdi import FTDI from pylabrobot.io.usb import USB @@ -27,7 +35,6 @@ from .confirmed_firmware_versions import CONFIRMED_FIRMWARE_VERSIONS from .errors import ( CommandRefusedError, - FrameError, describe_quit_codes, parse_quit_codes, parse_warning_codes, @@ -38,8 +45,7 @@ from .features.luminescence import Luminescence from .features.temperature_controller import TemperatureController from .framing import ( - FTDI_CR, - FTDI_STX, + FTDI_HEADER, USB_ACK, USB_CONTINUATION, USB_FIRST, @@ -47,9 +53,6 @@ USB_HOST_ACK, USB_PACKET_SIZE, USB_SINGLE, - ftdi_extract_payload, - ftdi_validate_frame, - ftdi_wrap_payload, usb_packet_type, usb_total_length, usb_wrap_payload, @@ -451,13 +454,13 @@ async def _ftdi_exchange(self, payload: bytes) -> bytes: """Send one payload over FTDI and return the validated reply payload.""" reads = self._get_queued_reads() if reads is None: - await self._ftdi_write(ftdi_wrap_payload(payload)) + await self._ftdi_write(ftdi_wrap_payload(payload, header=FTDI_HEADER)) frame = await self._ftdi_read_frame() else: self._discard_unread(reads) # bytes read before the command is sent are not its reply - await self._ftdi_write(ftdi_wrap_payload(payload)) + await self._ftdi_write(ftdi_wrap_payload(payload, header=FTDI_HEADER)) frame = await self._ftdi_read_queued_frame(reads) - ftdi_validate_frame(frame) + ftdi_validate_frame(frame, header=FTDI_HEADER) return ftdi_extract_payload(frame) # -- USB interface --------------------------------------------------------- diff --git a/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py index 5bcc8253f95..8090b151dc4 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py @@ -10,13 +10,13 @@ from pylabrobot.bmg_labtech.clariostar.driver import master as master_module from pylabrobot.bmg_labtech.clariostar.driver.errors import ( CommandRefusedError, - FrameError, parse_quit_codes, ) -from pylabrobot.bmg_labtech.clariostar.driver.framing import ( +from pylabrobot.bmg_labtech.clariostar.driver.master import CLARIOstarDriver +from pylabrobot.bmg_labtech.protocol.errors import FrameError +from pylabrobot.bmg_labtech.protocol.framing import ( ftdi_extract_payload, ) -from pylabrobot.bmg_labtech.clariostar.driver.master import CLARIOstarDriver from pylabrobot.io.ftdi import FTDI from pylabrobot.io.usb import USB diff --git a/pylabrobot/bmg_labtech/protocol/__init__.py b/pylabrobot/bmg_labtech/protocol/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/pylabrobot/bmg_labtech/protocol/errors.py b/pylabrobot/bmg_labtech/protocol/errors.py new file mode 100644 index 00000000000..4382a0219b4 --- /dev/null +++ b/pylabrobot/bmg_labtech/protocol/errors.py @@ -0,0 +1,9 @@ +"""Errors of BMG Labtech's wire framing.""" + + +class FrameError(Exception): + """A reply frame is malformed, truncated, or missing.""" + + +class ChecksumError(FrameError): + """An FTDI reply frame's checksum does not match its bytes.""" diff --git a/pylabrobot/bmg_labtech/protocol/framing.py b/pylabrobot/bmg_labtech/protocol/framing.py new file mode 100644 index 00000000000..0915994606f --- /dev/null +++ b/pylabrobot/bmg_labtech/protocol/framing.py @@ -0,0 +1,55 @@ +"""BMG Labtech FTDI wire framing, the same on BMG's readers. + +One frame per message, 8 bytes of overhead:: + + STX (1) | size (2 BE, whole frame) | header (1) | payload (n) | checksum (3 BE) | CR (1) + +The checksum is the 24-bit sum of every byte from STX through the end of the payload. The header +byte is set per reader (0x0C on the CLARIOstar). +""" + +from .errors import ChecksumError, FrameError + +FTDI_STX = 0x02 +FTDI_CR = 0x0D +FTDI_OVERHEAD = 8 # STX + size(2) + header + checksum(3) + CR + + +def ftdi_wrap_payload(payload: bytes, *, header: int) -> bytes: + """Frame a logical payload for the FTDI interface, with the reader's header byte.""" + frame = bytearray([FTDI_STX]) + frame.extend((len(payload) + FTDI_OVERHEAD).to_bytes(2, "big")) + frame.append(header) + frame.extend(payload) + frame.extend((sum(frame) & 0xFFFFFF).to_bytes(3, "big")) + frame.append(FTDI_CR) + return bytes(frame) + + +def ftdi_validate_frame(frame: bytes, *, header: int) -> None: + """Check an FTDI frame's delimiters, header byte, size field and checksum. + + Raises: + FrameError: on a wrong STX, CR, header or size field. + ChecksumError: on a checksum mismatch. + """ + if len(frame) < FTDI_OVERHEAD: + raise FrameError(f"Reply too short ({len(frame)} bytes): {frame.hex()}") + if frame[0] != FTDI_STX: + raise FrameError(f"Expected STX 0x{FTDI_STX:02x}, got 0x{frame[0]:02x}") + if frame[-1] != FTDI_CR: + raise FrameError(f"Expected CR 0x{FTDI_CR:02x}, got 0x{frame[-1]:02x}") + if frame[3] != header: + raise FrameError(f"Expected header 0x{header:02x} at byte 3, got 0x{frame[3]:02x}") + size = int.from_bytes(frame[1:3], "big") + if size != len(frame): + raise FrameError(f"Size field says {size} bytes, got {len(frame)}") + received = int.from_bytes(frame[-4:-1], "big") + computed = sum(frame[:-4]) & 0xFFFFFF + if received != computed: + raise ChecksumError(f"Checksum 0x{received:06x}, computed 0x{computed:06x}") + + +def ftdi_extract_payload(frame: bytes) -> bytes: + """Return the logical payload of a validated FTDI frame.""" + return frame[4:-4] diff --git a/pylabrobot/bmg_labtech/protocol/framing_tests.py b/pylabrobot/bmg_labtech/protocol/framing_tests.py new file mode 100644 index 00000000000..3d4693d9159 --- /dev/null +++ b/pylabrobot/bmg_labtech/protocol/framing_tests.py @@ -0,0 +1,58 @@ +"""FTDI framing tests, against frames recorded from a CLARIOstar.""" + +import unittest + +from pylabrobot.bmg_labtech.protocol.errors import ChecksumError, FrameError +from pylabrobot.bmg_labtech.protocol.framing import ( + ftdi_extract_payload, + ftdi_validate_frame, + ftdi_wrap_payload, +) + +CLARIOSTAR_HEADER = 0x0C +STATUS_QUERY = bytes.fromhex("0200090c800000970d") +STATUS_IDLE = bytes.fromhex("0200180c010506240000000000000000000000c00001160d") +INIT = bytes.fromhex("02000d0c010000100300002f0d") +RECORDED_FRAMES = [STATUS_QUERY, STATUS_IDLE, INIT] + + +class FTDIFramingTests(unittest.TestCase): + def test_wrap_reproduces_every_recorded_sent_frame(self): + for sent in (STATUS_QUERY, INIT): + self.assertEqual( + ftdi_wrap_payload(ftdi_extract_payload(sent), header=CLARIOSTAR_HEADER), sent + ) + + def test_every_recorded_frame_validates_with_its_header(self): + for frame in RECORDED_FRAMES: + ftdi_validate_frame(frame, header=CLARIOSTAR_HEADER) + self.assertEqual(len(ftdi_extract_payload(frame)), len(frame) - 8) + + def test_another_header_byte_fails(self): + for frame in RECORDED_FRAMES: + with self.assertRaises(FrameError): + ftdi_validate_frame(frame, header=0x08) + + def test_the_header_changes_only_byte_3_and_the_checksum(self): + framed = ftdi_wrap_payload(ftdi_extract_payload(STATUS_QUERY), header=0x08) + ftdi_validate_frame(framed, header=0x08) + differing = [i for i, (a, b) in enumerate(zip(framed, STATUS_QUERY)) if a != b] + self.assertEqual(differing, [3, 7]) + + def test_corrupted_byte_fails_checksum(self): + # Not a reader's reply: the recorded frame with one bit flipped. + frame = bytearray(STATUS_IDLE) + frame[6] ^= 0x01 + with self.assertRaises(ChecksumError): + ftdi_validate_frame(bytes(frame), header=CLARIOSTAR_HEADER) + + def test_truncated_frame_fails(self): + # Not a reader's reply: the recorded frame cut short. + with self.assertRaises(FrameError): + ftdi_validate_frame(STATUS_IDLE[:-1], header=CLARIOSTAR_HEADER) + with self.assertRaises(FrameError): + ftdi_validate_frame(STATUS_IDLE[:5], header=CLARIOSTAR_HEADER) + + +if __name__ == "__main__": + unittest.main()