Maintain digital volume balances also when on_underflow was used

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")
 
@@ -651,7 +651,7 @@ def test_evo_aspirate(self) -> None:
         lw = Labware("A", 4, 5, min_volume=10, max_volume=100)
         lw.add("A01", 50)
         with EvoWorklist() as wl:
-            wl.evo_aspirate(
+            vasp = wl.evo_aspirate(
                 lw,
                 "A01",
                 labware_position=(30, 2),
@@ -660,6 +660,8 @@ def test_evo_aspirate(self) -> None:
                 liquid_class="PowerSuck",
                 on_underflow="debug",
             )
+            # only (50 - 10) = 40 µL are considered aspiratable
+            assert vasp == [40]
         assert len(wl) == 1
         assert "B;Aspirate" in wl[0]
         # Underflow ignored, minimum volume remains

robotools/evotools/worklist.py

@@ -41,7 +41,7 @@ def evo_aspirate(
         arm: int = 0,
         label: Optional[str] = None,
         on_underflow: Literal["debug", "warn", "raise"] = "raise",
-    ) -> None:
+    ) -> list[float]:
         """Performs aspiration from the provided labware. Is identical to the aspirate command inside the EvoWARE.
         Thus, several wells in a single column can be targeted.
 
@@ -71,13 +71,18 @@ def evo_aspirate(
             - ``"debug"`` mentions the underflowing wells in a log message at DEBUG level.
             - ``"warn"`` emits an :class:`~robotools.liquidhandling.exceptions.VolumeUnderflowWarning`. This `can be captured in unit tests <https://docs.pytest.org/en/stable/how-to/capture-warnings.html#additional-use-cases-of-warnings-in-tests>`_.
             - ``"raise"`` raises a :class:`~robotools.liquidhandling.exceptions.VolumeUnderflowError` about underflowing wells.
+
+        Returns
+        -------
+        vaspirated
+            List of aspirated volumes according to previous filling volume and minima.
         """
         # diferentiate between what is needed for volume calculation and for pipetting commands
         wells_calc = np.array(wells).flatten("F")
         volumes_calc = np.array(volumes).flatten("F")
         if len(volumes_calc) == 1:
             volumes_calc = np.repeat(volumes_calc, len(wells_calc))
