From c2ad9950413a09fb79f1bb156295ebb283de92c8 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Fri, 23 Oct 2026 20:50:27 +0100 Subject: [PATCH] v1 `CLARIOstar`: refuse a read, z-scan or grid scan with the drawer out The idle wait before a RUN, FOCUS_WELL or PLATE_MAP_SCAN now reads the drawer bit of the idle status it ends on and raises "the drawer is open; close it first" before the command is sent; the command then goes without a second wait. The frames sent are unchanged. Co-Authored-By: Claude Opus 5.5 --- .../driver/features/absorbance_tests.py | 20 +++++++++++ .../driver/features/fluorescence.py | 14 ++++++-- .../driver/features/fluorescence_tests.py | 21 +++++++++++ .../clariostar/driver/features/measurement.py | 35 ++++++++++++++++--- 4 files changed, 83 insertions(+), 7 deletions(-) diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py index 302d1454468..3c90a230bdd 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/absorbance_tests.py @@ -27,6 +27,8 @@ ShakeSettings, ) from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( + STATUS_BUSY, + STATUS_DRAWER_OPEN, STATUS_IDLE, STATUS_QUERY_FRAME, FakeFTDI, @@ -1160,5 +1162,23 @@ async def test_simulated_data_must_match_the_measurement(self): await measure_with(self.absorbance, wavelengths, simulated_data=simulated_data) +class DrawerOpenTests(unittest.IsolatedAsyncioTestCase): + async def test_a_read_with_the_drawer_out_is_refused_before_the_run(self): + io = FakeFTDI() + io.replies = [STATUS_DRAWER_OPEN] + with self.assertRaisesRegex(RuntimeError, "the drawer is open"): + await make_absorbance(io, ground_truth_96_well_plate()).measure(wavelengths=[600]) + self.assertEqual(io.written, [STATUS_QUERY_FRAME]) + + async def test_the_drawer_is_read_from_the_idle_status_the_run_waits_for(self): + io = FakeFTDI() + io.replies = [STATUS_BUSY, STATUS_DRAWER_OPEN] + absorbance = make_absorbance(io, ground_truth_96_well_plate()) + absorbance._driver.default_poll_interval = 0 + with self.assertRaisesRegex(RuntimeError, "the drawer is open"): + await absorbance.measure(wavelengths=[600]) + self.assertEqual(io.written, [STATUS_QUERY_FRAME, STATUS_QUERY_FRAME]) + + if __name__ == "__main__": unittest.main() diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py index cfbaabd2be6..8c3cf7442eb 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence.py @@ -1252,10 +1252,14 @@ async def _run_z_profile(self, parameters: bytes, timeout: float) -> bytes: Raises: TimeoutError: if the z-scan takes longer than `timeout` seconds. - RuntimeError: if it ends without reporting its result ready. + RuntimeError: if the drawer is open (nothing is sent), or the z-scan ends without reporting + its result ready. """ driver = self._driver - await driver.send_command(driver.CommandFamily.FOCUS_WELL, parameters=parameters) + await self._wait_until_idle_with_drawer_in() + await driver.send_command( + driver.CommandFamily.FOCUS_WELL, parameters=parameters, when_idle=False + ) start = time.monotonic() stage: Optional[int] = None try: @@ -1495,9 +1499,13 @@ async def _run_grid( Raises: TimeoutError: if the scan takes longer than `timeout` seconds. + RuntimeError: if the drawer is open; nothing is sent. """ driver = self._driver - await driver.send_command(driver.CommandFamily.PLATE_MAP_SCAN, parameters=parameters) + await self._wait_until_idle_with_drawer_in() + await driver.send_command( + driver.CommandFamily.PLATE_MAP_SCAN, parameters=parameters, when_idle=False + ) start = time.monotonic() grids: List[bytes] = [] try: diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py index 2364f5bba2c..f7459fc30dc 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/fluorescence_tests.py @@ -40,6 +40,7 @@ ) from pylabrobot.bmg_labtech.clariostar.driver.framing import FTDI_HEADER from pylabrobot.bmg_labtech.clariostar.driver.master_tests import ( + STATUS_DRAWER_OPEN, STATUS_IDLE, STATUS_QUERY_FRAME, FakeFTDI, @@ -1912,5 +1913,25 @@ async def test_invalid_arguments_send_nothing(self): self.assertEqual(io.written, []) +class DrawerOpenTests(unittest.IsolatedAsyncioTestCase): + async def test_a_read_a_z_scan_and_a_grid_scan_with_the_drawer_out_are_refused(self): + plate, grid = fluorescence_plate(), grid_plate() + calls = { + "read": lambda feature: feature.measure(BASELINE, focal_height=10.0, gain=1000), + "z-scan": lambda feature: feature.measure_z_profile(BASELINE, well=plate.get_well("A1")), + "grid scan": lambda feature: feature.measure_well_centers( + GRID_CHANNEL, focal_height=15.0, gain=2239 + ), + } + for name, call in calls.items(): + with self.subTest(name): + io = FakeFTDI() + io.replies = [STATUS_DRAWER_OPEN] + feature = make_fluorescence(io, grid if name == "grid scan" else plate) + with self.assertRaisesRegex(RuntimeError, "the drawer is open"): + await call(feature) + self.assertEqual(io.written, [STATUS_QUERY_FRAME]) + + if __name__ == "__main__": unittest.main() diff --git a/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py b/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py index a6b294aa1c4..0e8d0f1100b 100644 --- a/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py +++ b/pylabrobot/bmg_labtech/clariostar/driver/features/measurement.py @@ -19,7 +19,7 @@ from ..errors import CommandRefusedError if TYPE_CHECKING: - from ..master import CLARIOstarDriver + from ..master import CLARIOstarDriver, CLARIOstarStatus logger = logging.getLogger(__name__) @@ -607,9 +607,35 @@ def _compute_run_end(self, flashes: int) -> bytes: # clears it (every cycle of a kinetic read too). _DATA_READY = 0x01 + async def _wait_until_idle_with_drawer_in(self) -> None: + """Poll status until the reader is idle, as `send_command` does before a command, and raise if + that status shows the drawer out: a run, a z-scan and a grid scan move the plate under the + optics. The command then goes without a second wait (`when_idle=False`). + + Raises: + RuntimeError: if the drawer is open; nothing is sent after the status. + """ + statuses: List["CLARIOstarStatus"] = [] + + def idle(status: "CLARIOstarStatus") -> bool: + statuses.append(status) + return not status.busy + + await self._driver._wait_until_ready( + timeout=self._driver.default_read_timeout, + poll_interval=self._driver.default_poll_interval, + until=idle, + ) + if statuses[-1].drawer_open: + raise RuntimeError("the drawer is open; close it first") + async def _unchecked_fw_run(self, parameters: bytes) -> None: - """Send RUN with `parameters`, the bytes after the family byte.""" - await self._driver.send_command(self._driver.CommandFamily.RUN, parameters=parameters) + """Send RUN with `parameters`, the bytes after the family byte, once the reader is idle with + the drawer in (`_wait_until_idle_with_drawer_in`).""" + await self._wait_until_idle_with_drawer_in() + await self._driver.send_command( + self._driver.CommandFamily.RUN, parameters=parameters, when_idle=False + ) async def _request_page(self, retries: int) -> bytes: """Request one data page (`05 02 00 00 00 00 00`) while the reader may still be busy. @@ -697,7 +723,8 @@ async def _run_measurement( Raises: TimeoutError: if the run takes longer than `timeout` seconds. - RuntimeError: if the run ends without reporting its data ready, or before all its pages. + RuntimeError: if the drawer is open (nothing is sent), or the run ends without reporting its + data ready, or before all its pages. ValueError: if a data page holds no values. """ await self._unchecked_fw_run(parameters)