diff --git a/pylabrobot/legacy/liquid_handling/backends/chatterbox.py b/pylabrobot/legacy/liquid_handling/backends/chatterbox.py index e147601de21..e134ff39cd3 100644 --- a/pylabrobot/legacy/liquid_handling/backends/chatterbox.py +++ b/pylabrobot/legacy/liquid_handling/backends/chatterbox.py @@ -230,13 +230,16 @@ async def drop_resource(self, drop: ResourceDrop): print(f"Dropping resource: {drop}") async def request_tip_presence(self) -> List[Optional[bool]]: - """Return tip presence based on the tip tracker state. + """Return simulated sleeve-sensor tip presence from committed tracker state. + + Pending pickup/drop operations are excluded so error recovery can distinguish + intended state from the last committed (simulated physical) state. Returns: A list of length `num_channels` where each element is `True` if a tip is mounted, `False` if not, or `None` if unknown. """ - return [self.head[ch].has_tip for ch in range(self.num_channels)] + return [self.head[ch].has_committed_tip for ch in range(self.num_channels)] def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool: return True diff --git a/pylabrobot/legacy/liquid_handling/backends/chatterbox_tests.py b/pylabrobot/legacy/liquid_handling/backends/chatterbox_tests.py index efac36b4f8e..4c82e964c93 100644 --- a/pylabrobot/legacy/liquid_handling/backends/chatterbox_tests.py +++ b/pylabrobot/legacy/liquid_handling/backends/chatterbox_tests.py @@ -4,6 +4,7 @@ from pylabrobot.legacy.liquid_handling.backends.chatterbox import ( LiquidHandlerChatterboxBackend, ) +from pylabrobot.legacy.liquid_handling.errors import ChannelizedError from pylabrobot.resources import ( Coordinate, cor_96_wellplate_360uL_Fb, @@ -65,3 +66,43 @@ async def test_dispense96(self): async def test_move(self): await self.lh.move_resource(self.plate, Coordinate(0, 0, 0)) + + async def test_failed_pickup_does_not_commit_pending_tips(self): + async def fail_pickup(*args, **kwargs): + raise RuntimeError("simulated pickup failure") + + self.backend.pick_up_tips = fail_pickup # type: ignore[method-assign] + with self.assertRaises(RuntimeError): + await self.lh.pick_up_tips(self.tip_rack["A1"]) + self.assertFalse(self.lh.head[0].has_tip) + + async def test_failed_drop_does_not_commit_pending_remove(self): + await self.lh.pick_up_tips(self.tip_rack["A1"]) + self.assertTrue(self.lh.head[0].has_tip) + + async def fail_drop(*args, **kwargs): + raise RuntimeError("simulated drop failure") + + self.backend.drop_tips = fail_drop # type: ignore[method-assign] + with self.assertRaises(RuntimeError): + await self.lh.drop_tips(self.tip_rack["A1"]) + self.assertTrue(self.lh.head[0].has_tip) + + async def test_failed_pickup_presence_query_overrides_channelized_error(self): + async def fail_pickup(*args, **kwargs): + raise ChannelizedError(errors={0: Exception("channel 0 failed")}) + + self.backend.pick_up_tips = fail_pickup # type: ignore[method-assign] + with self.assertRaises(ChannelizedError): + await self.lh.pick_up_tips(self.tip_rack["A1", "B1"]) + self.assertFalse(self.lh.head[0].has_tip) + self.assertFalse(self.lh.head[1].has_tip) + + async def test_request_tip_presence_uses_committed_state(self): + tip = self.tip_rack.get_item("A1").get_tip() + self.lh.head[0].add_tip(tip, commit=False) + self.assertEqual((await self.backend.request_tip_presence())[0], False) + + self.lh.head[0].commit() + self.lh.head[0].remove_tip(commit=False) + self.assertEqual((await self.backend.request_tip_presence())[0], True) diff --git a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py index 59d37fccec7..9dff5ff7098 100644 --- a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py @@ -291,13 +291,16 @@ async def request_working_envelopes_per_arm( # # # # # # # # 1_000 uL Channel: Basic Commands # # # # # # # # async def request_tip_presence(self) -> List[Optional[bool]]: - """Return mock tip presence based on the tip tracker state. + """Return mock sleeve-sensor tip presence from committed tracker state. + + Pending pickup/drop operations are excluded so error recovery can distinguish + intended state from the last committed (simulated physical) state. Returns: A list of length `num_channels` where each element is `True` if a tip is mounted, `False` if not, or `None` if unknown. """ - return [self.head[ch].has_tip for ch in range(self.num_channels)] + return [self.head[ch].has_committed_tip for ch in range(self.num_channels)] async def request_z_pos_channel_n(self, channel: int) -> float: return 285.0 @@ -406,7 +409,7 @@ async def head96_request_z_acceleration(self) -> float: return 400.0 async def head96_request_tip_presence(self) -> int: - """Mock 96-head tip presence from the tip tracker: 1 if any channel holds a tip, else 0. + """Mock 96-head tip presence from committed tracker state: 1 if any channel holds a tip. Raises if tip tracking is disabled, since the tracker is then not updated and has no state to report. """ @@ -415,7 +418,7 @@ async def head96_request_tip_presence(self) -> int: "cannot report 96-head tip presence with tip tracking disabled in simulation; " "enable it with set_tip_tracking(True) or call with requires_tip=False" ) - return int(any(tracker.has_tip for tracker in self.head96.values())) + return int(any(tracker.has_committed_tip for tracker in self.head96.values())) # # # # # # # # Extension: iSWAP # # # # # # # # @@ -470,18 +473,21 @@ async def slow_iswap(self, wrist_velocity: int = 20_000, gripper_velocity: int = # # # # # # # # Liquid Level Detection (LLD) # # # # # # # # async def request_tip_len_on_channel(self, channel_idx: int) -> float: - """Return tip length from the tip tracker. + """Return simulated measured tip length from committed tracker state. + + Pending pickup/drop operations are excluded so the query matches the last + committed (simulated physical) tip. Args: channel_idx: Index of the pipetting channel (0-indexed). Returns: - The tip length in mm from the tip tracker. + The tip length in mm from the committed tip tracker state. Raises: - NoTipError: If no tip is present on the channel (via tip tracker). + NoTipError: If no committed tip is present on the channel. """ - tip = self.head[channel_idx].get_tip() + tip = self.head[channel_idx].get_committed_tip() return tip.total_tip_length async def position_channels_in_y_direction(self, ys, make_space=True): diff --git a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_tests.py b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_tests.py index d2f822bcd4d..82c0e277898 100644 --- a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_tests.py +++ b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_tests.py @@ -32,6 +32,7 @@ set_tip_tracking, ) from pylabrobot.resources.barcode import Barcode +from pylabrobot.resources.errors import NoTipError from pylabrobot.resources.greiner import Greiner_384_wellplate_28ul_Fb from pylabrobot.resources.hamilton import STARDeck, STARLetDeck, hamilton_96_tiprack_300uL_filter @@ -2770,3 +2771,39 @@ async def test_rejects_target_on_absent_right_arm(self): # The default single-arm STAR has no right X-arm, so a right-arm target is rejected. with self.assertRaises(ValueError): self.star._check_x_arm_reachable(400.0, "right") + + +class TestSTARChatterboxTipPresenceRecovery(unittest.IsolatedAsyncioTestCase): + """Simulated STAR tip queries must report committed tracker state, not pending.""" + + async def asyncSetUp(self): + self.backend = STARChatterboxBackend() + self.deck = STARLetDeck() + self.lh = LiquidHandler(self.backend, deck=self.deck) + self.tip_car = TIP_CAR_480_A00(name="tip carrier") + self.tip_car[1] = self.tip_rack = hamilton_96_tiprack_300uL_filter(name="tip_rack_01") + self.deck.assign_child_resource(self.tip_car, rails=1) + await self.lh.setup() + + async def asyncTearDown(self): + await self.lh.stop() + + async def test_simulated_tip_queries_use_committed_state(self): + tip = self.tip_rack.get_item("A1").get_tip() + self.lh.head[0].add_tip(tip, commit=False) + self.assertEqual((await self.backend.request_tip_presence())[0], False) + with self.assertRaises(NoTipError): + await self.backend.request_tip_len_on_channel(0) + + self.lh.head[0].commit() + self.assertEqual(await self.backend.request_tip_len_on_channel(0), tip.total_tip_length) + self.lh.head[0].remove_tip(commit=False) + self.assertEqual(await self.backend.request_tip_len_on_channel(0), tip.total_tip_length) + + set_tip_tracking(enabled=True) + try: + self.assertEqual(await self.backend.head96_request_tip_presence(), 0) + self.lh.head96[0].add_tip(tip, commit=False) + self.assertEqual(await self.backend.head96_request_tip_presence(), 0) + finally: + set_tip_tracking(enabled=False) diff --git a/pylabrobot/resources/tip_tracker.py b/pylabrobot/resources/tip_tracker.py index 1a876ce86f5..878475b86ea 100644 --- a/pylabrobot/resources/tip_tracker.py +++ b/pylabrobot/resources/tip_tracker.py @@ -56,6 +56,11 @@ def has_tip(self) -> bool: """Whether the tip tracker has a tip. Note that this includes pending operations.""" return self._pending_tip is not None + @property + def has_committed_tip(self) -> bool: + """Whether the tip tracker has a committed tip. Pending operations are not included.""" + return self._tip is not None + def get_tip(self) -> "Tip": """Get the tip. Note that does includes pending operations. @@ -67,6 +72,16 @@ def get_tip(self) -> "Tip": raise NoTipError(f"{self.thing} does not have a tip.") return self._tip + def get_committed_tip(self) -> "Tip": + """Get the committed tip. Pending operations are not included. + + Raises: + NoTipError: If the tip tracker does not have a committed tip. + """ + if self._tip is None: + raise NoTipError(f"{self.thing} does not have a tip.") + return self._tip + def disable(self) -> None: """Disable the tip tracker.""" self._is_disabled = True diff --git a/pylabrobot/resources/tip_tracker_tests.py b/pylabrobot/resources/tip_tracker_tests.py index dabd924a528..6a4622fa05d 100644 --- a/pylabrobot/resources/tip_tracker_tests.py +++ b/pylabrobot/resources/tip_tracker_tests.py @@ -43,3 +43,29 @@ def test_remove_tip(self): with self.assertRaises(NoTipError): tracker.get_tip() + + def test_has_committed_tip_ignores_pending_add(self): + tracker = TipTracker(thing="tester") + tracker.add_tip(self.tip, commit=False) + self.assertEqual(tracker.has_tip, True) + self.assertEqual(tracker.has_committed_tip, False) + + def test_has_committed_tip_ignores_pending_remove(self): + tracker = TipTracker(thing="tester") + tracker.add_tip(self.tip) + self.assertEqual(tracker.has_committed_tip, True) + tracker.remove_tip(commit=False) + self.assertEqual(tracker.has_tip, False) + self.assertEqual(tracker.has_committed_tip, True) + + def test_get_committed_tip_ignores_pending_add(self): + tracker = TipTracker(thing="tester") + tracker.add_tip(self.tip, commit=False) + with self.assertRaises(NoTipError): + tracker.get_committed_tip() + + def test_get_committed_tip_ignores_pending_remove(self): + tracker = TipTracker(thing="tester") + tracker.add_tip(self.tip) + tracker.remove_tip(commit=False) + self.assertEqual(tracker.get_committed_tip(), self.tip)