From 2b5698d31872585e5684e14a61953a32ce231f58 Mon Sep 17 00:00:00 2001 From: Abdeltoto Date: Sat, 25 Apr 2026 20:13:46 -0400 Subject: [PATCH 1/3] Raise when thermocycler hold time is unavailable Made-with: Cursor --- pylabrobot/thermocycling/chatterbox.py | 4 ++-- pylabrobot/thermocycling/opentrons_backend.py | 5 ++++- .../thermocycling/opentrons_backend_tests.py | 8 +++++++ pylabrobot/thermocycling/thermocycler.py | 15 ++++++++----- .../thermocycling/thermocycler_tests.py | 21 +++++++++++++++++++ 5 files changed, 45 insertions(+), 8 deletions(-) diff --git a/pylabrobot/thermocycling/chatterbox.py b/pylabrobot/thermocycling/chatterbox.py index 1c45e40752d..4a3a3c84a23 100644 --- a/pylabrobot/thermocycling/chatterbox.py +++ b/pylabrobot/thermocycling/chatterbox.py @@ -131,11 +131,11 @@ async def run_protocol(self, protocol: Protocol, block_max_volume: float): async def get_hold_time(self) -> float: if not self._state.is_profile_running: - return 0.0 + raise RuntimeError("Hold time is not available. Is a profile running?") # Loop through all steps and print the full log instantly. if self._state.protocol is None: - return 0.0 + raise RuntimeError("Hold time is not available. Is a profile running?") self._state.is_profile_running = False self._state.current_step_index = self._state.total_steps - 1 diff --git a/pylabrobot/thermocycling/opentrons_backend.py b/pylabrobot/thermocycling/opentrons_backend.py index f2b444c786d..039708f348f 100644 --- a/pylabrobot/thermocycling/opentrons_backend.py +++ b/pylabrobot/thermocycling/opentrons_backend.py @@ -162,7 +162,10 @@ async def get_block_status(self) -> BlockStatus: return BlockStatus.IDLE async def get_hold_time(self) -> float: - return cast(float, self._find_module().get("holdTime", 0.0)) + hold_time = self._find_module().get("holdTime") + if hold_time is None: + raise RuntimeError("Hold time is not available. Is a profile running?") + return cast(float, hold_time) async def get_current_cycle_index(self) -> int: """Get the zero-based index of the current cycle from the Opentrons API.""" diff --git a/pylabrobot/thermocycling/opentrons_backend_tests.py b/pylabrobot/thermocycling/opentrons_backend_tests.py index b883b99b1f4..c68460a991e 100644 --- a/pylabrobot/thermocycling/opentrons_backend_tests.py +++ b/pylabrobot/thermocycling/opentrons_backend_tests.py @@ -109,3 +109,11 @@ async def test_getters_return_correct_data(self, mock_list_connected_modules): # assert await self.thermocycler_backend.get_total_cycle_count() == 10 assert await self.thermocycler_backend.get_current_step_index() == 0 # 1 - 1 = 0 (zero-based) assert await self.thermocycler_backend.get_total_step_count() == 3 + + @patch("pylabrobot.thermocycling.opentrons_backend.list_connected_modules") + async def test_get_hold_time_raises_if_not_running(self, mock_list_connected_modules): + mock_list_connected_modules.return_value = [{"id": "test_id", "data": {}}] + + with self.assertRaises(RuntimeError) as e: + await self.thermocycler_backend.get_hold_time() + self.assertEqual(str(e.exception), "Hold time is not available. Is a profile running?") diff --git a/pylabrobot/thermocycling/thermocycler.py b/pylabrobot/thermocycling/thermocycler.py index 622599d47a2..ad37f982c37 100644 --- a/pylabrobot/thermocycling/thermocycler.py +++ b/pylabrobot/thermocycling/thermocycler.py @@ -255,11 +255,16 @@ async def wait_for_lid(self, timeout: float = 1200, tolerance: float = 0.5, **ba async def is_profile_running(self, **backend_kwargs) -> bool: """Return True if a profile is still in progress.""" - hold = await self.get_hold_time(**backend_kwargs) - cycle = await self.get_current_cycle_index(**backend_kwargs) - total_cycles = await self.get_total_cycle_count(**backend_kwargs) - step = await self.get_current_step_index(**backend_kwargs) - total_steps = await self.get_total_step_count(**backend_kwargs) + try: + hold = await self.get_hold_time(**backend_kwargs) + cycle = await self.get_current_cycle_index(**backend_kwargs) + total_cycles = await self.get_total_cycle_count(**backend_kwargs) + step = await self.get_current_step_index(**backend_kwargs) + total_steps = await self.get_total_step_count(**backend_kwargs) + except NotImplementedError: + raise + except RuntimeError: + return False # if still holding in a step, it's running if hold and hold > 0: diff --git a/pylabrobot/thermocycling/thermocycler_tests.py b/pylabrobot/thermocycling/thermocycler_tests.py index 157d6762358..c16f91c4397 100644 --- a/pylabrobot/thermocycling/thermocycler_tests.py +++ b/pylabrobot/thermocycling/thermocycler_tests.py @@ -57,6 +57,14 @@ def test_thermocycler_serialization(self): deserialized = Thermocycler.deserialize(serialized) assert self.tc == deserialized + async def test_chatterbox_get_hold_time_raises_if_not_running(self): + """Chatterbox mirrors real backends when no profile is active.""" + self.tc.backend = ThermocyclerChatterboxBackend() + + with self.assertRaises(RuntimeError) as e: + await self.tc.get_hold_time() + self.assertEqual(str(e.exception), "Hold time is not available. Is a profile running?") + async def test_run_pcr_profile_builds_correct_profile(self): """Test that run_pcr_profile correctly builds the flat step list.""" @@ -137,3 +145,16 @@ async def test_is_profile_running_logic(self): self.tc.backend.get_total_step_count.return_value = total_steps # type: ignore print(f"Testing with hold={hold}, cycle={cycle}, total_cycles={total_cycles}, ") assert await self.tc.is_profile_running() is expected + + async def test_is_profile_running_returns_false_if_no_profile_running(self): + """RuntimeError from get_hold_time indicates that no profile is currently running.""" + self.tc.backend.get_hold_time.side_effect = RuntimeError("No profile is running") # type: ignore + + assert await self.tc.is_profile_running() is False + + async def test_is_profile_running_preserves_not_implemented_error(self): + """Unsupported backends should still surface NotImplementedError.""" + self.tc.backend.get_hold_time.side_effect = NotImplementedError # type: ignore + + with self.assertRaises(NotImplementedError): + await self.tc.is_profile_running() From cd117849b056ab0ab86ca8b4c8225d4522532f50 Mon Sep 17 00:00:00 2001 From: Abdeltoto Date: Sat, 25 Apr 2026 20:20:45 -0400 Subject: [PATCH 2/3] Propagate thermocycler runtime errors Made-with: Cursor --- pylabrobot/thermocycling/thermocycler.py | 15 +++++---------- pylabrobot/thermocycling/thermocycler_tests.py | 7 ++++--- 2 files changed, 9 insertions(+), 13 deletions(-) diff --git a/pylabrobot/thermocycling/thermocycler.py b/pylabrobot/thermocycling/thermocycler.py index ad37f982c37..622599d47a2 100644 --- a/pylabrobot/thermocycling/thermocycler.py +++ b/pylabrobot/thermocycling/thermocycler.py @@ -255,16 +255,11 @@ async def wait_for_lid(self, timeout: float = 1200, tolerance: float = 0.5, **ba async def is_profile_running(self, **backend_kwargs) -> bool: """Return True if a profile is still in progress.""" - try: - hold = await self.get_hold_time(**backend_kwargs) - cycle = await self.get_current_cycle_index(**backend_kwargs) - total_cycles = await self.get_total_cycle_count(**backend_kwargs) - step = await self.get_current_step_index(**backend_kwargs) - total_steps = await self.get_total_step_count(**backend_kwargs) - except NotImplementedError: - raise - except RuntimeError: - return False + hold = await self.get_hold_time(**backend_kwargs) + cycle = await self.get_current_cycle_index(**backend_kwargs) + total_cycles = await self.get_total_cycle_count(**backend_kwargs) + step = await self.get_current_step_index(**backend_kwargs) + total_steps = await self.get_total_step_count(**backend_kwargs) # if still holding in a step, it's running if hold and hold > 0: diff --git a/pylabrobot/thermocycling/thermocycler_tests.py b/pylabrobot/thermocycling/thermocycler_tests.py index c16f91c4397..eb0bc32c411 100644 --- a/pylabrobot/thermocycling/thermocycler_tests.py +++ b/pylabrobot/thermocycling/thermocycler_tests.py @@ -146,11 +146,12 @@ async def test_is_profile_running_logic(self): print(f"Testing with hold={hold}, cycle={cycle}, total_cycles={total_cycles}, ") assert await self.tc.is_profile_running() is expected - async def test_is_profile_running_returns_false_if_no_profile_running(self): - """RuntimeError from get_hold_time indicates that no profile is currently running.""" + async def test_is_profile_running_raises_if_no_profile_running(self): + """RuntimeError from get_hold_time should propagate.""" self.tc.backend.get_hold_time.side_effect = RuntimeError("No profile is running") # type: ignore - assert await self.tc.is_profile_running() is False + with self.assertRaises(RuntimeError): + await self.tc.is_profile_running() async def test_is_profile_running_preserves_not_implemented_error(self): """Unsupported backends should still surface NotImplementedError.""" From f02636cb156fcf50792b6c2e0efd37abcc00b560 Mon Sep 17 00:00:00 2001 From: Abdeltoto Date: Sat, 25 Apr 2026 20:39:24 -0400 Subject: [PATCH 3/3] Keep Chatterbox hold time behavior Made-with: Cursor --- pylabrobot/thermocycling/chatterbox.py | 4 ++-- pylabrobot/thermocycling/thermocycler_tests.py | 8 -------- 2 files changed, 2 insertions(+), 10 deletions(-) diff --git a/pylabrobot/thermocycling/chatterbox.py b/pylabrobot/thermocycling/chatterbox.py index 4a3a3c84a23..1c45e40752d 100644 --- a/pylabrobot/thermocycling/chatterbox.py +++ b/pylabrobot/thermocycling/chatterbox.py @@ -131,11 +131,11 @@ async def run_protocol(self, protocol: Protocol, block_max_volume: float): async def get_hold_time(self) -> float: if not self._state.is_profile_running: - raise RuntimeError("Hold time is not available. Is a profile running?") + return 0.0 # Loop through all steps and print the full log instantly. if self._state.protocol is None: - raise RuntimeError("Hold time is not available. Is a profile running?") + return 0.0 self._state.is_profile_running = False self._state.current_step_index = self._state.total_steps - 1 diff --git a/pylabrobot/thermocycling/thermocycler_tests.py b/pylabrobot/thermocycling/thermocycler_tests.py index eb0bc32c411..db0d4821c6f 100644 --- a/pylabrobot/thermocycling/thermocycler_tests.py +++ b/pylabrobot/thermocycling/thermocycler_tests.py @@ -57,14 +57,6 @@ def test_thermocycler_serialization(self): deserialized = Thermocycler.deserialize(serialized) assert self.tc == deserialized - async def test_chatterbox_get_hold_time_raises_if_not_running(self): - """Chatterbox mirrors real backends when no profile is active.""" - self.tc.backend = ThermocyclerChatterboxBackend() - - with self.assertRaises(RuntimeError) as e: - await self.tc.get_hold_time() - self.assertEqual(str(e.exception), "Hold time is not available. Is a profile running?") - async def test_run_pcr_profile_builds_correct_profile(self): """Test that run_pcr_profile correctly builds the flat step list."""