Container: accept lids via a Liddable mixin, centred on the top face

pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py

@@ -1880,22 +1880,22 @@ async def test_move_lid_across_rotated_resources(self):
     self.STAR._write_and_read_command.assert_has_calls(
       [
         _any_write_and_read_command_call(
-          "C0PPid0001xs04829xd0yj1142yd0zj2242zd0gr1th2800te2800gw4go1308gb1245gt20ga0gc0",
+          "C0PPid0001xs04829xd0yj1141yd0zj2242zd0gr1th2800te2800gw4go1308gb1245gt20ga0gc0",
         ),
         _any_write_and_read_command_call(
-          "C0PRid0002xs02318xd0yj1644yd0zj1983zd0th2800te2800gr4go1308ga0gc0",
+          "C0PRid0002xs02317xd0yj1644yd0zj1983zd0th2800te2800gr4go1308ga0gc0",
         ),
         _any_write_and_read_command_call(
-          "C0PPid0003xs02318xd0yj1644yd0zj1983zd0gr1th2800te2800gw4go0885gb0822gt20ga0gc0",
+          "C0PPid0003xs02317xd0yj1644yd0zj1983zd0gr1th2800te2800gw4go0885gb0822gt20ga0gc0",
         ),
         _any_write_and_read_command_call(
-          "C0PRid0004xs02315xd0yj3104yd0zj1983zd0th2800te2800gr3go0885ga0gc0",
+          "C0PRid0004xs02317xd0yj3104yd0zj1983zd0th2800te2800gr3go0885ga0gc0",
         ),
         _any_write_and_read_command_call(
-          "C0PPid0005xs02315xd0yj3104yd0zj1983zd0gr1th2800te2800gw4go0885gb0822gt20ga0gc0",
+          "C0PPid0005xs02317xd0yj3104yd0zj1983zd0gr1th2800te2800gw4go0885gb0822gt20ga0gc0",
         ),
         _any_write_and_read_command_call(
-          "C0PRid0006xs04829xd0yj1142yd0zj2242zd0th2800te2800gr4go0885ga0gc0",
+          "C0PRid0006xs04829xd0yj1141yd0zj2242zd0th2800te2800gr4go0885ga0gc0",
         ),
       ]
     )

pylabrobot/liquid_handling/liquid_handler.py

@@ -39,6 +39,7 @@
   Coordinate,
   Deck,
   Lid,
+  Liddable,
   Plate,
   PlateAdapter,
   PlateHolder,
@@ -91,6 +92,34 @@ class BlowOutVolumeError(Exception):
   pass
 
 
