diff --git a/pylabrobot/plate_reading/tecan/spark20m/spark_backend.py b/pylabrobot/plate_reading/tecan/spark20m/spark_backend.py index 093de094394..29594510a3c 100644 --- a/pylabrobot/plate_reading/tecan/spark20m/spark_backend.py +++ b/pylabrobot/plate_reading/tecan/spark20m/spark_backend.py @@ -1,3 +1,4 @@ +import asyncio import logging import statistics import time @@ -125,7 +126,13 @@ async def _run_measurement( """Execute a measurement: start background read, scan plate, collect results. This is the shared orchestration for all measurement types. + + Cleanup commands are sent immediately after scanning while the interrupt + endpoint is still active. Over USBIP the interrupt endpoint goes stale if + left unread for >2 s, so deferring cleanup until after the background reader + is stopped causes 15+ second hangs per command. """ + bg_task, stop_event, results = await self.reader.start_background_read(device) if bg_task is None or stop_event is None or results is None: @@ -134,11 +141,27 @@ async def _run_measurement( try: await self.measurement_control.prepare_instrument(measure_reference=True) await self.scan_plate_range(plate, wells, z) + + # The interrupt endpoint is hot right now (just received RespReady for + # the SCAN command). Send cleanup commands NOW, before any idle gap + # that would let the endpoint go stale. + await self.data_control.turn_all_interval_messages_off() + await self.measurement_control.end_measurement() + + # Smart drain: wait for remaining bulk data packets. + # Exit early once no new packets arrive for 0.5 s (after a 2 s minimum). + drain_start = time.monotonic() + last_count = len(results) + while time.monotonic() - drain_start < 5.0: + await asyncio.sleep(0.5) + current_count = len(results) + if current_count > last_count: + last_count = current_count + elif time.monotonic() - drain_start > 2.0: + break finally: stop_event.set() await bg_task - await self.data_control.turn_all_interval_messages_off() - await self.measurement_control.end_measurement() return results diff --git a/pylabrobot/plate_reading/tecan/spark20m/spark_reader_async.py b/pylabrobot/plate_reading/tecan/spark20m/spark_reader_async.py index 83a1d3256b0..e818111f28f 100644 --- a/pylabrobot/plate_reading/tecan/spark20m/spark_reader_async.py +++ b/pylabrobot/plate_reading/tecan/spark20m/spark_reader_async.py @@ -272,7 +272,7 @@ def _init_read( self, reader: USB, count: int = 512, - read_timeout: int = 2000, + read_timeout: int = 5000, ) -> "asyncio.Future[Any]": # Convert read_timeout from milliseconds to seconds for USB class. return asyncio.ensure_future( @@ -292,35 +292,39 @@ async def _get_response( ) -> Optional[Dict[str, Any]]: try: data = await read_task - - if data is None: - logging.warning("Read task returned None") - return None - - data_bytes = bytes(data) - logging.debug(f"Read task completed ({len(data_bytes)} bytes): {data_bytes.hex()}") - parsed = {} - if len(data_bytes) > 0: - try: - parsed = parse_single_spark_packet(data_bytes) - except ValueError as e: - logging.warning(f"Failed to parse packet: {e}") - # Treat as not ready/retry - if parsed.get("type") == "RespMessage": - self.msgs.append(parsed["payload"]) - elif parsed.get("type") == "RespError": - raise SparkError(parsed) + if data is not None: + data_bytes = bytes(data) + logging.debug(f"Read task completed ({len(data_bytes)} bytes): {data_bytes.hex()}") + if len(data_bytes) > 0: + try: + parsed = parse_single_spark_packet(data_bytes) + except ValueError as e: + logging.warning(f"Failed to parse packet: {e}") + + if parsed.get("type") == "RespMessage": + self.msgs.append(parsed["payload"]) + elif parsed.get("type") == "RespError": + raise SparkError(parsed) + else: + # Over USBIP the initial read can time out before the device responds. + # Fall through to the retry loop instead of giving up immediately. + logging.debug("Initial read returned None, entering retry loop...") deadline = time.monotonic() + timeout while parsed.get("type") != "RespReady" and time.monotonic() < deadline: try: - await asyncio.sleep(0.01) - logging.debug(f"Still busy, retrying... time left: {deadline - time.monotonic():.1f}s") + remaining = deadline - time.monotonic() + if remaining <= 0: + break + # Use min(1.0, remaining) — USBIP doesn't reliably honor sub-second + # USB timeouts, so 20ms reads return immediately as empty on some hosts. + usb_timeout = min(1.0, remaining) + logging.debug(f"Still busy, retrying... time left: {remaining:.1f}s") resp = await self._read_packet_in_executor( - reader=reader, endpoint=None, size=512, timeout=0.02 + reader=reader, endpoint=None, size=512, timeout=usb_timeout ) if resp: @@ -372,7 +376,7 @@ async def background_reader() -> None: f"Starting background reader for {device_type.name} {endpoint.name} (0x{endpoint.value:02x})" ) while not stop_event.is_set(): - await asyncio.sleep(0.2) # Avoid tight loop + await asyncio.sleep(0.05) # 50ms polling to capture fast spectrum data bursts try: # timeout in seconds data = await self._read_packet_in_executor(