From 2278698aa7413d4e6949fc201fb2f52ca23ce772 Mon Sep 17 00:00:00 2001 From: hazlamshamin Date: Wed, 26 Aug 2026 17:09:23 +0800 Subject: [PATCH 1/3] Add Tecan Infinite state and control APIs Expose reader-reported tray, plate sensor, temperature, shaking, and instrument states through the Infinite 200 PRO backend. Confirm state-changing commands with device readback. Do not reinitialize the reader when a lost response leaves tray movement or shaking state unknown. Treat timed busy frames as valid reader responses and confirm the final transport position. --- .../legacy/plate_reading/tecan/__init__.py | 11 +- .../plate_reading/tecan/infinite_backend.py | 482 +++++++++++++++++- .../tecan/infinite_backend_tests.py | 328 +++++++++++- 3 files changed, 804 insertions(+), 17 deletions(-) diff --git a/pylabrobot/legacy/plate_reading/tecan/__init__.py b/pylabrobot/legacy/plate_reading/tecan/__init__.py index 5208dffaab3..5d72349f847 100644 --- a/pylabrobot/legacy/plate_reading/tecan/__init__.py +++ b/pylabrobot/legacy/plate_reading/tecan/__init__.py @@ -1 +1,10 @@ -from .infinite_backend import ExperimentalTecanInfinite200ProBackend +from .infinite_backend import ( + ExperimentalTecanInfinite200ProBackend, + TecanInfiniteInstrumentState, + TecanInfiniteInstrumentStatus, + TecanInfinitePlatePosition, + TecanInfinitePlateSensorState, + TecanInfiniteResponseError, + TecanInfiniteShakingMode, + TecanInfiniteTemperatureStatus, +) diff --git a/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py b/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py index 06c6e08ca5c..e90f78d1e00 100644 --- a/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py +++ b/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py @@ -13,7 +13,7 @@ import time from abc import ABC, abstractmethod from dataclasses import dataclass -from typing import Dict, List, Optional, Sequence, Tuple +from typing import Dict, List, Literal, Optional, Sequence, Tuple, cast from pylabrobot.io.binary import Reader from pylabrobot.io.usb import USB @@ -24,6 +24,34 @@ logger = logging.getLogger(__name__) BIN_RE = re.compile(r"^(\d+),BIN:$") +TecanInfinitePlatePosition = Literal[ + "UNKNOWN", "INIT", "HOME", "IN", "OUT", "HEATING", "SHAKING", "FLOATING" +] +TecanInfinitePlateSensorState = Literal["FREE", "TAKEN", "UNDEFINED"] +TecanInfiniteTemperatureStatus = Literal["ON", "OFF"] +TecanInfiniteShakingMode = Literal["LINEAR", "ORBITAL"] +TecanInfiniteInstrumentState = Literal[ + "standby", "power_down", "power_up", "parked", "busy", "busy_in_background", "message", "unknown" +] + + +class TecanInfiniteResponseError(RuntimeError): + """Reports that reader communication did not confirm a query result or requested change.""" + + def __init__(self, command: str, responses: Sequence[str], reason: str) -> None: + self.command = command + self.responses = tuple(responses) + self.reason = reason + super().__init__(f"Tecan Infinite response to {command!r} {reason}: {self.responses!r}") + + +@dataclass(frozen=True) +class TecanInfiniteInstrumentStatus: + """Contains the interpreted reader state and the unmodified device response.""" + + state: TecanInfiniteInstrumentState + raw: str + def _integration_microseconds_to_seconds(value: int) -> float: # DLL/UI indicates integration time is stored in microseconds; UI displays ms by dividing by 1000. @@ -428,6 +456,15 @@ def _handle_bin(self, _payload_len: int, _blob: bytes) -> None: class ExperimentalTecanInfinite200ProBackend(PlateReaderBackend): """Backend shell for the Infinite 200 PRO.""" + _PLATE_POSITIONS = {"UNKNOWN", "INIT", "HOME", "IN", "OUT", "HEATING", "SHAKING", "FLOATING"} + _STATUS_STATES: Dict[str, TecanInfiniteInstrumentState] = { + "ST": "standby", + "PD": "power_down", + "PU": "power_up", + "PA": "parked", + } + _BUSY_STATUS_RE = re.compile(r"^(?P\+)?BY(?:#.[0-9]+|%[0-9]+|\$.*)?$") + _MODE_CAPABILITY_COMMANDS: Dict[str, List[str]] = { "ABS": [ "#BEAM DIAMETER", @@ -551,14 +588,410 @@ async def stop(self) -> None: async def open(self) -> None: """Open the reader drawer.""" - await self._send_command("ABSOLUTE MTP,OUT") - await self._send_command("BY#T5000") + await self._move_plate_transport("OUT") async def close(self, plate: Optional[Plate]) -> None: # noqa: ARG002 """Close the reader drawer.""" - await self._send_command("ABSOLUTE MTP,IN") - await self._send_command("BY#T5000") + await self._move_plate_transport("IN") + + async def get_plate_position(self) -> TecanInfinitePlatePosition: + """Read the position that the reader reports for the plate transport. + + The result can be ``IN``, ``OUT``, or another device state. It does not show whether a + microplate is present. It does not prove that the transport moved without an obstruction. + """ + + command = "?ABSOLUTE MTP,POS" + response = await self._query_state(command) + if response not in self._PLATE_POSITIONS: + raise TecanInfiniteResponseError(command, [response], "contained an unknown position") + return cast(TecanInfinitePlatePosition, response) + + async def get_plate_sensor_state(self) -> TecanInfinitePlateSensorState: + """Read the plate-position sensor state. + + After a successful inward movement, ``FREE`` means that the sensor position is unoccupied. + ``TAKEN`` means that a microplate occupies the sensor position. ``UNDEFINED`` means that the + reader does not report a known sensor state. Do not interpret ``UNDEFINED`` as an absent + microplate. While the tray is out, the reader can retain the result from the prior inward + movement. + """ + + command = "?SENSOR PLATEPOS" + response = await self._query_state(command) + if response not in {"FREE", "TAKEN", "UNDEFINED"}: + raise TecanInfiniteResponseError(command, [response], "contained an unknown sensor state") + return cast(TecanInfinitePlateSensorState, response) + + async def get_current_temperature(self) -> float: + """Read the current plate temperature in degrees Celsius.""" + + return await self._get_temperature("CURRENT") + + async def get_temperature_target(self) -> float: + """Read the target plate temperature in degrees Celsius.""" + + return await self._get_temperature("TARGET") + + async def get_temperature_status(self) -> TecanInfiniteTemperatureStatus: + """Read whether plate-temperature control is on or off.""" + + return await self._get_temperature_status() + + async def set_temperature(self, temperature: float) -> None: + """Set the plate-temperature target. Start temperature control. + + The reader accepts command values from 0.0 to 42.0 degrees Celsius in 0.1-degree increments. + Tecan specifies an achievable heating range from ambient temperature plus 5 degrees Celsius + to 42 degrees Celsius. This method reads the applied target and the control status before it + returns. + """ + + raw_temperature = self._encode_temperature(temperature) + await self._send_control_command(f"TEMPERATURE PLATE,TARGET={raw_temperature}") + applied_temperature = await self._get_temperature("TARGET", recover_on_timeout=False) + if applied_temperature != raw_temperature / 10.0: + raise TecanInfiniteResponseError( + "?TEMPERATURE PLATE,TARGET", + [str(applied_temperature)], + f"did not match the requested target {raw_temperature / 10.0}", + ) + + await self._send_control_command("TEMPERATURE PLATE,STATUS=ON") + applied_status = await self._get_temperature_status(recover_on_timeout=False) + if applied_status != "ON": + raise TecanInfiniteResponseError( + "?TEMPERATURE PLATE,STATUS", + [applied_status], + "did not confirm that temperature control was enabled", + ) + + async def stop_temperature_control(self) -> None: + """Stop plate-temperature control. Confirm that the reader reports ``OFF``.""" + + await self._send_control_command("TEMPERATURE PLATE,STATUS=OFF") + applied_status = await self._get_temperature_status(recover_on_timeout=False) + if applied_status != "OFF": + raise TecanInfiniteResponseError( + "?TEMPERATURE PLATE,STATUS", + [applied_status], + "did not confirm that temperature control was disabled", + ) + + async def get_shaking_mode(self) -> TecanInfiniteShakingMode: + """Read the configured shaking mode.""" + + command = "?SHAKING MODE" + response = await self._query_state(command) + if response not in {"LINEAR", "ORBITAL"}: + raise TecanInfiniteResponseError(command, [response], "contained an unknown shaking mode") + return cast(TecanInfiniteShakingMode, response) + + async def get_shaking_duration(self) -> Optional[int]: + """Read the shaking duration in seconds. Return ``None`` if the reader reports ``-1``.""" + + duration = await self._get_integer_state("?SHAKING TIME", "shaking duration") + if duration == -1: + return None + if duration < 0: + raise TecanInfiniteResponseError( + "?SHAKING TIME", [str(duration)], "contained an invalid shaking duration" + ) + return duration + + async def get_shaking_amplitude(self) -> Optional[float]: + """Read the shaking amplitude in millimeters. Return ``None`` if the reader reports ``-1``.""" + + raw_amplitude = await self._get_integer_state("?SHAKING AMPLITUDE", "shaking amplitude") + if raw_amplitude == -1: + return None + if raw_amplitude < 0: + raise TecanInfiniteResponseError( + "?SHAKING AMPLITUDE", [str(raw_amplitude)], "contained an invalid shaking amplitude" + ) + return raw_amplitude / 1000.0 + + async def shake( + self, + duration: int, + mode: TecanInfiniteShakingMode = "ORBITAL", + amplitude: float = 1.0, + ) -> None: + """Start a timed plate shake. Return after the reader reports standby. + + Set ``duration`` from 1 to 999 seconds. Set ``mode`` to ``LINEAR`` or ``ORBITAL``. Set + ``amplitude`` from 1.0 to 6.0 millimeters in 0.5-millimeter increments. The reader calculates + the frequency from the mode and amplitude. Cancellation does not stop the timed action after + the reader accepts the ``SHAKING ON`` command. + """ + + if isinstance(duration, bool) or not isinstance(duration, int) or not 1 <= duration <= 999: + raise ValueError("Shaking duration must be an integer from 1 to 999 seconds.") + if mode not in {"LINEAR", "ORBITAL"}: + raise ValueError("Shaking mode must be 'LINEAR' or 'ORBITAL'.") + if not math.isfinite(amplitude) or not 1.0 <= amplitude <= 6.0: + raise ValueError("Shaking amplitude must be from 1 to 6 mm.") + raw_amplitude = round(amplitude * 1000) + if not math.isclose(raw_amplitude / 1000.0, amplitude) or raw_amplitude % 500 != 0: + raise ValueError("Shaking amplitude must use 0.5 mm increments.") + + await self._set_shaking_mode(mode) + await self._set_shaking_amplitude(raw_amplitude, amplitude) + await self._set_shaking_duration(duration) + + try: + responses = await self._send_command( + "SHAKING ON", wait_for_terminal=False, recover_on_timeout=False + ) + except TimeoutError as error: + raise TecanInfiniteResponseError( + "SHAKING ON", + [], + "was not received. The shaking action may have started. The reader may still be shaking", + ) from error + self._require_busy_response("SHAKING ON", responses) + if len(responses) == 2: + return + try: + terminal = await self._read_command_response( + max_iterations=1, timeout=duration + 5, recover_on_timeout=False + ) + except TimeoutError as error: + raise TecanInfiniteResponseError( + "SHAKING ON", + responses, + "started, but completion could not be confirmed; the reader may still be shaking", + ) from error + if terminal != ["ST"]: + raise TecanInfiniteResponseError( + "SHAKING ON", terminal, "did not finish with the standby response" + ) + + async def get_instrument_status(self) -> TecanInfiniteInstrumentStatus: + """Read the reader status. Return the interpreted state and the unmodified response.""" + + response = await self._query_state("QQ") + if response in self._STATUS_STATES: + state = self._STATUS_STATES[response] + elif response.startswith("MSG"): + state = "message" + else: + busy_match = self._BUSY_STATUS_RE.fullmatch(response) + if busy_match is None: + state = "unknown" + elif busy_match.group("background"): + state = "busy_in_background" + else: + state = "busy" + return TecanInfiniteInstrumentStatus(state=state, raw=response) + + async def is_busy(self) -> bool: + """Return whether the reader reports a busy state. + + Return ``False`` for a known standby or power state. Raise an error if the response does not + identify a known busy or non-busy state. + """ + + status = await self.get_instrument_status() + if status.state in {"busy", "busy_in_background"}: + return True + if status.state in {"standby", "power_down", "power_up", "parked"}: + return False + raise TecanInfiniteResponseError( + "QQ", [status.raw], "did not report a known busy or non-busy state" + ) + + async def _move_plate_transport(self, destination: Literal["IN", "OUT"]) -> None: + """Move the plate transport. Confirm the final position that the reader reports.""" + + command = f"ABSOLUTE MTP,{destination}" + try: + responses = await self._send_command( + command, wait_for_terminal=False, recover_on_timeout=False + ) + except TimeoutError as error: + raise TecanInfiniteResponseError( + command, + [], + "was not received. The transport movement may have started. " + f"The transport may not be {destination}", + ) from error + self._require_busy_response(command, responses) + if len(responses) == 1: + try: + terminal = await self._read_command_response( + max_iterations=1, timeout=10, recover_on_timeout=False + ) + except TimeoutError as error: + raise TecanInfiniteResponseError( + command, + responses, + f"started, but completion could not be confirmed; the transport may not be {destination}", + ) from error + if terminal != ["ST"]: + raise TecanInfiniteResponseError( + command, terminal, "did not finish with the standby response" + ) + + applied_position = await self._query_state("?ABSOLUTE MTP,POS", recover_on_timeout=False) + if applied_position != destination: + raise TecanInfiniteResponseError( + "?ABSOLUTE MTP,POS", + [applied_position], + f"did not confirm the requested {destination} position", + ) + + async def _get_temperature( + self, + reading: Literal["CURRENT", "TARGET"], + *, + recover_on_timeout: bool = True, + ) -> float: + command = f"?TEMPERATURE PLATE,{reading}" + response = await self._query_state(command, recover_on_timeout=recover_on_timeout) + try: + return int(response) / 10.0 + except ValueError as error: + raise TecanInfiniteResponseError( + command, [response], "did not contain a temperature in tenths of a degree Celsius" + ) from error + + async def _get_temperature_status( + self, + *, + recover_on_timeout: bool = True, + ) -> TecanInfiniteTemperatureStatus: + """Read the plate-temperature status. + + If ``recover_on_timeout`` is ``False``, do not reinitialize the reader after a timeout. + """ + + command = "?TEMPERATURE PLATE,STATUS" + response = await self._query_state(command, recover_on_timeout=recover_on_timeout) + if response not in {"ON", "OFF"}: + raise TecanInfiniteResponseError( + command, [response], "contained an unknown temperature status" + ) + return cast(TecanInfiniteTemperatureStatus, response) + + @staticmethod + def _encode_temperature(temperature: float) -> int: + """Convert degrees Celsius to the reader value in tenths of a degree.""" + + if not math.isfinite(temperature): + raise ValueError("Temperature must be finite.") + if not 0.0 <= temperature <= 42.0: + raise ValueError("Temperature must be from 0 to 42 degrees Celsius.") + raw_temperature = round(temperature * 10) + if not math.isclose(raw_temperature / 10.0, temperature): + raise ValueError("Temperature must use 0.1 degree Celsius increments.") + return raw_temperature + + async def _get_integer_state( + self, command: str, description: str, *, recover_on_timeout: bool = True + ) -> int: + """Read one integer-valued device state.""" + + response = await self._query_state(command, recover_on_timeout=recover_on_timeout) + try: + return int(response) + except ValueError as error: + raise TecanInfiniteResponseError( + command, [response], f"did not contain a numeric {description}" + ) from error + + async def _set_shaking_mode(self, mode: TecanInfiniteShakingMode) -> None: + """Set the shaking mode. Confirm the applied mode from the reader response.""" + + await self._send_control_command(f"SHAKING MODE={mode}") + applied = await self._query_state("?SHAKING MODE", recover_on_timeout=False) + if applied != mode: + raise TecanInfiniteResponseError( + "?SHAKING MODE", [applied], f"did not match the requested mode {mode}" + ) + + async def _set_shaking_amplitude(self, raw_amplitude: int, amplitude: float) -> None: + """Set the amplitude in thousandths of a millimeter. Confirm the applied value.""" + + await self._send_control_command(f"SHAKING AMPLITUDE={raw_amplitude}") + applied = await self._get_integer_state( + "?SHAKING AMPLITUDE", "shaking amplitude", recover_on_timeout=False + ) + if applied != raw_amplitude: + raise TecanInfiniteResponseError( + "?SHAKING AMPLITUDE", + [str(applied)], + f"did not match the requested amplitude {amplitude}", + ) + + async def _set_shaking_duration(self, duration: int) -> None: + """Set the shaking duration. Confirm the applied value from the reader response.""" + + await self._send_control_command(f"SHAKING TIME={duration}") + applied = await self._get_integer_state( + "?SHAKING TIME", "shaking duration", recover_on_timeout=False + ) + if applied != duration: + raise TecanInfiniteResponseError( + "?SHAKING TIME", [str(applied)], f"did not match the requested duration {duration}" + ) + + async def _send_control_command(self, command: str) -> None: + """Send a command that can change reader state. Require an ``ST`` or ``+`` response. + + Do not reinitialize the reader after a timeout. The resulting reader state can be unknown. + """ + + try: + responses = await self._send_command( + command, wait_for_terminal=False, recover_on_timeout=False + ) + except TimeoutError as error: + raise TecanInfiniteResponseError( + command, + [], + "may have been accepted, but its outcome could not be confirmed", + ) from error + if responses not in (["ST"], ["+"]): + raise TecanInfiniteResponseError(command, responses, "did not confirm the requested change") + + @staticmethod + def _require_busy_response(command: str, responses: Sequence[str]) -> None: + """Require a timed busy response for an accepted timed action.""" + + if ( + not responses + or not responses[0].startswith("BY#T") + or len(responses) > 2 + or (len(responses) == 2 and responses[1] != "ST") + ): + raise TecanInfiniteResponseError(command, responses, "did not report a timed busy state") + + async def _query_state(self, command: str, *, recover_on_timeout: bool = True) -> str: + """Send one read-only query. Require one response frame that is not a device error. + + If ``recover_on_timeout`` is ``False``, do not reinitialize the reader after a timeout. + """ + + try: + responses = await self._send_command( + command, wait_for_terminal=False, recover_on_timeout=recover_on_timeout + ) + except TimeoutError as error: + if recover_on_timeout: + raise + raise TecanInfiniteResponseError( + command, + [], + "could not be read without reinitializing the device; its state is indeterminate", + ) from error + if len(responses) != 1: + raise TecanInfiniteResponseError(command, responses, "did not contain exactly one frame") + response = responses[0] + if response.startswith("ERR") or response == "-": + raise TecanInfiniteResponseError(command, responses, "reported a device error") + return response async def _run_scan( self, @@ -1029,11 +1462,17 @@ def _reset_stream_state(self) -> None: self._pending_bin_events.clear() self._parser = _StreamParser(allow_bare_ascii=True) - async def _read_packet(self, size: int) -> bytes: + async def _read_packet( + self, size: int, timeout: Optional[int] = None, *, recover_on_timeout: bool = True + ) -> bytes: try: - data = await self.io.read(size=size) + if timeout is None: + data = await self.io.read(size=size) + else: + data = await self.io.read(timeout=timeout, size=size) except TimeoutError: - await self._recover_transport() + if recover_on_timeout: + await self._recover_transport() raise return data @@ -1130,6 +1569,7 @@ async def _send_command( wait_for_terminal: bool = True, allow_timeout: bool = False, read_response: bool = True, + recover_on_timeout: bool = True, ) -> List[str]: logger.debug("[tecan] >> %s", command) framed = self._frame_command(command) @@ -1138,14 +1578,18 @@ async def _send_command( return [] if command.startswith(("#", "?")): try: - return await self._read_command_response(require_terminal=False) + return await self._read_command_response( + require_terminal=False, recover_on_timeout=recover_on_timeout + ) except TimeoutError: if allow_timeout: logger.warning("Timeout waiting for response to %s", command) return [] raise try: - frames = await self._read_command_response(require_terminal=wait_for_terminal) + frames = await self._read_command_response( + require_terminal=wait_for_terminal, recover_on_timeout=recover_on_timeout + ) except TimeoutError: if allow_timeout: logger.warning("Timeout waiting for response to %s", command) @@ -1163,13 +1607,18 @@ async def _drain(self, attempts: int = 4) -> None: break async def _read_command_response( - self, max_iterations: int = 8, require_terminal: bool = True + self, + max_iterations: int = 8, + require_terminal: bool = True, + timeout: Optional[int] = None, + *, + recover_on_timeout: bool = True, ) -> List[str]: """Read response frames and cache any binary payloads that arrive.""" frames: List[str] = [] saw_terminal = False for _ in range(max_iterations): - chunk = await self._read_packet(128) + chunk = await self._read_packet(128, timeout=timeout, recover_on_timeout=recover_on_timeout) if not chunk: break for event in self._parser.feed(chunk): @@ -1183,7 +1632,7 @@ async def _read_command_response( break if require_terminal and saw_terminal and not self._parser.has_pending_bin(): break - if require_terminal and not saw_terminal: + 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) return frames @@ -1341,4 +1790,11 @@ def _handle_bin(self, payload_len: int, blob: bytes) -> None: __all__ = [ "ExperimentalTecanInfinite200ProBackend", + "TecanInfiniteInstrumentState", + "TecanInfiniteInstrumentStatus", + "TecanInfinitePlatePosition", + "TecanInfinitePlateSensorState", + "TecanInfiniteResponseError", + "TecanInfiniteShakingMode", + "TecanInfiniteTemperatureStatus", ] diff --git a/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py b/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py index 284254d75b0..a823fb121f7 100644 --- a/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py +++ b/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py @@ -2,8 +2,11 @@ from unittest.mock import AsyncMock, call, patch from pylabrobot.io.usb import USB -from pylabrobot.legacy.plate_reading.tecan.infinite_backend import ( +from pylabrobot.legacy.plate_reading.tecan import ( ExperimentalTecanInfinite200ProBackend, + TecanInfiniteResponseError, +) +from pylabrobot.legacy.plate_reading.tecan.infinite_backend import ( _absorbance_od_calibrated, _AbsorbanceRunDecoder, _consume_leading_ascii_frame, @@ -671,27 +674,346 @@ def _frame(self, command: str) -> bytes: async def test_open(self): self.backend._ready = True + self.mock_usb.read.side_effect = [ + self._frame("BY#T5000"), + self._frame("ST"), + self._frame("OUT"), + ] await self.backend.open() self.mock_usb.write.assert_has_calls( [ call(self._frame("ABSOLUTE MTP,OUT")), - call(self._frame("BY#T5000")), + call(self._frame("?ABSOLUTE MTP,POS")), ] ) async def test_close(self): self.backend._ready = True + self.mock_usb.read.side_effect = [ + self._frame("BY#T5000"), + self._frame("ST"), + self._frame("IN"), + ] await self.backend.close(self.plate) self.mock_usb.write.assert_has_calls( [ call(self._frame("ABSOLUTE MTP,IN")), - call(self._frame("BY#T5000")), + call(self._frame("?ABSOLUTE MTP,POS")), + ] + ) + + async def test_transport_timeout_does_not_reinitialize_indeterminate_hardware(self): + self.mock_usb.read.side_effect = [self._frame("BY#T5000"), TimeoutError("move timed out")] + + with self.assertRaisesRegex(TecanInfiniteResponseError, "transport may not be OUT"): + await self.backend.open() + + self.mock_usb.stop.assert_not_awaited() + self.mock_usb.setup.assert_not_awaited() + + async def test_transport_initial_timeout_does_not_reinitialize_indeterminate_hardware(self): + self.mock_usb.read.side_effect = TimeoutError("initial response timed out") + + with self.assertRaisesRegex(TecanInfiniteResponseError, "was not received"): + await self.backend.open() + + self.mock_usb.stop.assert_not_awaited() + self.mock_usb.setup.assert_not_awaited() + + async def test_get_plate_position_reads_transport_position(self): + self.mock_usb.read.return_value = self._frame("IN") + + position = await self.backend.get_plate_position() + + self.assertEqual(position, "IN") + self.mock_usb.write.assert_awaited_once_with(self._frame("?ABSOLUTE MTP,POS")) + + async def test_get_plate_sensor_state_returns_reader_states(self): + for response in ("FREE", "TAKEN", "UNDEFINED"): + with self.subTest(response=response): + self.mock_usb.read.return_value = self._frame(response) + self.assertEqual(await self.backend.get_plate_sensor_state(), response) + + self.assertEqual( + self.mock_usb.write.await_args_list, + [call(self._frame("?SENSOR PLATEPOS"))] * 3, + ) + + async def test_get_plate_sensor_state_rejects_unknown_sensor_state(self): + self.mock_usb.read.return_value = self._frame("NOT_READY") + + with self.assertRaisesRegex(TecanInfiniteResponseError, "unknown sensor state"): + await self.backend.get_plate_sensor_state() + + async def test_temperature_queries_convert_tenths_of_a_degree(self): + self.mock_usb.read.side_effect = [self._frame("218"), self._frame("370"), self._frame("ON")] + + current = await self.backend.get_current_temperature() + target = await self.backend.get_temperature_target() + status = await self.backend.get_temperature_status() + + self.assertEqual(current, 21.8) + self.assertEqual(target, 37.0) + self.assertEqual(status, "ON") + self.mock_usb.write.assert_has_awaits( + [ + call(self._frame("?TEMPERATURE PLATE,CURRENT")), + call(self._frame("?TEMPERATURE PLATE,TARGET")), + call(self._frame("?TEMPERATURE PLATE,STATUS")), + ] + ) + + async def test_temperature_query_rejects_non_numeric_response(self): + self.mock_usb.read.return_value = self._frame("MSG001: warming") + + with self.assertRaisesRegex(TecanInfiniteResponseError, "tenths of a degree Celsius"): + await self.backend.get_current_temperature() + + async def test_set_temperature_applies_target_enables_control_and_verifies_both(self): + self.mock_usb.read.side_effect = [ + self._frame("ST"), + self._frame("370"), + self._frame("ST"), + self._frame("ON"), + ] + + await self.backend.set_temperature(37.0) + + self.mock_usb.write.assert_has_awaits( + [ + call(self._frame("TEMPERATURE PLATE,TARGET=370")), + call(self._frame("?TEMPERATURE PLATE,TARGET")), + call(self._frame("TEMPERATURE PLATE,STATUS=ON")), + call(self._frame("?TEMPERATURE PLATE,STATUS")), + ] + ) + + async def test_set_temperature_rejects_invalid_values_before_io(self): + invalid_cases = [ + (-0.1, "0 to 42 degrees Celsius"), + (42.1, "0 to 42 degrees Celsius"), + (37.05, "0.1 degree Celsius increments"), + ] + + for temperature, message in invalid_cases: + with self.subTest(temperature=temperature): + with self.assertRaisesRegex(ValueError, message): + await self.backend.set_temperature(temperature) + + self.mock_usb.write.assert_not_awaited() + + async def test_set_temperature_rejects_readback_mismatch(self): + self.mock_usb.read.side_effect = [self._frame("ST"), self._frame("369")] + + with self.assertRaisesRegex(TecanInfiniteResponseError, "requested target 37.0"): + await self.backend.set_temperature(37.0) + + async def test_set_temperature_stops_after_device_rejects_target(self): + self.mock_usb.read.return_value = self._frame("-") + + with self.assertRaisesRegex(TecanInfiniteResponseError, "did not confirm"): + await self.backend.set_temperature(37.0) + + self.mock_usb.write.assert_awaited_once_with(self._frame("TEMPERATURE PLATE,TARGET=370")) + + async def test_stop_temperature_control_verifies_disabled_state(self): + self.mock_usb.read.side_effect = [self._frame("ST"), self._frame("OFF")] + + await self.backend.stop_temperature_control() + + self.mock_usb.write.assert_has_awaits( + [ + call(self._frame("TEMPERATURE PLATE,STATUS=OFF")), + call(self._frame("?TEMPERATURE PLATE,STATUS")), + ] + ) + + async def test_shaking_queries_decode_device_values(self): + self.mock_usb.read.side_effect = [ + self._frame("ORBITAL"), + self._frame("2"), + self._frame("1500"), + ] + + self.assertEqual(await self.backend.get_shaking_mode(), "ORBITAL") + self.assertEqual(await self.backend.get_shaking_duration(), 2) + self.assertEqual(await self.backend.get_shaking_amplitude(), 1.5) + + async def test_shaking_queries_report_unconfigured_values_as_unavailable(self): + self.mock_usb.read.side_effect = [self._frame("-1"), self._frame("-1")] + + self.assertIsNone(await self.backend.get_shaking_duration()) + self.assertIsNone(await self.backend.get_shaking_amplitude()) + + async def test_shake_configures_readbacks_and_waits_for_completion(self): + self.mock_usb.read.side_effect = [ + self._frame("ST"), + self._frame("ORBITAL"), + self._frame("ST"), + self._frame("1000"), + self._frame("ST"), + self._frame("2"), + self._frame("BY#T2000"), + self._frame("ST"), + ] + + await self.backend.shake(duration=2, mode="ORBITAL", amplitude=1.0) + + self.mock_usb.write.assert_has_awaits( + [ + call(self._frame("SHAKING MODE=ORBITAL")), + call(self._frame("?SHAKING MODE")), + call(self._frame("SHAKING AMPLITUDE=1000")), + call(self._frame("?SHAKING AMPLITUDE")), + call(self._frame("SHAKING TIME=2")), + call(self._frame("?SHAKING TIME")), + call(self._frame("SHAKING ON")), ] ) + self.assertEqual(self.mock_usb.read.await_args_list[-1], call(timeout=7, size=128)) + + async def test_shake_accepts_coalesced_busy_and_completion_frames(self): + self.mock_usb.read.side_effect = [ + self._frame("ST"), + self._frame("LINEAR"), + self._frame("ST"), + self._frame("1500"), + self._frame("ST"), + self._frame("1"), + self._frame("BY#T1000") + self._frame("ST"), + ] + + await self.backend.shake(duration=1, mode="LINEAR", amplitude=1.5) + + self.assertEqual(self.mock_usb.read.await_count, 7) + + async def test_shake_timeout_does_not_reinitialize_indeterminate_hardware(self): + self.mock_usb.read.side_effect = [ + self._frame("ST"), + self._frame("ORBITAL"), + self._frame("ST"), + self._frame("1000"), + self._frame("ST"), + self._frame("1"), + self._frame("BY#T1000"), + TimeoutError("reader did not report completion"), + ] + + with self.assertRaisesRegex(TecanInfiniteResponseError, "reader may still be shaking"): + await self.backend.shake(duration=1) + + self.mock_usb.stop.assert_not_awaited() + self.mock_usb.setup.assert_not_awaited() + + async def test_shake_initial_timeout_does_not_reinitialize_indeterminate_hardware(self): + self.mock_usb.read.side_effect = [ + self._frame("ST"), + self._frame("ORBITAL"), + self._frame("ST"), + self._frame("1000"), + self._frame("ST"), + self._frame("1"), + TimeoutError("initial response timed out"), + ] + + with self.assertRaisesRegex(TecanInfiniteResponseError, "reader may still be shaking"): + await self.backend.shake(duration=1) + + self.mock_usb.stop.assert_not_awaited() + self.mock_usb.setup.assert_not_awaited() + + async def test_control_readback_timeout_does_not_reinitialize_indeterminate_hardware(self): + self.mock_usb.read.side_effect = [self._frame("ST"), TimeoutError("readback timed out")] + + with self.assertRaisesRegex(TecanInfiniteResponseError, "state is indeterminate"): + await self.backend.set_temperature(25.0) + + self.mock_usb.stop.assert_not_awaited() + self.mock_usb.setup.assert_not_awaited() + + async def test_normal_read_timeout_retains_transport_recovery(self): + self.mock_usb.read.side_effect = TimeoutError("transport timed out") + + with patch.object(self.backend, "_recover_transport", new_callable=AsyncMock) as recover: + with self.assertRaisesRegex(TimeoutError, "transport timed out"): + await self.backend._read_packet(128) + + recover.assert_awaited_once_with() + + async def test_shake_rejects_invalid_parameters_before_io(self): + invalid_cases = [ + {"duration": 0}, + {"duration": 1000}, + {"duration": True}, + {"duration": 1, "mode": "SIDEWAYS"}, + {"duration": 1, "amplitude": 0.5}, + {"duration": 1, "amplitude": 1.2}, + {"duration": 1, "amplitude": 6.5}, + ] + + for parameters in invalid_cases: + with self.subTest(parameters=parameters): + with self.assertRaises(ValueError): + await self.backend.shake(**parameters) # type: ignore[arg-type] + + self.mock_usb.write.assert_not_awaited() + + async def test_get_instrument_status_normalizes_known_states_and_retains_raw_reply(self): + cases = [ + ("ST", "standby"), + ("PD", "power_down"), + ("PU", "power_up"), + ("PA", "parked"), + ("BY", "busy"), + ("BY#T5000", "busy"), + ("BY%C50", "unknown"), + ("BY%50", "busy"), + ("BY$reading", "busy"), + ("+BY#C2", "busy_in_background"), + ("MSG001: service requested", "message"), + ("NEW_STATUS", "unknown"), + ] + + for raw, expected_state in cases: + with self.subTest(raw=raw): + self.mock_usb.read.return_value = self._frame(raw) + status = await self.backend.get_instrument_status() + self.assertEqual(status.state, expected_state) + self.assertEqual(status.raw, raw) + + async def test_is_busy_sends_one_status_query(self): + self.mock_usb.read.return_value = self._frame("BY#C4") + + self.assertTrue(await self.backend.is_busy()) + + self.mock_usb.write.assert_awaited_once_with(self._frame("QQ")) + + async def test_is_busy_rejects_indeterminate_status(self): + for response in ("MSG001: service requested", "NEW_STATUS"): + with self.subTest(response=response): + self.mock_usb.read.return_value = self._frame(response) + with self.assertRaisesRegex(TecanInfiniteResponseError, "known busy or non-busy state"): + await self.backend.is_busy() + + async def test_state_query_raises_device_error_with_raw_reply(self): + self.mock_usb.read.return_value = self._frame("ERR123: sensor failed") + + with self.assertRaises(TecanInfiniteResponseError) as raised: + await self.backend.get_plate_sensor_state() + + self.assertEqual(raised.exception.command, "?SENSOR PLATEPOS") + self.assertEqual(raised.exception.responses, ("ERR123: sensor failed",)) + + async def test_state_query_requires_exactly_one_response_frame(self): + for response in (b"", self._frame("IN") + self._frame("ST")): + with self.subTest(response=response): + self.mock_usb.read.return_value = response + with self.assertRaisesRegex(TecanInfiniteResponseError, "exactly one frame"): + await self.backend.get_plate_position() async def test_read_absorbance_commands(self): """Test that read_absorbance sends the correct configuration commands.""" From 368f7c8e69d16756274213e94d6c7146c750c607 Mon Sep 17 00:00:00 2001 From: hazlamshamin Date: Wed, 26 Aug 2026 19:13:56 +0800 Subject: [PATCH 2/3] Document Tecan temperature readback helper --- pylabrobot/legacy/plate_reading/tecan/infinite_backend.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py b/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py index e90f78d1e00..791a1de7ba1 100644 --- a/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py +++ b/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py @@ -848,6 +848,13 @@ async def _get_temperature( *, recover_on_timeout: bool = True, ) -> float: + """Read the current or target plate temperature in degrees Celsius. + + Set ``reading`` to ``CURRENT`` for the measured value or ``TARGET`` for the configured target. + The reader reports the value in tenths of a degree Celsius. If ``recover_on_timeout`` is + ``False``, do not reinitialize the reader after a timeout. + """ + command = f"?TEMPERATURE PLATE,{reading}" response = await self._query_state(command, recover_on_timeout=recover_on_timeout) try: From 45b051634d8f6d469ae6460b05497f1bd0b6fdb0 Mon Sep 17 00:00:00 2001 From: hazlamshamin Date: Fri, 28 Aug 2026 19:25:47 +0800 Subject: [PATCH 3/3] Rename Tecan state queries to request convention --- .../plate_reading/tecan/infinite_backend.py | 68 +++++++++---------- .../tecan/infinite_backend_tests.py | 46 ++++++------- 2 files changed, 57 insertions(+), 57 deletions(-) diff --git a/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py b/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py index 791a1de7ba1..01c803bb096 100644 --- a/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py +++ b/pylabrobot/legacy/plate_reading/tecan/infinite_backend.py @@ -595,7 +595,7 @@ async def close(self, plate: Optional[Plate]) -> None: # noqa: ARG002 await self._move_plate_transport("IN") - async def get_plate_position(self) -> TecanInfinitePlatePosition: + async def request_plate_position(self) -> TecanInfinitePlatePosition: """Read the position that the reader reports for the plate transport. The result can be ``IN``, ``OUT``, or another device state. It does not show whether a @@ -603,12 +603,12 @@ async def get_plate_position(self) -> TecanInfinitePlatePosition: """ command = "?ABSOLUTE MTP,POS" - response = await self._query_state(command) + response = await self._request_state(command) if response not in self._PLATE_POSITIONS: raise TecanInfiniteResponseError(command, [response], "contained an unknown position") return cast(TecanInfinitePlatePosition, response) - async def get_plate_sensor_state(self) -> TecanInfinitePlateSensorState: + async def request_plate_sensor_state(self) -> TecanInfinitePlateSensorState: """Read the plate-position sensor state. After a successful inward movement, ``FREE`` means that the sensor position is unoccupied. @@ -619,25 +619,25 @@ async def get_plate_sensor_state(self) -> TecanInfinitePlateSensorState: """ command = "?SENSOR PLATEPOS" - response = await self._query_state(command) + response = await self._request_state(command) if response not in {"FREE", "TAKEN", "UNDEFINED"}: raise TecanInfiniteResponseError(command, [response], "contained an unknown sensor state") return cast(TecanInfinitePlateSensorState, response) - async def get_current_temperature(self) -> float: + async def request_current_temperature(self) -> float: """Read the current plate temperature in degrees Celsius.""" - return await self._get_temperature("CURRENT") + return await self._request_temperature("CURRENT") - async def get_temperature_target(self) -> float: + async def request_target_temperature(self) -> float: """Read the target plate temperature in degrees Celsius.""" - return await self._get_temperature("TARGET") + return await self._request_temperature("TARGET") - async def get_temperature_status(self) -> TecanInfiniteTemperatureStatus: + async def request_temperature_status(self) -> TecanInfiniteTemperatureStatus: """Read whether plate-temperature control is on or off.""" - return await self._get_temperature_status() + return await self._request_temperature_status() async def set_temperature(self, temperature: float) -> None: """Set the plate-temperature target. Start temperature control. @@ -650,7 +650,7 @@ async def set_temperature(self, temperature: float) -> None: raw_temperature = self._encode_temperature(temperature) await self._send_control_command(f"TEMPERATURE PLATE,TARGET={raw_temperature}") - applied_temperature = await self._get_temperature("TARGET", recover_on_timeout=False) + applied_temperature = await self._request_temperature("TARGET", recover_on_timeout=False) if applied_temperature != raw_temperature / 10.0: raise TecanInfiniteResponseError( "?TEMPERATURE PLATE,TARGET", @@ -659,7 +659,7 @@ async def set_temperature(self, temperature: float) -> None: ) await self._send_control_command("TEMPERATURE PLATE,STATUS=ON") - applied_status = await self._get_temperature_status(recover_on_timeout=False) + applied_status = await self._request_temperature_status(recover_on_timeout=False) if applied_status != "ON": raise TecanInfiniteResponseError( "?TEMPERATURE PLATE,STATUS", @@ -671,7 +671,7 @@ async def stop_temperature_control(self) -> None: """Stop plate-temperature control. Confirm that the reader reports ``OFF``.""" await self._send_control_command("TEMPERATURE PLATE,STATUS=OFF") - applied_status = await self._get_temperature_status(recover_on_timeout=False) + applied_status = await self._request_temperature_status(recover_on_timeout=False) if applied_status != "OFF": raise TecanInfiniteResponseError( "?TEMPERATURE PLATE,STATUS", @@ -679,19 +679,19 @@ async def stop_temperature_control(self) -> None: "did not confirm that temperature control was disabled", ) - async def get_shaking_mode(self) -> TecanInfiniteShakingMode: + async def request_shaking_mode(self) -> TecanInfiniteShakingMode: """Read the configured shaking mode.""" command = "?SHAKING MODE" - response = await self._query_state(command) + response = await self._request_state(command) if response not in {"LINEAR", "ORBITAL"}: raise TecanInfiniteResponseError(command, [response], "contained an unknown shaking mode") return cast(TecanInfiniteShakingMode, response) - async def get_shaking_duration(self) -> Optional[int]: + async def request_shaking_duration(self) -> Optional[int]: """Read the shaking duration in seconds. Return ``None`` if the reader reports ``-1``.""" - duration = await self._get_integer_state("?SHAKING TIME", "shaking duration") + duration = await self._request_integer_state("?SHAKING TIME", "shaking duration") if duration == -1: return None if duration < 0: @@ -700,10 +700,10 @@ async def get_shaking_duration(self) -> Optional[int]: ) return duration - async def get_shaking_amplitude(self) -> Optional[float]: + async def request_shaking_amplitude(self) -> Optional[float]: """Read the shaking amplitude in millimeters. Return ``None`` if the reader reports ``-1``.""" - raw_amplitude = await self._get_integer_state("?SHAKING AMPLITUDE", "shaking amplitude") + raw_amplitude = await self._request_integer_state("?SHAKING AMPLITUDE", "shaking amplitude") if raw_amplitude == -1: return None if raw_amplitude < 0: @@ -768,10 +768,10 @@ async def shake( "SHAKING ON", terminal, "did not finish with the standby response" ) - async def get_instrument_status(self) -> TecanInfiniteInstrumentStatus: + async def request_instrument_status(self) -> TecanInfiniteInstrumentStatus: """Read the reader status. Return the interpreted state and the unmodified response.""" - response = await self._query_state("QQ") + response = await self._request_state("QQ") if response in self._STATUS_STATES: state = self._STATUS_STATES[response] elif response.startswith("MSG"): @@ -786,14 +786,14 @@ async def get_instrument_status(self) -> TecanInfiniteInstrumentStatus: state = "busy" return TecanInfiniteInstrumentStatus(state=state, raw=response) - async def is_busy(self) -> bool: + async def request_is_busy(self) -> bool: """Return whether the reader reports a busy state. Return ``False`` for a known standby or power state. Raise an error if the response does not identify a known busy or non-busy state. """ - status = await self.get_instrument_status() + status = await self.request_instrument_status() if status.state in {"busy", "busy_in_background"}: return True if status.state in {"standby", "power_down", "power_up", "parked"}: @@ -834,7 +834,7 @@ async def _move_plate_transport(self, destination: Literal["IN", "OUT"]) -> None command, terminal, "did not finish with the standby response" ) - applied_position = await self._query_state("?ABSOLUTE MTP,POS", recover_on_timeout=False) + applied_position = await self._request_state("?ABSOLUTE MTP,POS", recover_on_timeout=False) if applied_position != destination: raise TecanInfiniteResponseError( "?ABSOLUTE MTP,POS", @@ -842,7 +842,7 @@ async def _move_plate_transport(self, destination: Literal["IN", "OUT"]) -> None f"did not confirm the requested {destination} position", ) - async def _get_temperature( + async def _request_temperature( self, reading: Literal["CURRENT", "TARGET"], *, @@ -856,7 +856,7 @@ async def _get_temperature( """ command = f"?TEMPERATURE PLATE,{reading}" - response = await self._query_state(command, recover_on_timeout=recover_on_timeout) + response = await self._request_state(command, recover_on_timeout=recover_on_timeout) try: return int(response) / 10.0 except ValueError as error: @@ -864,7 +864,7 @@ async def _get_temperature( command, [response], "did not contain a temperature in tenths of a degree Celsius" ) from error - async def _get_temperature_status( + async def _request_temperature_status( self, *, recover_on_timeout: bool = True, @@ -875,7 +875,7 @@ async def _get_temperature_status( """ command = "?TEMPERATURE PLATE,STATUS" - response = await self._query_state(command, recover_on_timeout=recover_on_timeout) + response = await self._request_state(command, recover_on_timeout=recover_on_timeout) if response not in {"ON", "OFF"}: raise TecanInfiniteResponseError( command, [response], "contained an unknown temperature status" @@ -895,12 +895,12 @@ def _encode_temperature(temperature: float) -> int: raise ValueError("Temperature must use 0.1 degree Celsius increments.") return raw_temperature - async def _get_integer_state( + async def _request_integer_state( self, command: str, description: str, *, recover_on_timeout: bool = True ) -> int: """Read one integer-valued device state.""" - response = await self._query_state(command, recover_on_timeout=recover_on_timeout) + response = await self._request_state(command, recover_on_timeout=recover_on_timeout) try: return int(response) except ValueError as error: @@ -912,7 +912,7 @@ async def _set_shaking_mode(self, mode: TecanInfiniteShakingMode) -> None: """Set the shaking mode. Confirm the applied mode from the reader response.""" await self._send_control_command(f"SHAKING MODE={mode}") - applied = await self._query_state("?SHAKING MODE", recover_on_timeout=False) + applied = await self._request_state("?SHAKING MODE", recover_on_timeout=False) if applied != mode: raise TecanInfiniteResponseError( "?SHAKING MODE", [applied], f"did not match the requested mode {mode}" @@ -922,7 +922,7 @@ async def _set_shaking_amplitude(self, raw_amplitude: int, amplitude: float) -> """Set the amplitude in thousandths of a millimeter. Confirm the applied value.""" await self._send_control_command(f"SHAKING AMPLITUDE={raw_amplitude}") - applied = await self._get_integer_state( + applied = await self._request_integer_state( "?SHAKING AMPLITUDE", "shaking amplitude", recover_on_timeout=False ) if applied != raw_amplitude: @@ -936,7 +936,7 @@ async def _set_shaking_duration(self, duration: int) -> None: """Set the shaking duration. Confirm the applied value from the reader response.""" await self._send_control_command(f"SHAKING TIME={duration}") - applied = await self._get_integer_state( + applied = await self._request_integer_state( "?SHAKING TIME", "shaking duration", recover_on_timeout=False ) if applied != duration: @@ -975,7 +975,7 @@ def _require_busy_response(command: str, responses: Sequence[str]) -> None: ): raise TecanInfiniteResponseError(command, responses, "did not report a timed busy state") - async def _query_state(self, command: str, *, recover_on_timeout: bool = True) -> str: + async def _request_state(self, command: str, *, recover_on_timeout: bool = True) -> str: """Send one read-only query. Require one response frame that is not a device error. If ``recover_on_timeout`` is ``False``, do not reinitialize the reader after a timeout. diff --git a/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py b/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py index a823fb121f7..ae269c9d204 100644 --- a/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py +++ b/pylabrobot/legacy/plate_reading/tecan/infinite_backend_tests.py @@ -724,37 +724,37 @@ async def test_transport_initial_timeout_does_not_reinitialize_indeterminate_har self.mock_usb.stop.assert_not_awaited() self.mock_usb.setup.assert_not_awaited() - async def test_get_plate_position_reads_transport_position(self): + async def test_request_plate_position_reads_transport_position(self): self.mock_usb.read.return_value = self._frame("IN") - position = await self.backend.get_plate_position() + position = await self.backend.request_plate_position() self.assertEqual(position, "IN") self.mock_usb.write.assert_awaited_once_with(self._frame("?ABSOLUTE MTP,POS")) - async def test_get_plate_sensor_state_returns_reader_states(self): + async def test_request_plate_sensor_state_returns_reader_states(self): for response in ("FREE", "TAKEN", "UNDEFINED"): with self.subTest(response=response): self.mock_usb.read.return_value = self._frame(response) - self.assertEqual(await self.backend.get_plate_sensor_state(), response) + self.assertEqual(await self.backend.request_plate_sensor_state(), response) self.assertEqual( self.mock_usb.write.await_args_list, [call(self._frame("?SENSOR PLATEPOS"))] * 3, ) - async def test_get_plate_sensor_state_rejects_unknown_sensor_state(self): + async def test_request_plate_sensor_state_rejects_unknown_sensor_state(self): self.mock_usb.read.return_value = self._frame("NOT_READY") with self.assertRaisesRegex(TecanInfiniteResponseError, "unknown sensor state"): - await self.backend.get_plate_sensor_state() + await self.backend.request_plate_sensor_state() async def test_temperature_queries_convert_tenths_of_a_degree(self): self.mock_usb.read.side_effect = [self._frame("218"), self._frame("370"), self._frame("ON")] - current = await self.backend.get_current_temperature() - target = await self.backend.get_temperature_target() - status = await self.backend.get_temperature_status() + current = await self.backend.request_current_temperature() + target = await self.backend.request_target_temperature() + status = await self.backend.request_temperature_status() self.assertEqual(current, 21.8) self.assertEqual(target, 37.0) @@ -771,7 +771,7 @@ async def test_temperature_query_rejects_non_numeric_response(self): self.mock_usb.read.return_value = self._frame("MSG001: warming") with self.assertRaisesRegex(TecanInfiniteResponseError, "tenths of a degree Celsius"): - await self.backend.get_current_temperature() + await self.backend.request_current_temperature() async def test_set_temperature_applies_target_enables_control_and_verifies_both(self): self.mock_usb.read.side_effect = [ @@ -839,15 +839,15 @@ async def test_shaking_queries_decode_device_values(self): self._frame("1500"), ] - self.assertEqual(await self.backend.get_shaking_mode(), "ORBITAL") - self.assertEqual(await self.backend.get_shaking_duration(), 2) - self.assertEqual(await self.backend.get_shaking_amplitude(), 1.5) + self.assertEqual(await self.backend.request_shaking_mode(), "ORBITAL") + self.assertEqual(await self.backend.request_shaking_duration(), 2) + self.assertEqual(await self.backend.request_shaking_amplitude(), 1.5) async def test_shaking_queries_report_unconfigured_values_as_unavailable(self): self.mock_usb.read.side_effect = [self._frame("-1"), self._frame("-1")] - self.assertIsNone(await self.backend.get_shaking_duration()) - self.assertIsNone(await self.backend.get_shaking_amplitude()) + self.assertIsNone(await self.backend.request_shaking_duration()) + self.assertIsNone(await self.backend.request_shaking_amplitude()) async def test_shake_configures_readbacks_and_waits_for_completion(self): self.mock_usb.read.side_effect = [ @@ -962,7 +962,7 @@ async def test_shake_rejects_invalid_parameters_before_io(self): self.mock_usb.write.assert_not_awaited() - async def test_get_instrument_status_normalizes_known_states_and_retains_raw_reply(self): + async def test_request_instrument_status_normalizes_known_states_and_retains_raw_reply(self): cases = [ ("ST", "standby"), ("PD", "power_down"), @@ -981,29 +981,29 @@ async def test_get_instrument_status_normalizes_known_states_and_retains_raw_rep for raw, expected_state in cases: with self.subTest(raw=raw): self.mock_usb.read.return_value = self._frame(raw) - status = await self.backend.get_instrument_status() + status = await self.backend.request_instrument_status() self.assertEqual(status.state, expected_state) self.assertEqual(status.raw, raw) - async def test_is_busy_sends_one_status_query(self): + async def test_request_is_busy_sends_one_status_query(self): self.mock_usb.read.return_value = self._frame("BY#C4") - self.assertTrue(await self.backend.is_busy()) + self.assertTrue(await self.backend.request_is_busy()) self.mock_usb.write.assert_awaited_once_with(self._frame("QQ")) - async def test_is_busy_rejects_indeterminate_status(self): + async def test_request_is_busy_rejects_indeterminate_status(self): for response in ("MSG001: service requested", "NEW_STATUS"): with self.subTest(response=response): self.mock_usb.read.return_value = self._frame(response) with self.assertRaisesRegex(TecanInfiniteResponseError, "known busy or non-busy state"): - await self.backend.is_busy() + await self.backend.request_is_busy() async def test_state_query_raises_device_error_with_raw_reply(self): self.mock_usb.read.return_value = self._frame("ERR123: sensor failed") with self.assertRaises(TecanInfiniteResponseError) as raised: - await self.backend.get_plate_sensor_state() + await self.backend.request_plate_sensor_state() self.assertEqual(raised.exception.command, "?SENSOR PLATEPOS") self.assertEqual(raised.exception.responses, ("ERR123: sensor failed",)) @@ -1013,7 +1013,7 @@ async def test_state_query_requires_exactly_one_response_frame(self): with self.subTest(response=response): self.mock_usb.read.return_value = response with self.assertRaisesRegex(TecanInfiniteResponseError, "exactly one frame"): - await self.backend.get_plate_position() + await self.backend.request_plate_position() async def test_read_absorbance_commands(self): """Test that read_absorbance sends the correct configuration commands."""