diff --git a/pylabrobot/liquid_handling/backends/backend.py b/pylabrobot/liquid_handling/backends/backend.py index c57735806ce..dc3253986e3 100644 --- a/pylabrobot/liquid_handling/backends/backend.py +++ b/pylabrobot/liquid_handling/backends/backend.py @@ -152,6 +152,12 @@ async def move_channel_z(self, channel: int, z: float): raise NotImplementedError() + async def request_tip_presence(self) -> List[Optional[bool]]: + """Request the tip presence status for each channel. Returns a list of length `num_channels` + where each element is `True` if a tip is mounted, `False` if not, or `None` if unknown.""" + + raise NotImplementedError() + @abstractmethod def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool: """Check if the tip can be picked up by the specified channel. Does not consider diff --git a/pylabrobot/liquid_handling/backends/chatterbox.py b/pylabrobot/liquid_handling/backends/chatterbox.py index 7c1b03d4798..227803bc860 100644 --- a/pylabrobot/liquid_handling/backends/chatterbox.py +++ b/pylabrobot/liquid_handling/backends/chatterbox.py @@ -1,4 +1,4 @@ -from typing import List, Union +from typing import List, Optional, Union from pylabrobot.liquid_handling.backends.backend import ( LiquidHandlerBackend, @@ -229,5 +229,14 @@ async def move_picked_up_resource(self, move: ResourceMove): 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. + + 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)] + def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool: return True diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py index a2747c9f39c..c4d110dd13e 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py @@ -6397,19 +6397,15 @@ async def request_tip_bottom_z_position(self, channel_idx: int) -> float: # Extract z-coordinate and convert to mm return float(z_pos_query["rd"] / 10) - async def request_tip_presence(self) -> List[int]: - """Request query tip presence on each channel + async def request_tip_presence(self) -> List[Optional[bool]]: + """Request tip presence on each channel. Returns: - 0 = no tip, 1 = Tip in gripper (for each channel) + A list of length `num_channels` where each element is `True` if a tip is mounted, + `False` if not, or `None` if unknown. """ - warnings.warn( # TODO: remove 2026-06 - "`request_tip_presence` is deprecated and will be " - "removed in 2026-06 use `channels_sense_tip_presence` instead.", - DeprecationWarning, - stacklevel=2, - ) - return await self.channels_sense_tip_presence() + raw = await self.channels_sense_tip_presence() + return [bool(v) for v in raw] async def channels_sense_tip_presence(self) -> List[int]: """Measure tip presence on all single channels using their sleeve sensors. diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py index 7cea5c467cc..24f65570c95 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py +++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_chatterbox.py @@ -175,25 +175,22 @@ async def request_extended_configuration(self): # # # # # # # # 1_000 uL Channel: Basic Commands # # # # # # # # - async def request_tip_presence(self, mock_presence: Optional[List[int]] = None) -> List[int]: - """Check mock tip presence with optional list for user-modifiable tip presence. - (Mock MEM-READ command) - Args: - mock_presence: Optional list indicating tip presence for each channel. - 1 indicates tip present, 0 indicates no tip. - - Default: all tips present. + async def request_tip_presence(self) -> List[Optional[bool]]: + """Return mock tip presence based on the tip tracker state. Returns: - List of integers indicating tip presence for each channel. + A list of length `num_channels` where each element is `True` if a tip is mounted, + `False` if not, or `None` if unknown. """ - if mock_presence is None: - return [1 for channel_idx in range(self.num_channels)] + return [self.head[ch].has_tip for ch in range(self.num_channels)] - assert ( - len(mock_presence) == self.num_channels - ), "Length of mock_presence must match number of channels." - return mock_presence + async def channels_sense_tip_presence(self) -> List[int]: + """Return mock tip presence as integers (STAR firmware format). + + Returns: + List of integers where 0 = no tip, 1 = tip present (for each channel). + """ + return [int(self.head[ch].has_tip) for ch in range(self.num_channels)] async def request_z_pos_channel_n(self, channel: int) -> float: return 285.0 diff --git a/pylabrobot/liquid_handling/backends/hamilton/nimbus_backend.py b/pylabrobot/liquid_handling/backends/hamilton/nimbus_backend.py index c6f4d0a941d..ed6b5886e51 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/nimbus_backend.py +++ b/pylabrobot/liquid_handling/backends/hamilton/nimbus_backend.py @@ -1250,6 +1250,20 @@ async def stop(self): """Stop the backend and close connection.""" await HamiltonTCPBackend.stop(self) + async def request_tip_presence(self) -> List[Optional[bool]]: + """Request tip presence on each channel. + + Returns: + A list of length `num_channels` where each element is `True` if a tip is mounted, + `False` if not, or `None` if unknown. + """ + if self._pipette_address is None: + raise RuntimeError("Pipette address not discovered. Call setup() first.") + tip_status = await self.send_command(IsTipPresent(self._pipette_address)) + assert tip_status is not None, "IsTipPresent command returned None" + tip_present = tip_status.get("tip_present", []) + return [bool(v) for v in tip_present] + def _build_waste_position_params( self, use_channels: List[int], diff --git a/pylabrobot/liquid_handling/backends/hamilton/vantage_backend.py b/pylabrobot/liquid_handling/backends/hamilton/vantage_backend.py index 688b7735f99..61246e9cf61 100644 --- a/pylabrobot/liquid_handling/backends/hamilton/vantage_backend.py +++ b/pylabrobot/liquid_handling/backends/hamilton/vantage_backend.py @@ -3888,6 +3888,15 @@ async def query_tip_presence(self) -> List[bool]: presences_int = cast(List[int], resp["rt"]) return [bool(p) for p in presences_int] + async def request_tip_presence(self) -> List[Optional[bool]]: + """Request tip presence on each channel. + + 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 await self.query_tip_presence() + async def request_height_of_last_lld(self): """Request height of last LLD""" diff --git a/pylabrobot/liquid_handling/backends/serializing_backend.py b/pylabrobot/liquid_handling/backends/serializing_backend.py index 5766acf5534..63c83f0ce5a 100644 --- a/pylabrobot/liquid_handling/backends/serializing_backend.py +++ b/pylabrobot/liquid_handling/backends/serializing_backend.py @@ -226,5 +226,17 @@ async def move_channel_y(self, channel: int, y: float): async def move_channel_z(self, channel: int, z: float): await self.send_command(command="move_channel_z", data={"channel": channel, "z": z}) + async def request_tip_presence(self) -> List[Optional[bool]]: + """Request tip presence on each channel via the serialized command interface. + + Returns: + A list of length `num_channels` where each element is `True` if a tip is mounted, + `False` if not, or `None` if unknown. + """ + result = await self.send_command(command="request_tip_presence") + if result is not None and "tip_presence" in result: + return result["tip_presence"] + return [None] * self.num_channels + def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool: return True diff --git a/pylabrobot/liquid_handling/liquid_handler.py b/pylabrobot/liquid_handling/liquid_handler.py index 3b04ff68b77..4b8a7c44f82 100644 --- a/pylabrobot/liquid_handling/liquid_handler.py +++ b/pylabrobot/liquid_handling/liquid_handler.py @@ -494,16 +494,23 @@ async def pick_up_tips( del backend_kwargs[extra] # actually pick up the tips - error: Optional[Exception] = None + error: Optional[BaseException] = None try: await self.backend.pick_up_tips(ops=pickups, use_channels=use_channels, **backend_kwargs) - except Exception as e: + except BaseException as e: error = e # determine which channels were successful successes = [error is None] * len(pickups) - if error is not None and isinstance(error, ChannelizedError): - successes = [channel_idx not in error.errors for channel_idx in use_channels] + if error is not None: + try: + tip_presence = await self.backend.request_tip_presence() + successes = [tip_presence[ch] is True for ch in use_channels] + except Exception as tip_presence_error: + if not isinstance(tip_presence_error, NotImplementedError): + logger.warning("Failed to query tip presence after error: %s", tip_presence_error) + if isinstance(error, ChannelizedError): + successes = [channel_idx not in error.errors for channel_idx in use_channels] # commit or rollback the state trackers for channel, op, success in zip(use_channels, pickups, successes): @@ -633,16 +640,23 @@ async def drop_tips( del backend_kwargs[extra] # actually drop the tips - error: Optional[Exception] = None + error: Optional[BaseException] = None try: await self.backend.drop_tips(ops=drops, use_channels=use_channels, **backend_kwargs) - except Exception as e: + except BaseException as e: error = e # determine which channels were successful successes = [error is None] * len(drops) - if error is not None and isinstance(error, ChannelizedError): - successes = [channel_idx not in error.errors for channel_idx in use_channels] + if error is not None: + try: + tip_presence = await self.backend.request_tip_presence() + successes = [tip_presence[ch] is False for ch in use_channels] + except Exception as tip_presence_error: + if not isinstance(tip_presence_error, NotImplementedError): + logger.warning("Failed to query tip presence after error: %s", tip_presence_error) + if isinstance(error, ChannelizedError): + successes = [channel_idx not in error.errors for channel_idx in use_channels] # commit or rollback the state trackers for channel, op, success in zip(use_channels, drops, successes):