feat: add dispensing module with Formulatrix Mantis backend

_typos.toml

@@ -28,6 +28,7 @@ mis = "mis"
 RHE = "RHE"
 "ASEND" = "ASEND"
 caf = "caf"
+Occured = "Occured"
 
 [files]
 extend-exclude = [

pylabrobot/dispensing/__init__.py

@@ -0,0 +1,3 @@
+from .backend import DispenserBackend
+from .dispenser import Dispenser
+from .standard import DispenseOp

pylabrobot/dispensing/backend.py

@@ -0,0 +1,37 @@
+"""Abstract backend interface for chip-based contactless liquid dispensers."""
+
+from __future__ import annotations
+
+from abc import ABCMeta, abstractmethod
+from typing import List
+
+from pylabrobot.machines.backend import MachineBackend
+
+from .standard import DispenseOp
+
+
+class DispenserBackend(MachineBackend, metaclass=ABCMeta):
+  """Abstract class for a chip-based contactless liquid dispenser backend.
+
+  Subclasses must implement :meth:`setup`, :meth:`stop`, and :meth:`dispense`.
+  """
+
+  @abstractmethod
+  async def setup(self) -> None:
+    """Set up the dispenser (connect, home, initialize pressure, etc.)."""
+
+  @abstractmethod
+  async def stop(self) -> None:
+    """Shut down the dispenser and release all resources.
+
+    After calling this, :meth:`setup` should be callable again.
+    """
+
+  @abstractmethod
+  async def dispense(self, ops: List[DispenseOp], **backend_kwargs) -> None:
+    """Dispense liquid into the specified wells.
+
+    Args:
+      ops: A list of :class:`DispenseOp` describing each dispense target.
+      **backend_kwargs: Additional keyword arguments specific to the backend.
+    """

pylabrobot/dispensing/chatterbox.py

@@ -0,0 +1,21 @@
+"""Chatterbox backend for device-free testing of dispensers."""
+
+from typing import List
+
+from pylabrobot.dispensing.backend import DispenserBackend
+from pylabrobot.dispensing.standard import DispenseOp
+
+
+class DispenserChatterboxBackend(DispenserBackend):
+  """Chatterbox backend for device-free testing. Prints all operations."""
+
+  async def setup(self) -> None:
+    print("Setting up the dispenser.")
+
+  async def stop(self) -> None:
+    print("Stopping the dispenser.")
+
+  async def dispense(self, ops: List[DispenseOp], **backend_kwargs) -> None:
+    for op in ops:
+      chip_str = f" (chip {op.chip})" if op.chip is not None else ""
+      print(f"Dispensing {op.volume:.2f} µL into {op.resource.name}{chip_str}")

pylabrobot/dispensing/dispenser.py

@@ -0,0 +1,68 @@
+"""Front-end for chip-based contactless liquid dispensers."""
+
+from __future__ import annotations
+
+import logging
+from typing import List, Optional, Sequence, Union
+
+from pylabrobot.machines.machine import Machine, need_setup_finished
+from pylabrobot.resources import Well
+
+from .backend import DispenserBackend
+from .standard import DispenseOp
+
+logger = logging.getLogger(__name__)
+
+
+class Dispenser(Machine):
+  """Front-end for chip-based contactless liquid dispensers.
+
+  Dispensers use disposable silicon chips with microvalves and pressure-driven
+  dispensing to deliver nanoliter-to-microliter volumes into microplate wells
+  without contacting the liquid.
+
+  Example::
+
+    >>> from pylabrobot.dispensing.mantis import MantisBackend
+    >>> d = Dispenser(backend=MantisBackend(serial_number="M-000438"))
+    >>> await d.setup()
+    >>> await d.dispense(plate["A1:H12"], volume=5.0, chip=3)
+    >>> await d.stop()
+  """
+
+  def __init__(self, backend: DispenserBackend) -> None:
+    super().__init__(backend=backend)
+    self.backend: DispenserBackend = backend  # fix type for IDE
+
+  @need_setup_finished
+  async def dispense(
+    self,
+    resources: Union[Well, Sequence[Well]],
+    volume: float,
+    chip: Optional[int] = None,
+    **backend_kwargs,
+  ) -> None:
+    """Dispense liquid into target wells.
+
+    Args:
+      resources: Target well(s) to dispense into.
+      volume: Volume in µL to dispense per well.
+      chip: Chip number to use (1-6). If ``None``, the backend selects automatically.
+      **backend_kwargs: Additional keyword arguments passed to the backend.
+
+    Raises:
+      RuntimeError: If setup has not been called.
+      ValueError: If *volume* is not positive.
+    """
+    if isinstance(resources, Well):
+      resources = [resources]
+
+    if volume <= 0:
+      raise ValueError(f"Volume must be positive, got {volume}")
+
+    ops: List[DispenseOp] = [
+      DispenseOp(resource=well, volume=volume, chip=chip) for well in resources
+    ]
+
+    logger.info("Dispensing %.2f µL into %d well(s)", volume, len(ops))
+    await self.backend.dispense(ops, **backend_kwargs)

pylabrobot/dispensing/dispenser_tests.py

@@ -0,0 +1,79 @@
+"""Tests for the Dispenser front-end."""
+
+import unittest
+from unittest.mock import AsyncMock
+
+from pylabrobot.dispensing.backend import DispenserBackend
+from pylabrobot.dispensing.dispenser import Dispenser
+from pylabrobot.resources import Cor_96_wellplate_360ul_Fb
+
+
+class MockDispenserBackend(DispenserBackend):
+  """Mock backend for testing."""
+
+  async def setup(self) -> None:
+    pass
+
+  async def stop(self) -> None:
+    pass
+
+  async def dispense(self, ops, **backend_kwargs) -> None:
+    pass
+
+
+class TestDispenser(unittest.IsolatedAsyncioTestCase):
+  async def asyncSetUp(self) -> None:
+    self.backend = AsyncMock(spec=MockDispenserBackend)
+    self.dispenser = Dispenser(backend=self.backend)
+    await self.dispenser.setup()
+
+  async def asyncTearDown(self) -> None:
+    await self.dispenser.stop()
+
+  async def test_dispense_single_well(self):
+    plate = Cor_96_wellplate_360ul_Fb(name="test_plate")
+    await self.dispenser.dispense(plate["A1"][0], volume=5.0, chip=3)
+    self.backend.dispense.assert_called_once()
+    ops = self.backend.dispense.call_args[0][0]
+    self.assertEqual(len(ops), 1)
+    self.assertEqual(ops[0].volume, 5.0)
+    self.assertEqual(ops[0].chip, 3)
+
+  async def test_dispense_multiple_wells(self):
+    plate = Cor_96_wellplate_360ul_Fb(name="test_plate")
+    wells = plate["A1"] + plate["A2"] + plate["B1"]
+    await self.dispenser.dispense(wells, volume=10.0)
+    self.backend.dispense.assert_called_once()
+    ops = self.backend.dispense.call_args[0][0]
+    self.assertEqual(len(ops), 3)
+    for op in ops:
+      self.assertEqual(op.volume, 10.0)
+      self.assertIsNone(op.chip)
+
+  async def test_dispense_negative_volume_raises(self):
+    plate = Cor_96_wellplate_360ul_Fb(name="test_plate")
+    with self.assertRaises(ValueError):
+      await self.dispenser.dispense(plate["A1"][0], volume=-1.0)
+
+  async def test_dispense_zero_volume_raises(self):
+    plate = Cor_96_wellplate_360ul_Fb(name="test_plate")
+    with self.assertRaises(ValueError):
+      await self.dispenser.dispense(plate["A1"][0], volume=0.0)
+
+  async def test_dispense_before_setup_raises(self):
+    backend = AsyncMock(spec=MockDispenserBackend)
+    dispenser = Dispenser(backend=backend)
+    plate = Cor_96_wellplate_360ul_Fb(name="test_plate")
+    with self.assertRaises(RuntimeError):
+      await dispenser.dispense(plate["A1"][0], volume=5.0)
+
+  async def test_backend_kwargs_forwarded(self):
+    plate = Cor_96_wellplate_360ul_Fb(name="test_plate")
+    await self.dispenser.dispense(plate["A1"][0], volume=5.0, custom_param="value")
+    self.backend.dispense.assert_called_once()
+    kwargs = self.backend.dispense.call_args[1]
+    self.assertEqual(kwargs["custom_param"], "value")
+
+
+if __name__ == "__main__":
+  unittest.main()

pylabrobot/dispensing/mantis/__init__.py

@@ -0,0 +1 @@
+from .mantis_backend import MantisBackend