STARBackend: reject aspirations whose pre-wetting or transport-air peak overfills the tip

pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py

@@ -3479,6 +3479,27 @@ async def aspirate(
         f"Well bottom: {well_bottoms}, liquid height: {liquid_heights}, surface_following_distance: {surface_following_distance}, minimum_height: {minimum_height}"
       )
 
+    # A tip fills to one of two transient peaks that never coexist: volume + pre_wetting
+    # (pre-wet over-aspiration) and volume + transport_air (air gap drawn above the liquid);
+    # blow-out air counts toward neither.
+    over_capacity = []
+    for i, op in enumerate(ops):
+      for label, extra in (
+        ("pre-wetting", pre_wetting_volume[i]),
+        ("transport-air", transport_air_volume[i]),
+      ):
+        peak = volumes[i] + extra
+        if peak > op.tip.maximal_volume:
+          over_capacity.append((i, label, peak, op.tip.maximal_volume))
+    if over_capacity:
+      raise ValueError(
+        "Aspiration would exceed tip capacity: "
+        + "; ".join(
+          f"channel {i} {label} peak {peak:.1f} uL > tip maximal volume {tip_max:.1f} uL"
+          for i, label, peak, tip_max in over_capacity
+        )
+      )
+
     try:
       return await self.aspirate_pip(
         aspiration_type=[0 for _ in range(n)],

pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py

@@ -988,6 +988,41 @@ async def test_single_channel_aspiration(self):
       ]
     )
 
+  async def test_aspirate_rejects_over_tip_capacity(self):
+    self.lh.update_head_state({0: self.tip_rack.get_tip("A1")})  # 300 uL filter tip, max 360
+    assert self.plate.lid is not None
+    self.plate.lid.unassign()
+    well = self.plate.get_item("A1")
+    # pre-wetting peak (volume + pre_wetting) exceeds the tip max
+    well.tracker.set_volume(300)
+    with self.assertRaises(ValueError):
+      await self.lh.aspirate([well], vols=[100], pre_wetting_volume=[300])
+    # transport-air peak (volume + transport_air) exceeds the tip max
+    well.tracker.set_volume(300)
+    with self.assertRaises(ValueError):
+      await self.lh.aspirate([well], vols=[100], transport_air_volume=[300])
+
+  async def test_aspirate_capacity_ignores_blow_out(self):
+    self.lh.update_head_state({0: self.tip_rack.get_tip("A1")})  # 300 uL filter tip, max 360
+    assert self.plate.lid is not None
+    self.plate.lid.unassign()
+    well = self.plate.get_item("A1")
+    well.tracker.set_volume(200)
+    # blow-out counts toward neither peak, so a large blow-out stays within capacity
+    await self.lh.aspirate([well], vols=[100], blow_out_air_volume=[300])
+
+  async def test_aspirate_capacity_boundary_is_exclusive(self):
+    self.lh.update_head_state({0: self.tip_rack.get_tip("A1")})  # 300 uL filter tip, max 360
+    assert self.plate.lid is not None
+    self.plate.lid.unassign()
+    well = self.plate.get_item("A1")
+    # a peak exactly at the tip's maximal volume is allowed; just above it is rejected (>)
+    well.tracker.set_volume(400)
+    await self.lh.aspirate([well], vols=[360], disable_volume_correction=[True])
+    well.tracker.set_volume(400)
+    with self.assertRaises(ValueError):
+      await self.lh.aspirate([well], vols=[361], disable_volume_correction=[True])
+
   async def test_single_channel_aspiration_liquid_height(self):
     self.lh.update_head_state({0: self.tip_rack.get_tip("A1")})
     # TODO: Hamilton liquid classes