Add container no-go zones with spread-aware channel distribution

docs/user_guide/00_liquid-handling/_liquid-handling.rst

@@ -24,3 +24,4 @@ Examples:
    moving-channels-around
    tutorial_tip_inventory_consolidation
    mixing
+   container_no_go_zones

pylabrobot/liquid_handling/backends/backend.py

@@ -18,6 +18,7 @@
   SingleChannelAspiration,
   SingleChannelDispense,
 )
+from pylabrobot.liquid_handling.utils import GENERIC_LH_MIN_SPACING_BETWEEN_CHANNELS
 from pylabrobot.machines.backend import MachineBackend
 from pylabrobot.resources import Deck, Tip
 from pylabrobot.resources.tip_tracker import TipTracker
@@ -158,6 +159,19 @@ async def request_tip_presence(self) -> List[Optional[bool]]:
 
     raise NotImplementedError()
 
+  def get_channel_spacings(self, use_channels: List[int]) -> List[float]:
+    """Get the minimum spacing between each adjacent pair of channels.
+
+    Args:
+      use_channels: The channels being used, in order.
+
+    Returns:
+      List of minimum spacings (mm) between each adjacent pair. Length is
+      ``len(use_channels) - 1``. Defaults to ``GENERIC_LH_MIN_SPACING_BETWEEN_CHANNELS`` (9mm)
+      for all pairs. Backends with variable channel spacing should override this.
+    """
+    return [GENERIC_LH_MIN_SPACING_BETWEEN_CHANNELS] * max(len(use_channels) - 1, 0)
+
   @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/hamilton/STAR_backend.py

@@ -2193,10 +2193,6 @@ def _compute_channels_in_resource_locations(
           )
           min_ch = min(use_channels)
           offsets = [all_offsets[ch - min_ch] for ch in use_channels]
-
-          if num_channels_in_span % 2 != 0:
-            y_offset = 5.5
-            offsets = [offset + Coordinate(0, y_offset, 0) for offset in offsets]
         # else: container too small to fit all channels — fall back to center offsets.
         # Y sub-batching will serialize channels that can't coexist.
 
