feat: add default 96-head offset

pylabrobot/liquid_handling/liquid_handler.py

@@ -61,6 +61,7 @@
 from pylabrobot.resources.errors import CrossContaminationError, HasTipError
 from pylabrobot.resources.liquid import Liquid
 from pylabrobot.resources.rotation import Rotation
+from pylabrobot.serializer import deserialize, serialize
 from pylabrobot.tilting.tilter import Tilter
 
 from .backends import LiquidHandlerBackend
@@ -118,12 +119,18 @@ class LiquidHandler(Resource, Machine):
   defined in `pyhamilton.liquid_handling.backends`) to communicate with the liquid handler.
   """
 
-  def __init__(self, backend: LiquidHandlerBackend, deck: Deck):
+  def __init__(
+    self,
+    backend: LiquidHandlerBackend,
+    deck: Deck,
+    default_offset_head96: Optional[Coordinate] = None,
+  ):
     """Initialize a LiquidHandler.
 
     Args:
       backend: Backend to use.
       deck: Deck to use.
+      default_offset_head96: Base offset applied to all 96-head operations.
     """
 
     Resource.__init__(
@@ -149,6 +156,10 @@ def __init__(self, backend: LiquidHandlerBackend, deck: Deck):
 
     self._blow_out_air_volume: Optional[List[Optional[float]]] = None
 
+    # Default offset applied to all 96-head operations. Any offset passed to a 96-head method is
+    # added to this value.
+    self.default_offset_head96: Coordinate = default_offset_head96 or Coordinate.zero()
+
     # assign deck as only child resource, and set location of self to origin.
     self.location = Coordinate.zero()
     super().assign_child_resource(deck, location=deck.location or Coordinate.zero())
@@ -1339,10 +1350,13 @@ async def pick_up_tips96(
 
     Args:
       tip_rack: The tip rack to pick up tips from.
-      offset: The offset to use when picking up tips, optional.
+      offset: Additional offset to use when picking up tips. This is added to
+        :attr:`default_offset_head96`.
       backend_kwargs: Additional keyword arguments for the backend, optional.
     """
 
