fix(spark): improve USB/USBIP reliability for Tecan Spark 20M

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
 

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(