-        labware.remove(wells_calc, volumes_calc, label, on_underflow=on_underflow)
+        vasp = labware.remove(wells_calc, volumes_calc, label, on_underflow=on_underflow)
         self.comment(label)
         cmd = commands.evo_aspirate(
             n_rows=labware.n_rows,
@@ -91,7 +96,7 @@ def evo_aspirate(
             max_volume=self.max_volume,
         )
         self.append(cmd)
-        return
+        return vasp
 
     def evo_dispense(
         self,
@@ -326,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/liquidhandling/labware.py

@@ -295,7 +295,7 @@ def remove(
         label: Optional[str] = None,
         *,
         on_underflow: Literal["debug", "warn", "raise"] = "raise",
-    ) -> None:
+    ) -> list[float]:
         """Removes volumes from wells.
 
         Parameters
@@ -314,13 +314,19 @@ def remove(
             - ``"debug"`` mentions the underflowing wells in a log message at DEBUG level.
             - ``"warn"`` emits an :class:`~robotools.liquidhandling.exceptions.VolumeUnderflowWarning`. This `can be captured in unit tests <https://docs.pytest.org/en/stable/how-to/capture-warnings.html#additional-use-cases-of-warnings-in-tests>`_.
             - ``"raise"`` raises a :class:`~robotools.liquidhandling.exceptions.VolumeUnderflowError` about underflowing wells.
+
+        Returns
+        -------
+        vaspirated
+            List of aspirated volumes according to previous filling volume and minima.
         """
         wells = np.array(wells).flatten("F")
         volumes = np.array(volumes).flatten("F")
         if len(volumes) == 1:
             volumes = np.repeat(volumes, len(wells))
         assert len(volumes) == len(wells), "Number of volumes must number of wells"
         assert np.all(volumes >= 0), "Volumes must be positive or zero."
+        vaspirated = []
         for well, volume in zip(wells, volumes):
             idx = self.indices[well]
             v_original = self._volumes[idx]
@@ -337,11 +343,11 @@ def remove(
                         raise VolumeUnderflowError(
                             self.name, well, v_original, volume, self.min_volume, label
                         )
-                self._volumes[idx] = self.min_volume
-            else:
-                self._volumes[idx] -= volume
+                volume = v_original - self.min_volume
+            self._volumes[idx] -= volume
+            vaspirated.append(volume)
         self.log(label)
-        return
+        return vaspirated
 
     def log(self, label: Optional[str]) -> None:
         """Logs the current volumes to the history.

robotools/liquidhandling/test_labware.py

@@ -181,9 +181,15 @@ def test_warn_on_underflow(self) -> None:
         plate = Labware("TestPlate", 4, 6, min_volume=100, max_volume=250, initial_volumes=150)
         wells = ["A01", "A02", "B04"]
         with pytest.warns(VolumeUnderflowWarning, match="150.0 - 500 < 100"):
-            plate.remove(wells, 500, on_underflow="warn")
+            vaspirated = plate.remove(wells, [500, 500, 30], on_underflow="warn")
+        # At least the min volume always remains
         assert plate.volumes[0, 0] == 100
+        assert plate.volumes[0, 1] == 100
+        assert plate.volumes[1, 3] == 120
         assert len(plate.history) == 2
+        # No more than the difference from current to min volume is aspirated
+        assert isinstance(vaspirated, list)
+        assert vaspirated == [50.0, 50.0, 30.0]
         return
 
     def test_remove_debug_underflows(self, caplog) -> None:
@@ -308,8 +314,9 @@ def test_trough_add_too_much(self) -> None:
 
     def test_trough_remove_valid(self) -> None:
         trough = Trough("TestTrough", 3, 4, min_volume=1000, max_volume=30000, initial_volumes=3000)
-        # adding into the first column (which is actually one well)
-        trough.remove(["A01", "B01"], 50)
+        # removing from the first column (which is actually one well)
+        vasp = trough.remove(["A01", "B01"], 50)
+        assert vasp == [50, 50]
         np.testing.assert_array_equal(trough.volumes, np.array([[2900, 3000, 3000, 3000]]))
         # adding to the last row (separate wells)
         trough.remove(["C01", "C02", "C03"], 50)

robotools/worklists/base.py

@@ -451,7 +451,7 @@ def aspirate(
         label: Optional[str] = None,
         on_underflow: Literal["debug", "warn", "raise"] = "raise",
         **kwargs,
-    ) -> None:
+    ) -> list[float]:
         """Performs aspiration from the provided labware.
 
         Parameters
@@ -476,17 +476,22 @@ def aspirate(
             Additional keyword arguments to pass to `aspirate_well`.
             Most prominent example: `liquid_class`.
             Take a look at `Worklist.aspirate_well` for the full list of options.
+
+        Returns
+        -------
+        vaspirated
+            List of aspirated volumes according to previous filling volume and minima.
         """
         wells = numpy.array(wells).flatten("F")
         volumes = numpy.array(volumes).flatten("F")
         if len(volumes) == 1:
             volumes = numpy.repeat(volumes, len(wells))
-        labware.remove(wells, volumes, label, on_underflow=on_underflow)
+        vaspirated = labware.remove(wells, volumes, label, on_underflow=on_underflow)
         self.comment(label)
         for well, volume in zip(wells, volumes):
             if volume > 0:
                 self.aspirate_well(labware.name, self._get_well_position(labware, well), volume, **kwargs)
-        return
+        return vaspirated
 
     def dispense(
         self,
@@ -496,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.
@@ -507,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`.
@@ -521,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:
@@ -643,6 +663,7 @@ def distribute(
         # update volume tracking
         n_dst = len(dst_wells)
         source.remove(source.wells[0, source_column], volume * n_dst, label=label)
+        # ℹ No need to capture vasp volumes because underflows are verboten in this operation.
         src_composition = source.get_well_composition(source.wells[0, source_column])
         destination.add(destination_wells, volume, label=label, compositions=[src_composition] * n_dst)
         return

robotools/worklists/test_base.py

@@ -391,7 +391,9 @@ class TestStandardLabwareWorklist:
     def test_aspirate(self, wl_cls) -> None:
         source = Labware("SourceLW", rows=3, columns=3, min_volume=10, max_volume=200, initial_volumes=200)
         with wl_cls() as wl:
-            wl.aspirate(source, ["A01", "A02", "C02"], 50, label=None)
+            vasp = wl.aspirate(source, ["A01", "A02", "C02"], 50, label=None)
+            assert isinstance(vasp, list)
+            assert vasp == [50, 50, 50]
             wl.aspirate(source, ["A03", "B03", "C03"], [10, 20, 30.5], label="second aspirate")
             assert wl == [
                 "A;SourceLW;;;1;;50.00;;;;",