Digitally transfer only the aspiratable volumes in worklist methods

Closes #102

Author: michaelosthege

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "robotools"
-version = "1.13.0"
+version = "1.13.1"
 description = "Pythonic in-silico liquid handling and creation of Tecan FreedomEVO worklists."
 readme = "README.md"
 requires-python = ">=3.10"

robotools/evotools/test_worklist.py

@@ -598,11 +598,11 @@ def test_transfer_on_underflow(self):
         with EvoWorklist() as wl:
             wl.transfer(A, "A01", A, "A02", 600, on_underflow="debug")
             assert A.volumes[0, 0] == 100
-            assert A.volumes[0, 1] == 1100
+            assert A.volumes[0, 1] == 900  # only the aspiratable volume is transferred
             with pytest.warns(VolumeUnderflowWarning, match="500.0 - 600.0 < 100"):
                 wl.transfer(A, "B01", A, "B02", 600, on_underflow="warn")
                 assert A.volumes[1, 0] == 100
-                assert A.volumes[1, 1] == 1100
+                assert A.volumes[1, 1] == 900
             with pytest.raises(VolumeUnderflowError, match="500.0 - 600.0 < 100"):
                 wl.transfer(A, "C01", A, "C02", 600, on_underflow="raise")
 

robotools/evotools/worklist.py

@@ -331,13 +331,16 @@ def transfer(
                     if len(vs) > p:
                         v = vs[p]
                         if v > 0:
-                            self.aspirate(source, s, v, label=None, on_underflow=on_underflow, **kwargs)
+                            vasp = self.aspirate(
+                                source, s, v, label=None, on_underflow=on_underflow, **kwargs
+                            )
                             self.dispense(
                                 destination,
                                 d,
                                 v,
                                 label=None,
                                 compositions=[source.get_well_composition(s)],
+                                vtrack=vasp,
                                 **kwargs,
                             )
                             nsteps += 1

robotools/fluenttools/test_worklist.py

@@ -52,10 +52,10 @@ def test_transfer_on_underflow(self):
         with FluentWorklist() as wl:
             wl.transfer(A, "A01", A, "A02", 600, on_underflow="debug")
             assert A.volumes[0, 0] == 100
-            assert A.volumes[0, 1] == 1100
+            assert A.volumes[0, 1] == 900  # only the aspiratable volume is transferred
             with pytest.warns(VolumeUnderflowWarning, match="500.0 - 600.0 < 100"):
                 wl.transfer(A, "B01", A, "B02", 600, on_underflow="warn")
                 assert A.volumes[1, 0] == 100
-                assert A.volumes[1, 1] == 1100
+                assert A.volumes[1, 1] == 900
             with pytest.raises(VolumeUnderflowError, match="500.0 - 600.0 < 100"):
                 wl.transfer(A, "C01", A, "C02", 600, on_underflow="raise")

robotools/fluenttools/worklist.py

@@ -136,13 +136,16 @@ def transfer(
                     if len(vs) > p:
                         v = vs[p]
                         if v > 0:
-                            self.aspirate(source, s, v, label=None, on_underflow=on_underflow, **kwargs)
+                            vasp = self.aspirate(
+                                source, s, v, label=None, on_underflow=on_underflow, **kwargs
+                            )
                             self.dispense(
                                 destination,
                                 d,
                                 v,
                                 label=None,
                                 compositions=[source.get_well_composition(s)],
+                                vtrack=vasp,
                                 **kwargs,
                             )
                             nsteps += 1

robotools/worklists/base.py

@@ -501,6 +501,7 @@ def dispense(
         *,
         label: Optional[str] = None,
         compositions: Optional[List[Optional[Dict[str, float]]]] = None,
+        vtrack: Union[float, Sequence[float], numpy.ndarray] | None = None,
         **kwargs,
     ) -> None:
         """Performs dispensing into the provided labware.
@@ -512,11 +513,16 @@ def dispense(
         wells : str or iterable
             List of well ids
         volumes : float or iterable
-            Volume(s) to dispense
+            Volume(s) to dispense in the command
         label : str
             Label of the operation to log into labware history
         compositions : list
             Iterable of liquid compositions
+        vtrack
+            Volume to add to the destination in digital volume tracking.
+            Defaults to ``volume``.
+            Used by ``transfer`` commands to account for ``on_underflow`` situations
+            where more volume was aspirated than considered aspiratable based on volume tracking.
         kwargs
             Additional keyword arguments to pass to `dispense_well`.
             Most prominent example: `liquid_class`.
@@ -526,7 +532,16 @@ def dispense(
         volumes = numpy.array(volumes).flatten("F")
         if len(volumes) == 1:
             volumes = numpy.repeat(volumes, len(wells))
-        labware.add(wells, volumes, label, compositions=compositions)
+
+        # Digital volume transfer overrides may be provided
+        if vtrack is None:
+            vtrack = volumes
+        else:
+            vtrack = numpy.array(vtrack).flatten("F")
+            if len(vtrack) == 1:
+                vtrack = numpy.repeat(vtrack, len(wells))
+
+        labware.add(wells, vtrack, label, compositions=compositions)
         self.comment(label)
         for well, volume in zip(wells, volumes):
             if volume > 0: