From a122e30d92112c0b14a966ac4375aa30c56e41eb Mon Sep 17 00:00:00 2001 From: VivienP Date: Mon, 17 Aug 2026 16:52:17 +0200 Subject: [PATCH] fix(liquid-handling): report committed tip state from chatterbox sensors request_tip_presence used pending has_tip, so recovery committed failed pickups and drops. Add TipTracker.has_committed_tip and query that instead. Co-authored-by: Cursor --- .../liquid_handling/backends/chatterbox.py | 7 +++- .../backends/chatterbox_tests.py | 42 +++++++++++++++++++ .../backends/hamilton/STAR_chatterbox.py | 11 +++-- pylabrobot/resources/tip_tracker.py | 5 +++ pylabrobot/resources/tip_tracker_tests.py | 14 +++++++ 5 files changed, 73 insertions(+), 6 deletions(-) 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..a897bfd711c 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,44 @@ 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_multi_channel_pickup_rolls_back_all_channels(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", "B1"]) + self.assertFalse(self.lh.head[0].has_tip) + self.assertFalse(self.lh.head[1].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) diff --git a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py index 2a426177ba0..8a4393e9758 100644 --- a/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py @@ -288,13 +288,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 @@ -403,7 +406,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. """ @@ -412,7 +415,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 # # # # # # # # diff --git a/pylabrobot/resources/tip_tracker.py b/pylabrobot/resources/tip_tracker.py index 1a876ce86f5..eb0d9378dc2 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. diff --git a/pylabrobot/resources/tip_tracker_tests.py b/pylabrobot/resources/tip_tracker_tests.py index dabd924a528..9912c2fec65 100644 --- a/pylabrobot/resources/tip_tracker_tests.py +++ b/pylabrobot/resources/tip_tracker_tests.py @@ -43,3 +43,17 @@ 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)