@@ -2307,8 +2303,8 @@ async def probe_liquid_heights(
     Notes:
       - All specified channels must have tips attached
       - Containers at different X positions are probed in sequential groups (single X carriage)
-      - For single containers with odd channel counts, Y-offsets are applied to avoid
-        center dividers (Hamilton 1000 uL spacing: 9mm, offset: 5.5mm)
+      - For single containers with no-go zones, Y-offsets are computed to avoid
+        obstructed regions (e.g. center dividers in troughs)
     """
 
     if move_to_z_safety_after is not None:
@@ -4524,6 +4520,12 @@ async def move_channel_z_relative(self, channel: int, distance: float):
     current_z = await self.request_z_pos_channel_n(channel)
     await self.move_channel_z(channel, current_z + distance)
 
+  def get_channel_spacings(self, use_channels: List[int]) -> List[float]:
+    sorted_channels = sorted(use_channels)
+    return [
+      self._min_spacing_between(lo, hi) for lo, hi in zip(sorted_channels[:-1], sorted_channels[1:])
+    ]
+
   def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool:
     if not isinstance(tip, HamiltonTip):
       return False

pylabrobot/liquid_handling/liquid_handler.py

@@ -30,8 +30,7 @@
   get_strictness,
 )
 from pylabrobot.liquid_handling.utils import (
-  get_tight_single_resource_liquid_op_offsets,
-  get_wide_single_resource_liquid_op_offsets,
+  compute_channel_offsets,
 )
 from pylabrobot.machines.machine import Machine, need_setup_finished
 from pylabrobot.plate_reading import PlateReader
@@ -348,6 +347,20 @@ def _check_args(
 
     return extra
 
+  def _compute_spread_offsets(
+    self,
+    resource: Resource,
+    use_channels: List[int],
+    spread: str,
+  ) -> List[Coordinate]:
+    """Compute channel spread offsets for a single-resource multi-channel operation."""
+    return compute_channel_offsets(
+      resource=resource,
+      num_channels=len(use_channels),
+      spread=spread,
+      channel_spacings=self.backend.get_channel_spacings(use_channels),
+    )
+
   def _make_sure_channels_exist(self, channels: List[int]):
     """Checks that the channels exist."""
     invalid_channels = [c for c in channels if c not in self.head]
@@ -768,10 +781,7 @@ async def discard_tips(
       raise RuntimeError("No tips have been picked up and no channels were specified.")
 
     trash = self.deck.get_trash_area()
-    trash_offsets = get_tight_single_resource_liquid_op_offsets(
-      trash,
-      num_channels=n,
-    )
+    trash_offsets = compute_channel_offsets(trash, num_channels=n, spread="tight")
     # add trash_offsets to offsets if defined, otherwise use trash_offsets
     # too advanced for mypy
     offsets = [
@@ -947,18 +957,8 @@ async def aspirate(
     if len(set(resources)) == 1:
       resource = resources[0]
       resources = [resource] * len(use_channels)
-      if spread == "tight":
-        center_offsets = get_tight_single_resource_liquid_op_offsets(
-          resource=resource, num_channels=len(use_channels)
-        )
-      elif spread == "wide":
-        center_offsets = get_wide_single_resource_liquid_op_offsets(
-          resource=resource, num_channels=len(use_channels)
-        )
-      elif spread == "custom":
-        center_offsets = [Coordinate.zero()] * len(use_channels)
-      else:
-        raise ValueError("Invalid value for 'spread'. Must be 'tight', 'wide', or 'custom'.")
+
+      center_offsets = self._compute_spread_offsets(resource, use_channels, spread)
 
       # add user defined offsets to the computed centers
       offsets = [c + o for c, o in zip(center_offsets, offsets)]
@@ -1130,18 +1130,8 @@ async def dispense(
     if len(set(resources)) == 1:
       resource = resources[0]
       resources = [resource] * len(use_channels)
-      if spread == "tight":
-        center_offsets = get_tight_single_resource_liquid_op_offsets(
-          resource=resource, num_channels=len(use_channels)
-        )
-      elif spread == "wide":
-        center_offsets = get_wide_single_resource_liquid_op_offsets(
-          resource=resource, num_channels=len(use_channels)
-        )
-      elif spread == "custom":
-        center_offsets = [Coordinate.zero()] * len(use_channels)
-      else:
-        raise ValueError("Invalid value for 'spread'. Must be 'tight', 'wide', or 'custom'.")
+
+      center_offsets = self._compute_spread_offsets(resource, use_channels, spread)
 
       # add user defined offsets to the computed centers
       offsets = [c + o for c, o in zip(center_offsets, offsets)]

pylabrobot/liquid_handling/liquid_handler_tests.py

@@ -1207,3 +1207,121 @@ async def test_load_state_backward_compatible(self):
     # Old state format without head96_state or arm_state
     old_state = {"head_state": {c: self.lh.head[c].serialize() for c in self.lh.head}}
     self.lh.load_state(old_state)  # should not raise
+
+
+class TestNoGoZoneIntegration(unittest.IsolatedAsyncioTestCase):
+  """Integration tests for no-go zone channel distribution through LiquidHandler."""
+
+  async def asyncSetUp(self):
+    self.backend = _create_mock_backend(num_channels=8)
+    self.backend.get_channel_spacings.return_value = [9.0]
+    self.deck = STARLetDeck()
+    self.lh = LiquidHandler(backend=self.backend, deck=self.deck)
+
+    # A trough-like container with a center divider
+    from pylabrobot.resources.trough import Trough
+
+    self.trough = Trough(
+      name="trough",
+      size_x=19.0,
+      size_y=90.0,
+      size_z=65.0,
+      max_volume=60_000,
+      no_go_zones=[(Coordinate(0, 44, 0), Coordinate(19, 46, 65))],
+    )
+    self.deck.assign_child_resource(self.trough, location=Coordinate(100, 100, 0))
+
+    self.tip_rack = hamilton_96_tiprack_300uL_filter(name="tip_rack")
+    self.deck.assign_child_resource(self.tip_rack, location=Coordinate(0, 0, 0))
+    await self.lh.setup()
+
+  async def test_aspirate_2_channels_avoids_no_go_zone(self):
+    """2 channels on a trough with a center divider should be placed in separate compartments."""
+    t0 = self.tip_rack.get_item("A1").get_tip()
+    t1 = self.tip_rack.get_item("B1").get_tip()
+    self.lh.update_head_state({0: t0, 1: t1})
+    self.trough.tracker.set_volume(50_000)
+
+    await self.lh.aspirate([self.trough] * 2, vols=[100, 100], use_channels=[0, 1])
+
+    ops = self.backend.aspirate.call_args.kwargs["ops"]
+    y_offsets = [op.offset.y for op in ops]
+    # Both offsets should be non-zero (not centered) and in opposite compartments
+    self.assertEqual(len(y_offsets), 2)
+    # One positive (back compartment), one negative (front compartment)
+    self.assertTrue(y_offsets[0] > 0 and y_offsets[1] < 0, f"offsets: {y_offsets}")
+    # Neither should be near the divider (Y=44-46, container center=45, so offset ~0 is bad)
+    for y in y_offsets:
+      self.assertGreater(abs(y), 5, f"offset {y} too close to divider")
+
+  async def test_single_channel_respects_no_go_zone(self):
+    """Single channel on a container with no-go zones should be placed in a safe compartment."""
+    t = self.tip_rack.get_item("A1").get_tip()
+    self.lh.update_head_state({0: t})
+    self.trough.tracker.set_volume(50_000)
+
+    await self.lh.aspirate([self.trough], vols=[100], use_channels=[0])
+
+    ops = self.backend.aspirate.call_args.kwargs["ops"]
+    # Single channel should be placed in a compartment, not at container center
+    # (container center Y=45 is inside the no-go zone at Y=44-46)
+    self.assertNotAlmostEqual(ops[0].offset.y, 0.0)
+    # Should be in the back compartment center: (48+88)/2 = 68, offset = 68-45 = 23
+    self.assertAlmostEqual(ops[0].offset.y, 23.0)
+
+  async def test_dispense_uses_no_go_zones(self):
+    """Dispense should also use no-go zone logic."""
+    t0 = self.tip_rack.get_item("A1").get_tip()
+    t1 = self.tip_rack.get_item("B1").get_tip()
+    self.lh.update_head_state({0: t0, 1: t1})
+    t0.tracker.add_liquid(volume=100)
+    t1.tracker.add_liquid(volume=100)
+
+    await self.lh.dispense([self.trough] * 2, vols=[100, 100], use_channels=[0, 1])
+
+    ops = self.backend.dispense.call_args.kwargs["ops"]
+    y_offsets = [op.offset.y for op in ops]
+    self.assertTrue(y_offsets[0] > 0 and y_offsets[1] < 0, f"offsets: {y_offsets}")
+
+  async def test_no_go_zones_skipped_for_custom_spread(self):
+    """spread='custom' should bypass no-go zone logic."""
+    t0 = self.tip_rack.get_item("A1").get_tip()
+    t1 = self.tip_rack.get_item("B1").get_tip()
+    self.lh.update_head_state({0: t0, 1: t1})
+    self.trough.tracker.set_volume(50_000)
+
+    await self.lh.aspirate([self.trough] * 2, vols=[100, 100], use_channels=[0, 1], spread="custom")
+
+    ops = self.backend.aspirate.call_args.kwargs["ops"]
+    # Custom spread: offsets should be zero (user controls positioning)
+    for op in ops:
+      self.assertAlmostEqual(op.offset.y, 0.0)
+
+  async def test_no_go_zones_tight_vs_wide(self):
+    """spread='tight' should pack channels closer than spread='wide' within compartments."""
+    tips = [self.tip_rack.get_item(f"{chr(65 + i)}1").get_tip() for i in range(4)]
+    self.lh.update_head_state({i: t for i, t in enumerate(tips)})
+    self.trough.tracker.set_volume(50_000)
+    self.backend.get_channel_spacings.return_value = [9.0, 9.0, 9.0]
+
+    # wide (default): channels spread far apart within each compartment
+    await self.lh.aspirate(
+      [self.trough] * 4, vols=[100] * 4, use_channels=[0, 1, 2, 3], spread="wide"
+    )
+    wide_ops = self.backend.aspirate.call_args.kwargs["ops"]
+    wide_offsets = sorted([op.offset.y for op in wide_ops])
+
+    self.lh.update_head_state({i: t for i, t in enumerate(tips)})
+    self.trough.tracker.set_volume(50_000)
+
+    # tight: channels packed at minimum spacing within each compartment
+    await self.lh.aspirate(
+      [self.trough] * 4, vols=[100] * 4, use_channels=[0, 1, 2, 3], spread="tight"
+    )
+    tight_ops = self.backend.aspirate.call_args.kwargs["ops"]
+    tight_offsets = sorted([op.offset.y for op in tight_ops])
+
+    # within each compartment, wide channels should be further apart than tight
+    wide_gap_lower = abs(wide_offsets[1] - wide_offsets[0])
+    tight_gap_lower = abs(tight_offsets[1] - tight_offsets[0])
+    self.assertGreater(wide_gap_lower, tight_gap_lower)