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