From b496d5bdc65cafb8a49fa04ee4ffafb6936e4511 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Fri, 23 Oct 2026 16:57:05 +0100 Subject: [PATCH] BMG `transport`: the FTDI interface shared by BMG's readers The FTDI link of CLARIOstarDriver - opening the port, writing a frame and reading exactly one reply frame back through the queued bulk-in reads or libftdi, purge and drain - moves to pylabrobot/bmg_labtech/transport/ as BMGFTDIInterface, taking the reader's header byte, baud rate and reply-read timeouts; the queued reads move with it. The CLARIOstar keeps its measured timeouts, the exchange lock, the retries and its USB interface, builds the FTDI interface on first use over its current io, and sends and reads the same bytes. Co-Authored-By: Claude Opus 5.5 --- .../driver/ftdi_queued_reads_tests.py | 38 +-- .../bmg_labtech/clariostar/driver/master.py | 272 ++-------------- .../clariostar/driver/master_tests.py | 12 +- pylabrobot/bmg_labtech/transport/__init__.py | 0 .../bmg_labtech/transport/ftdi_interface.py | 291 ++++++++++++++++++ .../driver => transport}/ftdi_queued_reads.py | 0 6 files changed, 349 insertions(+), 264 deletions(-) create mode 100644 pylabrobot/bmg_labtech/transport/__init__.py create mode 100644 pylabrobot/bmg_labtech/transport/ftdi_interface.py rename pylabrobot/bmg_labtech/{clariostar/driver => transport}/ftdi_queued_reads.py (100%) 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 bccbcbdfd27..54ba0d9592e 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads_tests.py @@ -15,8 +15,6 @@ from typing import Any, Callable, Deque, List, Optional, Tuple from unittest.mock import patch -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.features.absorbance_tests import ( RUN_REPLY, STATUS_MEASURED, @@ -25,7 +23,6 @@ ground_truth_96_well_plate, make_absorbance, ) -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 ( STATUS_BUSY, @@ -36,6 +33,9 @@ ) from pylabrobot.bmg_labtech.protocol.errors import FrameError from pylabrobot.bmg_labtech.protocol.framing import ftdi_extract_payload +from pylabrobot.bmg_labtech.transport import ftdi_interface as ftdi_interface_module +from pylabrobot.bmg_labtech.transport import ftdi_queued_reads +from pylabrobot.bmg_labtech.transport.ftdi_queued_reads import QueuedReads from pylabrobot.io.validation_utils import LOG_LEVEL_IO DATA_REQUEST = bytes.fromhex("05020000000000") @@ -307,10 +307,10 @@ def keep_reads_queued( driver: CLARIOstarDriver, libusb: FakeLibusb, transfers: int = ftdi_queued_reads._TRANSFERS ) -> QueuedReads: """Start `transfers` pending transfers on `libusb` for `driver` to read through.""" - driver._PACKET_READ_TIMEOUT, driver._FRAME_STALL_TIMEOUT = 1.0, 0.3 + driver._ftdi._packet_read_timeout, driver._ftdi._frame_stall_timeout = 1.0, 0.3 reads = QueuedReads(libusb.functions, 1, 2, transfers=transfers) reads.start() - driver._queued_reads = reads + driver._ftdi._queued_reads = reads return reads @@ -329,8 +329,8 @@ async def test_a_stall_loses_bytes_with_libftdi_but_not_with_transfers_pending(s chip = ChipModel(packets, stall=stall) driver = make_driver(LibftdiFTDI(chip)) - driver._PACKET_READ_TIMEOUT = 1.0 # the driver's own: the page starts after 34 ms - with patch.object(master_module.time, "time", side_effect=lambda: chip.time): + driver._ftdi._packet_read_timeout = 1.0 # the driver's own: the page starts after 34 ms + with patch.object(ftdi_interface_module.time, "time", side_effect=lambda: chip.time): with self.assertRaises(FrameError): await driver.exchange(DATA_REQUEST) self.assertGreater(chip.dropped, 0) @@ -339,14 +339,14 @@ async def test_a_stall_loses_bytes_with_libftdi_but_not_with_transfers_pending(s io, libusb = LibftdiFTDI(chip), FakeLibusb(chip) driver = make_queued_driver(io, libusb) # No sleep and no timeout ends a wait: the event thread's callbacks wake the read. - driver._PACKET_READ_TIMEOUT = driver._FRAME_STALL_TIMEOUT = 60.0 - with patch.object(master_module.asyncio, "sleep", side_effect=AssertionError("polled")): + driver._ftdi._packet_read_timeout = driver._ftdi._frame_stall_timeout = 60.0 + with patch.object(ftdi_interface_module.asyncio, "sleep", side_effect=AssertionError("polled")): with self.assertNoLogs(ftdi_queued_reads.logger, "WARNING"): # the stall used 6 of 125 reply = await asyncio.wait_for(driver.exchange(DATA_REQUEST), 10) self.assertEqual(reply, ftdi_extract_payload(page)) self.assertEqual(chip.dropped, 0) self.assertEqual(io.written, [DATA_REQUEST_FRAME]) - driver._stop_queued_reads() + driver._ftdi._stop_queued_reads() self.assertEqual(sorted(libusb.freed), sorted(libusb.allocated)) self.assertEqual(len(libusb.allocated), 125) # 0.25 s of 2 ms latency periods @@ -355,7 +355,7 @@ async def test_a_stall_longer_than_the_transfers_cover_is_warned(self): _, packets, stall = load_spectrum_page_transfers() chip = ChipModel(packets, stall=stall) driver = make_queued_driver(LibftdiFTDI(chip), FakeLibusb(chip), transfers=4) - self.addCleanup(driver._stop_queued_reads) + self.addCleanup(driver._ftdi._stop_queued_reads) with self.assertLogs("pylabrobot", "WARNING") as logs: with self.assertRaises(FrameError): await driver.exchange(DATA_REQUEST) @@ -371,7 +371,7 @@ async def test_bytes_read_before_a_command_are_not_its_reply(self): chip = ChipModel([(0.0, STATUS_BUSY)], [(0.004, STATUS_IDLE)]) io = LibftdiFTDI(chip) driver = make_queued_driver(io, FakeLibusb(chip)) - self.addCleanup(driver._stop_queued_reads) + self.addCleanup(driver._ftdi._stop_queued_reads) chip.start() await wait_until(lambda: not chip.streaming()) sent = chip.time @@ -385,16 +385,16 @@ async def test_transfers_that_failed_leave_reading_to_libftdi(self): io, libusb = FakeFTDI(), FakeLibusb(ChipModel(), status=5) io.replies = [STATUS_IDLE] driver = make_queued_driver(io, libusb) - reads = driver._queued_reads + reads = driver._ftdi._queued_reads assert reads is not None deadline = time.monotonic() + 5.0 while reads.running: self.assertLess(time.monotonic(), deadline) await reads.wait(1.0) # the last transfer to fail wakes it - with self.assertLogs(master_module.logger, "WARNING"): + with self.assertLogs(ftdi_interface_module.logger, "WARNING"): status = await driver.request_status() self.assertFalse(status.busy) - self.assertIsNone(driver._queued_reads) + self.assertIsNone(driver._ftdi._queued_reads) self.assertEqual(sorted(libusb.freed), sorted(libusb.allocated)) async def test_a_cancelled_spectrum_leaves_the_reads_running_and_the_link_clean(self): @@ -407,7 +407,7 @@ async def test_a_cancelled_spectrum_leaves_the_reads_running_and_the_link_clean( absorbance = make_absorbance(io, ground_truth_96_well_plate()) driver = absorbance._driver reads = keep_reads_queued(driver, libusb) - self.addCleanup(driver._stop_queued_reads) + self.addCleanup(driver._ftdi._stop_queued_reads) with self.assertLogs("pylabrobot", LOG_LEVEL_IO) as logs: task = asyncio.create_task( absorbance.measure_spectrum(start_wavelength=220, end_wavelength=1000, step_size=1) @@ -429,7 +429,7 @@ async def test_a_cancelled_spectrum_leaves_the_reads_running_and_the_link_clean( late = ast.literal_eval(messages[first_discard].split(" discarded ")[1]) self.assertEqual(taken + late, page) # the drained rest, before the STOP's write self.assertGreater(len(late), len(page) // 2) - self.assertIs(driver._get_queued_reads(), reads) + self.assertIs(driver._ftdi._get_queued_reads(), reads) self.assertTrue(reads.running) self.assertEqual(reads.take(), (b"", 0)) self.assertEqual(chip.dropped, 0) @@ -445,8 +445,8 @@ async def test_a_cancelled_spectrum_through_libftdi_leaves_the_link_clean(self): io = LibftdiFTDI(chip) absorbance = make_absorbance(io, ground_truth_96_well_plate()) driver = absorbance._driver - driver._PACKET_READ_TIMEOUT, driver._FRAME_STALL_TIMEOUT = 1.0, 0.3 - self.assertIsNone(driver._queued_reads) + driver._ftdi._packet_read_timeout, driver._ftdi._frame_stall_timeout = 1.0, 0.3 + self.assertIsNone(driver._ftdi._queued_reads) with self.assertLogs("pylabrobot", LOG_LEVEL_IO) as logs: task = asyncio.create_task( absorbance.measure_spectrum(start_wavelength=220, end_wavelength=1000, step_size=1) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/master.py b/pylabrobot/bmg_labtech/clariostar/driver/master.py index 10762c6f7c8..24aa35ed60f 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/master.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/master.py @@ -12,17 +12,9 @@ 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.bmg_labtech.transport.ftdi_interface import BMGFTDIInterface from pylabrobot.io.ftdi import FTDI from pylabrobot.io.usb import USB -from pylabrobot.io.validation_utils import LOG_LEVEL_IO from .configuration import ( BARCODE_READER_POSITIONS, @@ -57,7 +49,6 @@ usb_total_length, usb_wrap_payload, ) -from .ftdi_queued_reads import LATENCY_TIMER, QueuedReads from .parameter_table import ( PARAMETER_TABLE, REPLY_OFFSET, @@ -111,7 +102,6 @@ class CLARIOstarDriver: # Seconds between status polls while waiting, with a ~38 ms exchange. default_poll_interval: float = 0.25 - _FTDI_READ_CHUNK = 25 # bytes per read; a shorter read marks the end of a burst _PACKET_READ_TIMEOUT = 1.0 # seconds to wait for the start of an FTDI reply # Seconds without new bytes before abandoning a started frame. The longest pause measured # inside a reply, over about 26,000 replies including measurement data, was 0.44 s. @@ -256,8 +246,8 @@ def __init__( # overlap. self._lock = asyncio.Lock() self._last_warning: Optional[Tuple[int, int, int, int, int]] = None - # FTDI only: bulk-in transfers kept pending from setup to stop; None reads through libftdi. - self._queued_reads: Optional[QueuedReads] = None + # Built on first use over the current io (`_ftdi`). + self._ftdi_interface: Optional[BMGFTDIInterface] = None self.drawer = Drawer(self) # Built by `discover` only if the reader reports a temperature controller, absorbance, a # fluorescence channel, and a luminescence channel. @@ -269,199 +259,28 @@ def __init__( # -- FTDI interface -------------------------------------------------------- @property - def _ftdi(self) -> FTDI: - """The io, as an FTDI.""" - return cast(FTDI, self.io) - - def _start_queued_reads(self) -> None: - """Keep bulk-in transfers pending on the open FTDI chip; if they cannot start, read with - libftdi (`_ftdi_read_frame`). - - Not while capture or validation is active: their replay needs every read to pass `FTDI.read`. - The queued reads belong to the event loop that ran setup. - """ - if get_capture_or_validation_active(): - logger.info("capture or validation active: reading with libftdi") - return - try: - reads = QueuedReads.from_device(self._ftdi.dev) - reads.start() - except Exception as e: # no open pylibftdi device, no libusb handle, nothing submitted - logger.warning("bulk-in transfers not kept pending (%s); reading with libftdi", e) - return - self._queued_reads = reads - - def _stop_queued_reads(self) -> None: - """Cancel and free the pending bulk-in transfers, if any; libftdi reads from then on.""" - reads, self._queued_reads = self._queued_reads, None - if reads is not None: - reads.stop() - - async def _ftdi_setup(self) -> None: - """Open the FTDI port at 125000 baud, 8N1, latency 2 ms, and keep bulk-in transfers pending.""" - self._stop_queued_reads() # a second setup reopens the device under them - await self._ftdi.setup() - await self._ftdi.set_baudrate(125000) - await self._ftdi.set_line_property(8, 0, 0) # 8N1 - await self._ftdi.set_latency_timer(LATENCY_TIMER) - self._start_queued_reads() - - async def _ftdi_write(self, frame: bytes) -> None: - """Write a whole frame; raise OSError on a short write.""" - written = await self._ftdi.write(frame) - if written != len(frame): - raise OSError(f"Short write: sent {written} of {len(frame)} bytes") - - async def _ftdi_read_frame(self) -> bytes: - """Read one FTDI reply frame. - - CR (0x0D) also occurs inside payloads, so once the header is in, the size field decides - when the frame is complete. Bytes before an STX are dropped to resync. Before the header, - ``_PACKET_READ_TIMEOUT`` bounds the wait; after it, only a silence of - ``_FRAME_STALL_TIMEOUT`` or the ``_FRAME_HARD_CAP`` ends the read, as long replies - stream in bursts. - - Returns: - The frame, possibly short of its size field on a stall; the caller validates it. - """ - data = b"" - expected_size: Optional[int] = None - end_byte_found = False - start = time.time() - last_data_time = start - - while True: - chunk = await self._ftdi.read(self._FTDI_READ_CHUNK) - if chunk: - data += chunk - last_data_time = time.time() - - if expected_size is None and data[0] != FTDI_STX: - index = data.find(FTDI_STX) - data = data[index:] if index != -1 else b"" - - end_byte_found = bool(data) and data[-1] == FTDI_CR - - if expected_size is None and len(data) >= 3: - expected_size = int.from_bytes(data[1:3], "big") - start = time.time() - - if len(chunk) < self._FTDI_READ_CHUNK and end_byte_found: - if expected_size is None or len(data) >= expected_size: - break - if expected_size is not None and len(data) >= expected_size: - break - - # The size is known, so ask for exactly the rest of the frame. - if expected_size is not None: - remainder = await self._ftdi.read(expected_size - len(data)) - if remainder: - data += remainder - last_data_time = time.time() - end_byte_found = data[-1] == FTDI_CR - if len(data) >= expected_size: - break - else: - if end_byte_found and (expected_size is None or len(data) >= expected_size): - break - now = time.time() - if expected_size is None: - timed_out = now - start > self._PACKET_READ_TIMEOUT - else: - timed_out = ( - now - last_data_time > self._FRAME_STALL_TIMEOUT or now - start > self._FRAME_HARD_CAP - ) - if timed_out: - if expected_size is not None and 0 < len(data) < expected_size: - data += await self._ftdi.read(expected_size - len(data)) - break - await asyncio.sleep(0.0001) - - if expected_size is not None and len(data) >= expected_size: - return data[:expected_size] - return data - - def _get_queued_reads(self) -> Optional[QueuedReads]: - """The queued reads while they run; None once they stopped by themselves (e.g. the device - was removed), and libftdi reads from then on.""" - reads = self._queued_reads - if reads is not None and not reads.running: - logger.warning("bulk-in transfers stopped (%s); reading with libftdi", reads.error) - self._stop_queued_reads() - return None - return reads - - def _discard_unread(self, reads: QueuedReads) -> None: - """Drop the bytes the queued reads hold unread, logging them at the I/O level.""" - data, _ = reads.take() - if data: - logger.log(LOG_LEVEL_IO, "[%s] discarded %s", self._ftdi.device_id, data) - - async def _ftdi_read_queued_frame(self, reads: QueuedReads) -> bytes: - """Read one FTDI reply frame from the queued reads, with `_ftdi_read_frame`'s rules. + def _ftdi(self) -> BMGFTDIInterface: + """The FTDI interface over the current io; one built over an earlier io is stopped first. - Bytes before an STX are dropped and the size field ends the frame. It waits until the pending - reads receive bytes: ``_PACKET_READ_TIMEOUT`` bounds the wait for the header; after it, a - silence of ``_FRAME_STALL_TIMEOUT`` or the ``_FRAME_HARD_CAP``. Packets flagging a line error - (an overrun loses bytes) are warned about. The IO log gets one line per reply: every byte - taken, dropped ones included. - - Returns: - The frame, possibly short of its size field on a stall; the caller validates it. + Raises: + RuntimeError: if the reader is connected over USB. """ - data = bytearray() - received = bytearray() - expected_size: Optional[int] = None - line_errors = 0 - start = time.time() - last_data_time = start - try: - while True: - chunk, errors = reads.take() - line_errors += errors - now = time.time() - if chunk: - received += chunk - data += chunk - last_data_time = now - if expected_size is None: - index = data.find(FTDI_STX) - del data[: index if index != -1 else len(data)] - if len(data) >= 3: - expected_size = int.from_bytes(data[1:3], "big") - start = now - if expected_size is not None and len(data) >= expected_size: - break - elif not reads.running: - break - if expected_size is None: - deadline = start + self._PACKET_READ_TIMEOUT - else: - deadline = min(last_data_time + self._FRAME_STALL_TIMEOUT, start + self._FRAME_HARD_CAP) - if now > deadline: - break - await reads.wait(deadline - now) - finally: # a cancelled read's bytes too - if received: - logger.log(LOG_LEVEL_IO, "[%s] read %s", self._ftdi.device_id, bytes(received)) - if line_errors: - logger.warning("%d packets of this reply flagged a line error (bytes lost)", line_errors) - if expected_size is not None and len(data) >= expected_size: - return bytes(data[:expected_size]) - return bytes(data) - - 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, 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, header=FTDI_HEADER)) - frame = await self._ftdi_read_queued_frame(reads) - ftdi_validate_frame(frame, header=FTDI_HEADER) - return ftdi_extract_payload(frame) + if self.use_io != "ftdi": + raise RuntimeError("the CLARIOstar is connected over USB, not FTDI") + interface = self._ftdi_interface + if interface is None or interface.io is not self.io: + if interface is not None: + interface.stop() + interface = BMGFTDIInterface( + cast(FTDI, self.io), + header=FTDI_HEADER, + baudrate=125000, + packet_read_timeout=self._PACKET_READ_TIMEOUT, + frame_stall_timeout=self._FRAME_STALL_TIMEOUT, + frame_hard_cap=self._FRAME_HARD_CAP, + ) + self._ftdi_interface = interface + return interface # -- USB interface --------------------------------------------------------- @@ -522,44 +341,16 @@ async def _usb_exchange(self, payload: bytes) -> bytes: async def _purge_rx(self) -> None: """Discard unread reply bytes.""" if self.use_io == "ftdi": - await self._ftdi.usb_purge_rx_buffer() - if self._queued_reads is not None: - self._discard_unread(self._queued_reads) + await self._ftdi.purge_rx() else: await self._usb.drain() async def _drain_rx(self) -> None: - """Discard reply bytes until none has come for ``_FRAME_STALL_TIMEOUT``, at most - ``_FRAME_HARD_CAP``: the rest of a reply the reader is still sending. - - The reader sends a reply it has started to its end, a data page too (61,508 B over about - 7 s); after a cancelled read its rest would reach the next command's read. The queued reads' - bytes are logged on one I/O line, libftdi's by `FTDI.read`. - """ - if self.use_io != "ftdi": + """Discard the rest of a reply the reader is still sending (`BMGFTDIInterface.drain_rx`).""" + if self.use_io == "ftdi": + await self._ftdi.drain_rx() + else: await self._usb.drain() - return - reads = self._get_queued_reads() - drained = bytearray() - start = last_data_time = time.monotonic() - while True: - if reads is not None: - data, _ = reads.take() - else: - data = await self._ftdi.read(self._FTDI_READ_CHUNK) - now = time.monotonic() - if data: - drained += data - last_data_time = now - deadline = min(last_data_time + self._FRAME_STALL_TIMEOUT, start + self._FRAME_HARD_CAP) - if now >= deadline: - break - if reads is not None: - await reads.wait(deadline - now) - elif not data: - await asyncio.sleep(LATENCY_TIMER / 1000) # the chip sends at least once per period - if reads is not None and drained: - logger.log(LOG_LEVEL_IO, "[%s] discarded %s", self._ftdi.device_id, bytes(drained)) async def exchange(self, payload: bytes) -> bytes: """Send one logical payload and return the reply payload. @@ -569,7 +360,7 @@ async def exchange(self, payload: bytes) -> bytes: """ async with self._lock: if self.use_io == "ftdi": - return await self._ftdi_exchange(payload) + return await self._ftdi.exchange(payload) return await self._usb_exchange(payload) async def _exchange_with_retries(self, payload: bytes, retries: int) -> bytes: @@ -1135,7 +926,7 @@ async def setup(self) -> None: """ logger.debug("setting up the CLARIOstar on %s ...", self._describe_link()) if self.use_io == "ftdi": - await self._ftdi_setup() + await self._ftdi.setup() else: logger.warning( "the CLARIOstar USB interface has not been tested on hardware with this driver; please " @@ -1230,5 +1021,6 @@ async def stop(self, accept_plate_left_in_device: bool = False) -> None: drawer stays open and the connection stays open. """ await self._shut_down(accept_plate_left_in_device) - self._stop_queued_reads() # before the device handle under them is closed + if self._ftdi_interface is not None: + self._ftdi_interface.stop() # before the device handle under them is closed await self.io.stop() diff --git a/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py index 8090b151dc4..b54155264f4 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/master_tests.py @@ -17,6 +17,7 @@ from pylabrobot.bmg_labtech.protocol.framing import ( ftdi_extract_payload, ) +from pylabrobot.bmg_labtech.transport import ftdi_interface as ftdi_interface_module from pylabrobot.io.ftdi import FTDI from pylabrobot.io.usb import USB @@ -153,8 +154,8 @@ def parse_status(frame: bytes): def make_driver(io: FakeFTDI) -> CLARIOstarDriver: driver = CLARIOstarDriver(io=io) - driver._PACKET_READ_TIMEOUT = 0.01 - driver._FRAME_STALL_TIMEOUT = 0.05 + driver._ftdi._packet_read_timeout = 0.01 + driver._ftdi._frame_stall_timeout = 0.05 driver._RETRY_SETTLE = 0 driver.default_poll_interval = 0 driver.drawer.default_poll_interval = 0 @@ -265,9 +266,10 @@ async def test_a_long_read_does_not_count_as_silence(self): ] clock = [0.0] driver = make_driver(SessionReadsFTDI(reads, clock)) - driver._PACKET_READ_TIMEOUT, driver._FRAME_STALL_TIMEOUT = 1.0, 2.0 # the driver's own - with patch.object(master_module.time, "time", side_effect=lambda: clock[0]): - frame = await driver._ftdi_read_frame() + driver._ftdi._packet_read_timeout = CLARIOstarDriver._PACKET_READ_TIMEOUT + driver._ftdi._frame_stall_timeout = CLARIOstarDriver._FRAME_STALL_TIMEOUT + with patch.object(ftdi_interface_module.time, "time", side_effect=lambda: clock[0]): + frame = await driver._ftdi._read_frame() self.assertEqual(frame, b"".join(data for _, data in reads)) self.assertGreaterEqual(clock[0], reads[-1][0] + 2.0) # 2 s of silence after the last bytes diff --git a/pylabrobot/bmg_labtech/transport/__init__.py b/pylabrobot/bmg_labtech/transport/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/pylabrobot/bmg_labtech/transport/ftdi_interface.py b/pylabrobot/bmg_labtech/transport/ftdi_interface.py new file mode 100644 index 00000000000..86236b57a59 --- /dev/null +++ b/pylabrobot/bmg_labtech/transport/ftdi_interface.py @@ -0,0 +1,291 @@ +"""A BMG reader's FTDI interface: it writes a frame and reads exactly one reply frame back. + +The reply ends where its size field says (CR also occurs inside payloads). Replies are read +through bulk-in transfers kept pending from `setup` to `stop` (`ftdi_queued_reads`), or with +libftdi when those cannot run. +""" + +import asyncio +import logging +import time +from typing import Optional + +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.validation_utils import LOG_LEVEL_IO + +from .ftdi_queued_reads import LATENCY_TIMER, QueuedReads + +logger = logging.getLogger(__name__) + + +class BMGFTDIInterface: + """Exchanges BMG frames with one reader over FTDI, one reply frame per frame written.""" + + _FTDI_READ_CHUNK = 25 # bytes per read; a shorter read marks the end of a burst + + def __init__( + self, + io: FTDI, + *, + header: int, + baudrate: int, + packet_read_timeout: float, + frame_stall_timeout: float, + frame_hard_cap: float, + ): + """ + Args: + io: the reader's FTDI connection. + header: the reader's frame header byte. + baudrate: the reader's baud rate. + packet_read_timeout: seconds to wait for the start of a reply. + frame_stall_timeout: seconds without new bytes before a started frame is abandoned. + frame_hard_cap: seconds; ceiling on reading one frame. + """ + self.io = io + self.header = header + self.baudrate = baudrate + self._packet_read_timeout = packet_read_timeout + self._frame_stall_timeout = frame_stall_timeout + self._frame_hard_cap = frame_hard_cap + # Bulk-in transfers kept pending from setup to stop; None reads through libftdi. + self._queued_reads: Optional[QueuedReads] = None + + def _start_queued_reads(self) -> None: + """Keep bulk-in transfers pending on the open FTDI chip; if they cannot start, read with + libftdi (`_read_frame`). + + Not while capture or validation is active: their replay needs every read to pass `FTDI.read`. + The queued reads belong to the event loop that ran setup. + """ + if get_capture_or_validation_active(): + logger.info("capture or validation active: reading with libftdi") + return + try: + reads = QueuedReads.from_device(self.io.dev) + reads.start() + except Exception as e: # no open pylibftdi device, no libusb handle, nothing submitted + logger.warning("bulk-in transfers not kept pending (%s); reading with libftdi", e) + return + self._queued_reads = reads + + def _stop_queued_reads(self) -> None: + """Cancel and free the pending bulk-in transfers, if any; libftdi reads from then on.""" + reads, self._queued_reads = self._queued_reads, None + if reads is not None: + reads.stop() + + async def setup(self) -> None: + """Open the FTDI port at the reader's baud rate, 8N1, latency 2 ms, and keep bulk-in + transfers pending.""" + self._stop_queued_reads() # a second setup reopens the device under them + await self.io.setup() + await self.io.set_baudrate(self.baudrate) + await self.io.set_line_property(8, 0, 0) # 8N1 + await self.io.set_latency_timer(LATENCY_TIMER) + self._start_queued_reads() + + async def _write(self, frame: bytes) -> None: + """Write a whole frame; raise OSError on a short write.""" + written = await self.io.write(frame) + if written != len(frame): + raise OSError(f"Short write: sent {written} of {len(frame)} bytes") + + async def _read_frame(self) -> bytes: + """Read one FTDI reply frame. + + CR (0x0D) also occurs inside payloads, so once the header is in, the size field decides + when the frame is complete. Bytes before an STX are dropped to resync. Before the header, + ``packet_read_timeout`` bounds the wait; after it, only a silence of + ``frame_stall_timeout`` or the ``frame_hard_cap`` ends the read, as long replies + stream in bursts. + + Returns: + The frame, possibly short of its size field on a stall; the caller validates it. + """ + data = b"" + expected_size: Optional[int] = None + end_byte_found = False + start = time.time() + last_data_time = start + + while True: + chunk = await self.io.read(self._FTDI_READ_CHUNK) + if chunk: + data += chunk + last_data_time = time.time() + + if expected_size is None and data[0] != FTDI_STX: + index = data.find(FTDI_STX) + data = data[index:] if index != -1 else b"" + + end_byte_found = bool(data) and data[-1] == FTDI_CR + + if expected_size is None and len(data) >= 3: + expected_size = int.from_bytes(data[1:3], "big") + start = time.time() + + if len(chunk) < self._FTDI_READ_CHUNK and end_byte_found: + if expected_size is None or len(data) >= expected_size: + break + if expected_size is not None and len(data) >= expected_size: + break + + # The size is known, so ask for exactly the rest of the frame. + if expected_size is not None: + remainder = await self.io.read(expected_size - len(data)) + if remainder: + data += remainder + last_data_time = time.time() + end_byte_found = data[-1] == FTDI_CR + if len(data) >= expected_size: + break + else: + if end_byte_found and (expected_size is None or len(data) >= expected_size): + break + now = time.time() + if expected_size is None: + timed_out = now - start > self._packet_read_timeout + else: + timed_out = ( + now - last_data_time > self._frame_stall_timeout or now - start > self._frame_hard_cap + ) + if timed_out: + if expected_size is not None and 0 < len(data) < expected_size: + data += await self.io.read(expected_size - len(data)) + break + await asyncio.sleep(0.0001) + + if expected_size is not None and len(data) >= expected_size: + return data[:expected_size] + return data + + def _get_queued_reads(self) -> Optional[QueuedReads]: + """The queued reads while they run; None once they stopped by themselves (e.g. the device + was removed), and libftdi reads from then on.""" + reads = self._queued_reads + if reads is not None and not reads.running: + logger.warning("bulk-in transfers stopped (%s); reading with libftdi", reads.error) + self._stop_queued_reads() + return None + return reads + + def _discard_unread(self, reads: QueuedReads) -> None: + """Drop the bytes the queued reads hold unread, logging them at the I/O level.""" + data, _ = reads.take() + if data: + logger.log(LOG_LEVEL_IO, "[%s] discarded %s", self.io.device_id, data) + + async def _read_queued_frame(self, reads: QueuedReads) -> bytes: + """Read one FTDI reply frame from the queued reads, with `_read_frame`'s rules. + + Bytes before an STX are dropped and the size field ends the frame. It waits until the pending + reads receive bytes: ``packet_read_timeout`` bounds the wait for the header; after it, a + silence of ``frame_stall_timeout`` or the ``frame_hard_cap``. Packets flagging a line error + (an overrun loses bytes) are warned about. The IO log gets one line per reply: every byte + taken, dropped ones included. + + Returns: + The frame, possibly short of its size field on a stall; the caller validates it. + """ + data = bytearray() + received = bytearray() + expected_size: Optional[int] = None + line_errors = 0 + start = time.time() + last_data_time = start + try: + while True: + chunk, errors = reads.take() + line_errors += errors + now = time.time() + if chunk: + received += chunk + data += chunk + last_data_time = now + if expected_size is None: + index = data.find(FTDI_STX) + del data[: index if index != -1 else len(data)] + if len(data) >= 3: + expected_size = int.from_bytes(data[1:3], "big") + start = now + if expected_size is not None and len(data) >= expected_size: + break + elif not reads.running: + break + if expected_size is None: + deadline = start + self._packet_read_timeout + else: + deadline = min(last_data_time + self._frame_stall_timeout, start + self._frame_hard_cap) + if now > deadline: + break + await reads.wait(deadline - now) + finally: # a cancelled read's bytes too + if received: + logger.log(LOG_LEVEL_IO, "[%s] read %s", self.io.device_id, bytes(received)) + if line_errors: + logger.warning("%d packets of this reply flagged a line error (bytes lost)", line_errors) + if expected_size is not None and len(data) >= expected_size: + return bytes(data[:expected_size]) + return bytes(data) + + async def 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._write(ftdi_wrap_payload(payload, header=self.header)) + frame = await self._read_frame() + else: + self._discard_unread(reads) # bytes read before the command is sent are not its reply + await self._write(ftdi_wrap_payload(payload, header=self.header)) + frame = await self._read_queued_frame(reads) + ftdi_validate_frame(frame, header=self.header) + return ftdi_extract_payload(frame) + + async def purge_rx(self) -> None: + """Discard unread reply bytes.""" + await self.io.usb_purge_rx_buffer() + if self._queued_reads is not None: + self._discard_unread(self._queued_reads) + + async def drain_rx(self) -> None: + """Discard reply bytes until none has come for ``frame_stall_timeout``, at most + ``frame_hard_cap``: the rest of a reply the reader is still sending. + + The reader sends a reply it has started to its end, a data page too (61,508 B over about + 7 s); after a cancelled read its rest would reach the next command's read. The queued reads' + bytes are logged on one I/O line, libftdi's by `FTDI.read`. + """ + reads = self._get_queued_reads() + drained = bytearray() + start = last_data_time = time.monotonic() + while True: + if reads is not None: + data, _ = reads.take() + else: + data = await self.io.read(self._FTDI_READ_CHUNK) + now = time.monotonic() + if data: + drained += data + last_data_time = now + deadline = min(last_data_time + self._frame_stall_timeout, start + self._frame_hard_cap) + if now >= deadline: + break + if reads is not None: + await reads.wait(deadline - now) + elif not data: + await asyncio.sleep(LATENCY_TIMER / 1000) # the chip sends at least once per period + if reads is not None and drained: + logger.log(LOG_LEVEL_IO, "[%s] discarded %s", self.io.device_id, bytes(drained)) + + def stop(self) -> None: + """Cancel the pending bulk-in transfers, before the FTDI device under them is closed.""" + self._stop_queued_reads() diff --git a/pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads.py b/pylabrobot/bmg_labtech/transport/ftdi_queued_reads.py similarity index 100% rename from pylabrobot/bmg_labtech/clariostar/driver/ftdi_queued_reads.py rename to pylabrobot/bmg_labtech/transport/ftdi_queued_reads.py