Add request_tip_presence() for hardware-based error recovery

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

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

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.

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

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],

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"""
 

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

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):