diff --git a/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py b/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py index ef4d2bac6d6..9580dbe4705 100644 --- a/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py +++ b/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\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 diff --git a/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py b/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py index b893ae382d7..6f9b938bdfb 100644 --- a/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py +++ b/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 = [