Probe y-position of resource using channel with cLLD

pylabrobot/liquid_handling/backends/hamilton/STAR.py

@@ -7222,6 +7222,7 @@ def z_drive_increment_to_mm(value_increments: int) -> float:
   async def clld_probe_z_height_using_channel(
     self,
     channel_idx: int,  # 0-based indexing of channels!
+    tip_len: Optional[float] = None,  # mm
     lowest_immers_pos: float = 99.98,  # mm
     start_pos_search: float = 330.0,  # mm
     channel_speed: float = 10.0,  # mm
@@ -7239,7 +7240,7 @@ async def clld_probe_z_height_using_channel(
     Args:
       channel_idx: The index of the channel to use for probing. Backmost channel = 0.
       lowest_immers_pos: The lowest immersion position in mm.
-      start_pos_lld_search: The start position for z-touch search in mm.
+      start_pos_lld_search: The start position for clld search in mm.
       channel_speed: The speed of channel movement in mm/sec.
       channel_acceleration: The acceleration of the channel in mm/sec**2.
       detection_edge: The edge steepness at capacitive LLD detection.
@@ -7317,6 +7318,197 @@ async def clld_probe_z_height_using_channel(
 
     return result_in_mm
 
+  async def clld_probe_y_position_using_channel(
+    self,
+    channel_idx: int,  # 0-based indexing of channels!
+    probing_direction: Literal["forward", "backward"],
+    start_pos_search: Optional[float] = None,  # mm
+    end_pos_search: Optional[float] = None,  # mm
+    channel_speed: float = 10.0,  # mm/sec
+    channel_acceleration_int: Literal[1, 2, 3, 4] = 4,  # * 5_000 steps/sec**2 == 926 mm/sec**2
+    detection_edge: int = 10,
+    current_limit_int: Literal[1, 2, 3, 4, 5, 6, 7] = 7,
+    post_detection_dist: float = 2.0,  # mm
+  ) -> float:
+    """
+    Probes the y-position at which a conductive material is detected using the channel's capacitive
+    Liquid Level Detection (cLLD) capability.
+
+    This method aims to provide safe probing within defined boundaries to avoid collisions or damage
+    to the system. It is specifically designed for conductive materials.
+
+    Args:
+      channel_idx: Index of the channel to use for probing (0-based).
+        The backmost channel is 0.
+      probing_direction: Direction to move
+        the channel during probing. "forward" increases y-position,
+        "backward" decreases y-position.
+      start_pos_search: Initial y-position for the search
+        (in mm). Defaults to the current y-position of the channel.
+      end_pos_search: Final y-position for the search (in mm).
+        Defaults to the maximum y-position the channel can move to safely.
+      channel_speed: Speed of the channel's movement (in mm/sec).
+        Defaults to 10.0 mm/sec (i.e. slow default for safety).
+      channel_acceleration_int: Acceleration level,
+        corresponding to 1-4 (* 5,000 steps/sec²). Defaults to 4.
+      detection_edge: Steepness of the edge for capacitive detection.
+        Must be between 0 and 1024. Defaults to 10.
+      current_limit_int: Current limit level,
+        from 1 to 7. Defaults to 7.
+      post_detection_dist: Distance to move away from the detected
+        material after detection (in mm). Defaults to 2.0 mm.
+
+    Returns:
+      The detected y-position of the conductive material (in mm).
+
+    Raises:
+      ValueError:
+        - If the probing direction is invalid.
+        - If the specified start or end positions are outside the safe range.
+        - If no conductive material is detected during the probing process.
+    """
+
+    assert probing_direction in [
+      "forward",
+      "backward",
+    ], f"Probing direction must be either 'forward' or 'backward', is {probing_direction}."
+
+    # Anti-channel-crash feature
+    if channel_idx > 0:
+      channel_idx_minus_one_y_pos = await self.request_y_pos_channel_n(channel_idx - 1)
+    else:
+      channel_idx_minus_one_y_pos = STAR.y_drive_increment_to_mm(13_714) + 9  # y-position=635 mm
+    if channel_idx < (self.num_channels - 1):
+      channel_idx_plus_one_y_pos = await self.request_y_pos_channel_n(channel_idx + 1)
+    else:
+      channel_idx_plus_one_y_pos = 6
+      # Insight: STAR machines appear to lose connection to a channel below y-position=6 mm
+
+    max_safe_upper_y_pos = channel_idx_minus_one_y_pos - 9
+    max_safe_lower_y_pos = channel_idx_plus_one_y_pos + 9 if channel_idx_plus_one_y_pos != 0 else 6
+
+    # Enable safe start and end positions
+    if start_pos_search:
+      assert max_safe_lower_y_pos <= start_pos_search <= max_safe_upper_y_pos, (
+        f"Start position for y search must be between \n{max_safe_lower_y_pos} and "
+        + f"{max_safe_upper_y_pos} mm, is {end_pos_search} mm. Otherwise channel will crash."
+      )
+      await self.move_channel_y(y=start_pos_search, channel=channel_idx)
+
+    if end_pos_search:
+      assert max_safe_lower_y_pos <= end_pos_search <= max_safe_upper_y_pos, (
+        f"End position for y search must be between \n{max_safe_lower_y_pos} and "
+        + f"{max_safe_upper_y_pos} mm, is {end_pos_search} mm. Otherwise channel will crash."
+      )
+
+    tip = self.head[channel_idx]
+    if not isinstance(tip, HamiltonTip):
+      raise TypeError(
+        f"Channel {channel_idx} does not have a HamiltonTip attached, "
+        f"found {type(tip)} instead."
+      )
+    if tip.tip_diameter_bottom is None:
+      raise ValueError(
+        f"Tip {tip.tip_name} on channel {channel_idx} does not have a bottom diameter set."
+      )
+    tip_bottom_diameter = tip.tip_diameter_bottom
+
+    # Set safe y-search end position based on the probing direction
+    current_channel_y_pos = await self.request_y_pos_channel_n(channel_idx)
+    if probing_direction == "forward":
+      max_y_search_pos = end_pos_search or max_safe_upper_y_pos
+      if max_y_search_pos < current_channel_y_pos:
+        raise ValueError(
+          f"Channel {channel_idx} cannot move forwards: "
+          f"End position = {max_y_search_pos} < current position = {current_channel_y_pos}"
+          f"\nDid you mean to move backwards?"
+        )
+    else:  # probing_direction == "backwards"
+      max_y_search_pos = end_pos_search or max_safe_lower_y_pos
+      if max_y_search_pos > current_channel_y_pos:
+        raise ValueError(
+          f"Channel {channel_idx} cannot move backwards: "
+          f"End position = {max_y_search_pos} > current position = {current_channel_y_pos}"
+          f"\nDid you mean to move forwards?"
+        )
+
+    # Convert mm to increments
+    max_y_search_pos_increments = STAR.mm_to_y_drive_increment(max_y_search_pos)
+    channel_speed_increments = STAR.mm_to_y_drive_increment(channel_speed)
+
+    # Machine-compatability check of calculated parameters
+    assert 0 <= max_y_search_pos_increments <= 13_714, (
+      "Maximum y search position must be between \n0 and"
+      + f"{STAR.y_drive_increment_to_mm(13_714)+9} mm, is {max_y_search_pos_increments} mm"
+    )
+    assert 20 <= channel_speed_increments <= 8_000, (
+      f"LLD search speed must be between \n{STAR.y_drive_increment_to_mm(20)}"
+      + f"and {STAR.y_drive_increment_to_mm(8_000)} mm/sec, is {channel_speed} mm/sec"
+    )
+    assert channel_acceleration_int in [1, 2, 3, 4], (
+      "Channel speed must be in [1, 2, 3, 4] (* 5_000 steps/sec**2)"
+      + f", is {channel_speed} mm/sec"
+    )
+    assert (
+      0 <= detection_edge <= 1_0234
+    ), "Edge steepness at capacitive LLD detection must be between 0 and 1023"
+    assert current_limit_int in [
+      1,
+      2,
+      3,
+      4,
+      5,
+      6,
+      7,
+    ], f"Currrent limit must be in [1, 2, 3, 4, 5, 6, 7], is {channel_speed} mm/sec"
+
+    # Move channel for cLLD (Note: does not return detected y-position!)
+    await self.send_command(
+      module=f"P{channel_idx+1}",
+      command="YL",
+      ya=f"{max_y_search_pos_increments:05}",  # Maximum search position [steps]
+      gt=f"{detection_edge:04}",  # Edge steepness at capacitive LLD detection
+      gl=f"{0:04}",  # Offset after edge detection -> always 0 to measure y-pos!
+      yv=f"{channel_speed_increments:04}",  # Max speed [steps/second]
+      yr=f"{channel_acceleration_int}",  # Acceleration ramp [yr * 5_000 steps/second**2]
+      yw=f"{current_limit_int}",  # Current limit
+    )
+
+    detected_material_y_pos = await self.request_y_pos_channel_n(channel_idx)
+
+    # Dynamically evaluate post-detection distance to avoid crashes
+    if probing_direction == "forward":
+      if channel_idx == self.num_channels - 1:  # safe default
+        adjacent_y_pos = 6.0
+      else:  # next channel
+        adjacent_y_pos = await self.request_y_pos_channel_n(channel_idx + 1)
+
+      max_safe_y_mov_dist_post_detection = detected_material_y_pos - adjacent_y_pos - 9.0
+      move_target = detected_material_y_pos - min(
+        post_detection_dist, max_safe_y_mov_dist_post_detection
+      )
+
+    else:  # probing_direction == "backwards"
+      if channel_idx == 0:  # safe default
+        adjacent_y_pos = STAR.y_drive_increment_to_mm(13_714) + 9  # y-position=635 mm
+      else:  #  previous channel
+        adjacent_y_pos = await self.request_y_pos_channel_n(channel_idx - 1)
+
+      max_safe_y_mov_dist_post_detection = adjacent_y_pos - detected_material_y_pos - 9.0
+      move_target = detected_material_y_pos + min(
+        post_detection_dist, max_safe_y_mov_dist_post_detection
+      )
+
+    await self.move_channel_y(y=move_target, channel=channel_idx)
+
+    # Correct for tip_bottom_diameter
+    if probing_direction == "forward":
+      material_y_pos = detected_material_y_pos + tip_bottom_diameter / 2
+    else:  # probing_direction == "backwards"
+      material_y_pos = detected_material_y_pos - tip_bottom_diameter / 2
+
+    return material_y_pos
+
   async def ztouch_probe_z_height_using_channel(
     self,
     channel_idx: int,  # 0-based indexing of channels!

pylabrobot/resources/hamilton/tip_creators.py

@@ -6,7 +6,7 @@
 """
 
 import enum
-from typing import Union
+from typing import Optional, Union
 
 from pylabrobot.resources.tip import Tip
 
@@ -46,7 +46,13 @@ def __init__(
     maximal_volume: float,
     tip_size: Union[TipSize, str],  # union for deserialization, will probably refactor
     pickup_method: Union[TipPickupMethod, str],  # union for deserialization, will probably refactor
+    tip_diameter_bottom: Optional[float],
   ):
+    """Create a new Hamilton tip.
+
+    Args:
+      tip_diameter_bottom: diameter of the tip at the bottom. None if not known.
+    """
     if isinstance(tip_size, str):
       tip_size = TipSize[tip_size]
     if isinstance(pickup_method, str):
@@ -72,6 +78,7 @@ def __init__(
 
     self.pickup_method = pickup_method
     self.tip_size = tip_size
+    self.tip_diameter_bottom = tip_diameter_bottom
 
   def __repr__(self) -> str:
     return (
@@ -129,6 +136,7 @@ def standard_volume_tip_no_filter() -> HamiltonTip:
     maximal_volume=400,
     tip_size=TipSize.STANDARD_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=1.2,
   )
 
 
@@ -140,6 +148,7 @@ def standard_volume_tip_with_filter() -> HamiltonTip:
     maximal_volume=360,
     tip_size=TipSize.STANDARD_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=1.2,
   )
 
 
@@ -151,6 +160,7 @@ def slim_standard_volume_tip_with_filter() -> HamiltonTip:
     maximal_volume=360,
     tip_size=TipSize.HIGH_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=None,
   )
 
 
@@ -162,6 +172,7 @@ def low_volume_tip_no_filter() -> HamiltonTip:
     maximal_volume=15,
     tip_size=TipSize.LOW_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=0.8,
   )
 
 
@@ -173,6 +184,7 @@ def low_volume_tip_with_filter() -> HamiltonTip:
     maximal_volume=10,
     tip_size=TipSize.LOW_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=0.8,
   )
 
 
@@ -184,6 +196,7 @@ def high_volume_tip_no_filter() -> HamiltonTip:
     maximal_volume=1250,
     tip_size=TipSize.HIGH_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=1.2,
   )
 
 
@@ -195,6 +208,7 @@ def high_volume_tip_with_filter() -> HamiltonTip:
     maximal_volume=1065,
     tip_size=TipSize.HIGH_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=1.2,
   )
 
 
@@ -206,6 +220,7 @@ def wide_high_volume_tip_with_filter() -> HamiltonTip:
     maximal_volume=1065,
     tip_size=TipSize.HIGH_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=12,
   )
 
 
@@ -217,6 +232,7 @@ def ultrawide_high_volume_tip_with_filter() -> HamiltonTip:
     maximal_volume=1065,
     tip_size=TipSize.HIGH_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=32,
   )
 
 
@@ -228,6 +244,7 @@ def four_ml_tip_with_filter() -> HamiltonTip:
     maximal_volume=4367,
     tip_size=TipSize.XL,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=None,
   )
 
 
@@ -239,6 +256,7 @@ def five_ml_tip_with_filter() -> HamiltonTip:
     maximal_volume=5420,
     tip_size=TipSize.XL,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=None,
   )
 
 
@@ -255,6 +273,7 @@ def five_ml_tip() -> HamiltonTip:
     maximal_volume=5420,
     tip_size=TipSize.XL,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=None,
   )
 
 
@@ -266,6 +285,7 @@ def fifty_ul_tip_with_filter() -> HamiltonTip:
     maximal_volume=60,
     tip_size=TipSize.STANDARD_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=0.7,
   )
 
 
@@ -277,4 +297,5 @@ def fifty_ul_tip_no_filter() -> HamiltonTip:
     maximal_volume=65,
     tip_size=TipSize.STANDARD_VOLUME,
     pickup_method=TipPickupMethod.OUT_OF_RACK,
+    tip_diameter_bottom=0.7,
   )