STARBackend.{aspirate,disense} auto_surface_following_distance parameter

pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py

@@ -1723,6 +1723,7 @@ async def aspirate(
     liquid_surfaces_no_lld: Optional[List[float]] = None,
     # PLR:
     probe_liquid_height: bool = False,
+    auto_surface_following_distance: bool = False,
     # remove >2026-01
     mix_volume: Optional[List[float]] = None,
     mix_cycles: Optional[List[int]] = None,
@@ -1780,6 +1781,7 @@ async def aspirate(
       liquid_surface_no_lld: Liquid surface at function without LLD [mm]. Must be between 0 and 360. Defaults to well bottom + liquid height. Should use absolute z.
 
       probe_liquid_height: PLR-specific parameter. If True, probe the liquid height using cLLD before aspirating to set the liquid_height of every operation instead of using the default 0. Liquid heights must not be set when using this function.
+      auto_surface_following_distance: automatically compute the surface following distance based on the container height<->volume functions. Requires liquid height to be specified or `probe_liquid_height=True`.
     """
 
     # # # TODO: delete > 2026-01 # # #
@@ -1867,7 +1869,6 @@ async def aspirate(
     immersion_depth = [
       im * (-1 if immersion_depth_direction[i] else 1) for i, im in enumerate(immersion_depth)
     ]
-    surface_following_distance = _fill_in_defaults(surface_following_distance, [0.0] * n)
     flow_rates = [
       op.flow_rate or (hlc.aspiration_flow_rate if hlc is not None else 100.0)
       for op, hlc in zip(ops, hamilton_liquid_classes)
@@ -1953,6 +1954,48 @@ async def aspirate(
       wb + lh for wb, lh in zip(well_bottoms, liquid_heights)
     ]
 
+    if auto_surface_following_distance:
+      if any(op.liquid_height is None for op in ops) and not probe_liquid_height:
+        raise ValueError(
+          "To use auto_surface_following_distance all liquid heights must be set or probe_liquid_height must be True."
+        )
+
+      if any(not op.resource.supports_compute_height_volume_functions() for op in ops):
+        raise ValueError(
+          "automatic_surface_following can only be used with containers that support height<->volume functions."
+        )
+
+      current_volumes = [
+        op.resource.compute_volume_from_height(liquid_heights[i]) for i, op in enumerate(ops)
+      ]
+
+      # compute new liquid_height after aspiration
+      liquid_height_after_aspiration = [
+        op.resource.compute_height_from_volume(current_volumes[i] - op.volume)
+        for i, op in enumerate(ops)
+      ]
+
+      # compute new surface_following_distance
+      surface_following_distance = [
+        liquid_heights[i] - liquid_height_after_aspiration[i]
+        for i in range(len(liquid_height_after_aspiration))
+      ]
+    else:
+      surface_following_distance = _fill_in_defaults(surface_following_distance, [0.0] * n)
+
+    # check if the surface_following_distance would fall below the minimum height
+    if any(
+      ops[i].resource.get_absolute_location(z="cavity_bottom").z
+      + liquid_heights[i]
+      - surface_following_distance[i]
+      < minimum_height[i]
+      for i in range(n)
+    ):
+      raise ValueError(
+        f"automatic_surface_following would result in a surface_following_distance that goes below the minimum_height. "
+        f"Well bottom: {well_bottoms[i]}, surface_following_distance: {surface_following_distance[i]}, minimum_height: {minimum_height[i]}"
+      )
+
     try:
       return await self.aspirate_pip(
         aspiration_type=[0 for _ in range(n)],
@@ -2052,6 +2095,7 @@ async def dispense(
     empty: Optional[List[bool]] = None,  # truly "empty", does not exist in liquid editor, dm4
     # PLR specific
     probe_liquid_height: bool = False,
+    auto_surface_following_distance: bool = False,
     # remove  in the future
     immersion_depth_direction: Optional[List[int]] = None,
     mix_volume: Optional[List[float]] = None,
@@ -2108,6 +2152,7 @@ async def dispense(
         documentation. Dispense mode 4.
 
       probe_liquid_height: PLR-specific parameter. If True, probe the liquid height using cLLD before aspirating to set the liquid_height of every operation instead of using the default 0. Liquid heights must not be set when using this function.
+      auto_surface_following_distance: automatically compute the surface following distance based on the container height<->volume functions. Requires liquid height to be specified or `probe_liquid_height=True`.
     """
 
     n = len(ops)
@@ -2203,7 +2248,6 @@ async def dispense(
     immersion_depth = [
       im * (-1 if immersion_depth_direction[i] else 1) for i, im in enumerate(immersion_depth)
     ]
-    surface_following_distance = _fill_in_defaults(surface_following_distance, [0.0] * n)
     flow_rates = [
       op.flow_rate or (hlc.dispense_flow_rate if hlc is not None else 120.0)
       for op, hlc in zip(ops, hamilton_liquid_classes)
@@ -2271,6 +2315,35 @@ async def dispense(
     else:
       liquid_heights = [op.liquid_height or 0 for op in ops]
 
+    if auto_surface_following_distance:
+      if any(op.liquid_height is None for op in ops) and not probe_liquid_height:
+        raise ValueError(
+          "To use auto_surface_following_distance all liquid heights must be set or probe_liquid_height must be True."
+        )
+
+      if any(not op.resource.supports_compute_height_volume_functions() for op in ops):
+        raise ValueError(
+          "automatic_surface_following can only be used with containers that support height<->volume functions."
+        )
+
+      current_volumes = [
+        op.resource.compute_volume_from_height(liquid_heights[i]) for i, op in enumerate(ops)
+      ]
+
+      # compute new liquid_height after aspiration
+      liquid_height_after_aspiration = [
+        op.resource.compute_height_from_volume(current_volumes[i] + op.volume)
+        for i, op in enumerate(ops)
+      ]
+
+      # compute new surface_following_distance
+      surface_following_distance = [
+        liquid_height_after_aspiration[i] - liquid_heights[i]
+        for i in range(len(liquid_height_after_aspiration))
+      ]
+    else:
+      surface_following_distance = _fill_in_defaults(surface_following_distance, [0.0] * n)
+
     liquid_surfaces_no_lld = liquid_surface_no_lld or [
       wb + lh for wb, lh in zip(well_bottoms, liquid_heights)
     ]