BMG protocot: the FTDI framing shared by BMG's readers, header byte per reader

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,

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),

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

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,

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

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
 

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,

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):

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,

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

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 ---------------------------------------------------------------------
 

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):

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")