+def _lidded_ancestor(resource: Resource) -> Optional[Liddable]:
+  """Return the nearest lidded resource at or above ``resource``, walking up the parent chain.
+
+  A lid anywhere in the ancestry blocks pipetting, not just one on the direct parent: a well is
+  enclosed by its plate, but plates may in turn sit in lidded holders or nested containers. Returns
+  ``resource`` itself if it carries a lid, else the closest lidded ancestor, else ``None``.
+  """
+  current: Optional[Resource] = resource
+  while current is not None:
+    if isinstance(current, Liddable) and current.has_lid():
+      return current
+    current = current.parent
+  return None
+
+
+def _check_no_lid(resource: Resource, action: str) -> None:
+  """Raise if ``resource`` or any ancestor carries a lid. ``action`` is a verb phrase for the error."""
+  lidded = _lidded_ancestor(resource)
+  if lidded is None:
+    return
+  if lidded is resource:
+    raise ValueError(f"Cannot {action} {resource.name!r}: it has a lid. Remove the lid first.")
+  raise ValueError(
+    f"Cannot {action} {resource.name!r}: its enclosing resource {lidded.name!r} has a lid. "
+    "Remove the lid first."
+  )
+
+
 class LiquidHandler(Resource, Machine):
   """
   Front end for liquid handlers.
@@ -946,8 +975,7 @@ async def aspirate(
 
     # Checks
     for resource in resources:
-      if isinstance(resource.parent, Plate) and resource.parent.has_lid():
-        raise ValueError("Aspirating from a well with a lid is not supported.")
+      _check_no_lid(resource, "aspirate from")
 
     self._make_sure_channels_exist(use_channels)
     for n, p in [
@@ -1160,8 +1188,7 @@ async def dispense(
             raise BlowOutVolumeError("Blowout volume is larger than aspirated volume")
 
     for resource in resources:
-      if isinstance(resource.parent, Plate) and resource.parent.has_lid():
-        raise ValueError("Dispensing to plate with lid")
+      _check_no_lid(resource, "dispense to")
 
     for n, p in [
       ("resources", resources),
@@ -1732,10 +1759,10 @@ async def aspirate96(
     # Convert Plate to either one Container (single well) or a list of Wells
     containers: Sequence[Container]
     if isinstance(resource, Plate):
-      if resource.has_lid():
-        raise ValueError("Aspirating from plate with lid")
+      _check_no_lid(resource, "aspirate from")
       containers = resource.get_all_items() if resource.num_items > 1 else [resource.get_item(0)]
     elif isinstance(resource, Container):
+      _check_no_lid(resource, "aspirate from")
       containers = [resource]
     elif isinstance(resource, list) and all(isinstance(w, Well) for w in resource):
       containers = resource
@@ -1882,10 +1909,10 @@ async def dispense96(
     # Convert Plate to either one Container (single well) or a list of Wells
     containers: Sequence[Container]
     if isinstance(resource, Plate):
-      if resource.has_lid():
-        raise ValueError("Dispensing to plate with lid is not possible. Remove the lid first.")
+      _check_no_lid(resource, "dispense to")
       containers = resource.get_all_items() if resource.num_items > 1 else [resource.get_item(0)]
     elif isinstance(resource, Container):
+      _check_no_lid(resource, "dispense to")
       containers = [resource]
     elif isinstance(resource, list) and all(isinstance(w, Well) for w in resource):
       containers = resource
@@ -2191,6 +2218,9 @@ async def drop_resource(
       ).rotated(destination.get_absolute_rotation())
     elif isinstance(destination, Coordinate):
       to_location = destination
+    elif isinstance(destination, Trash):
+      # discarded, never seated - kept above the branches that place a resource on its destination
+      to_location = destination.get_location_wrt(self.deck)
     elif isinstance(destination, ResourceHolder):
       if destination.resource is not None and destination.resource is not resource:
         raise RuntimeError("Destination already has a plate")
@@ -2206,13 +2236,13 @@ async def drop_resource(
         resource.rotated(z=resource_rotation_wrt_destination_wrt_local)
       ).rotated(destination.get_absolute_rotation())
       to_location = destination.get_location_wrt(self.deck) + adjusted_plate_anchor
-    elif isinstance(destination, Plate) and isinstance(resource, Lid):
+    elif isinstance(destination, Liddable) and isinstance(resource, Lid):
       lid = resource
-      plate_location = destination.get_location_wrt(self.deck)
+      parent_location = destination.get_location_wrt(self.deck)
       child_wrt_parent = destination.get_lid_location(
         lid.rotated(z=resource_rotation_wrt_destination_wrt_local)
       ).rotated(destination.get_absolute_rotation())
-      to_location = plate_location + child_wrt_parent
+      to_location = parent_location + child_wrt_parent
     else:
       to_location = destination.get_location_wrt(self.deck)
 
@@ -2242,6 +2272,9 @@ async def drop_resource(
     if isinstance(destination, Coordinate):
       to_location -= self.deck.location  # passed as an absolute location, but stored as relative
       self.deck.assign_child_resource(resource, location=to_location)
+    elif isinstance(destination, Trash):
+      # discarded: `resource.unassign()` above already detached it, so leave it detached
+      pass
     elif isinstance(destination, PlateHolder):  # .zero() resources
       destination.assign_child_resource(resource)
     elif isinstance(destination, ResourceHolder):  # .zero() resources
@@ -2258,10 +2291,8 @@ async def drop_resource(
       destination.assign_child_resource(
         resource, location=destination.compute_plate_location(resource)
       )
-    elif isinstance(destination, Plate) and isinstance(resource, Lid):
+    elif isinstance(destination, Liddable) and isinstance(resource, Lid):
       destination.assign_child_resource(resource)
-    elif isinstance(destination, Trash):
-      pass  # don't assign to trash, resource will simply be unassigned
     else:
       destination.assign_child_resource(resource, location=to_location)
 
@@ -2348,7 +2379,7 @@ async def move_resource(
   async def move_lid(
     self,
     lid: Lid,
-    to: Union[Plate, ResourceStack, Coordinate],
+    to: Union[Liddable, ResourceStack, Coordinate],
     intermediate_locations: Optional[List[Coordinate]] = None,
     pickup_offset: Coordinate = Coordinate.zero(),
     destination_offset: Coordinate = Coordinate.zero(),

pylabrobot/liquid_handling/liquid_handler_tests.py

@@ -24,11 +24,14 @@
   Coordinate,
   Deck,
   Lid,
+  PetriDish,
   Plate,
+  Resource,
   ResourceNotFoundError,
   ResourceStack,
   TipRack,
   cor_96_wellplate_360uL_Fb,
+  hamilton_1_trough_200mL_Vb,
   nest_1_troughplate_195000uL_Vb,
   no_tip_tracking,
   set_tip_tracking,
@@ -310,6 +313,27 @@ async def test_move_lid(self):
       == lid.get_absolute_location().z
     )
 
+  async def test_move_lid_to_trash(self):
+    # a lid moved to the trash is discarded, not seated on it
+    plate = Plate("plate", size_x=100, size_y=100, size_z=15, ordered_items={})
+    self.deck.assign_child_resource(plate, location=Coordinate(0, 0, 100))
+    lid = Lid(name="lid", size_x=100, size_y=100, size_z=10, nesting_z_height=10)
+    plate.assign_child_resource(lid)
+    trash = self.deck.get_trash_area()
+
+    with unittest.mock.patch.object(
+      self.lh.backend, "drop_resource", wraps=self.lh.backend.drop_resource
+    ) as mock_drop:
+      await self.lh.move_lid(lid, trash)
+
+    self.assertEqual(
+      mock_drop.call_args.kwargs["drop"].destination,
+      trash.get_location_wrt(self.deck),
+    )
+    self.assertFalse(plate.has_lid())
+    self.assertIsNone(lid.parent)
+    self.assertNotIn(lid, trash.children)
+
   async def test_move_plate_onto_resource_stack_with_lid(self):
     plate = Plate("plate", size_x=100, size_y=100, size_z=15, ordered_items={})
     lid = Lid(
@@ -915,6 +939,57 @@ async def test_aspirate_with_lid(self):
     with self.assertRaises(ValueError):
       await self.lh.aspirate([well], vols=[10])
 
+  async def test_aspirate_from_trough_with_lid(self):
+    trough = hamilton_1_trough_200mL_Vb(name="trough")
+    trough.lid = Lid(
+      "trough_lid",
+      size_x=trough.get_size_x() + 2,
+      size_y=trough.get_size_y() + 2,
+      size_z=10,
+      nesting_z_height=4,
+    )
+    t = self.tip_rack.get_item("A1").get_tip()
+    self.lh.update_head_state({0: t})
+    with self.assertRaises(ValueError):
+      await self.lh.aspirate([trough], vols=[10])
+
+  async def test_aspirate_from_petri_dish_with_lid(self):
+    dish = PetriDish("petri_dish", diameter=90, height=15)
+    dish.lid = Lid(
+      "petri_dish_lid",
+      size_x=dish.get_size_x() + 4,
+      size_y=dish.get_size_y() + 4,
+      size_z=10,
+      nesting_z_height=4,
+    )
+    t = self.tip_rack.get_item("A1").get_tip()
+    self.lh.update_head_state({0: t})
+    with self.assertRaises(ValueError):
+      await self.lh.aspirate([dish], vols=[10])
+
+  async def test_lidded_ancestor_walks_full_chain(self):
+    # A lid on any ancestor blocks pipetting, not just the direct parent: nest a target two levels
+    # under a lidded container and check the walk finds it.
+    from pylabrobot.liquid_handling.liquid_handler import _lidded_ancestor
+
+    trough = hamilton_1_trough_200mL_Vb(name="trough")
+    trough.lid = Lid(
+      "trough_lid",
+      size_x=trough.get_size_x() + 2,
+      size_y=trough.get_size_y() + 2,
+      size_z=10,
+      nesting_z_height=4,
+    )
+    middle = Resource("middle", size_x=1, size_y=1, size_z=1)
+    target = Resource("target", size_x=1, size_y=1, size_z=1)
+    trough.assign_child_resource(middle, location=Coordinate.zero())
+    middle.assign_child_resource(target, location=Coordinate.zero())
+
+    self.assertIs(_lidded_ancestor(target), trough)
+    self.assertIs(_lidded_ancestor(middle), trough)
+    trough.lid = None
+    self.assertIsNone(_lidded_ancestor(target))
+
   @pytest.mark.filterwarnings("ignore:Extra arguments to backend")
   async def test_strictness(self):
     # Create a mock backend with a custom pick_up_tips that checks arguments

pylabrobot/resources/__init__.py

@@ -28,12 +28,13 @@
 from .greiner import *
 from .hamilton import *
 from .itemized_resource import ItemizedResource
+from .lid import Lid, Liddable
 from .liquid import Liquid
 from .nest import *
 from .opentrons import *
 from .perkin_elmer import *
 from .petri_dish import PetriDish, PetriDishHolder
-from .plate import Lid, Plate
+from .plate import Plate
 from .plate_adapter import PlateAdapter
 from .porvair import *
 from .powder import Powder

pylabrobot/resources/container.py

@@ -4,12 +4,16 @@
 from pylabrobot.utils.interpolation import interpolate_1d
 
 from .coordinate import Coordinate
+from .lid import Liddable
 from .resource import Resource
 from .volume_tracker import VolumeTracker
 
 
-class Container(Resource):
-  """A container is an abstract base class for a resource that can hold liquid."""
+class Container(Liddable, Resource):
+  """A container is an abstract base class for a resource that can hold liquid.
+
+  Via the :class:`Liddable` mixin, a container (trough, tube, petri dish, well) can host a lid.
+  """
 
   def __init__(
     self,