Wait for Tecan Infinite terminal responses

pylabrobot/legacy/plate_reading/tecan/infinite_backend.py

@@ -23,6 +23,7 @@
 
 logger = logging.getLogger(__name__)
 BIN_RE = re.compile(r"^(\d+),BIN:$")
+TIMED_BUSY_RE = re.compile(r"^\+?BY#T(?P<milliseconds>\d+)$")
 
 TecanInfinitePlatePosition = Literal[
   "UNKNOWN", "INIT", "HOME", "IN", "OUT", "HEATING", "SHAKING", "FLOATING"
@@ -456,6 +457,7 @@ def _handle_bin(self, _payload_len: int, _blob: bytes) -> None:
 class ExperimentalTecanInfinite200ProBackend(PlateReaderBackend):
   """Backend shell for the Infinite 200 PRO."""
 
+  _DEFAULT_READ_TIMEOUT_S = 30
   _PLATE_POSITIONS = {"UNKNOWN", "INIT", "HOME", "IN", "OUT", "HEATING", "SHAKING", "FLOATING"}
   _STATUS_STATES: Dict[str, TecanInfiniteInstrumentState] = {
     "ST": "standby",
@@ -549,7 +551,7 @@ def __init__(
       id_product=self.PRODUCT_ID,
       human_readable_device_name="Tecan Infinite 200 PRO",
       packet_read_timeout=3,
-      read_timeout=30,
+      read_timeout=self._DEFAULT_READ_TIMEOUT_S,
     )
     self.counts_per_mm_x = counts_per_mm_x
     self.counts_per_mm_y = counts_per_mm_y
@@ -1104,16 +1106,16 @@ async def read_absorbance(
       await self._end_run()
 
   async def _clear_mode_settings(self, excitation: bool = False, emission: bool = False) -> None:
-    """Clear mode settings before configuring a new scan."""
+    """Clear the mode settings before a new scan. Confirm each change."""
     if excitation:
-      await self._send_command("EXCITATION CLEAR", allow_timeout=True)
+      await self._send_control_command("EXCITATION CLEAR")
     if emission:
-      await self._send_command("EMISSION CLEAR", allow_timeout=True)
-    await self._send_command("TIME CLEAR", allow_timeout=True)
-    await self._send_command("GAIN CLEAR", allow_timeout=True)
-    await self._send_command("READS CLEAR", allow_timeout=True)
-    await self._send_command("POSITION CLEAR", allow_timeout=True)
-    await self._send_command("MIRROR CLEAR", allow_timeout=True)
+      await self._send_control_command("EMISSION CLEAR")
+    await self._send_control_command("TIME CLEAR")
+    await self._send_control_command("GAIN CLEAR")
+    await self._send_control_command("READS CLEAR")
+    await self._send_control_command("POSITION CLEAR")
+    await self._send_control_command("MIRROR CLEAR")
 
   async def _configure_absorbance(
     self,
@@ -1250,7 +1252,7 @@ async def _configure_fluorescence(
 
     # UI issues the entire FI configuration twice before PREPARE REF.
     for _ in range(2):
-      await self._send_command("MODE FI.TOP", allow_timeout=True)
+      await self._send_command("MODE FI.TOP")
       await self._clear_mode_settings(excitation=True, emission=True)
       await self._send_command(
         f"EXCITATION 0,FI,{ex_decitenth},{excitation_bandwidth},0", allow_timeout=True
@@ -1621,16 +1623,26 @@ async def _read_command_response(
     *,
     recover_on_timeout: bool = True,
   ) -> List[str]:
-    """Read response frames and cache any binary payloads that arrive."""
+    """Read response frames and cache binary payloads.
+
+    Raise ``TimeoutError`` if a required terminal frame does not arrive.
+    """
     frames: List[str] = []
     saw_terminal = False
+    default_timeout = self._DEFAULT_READ_TIMEOUT_S if timeout is None else timeout
+    next_read_timeout = timeout
     for _ in range(max_iterations):
-      chunk = await self._read_packet(128, timeout=timeout, recover_on_timeout=recover_on_timeout)
+      chunk = await self._read_packet(
+        128, timeout=next_read_timeout, recover_on_timeout=recover_on_timeout
+      )
       if not chunk:
         break
       for event in self._parser.feed(chunk):
         if event.text is not None:
           frames.append(event.text)
+          busy_timeout = self._timed_busy_timeout(event.text)
+          if busy_timeout is not None:
+            next_read_timeout = max(default_timeout, busy_timeout)
           if self._is_terminal_frame(event.text):
             saw_terminal = True
         elif event.payload_len is not None and event.blob is not None:
@@ -1642,12 +1654,24 @@ async def _read_command_response(
     if require_terminal and not saw_terminal and recover_on_timeout:
       # best effort: drain once more so pending ST doesn't leak into next command
       await self._drain(1)
+    if require_terminal and not saw_terminal:
+      raise TimeoutError("Timed out waiting for a terminal response frame.")
     return frames
 
   @staticmethod
   def _is_terminal_frame(text: str) -> bool:
     """Return True if the ASCII frame is a terminal marker."""
-    return text in {"ST", "+", "-"} or text.startswith("BY#T")
+    return text in {"ST", "+", "-"}
+
+  @staticmethod
+  def _timed_busy_timeout(text: str) -> Optional[int]:
+    """Return the advertised timed-busy timeout in seconds, rounded up."""
+
+    match = TIMED_BUSY_RE.fullmatch(text)
+    if match is None:
+      return None
+    milliseconds = int(match.group("milliseconds"))
+    return max(1, math.ceil(milliseconds / 1000))
 
 
 @dataclass

pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py

@@ -641,9 +641,15 @@ def test_terminal_frames(self):
     self.assertTrue(ExperimentalTecanInfinite200ProBackend._is_terminal_frame("ST"))
     self.assertTrue(ExperimentalTecanInfinite200ProBackend._is_terminal_frame("+"))
     self.assertTrue(ExperimentalTecanInfinite200ProBackend._is_terminal_frame("-"))
-    self.assertTrue(ExperimentalTecanInfinite200ProBackend._is_terminal_frame("BY#T5000"))
+    self.assertFalse(ExperimentalTecanInfinite200ProBackend._is_terminal_frame("BY#T5000"))
     self.assertFalse(ExperimentalTecanInfinite200ProBackend._is_terminal_frame("OK"))
 
+  def test_timed_busy_timeout(self):
+    self.assertEqual(ExperimentalTecanInfinite200ProBackend._timed_busy_timeout("BY#T5000"), 5)
+    self.assertEqual(ExperimentalTecanInfinite200ProBackend._timed_busy_timeout("BY#T5001"), 6)
+    self.assertEqual(ExperimentalTecanInfinite200ProBackend._timed_busy_timeout("+BY#T0"), 1)
+    self.assertIsNone(ExperimentalTecanInfinite200ProBackend._timed_busy_timeout("BY#A5000"))
+
 
 class TestTecanInfiniteCommands(unittest.IsolatedAsyncioTestCase):
   """Tests that verify correct commands are sent to the device."""
@@ -673,6 +679,56 @@ def _frame(self, command: str) -> bytes:
     """Helper to frame a command."""
     return ExperimentalTecanInfinite200ProBackend._frame_command(command)
 
+  async def test_timed_command_waits_for_standby_before_next_command(self):
+    self.mock_usb.read.side_effect = [
+      self._frame("BY#T5000"),
+      self._frame("ST"),
+      self._frame("+"),
+    ]
+
+    mode_responses = await self.backend._send_command("MODE FI.TOP")
+    clear_responses = await self.backend._send_command("EXCITATION CLEAR")
+
+    self.assertEqual(mode_responses, ["BY#T5000", "ST"])
+    self.assertEqual(clear_responses, ["+"])
+    self.assertEqual(self.mock_usb.read.await_args_list[1], call(timeout=30, size=128))
+    self.assertEqual(self.mock_usb.read.await_count, 3)
+
+  async def test_timed_command_uses_longer_advertised_timeout(self):
+    self.mock_usb.read.side_effect = [
+      self._frame("BY#T80000"),
+      self._frame("ST"),
+    ]
+
+    responses = await self.backend._send_command("INIT FORCE")
+
+    self.assertEqual(responses, ["BY#T80000", "ST"])
+    self.assertEqual(self.mock_usb.read.await_args_list[1], call(timeout=80, size=128))
+
+  async def test_required_terminal_response_does_not_accept_busy_frame(self):
+    self.mock_usb.read.return_value = self._frame("BY#T5000")
+
+    with self.assertRaisesRegex(TimeoutError, "terminal response frame"):
+      await self.backend._read_command_response(max_iterations=1, recover_on_timeout=False)
+
+  async def test_clear_mode_timeout_does_not_reinitialize_or_continue(self):
+    self.mock_usb.read.side_effect = TimeoutError("clear response timed out")
+
+    with self.assertRaisesRegex(TecanInfiniteResponseError, "outcome could not be confirmed"):
+      await self.backend._clear_mode_settings(excitation=True)
+
+    self.mock_usb.write.assert_awaited_once_with(self._frame("EXCITATION CLEAR"))
+    self.mock_usb.stop.assert_not_awaited()
+    self.mock_usb.setup.assert_not_awaited()
+
+  async def test_clear_mode_rejects_device_error(self):
+    self.mock_usb.read.return_value = self._frame("-")
+
+    with self.assertRaisesRegex(TecanInfiniteResponseError, "did not confirm"):
+      await self.backend._clear_mode_settings(excitation=True)
+
+    self.mock_usb.write.assert_awaited_once_with(self._frame("EXCITATION CLEAR"))
+
   async def test_open(self):
     self.backend._ready = True
     self.mock_usb.read.side_effect = [