+    offset = self.default_offset_head96 + offset
+
     self._log_command(
       "pick_up_tips96",
       tip_rack=tip_rack,
@@ -1406,13 +1420,16 @@ async def drop_tips96(
 
     Args:
       resource: The tip rack to drop tips to.
-      offset: The offset to use when dropping tips.
+      offset: Additional offset to use when dropping tips. This is added to
+        :attr:`default_offset_head96`.
       allow_nonzero_volume: If `True`, the tip will be dropped even if its volume is not zero (there
         is liquid in the tip). If `False`, a RuntimeError will be raised if the tip has nonzero
         volume.
       backend_kwargs: Additional keyword arguments for the backend, optional.
     """
 
+    offset = self.default_offset_head96 + offset
+
     self._log_command(
       "drop_tips96",
       resource=resource,
@@ -1566,7 +1583,8 @@ async def aspirate96(
       resource (Union[Plate, Container, List[Well]]): Resource object or list of wells.
       volume (float): The volume to aspirate through each channel
       offset (Coordinate): Adjustment to where the 96 head should go to aspirate relative to where
-        the plate or container is defined to be. Defaults to Coordinate.zero().
+        the plate or container is defined to be. Added to :attr:`default_offset_head96`.
+        Defaults to :func:`Coordinate.zero`.
       flow_rate ([Optional[float]]): The flow rate to use when aspirating, in ul/s. If `None`, the
         backend default will be used.
       liquid_height ([Optional[float]]): The height of the liquid in the well wrt the bottom, in
@@ -1576,6 +1594,8 @@ async def aspirate96(
       backend_kwargs: Additional keyword arguments for the backend, optional.
     """
 
+    offset = self.default_offset_head96 + offset
+
     self._log_command(
       "aspirate96",
       resource=resource,
@@ -1684,7 +1704,7 @@ async def aspirate96(
 
     try:
       await self.backend.aspirate96(aspiration=aspiration, **backend_kwargs)
-    except Exception as error:
+    except Exception:
       for channel in self.head96.values():
         channel.get_tip().tracker.rollback()
       for container in containers:
@@ -1719,7 +1739,8 @@ async def dispense96(
       resource (Union[Plate, Container, List[Well]]): Resource object or list of wells.
       volume (float): The volume to dispense through each channel
       offset (Coordinate): Adjustment to where the 96 head should go to aspirate relative to where
-        the plate or container is defined to be. Defaults to Coordinate.zero().
+        the plate or container is defined to be. Added to :attr:`default_offset_head96`.
+        Defaults to :func:`Coordinate.zero`.
       flow_rate ([Optional[float]]): The flow rate to use when dispensing, in ul/s. If `None`, the
         backend default will be used.
       liquid_height ([Optional[float]]): The height of the liquid in the well wrt the bottom, in
@@ -1729,6 +1750,8 @@ async def dispense96(
       backend_kwargs: Additional keyword arguments for the backend, optional.
     """
 
+    offset = self.default_offset_head96 + offset
+
     self._log_command(
       "dispense96",
       resource=resource,
@@ -1829,7 +1852,7 @@ async def dispense96(
 
     try:
       await self.backend.dispense96(dispense=dispense, **backend_kwargs)
-    except Exception as error:
+    except Exception:
       for channel in self.head96.values():
         channel.get_tip().tracker.rollback()
       for container in containers:
@@ -2325,7 +2348,11 @@ async def move_plate(
     )
 
   def serialize(self):
-    return {**Resource.serialize(self), **Machine.serialize(self)}
+    return {
+      **Resource.serialize(self),
+      **Machine.serialize(self),
+      "default_offset_head96": serialize(self.default_offset_head96),
+    }
 
   @classmethod
   def deserialize(cls, data: dict, allow_marshal: bool = False) -> LiquidHandler:
@@ -2338,7 +2365,18 @@ def deserialize(cls, data: dict, allow_marshal: bool = False) -> LiquidHandler:
     deck_data = data["children"][0]
     deck = Deck.deserialize(data=deck_data, allow_marshal=allow_marshal)
     backend = LiquidHandlerBackend.deserialize(data=data["backend"])
-    return cls(deck=deck, backend=backend)
+
+    if "default_offset_head96" in data:
+      default_offset = deserialize(data["default_offset_head96"], allow_marshal=allow_marshal)
+      assert isinstance(default_offset, Coordinate)
+    else:
+      default_offset = Coordinate.zero()
+
+    return cls(
+      deck=deck,
+      backend=backend,
+      default_offset_head96=default_offset,
+    )
 
   @classmethod
   def load(cls, path: str) -> LiquidHandler:

pylabrobot/liquid_handling/liquid_handler_tests.py

@@ -42,6 +42,7 @@
   set_volume_tracking,
 )
 from pylabrobot.resources.well import Well
+from pylabrobot.serializer import serialize
 
 from . import backends
 from .liquid_handler import LiquidHandler
@@ -491,6 +492,51 @@ async def test_offsets_tips(self):
       },
     )
 
+  async def test_default_offset_head96(self):
+    self.lh.default_offset_head96 = Coordinate(1, 2, 3)
+
+    await self.lh.pick_up_tips96(self.tip_rack)
+    cmd = self.get_first_command("pick_up_tips96")
+    self.assertIsNotNone(cmd)
+    self.assertEqual(cmd["kwargs"]["pickup"].offset, Coordinate(1, 2, 3))  # type: ignore
+    self.backend.clear()
+
+    # aspirate with extra offset; effective offset should be default + provided
+    await self.lh.aspirate96(self.plate, volume=10, offset=Coordinate(1, 0, 0))
+    cmd = self.get_first_command("aspirate96")
+    self.assertIsNotNone(cmd)
+    self.assertEqual(cmd["kwargs"]["aspiration"].offset, Coordinate(2, 2, 3))  # type: ignore
+    self.backend.clear()
+
+    # dispense without providing offset uses default
+    await self.lh.dispense96(self.plate, volume=10)
+    cmd = self.get_first_command("dispense96")
+    self.assertIsNotNone(cmd)
+    self.assertEqual(cmd["kwargs"]["dispense"].offset, Coordinate(1, 2, 3))  # type: ignore
+    self.backend.clear()
+
+    await self.lh.drop_tips96(self.tip_rack, offset=Coordinate(0, 1, 0))
+    cmd = self.get_first_command("drop_tips96")
+    self.assertIsNotNone(cmd)
+    self.assertEqual(cmd["kwargs"]["drop"].offset, Coordinate(1, 3, 3))  # type: ignore
+
+  async def test_default_offset_head96_initializer(self):
+    backend = backends.SaverBackend(num_channels=8)
+    deck = STARLetDeck()
+    lh = LiquidHandler(
+      backend=backend,
+      deck=deck,
+      default_offset_head96=Coordinate(1, 2, 3),
+    )
+    self.assertEqual(lh.default_offset_head96, Coordinate(1, 2, 3))
+
+  async def test_default_offset_head96_serialization(self):
+    self.lh.default_offset_head96 = Coordinate(1, 2, 3)
+    data = self.lh.serialize()
+    self.assertEqual(data["default_offset_head96"], serialize(Coordinate(1, 2, 3)))
+    new_lh = LiquidHandler.deserialize(data)
+    self.assertEqual(new_lh.default_offset_head96, Coordinate(1, 2, 3))
+
   async def test_with_use_channels(self):
     tip_spot = self.tip_rack.get_item("A1")
     tip = tip_spot.get_tip()