Add tip inventory probe function (#573)

Author: rickwierenga

pylabrobot/liquid_handling/liquid_handler.py

@@ -11,6 +11,7 @@
 import warnings
 from typing import (
   Any,
+  Awaitable,
   Callable,
   Dict,
   List,
@@ -82,6 +83,12 @@
 logger = logging.getLogger("pylabrobot")
 
 
+TipPresenceProbingMethod = Callable[
+  [List[TipSpot], Optional[List[int]]],
+  Awaitable[Dict[str, bool]],
+]
+
+
 def check_contaminated(liquid_history_tip, liquid_history_well):
   """Helper function used to check if adding a liquid to the container
   would result in cross contamination"""
@@ -2411,3 +2418,49 @@ async def probe_tip_presence_via_pickup(
           print(f"Warning: drop_tips failed for cluster at x={cluster[0][0].location.x}: {e}")
 
     return {ts.name: flag for ts, flag in zip(tip_spots, presence_flags)}
+
+  async def probe_tip_inventory(
+    self,
+    tip_spots: List[TipSpot],
+    probing_fn: Optional[TipPresenceProbingMethod] = None,
+    use_channels: Optional[List[int]] = None,
+  ) -> Dict[str, bool]:
+    """Probe the presence of tips in multiple tip spots.
+
+    The provided ``probing_fn`` is used for probing batches of tip spots. The
+    default uses :meth:`probe_tip_presence_via_pickup`.
+
+    Examples:
+      Probe all tip spots in one or more tip racks.
+
+      >>> import pylabrobot.resources.functional as F
+      >>> spots = F.get_all_tip_spots([tip_rack_1, tip_rack_2])
+      >>> presence = await lh.probe_tip_inventory(spots)
+
+    Args:
+      tip_spots:
+        Tip spots to probe for presence of a tip.
+      probing_fn:
+        Function used to probe a batch of tip spots. Must accept ``tip_spots`` and
+        ``use_channels`` and return a mapping of tip spot names to boolean flags.
+
+    Returns:
+      Mapping from tip spot names to whether a tip is present.
+    """
+
+    if probing_fn is None:
+      probing_fn = self.probe_tip_presence_via_pickup
+
+    results: Dict[str, bool] = {}
+
+    num_channels = self.backend.num_channels
+    if use_channels is None:
+      use_channels = list(range(num_channels))
+
+    for i in range(0, len(tip_spots), num_channels):
+      subset = tip_spots[i : i + num_channels]
+      use_channels = list(range(len(subset)))
+      batch_result = await probing_fn(subset, use_channels)
+      results.update(batch_result)
+
+    return results