Add on_underflow option to enable ignoring or just warning about volume underflows

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "robotools"
-version = "1.12.0"
+version = "1.13.0"
 description = "Pythonic in-silico liquid handling and creation of Tecan FreedomEVO worklists."
 readme = "README.md"
 requires-python = ">=3.10"
@@ -15,8 +15,6 @@ authors = [
 classifiers = [
     "Programming Language :: Python",
     "Operating System :: OS Independent",
-    "Programming Language :: Python :: 3.8",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
@@ -26,6 +24,7 @@ classifiers = [
 ]
 dependencies = [
     "numpy",
+    "typing_extensions",
 ]
 
 [project.urls]

robotools/__init__.py

@@ -10,7 +10,9 @@
     Trough,
     VolumeOverflowError,
     VolumeUnderflowError,
+    VolumeUnderflowWarning,
     VolumeViolationException,
+    VolumeViolationWarning,
 )
 from .transform import (
     WellRandomizer,

robotools/evotools/test_worklist.py

@@ -3,6 +3,10 @@
 
 from robotools.evotools import EvoWorklist, Labwares
 from robotools.evotools.types import Tip
+from robotools.liquidhandling.exceptions import (
+    VolumeUnderflowError,
+    VolumeUnderflowWarning,
+)
 from robotools.liquidhandling.labware import Labware, Trough
 from robotools.worklists.exceptions import InvalidOperationError
 
@@ -589,6 +593,19 @@ def test_history_condensation_within_labware(self) -> None:
         )
         return
 
+    def test_transfer_on_underflow(self):
+        A = Labware("A", 3, 2, min_volume=100, max_volume=2000, initial_volumes=500)
+        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
+            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
+            with pytest.raises(VolumeUnderflowError, match="500.0 - 600.0 < 100"):
+                wl.transfer(A, "C01", A, "C02", 600, on_underflow="raise")
+
 
 class TestTroughLabwareWorklist:
     def test_aspirate(self) -> None:
@@ -639,12 +656,14 @@ def test_evo_aspirate(self) -> None:
                 "A01",
                 labware_position=(30, 2),
                 tips=[Tip.T2],
-                volumes=20,
+                volumes=200,
                 liquid_class="PowerSuck",
+                on_underflow="debug",
             )
         assert len(wl) == 1
         assert "B;Aspirate" in wl[0]
-        assert lw.volumes[0, 0] == 30
+        # Underflow ignored, minimum volume remains
+        assert lw.volumes[0, 0] == 10
         pass
 
     def test_evo_dispense(self) -> None:

robotools/evotools/worklist.py

@@ -40,6 +40,7 @@ def evo_aspirate(
         *,
         arm: int = 0,
         label: Optional[str] = None,
+        on_underflow: Literal["debug", "warn", "raise"] = "raise",
     ) -> None:
         """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.
@@ -62,13 +63,21 @@ def evo_aspirate(
             Which LiHa to use, if more than one is available
         label : str
             Label of the operation to log into labware history
+        on_underflow
+            What to do about volume underflows (going below ``vmin``) in non-empty wells.
+
+            Options:
+
+            - ``"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.
         """
         # 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)
+        labware.remove(wells_calc, volumes_calc, label, on_underflow=on_underflow)
         self.comment(label)
         cmd = commands.evo_aspirate(
             n_rows=labware.n_rows,
@@ -221,6 +230,7 @@ def transfer(
         label: Optional[str] = None,
         wash_scheme: Literal[1, 2, 3, 4, "flush", "reuse"] = 1,
         partition_by: str = "auto",
+        on_underflow: Literal["debug", "warn", "raise"] = "raise",
         **kwargs,
     ) -> None:
         """Transfer operation between two labwares.
@@ -251,6 +261,14 @@ def transfer(
                 'auto': partitioning by source unless the source is a Trough
                 'source': partitioning by source columns
                 'destination': partitioning by destination columns
+        on_underflow
+            What to do about volume underflows (going below ``vmin``) in non-empty wells.
+
+            Options:
+
+            - ``"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.
         kwargs
             Additional keyword arguments to pass to aspirate and dispense.
             Most prominent example: `liquid_class`.
@@ -308,7 +326,7 @@ def transfer(
                     if len(vs) > p:
                         v = vs[p]
                         if v > 0:
-                            self.aspirate(source, s, v, label=None, **kwargs)
+                            self.aspirate(source, s, v, label=None, on_underflow=on_underflow, **kwargs)
                             self.dispense(
                                 destination,
                                 d,

robotools/fluenttools/test_worklist.py

@@ -1,6 +1,10 @@
 import pytest
 
 from robotools.fluenttools.worklist import FluentWorklist
+from robotools.liquidhandling.exceptions import (
+    VolumeUnderflowError,
+    VolumeUnderflowWarning,
+)
 from robotools.liquidhandling.labware import Labware
 
 
@@ -42,3 +46,16 @@ def test_transfer_flush(self):
         assert len(wl) == 3
         assert wl[-1] == "F;"
         pass
+
+    def test_transfer_on_underflow(self):
+        A = Labware("A", 3, 2, min_volume=100, max_volume=2000, initial_volumes=500)
+        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
+            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
+            with pytest.raises(VolumeUnderflowError, match="500.0 - 600.0 < 100"):
+                wl.transfer(A, "C01", A, "C02", 600, on_underflow="raise")

robotools/fluenttools/worklist.py

@@ -43,6 +43,7 @@ def transfer(
         label: Optional[str] = None,
         wash_scheme: Literal[1, 2, 3, 4, "flush", "reuse"] = 1,
         partition_by: str = "auto",
+        on_underflow: Literal["debug", "warn", "raise"] = "raise",
         **kwargs,
     ) -> None:
         """Transfer operation between two labwares.
@@ -73,6 +74,14 @@ def transfer(
                 'auto': partitioning by source unless the source is a Trough
                 'source': partitioning by source columns
                 'destination': partitioning by destination columns
+        on_underflow
+            What to do about volume underflows (going below ``vmin``) in non-empty wells.
+
+            Options:
+
+            - ``"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.
         kwargs
             Additional keyword arguments to pass to aspirate and dispense.
             Most prominent example: `liquid_class`.
@@ -127,7 +136,7 @@ def transfer(
                     if len(vs) > p:
                         v = vs[p]
                         if v > 0:
-                            self.aspirate(source, s, v, label=None, **kwargs)
+                            self.aspirate(source, s, v, label=None, on_underflow=on_underflow, **kwargs)
                             self.dispense(
                                 destination,
                                 d,

robotools/liquidhandling/__init__.py

@@ -1,6 +1,8 @@
 from robotools.liquidhandling.exceptions import (
     VolumeOverflowError,
     VolumeUnderflowError,
+    VolumeUnderflowWarning,
     VolumeViolationException,
+    VolumeViolationWarning,
 )
 from robotools.liquidhandling.labware import Labware, Trough

robotools/liquidhandling/exceptions.py

@@ -5,14 +5,20 @@
 __all__ = (
     "VolumeOverflowError",
     "VolumeUnderflowError",
+    "VolumeUnderflowWarning",
     "VolumeViolationException",
+    "VolumeViolationWarning",
 )
 
 
 class VolumeViolationException(Exception):
     """Error indicating a violation of volume constraints."""
 
 
+class VolumeViolationWarning(UserWarning):
+    """Warning indicating the possible violation of volume constratins."""
+
+
 class VolumeOverflowError(VolumeViolationException):
     """Error that indicates the planned overflow of a well."""
 
@@ -33,6 +39,10 @@ def __init__(
             super().__init__(f'Too much volume for "{labware}".{well}: {current} + {change} > {threshold}')
 
 
+class VolumeUnderflowWarning(VolumeViolationWarning):
+    """Warning indicating the possible underflow of a well."""
+
+
 class VolumeUnderflowError(VolumeViolationException):
     """Error that indicates the planned underflow of a well."""
 

robotools/liquidhandling/labware.py

@@ -1,8 +1,9 @@
 """Object-oriented, stateful labware representations."""
 
 
+import logging
 import warnings
-from typing import Dict, List, Mapping, Optional, Sequence, Tuple, Union
+from typing import Dict, List, Literal, Mapping, Optional, Sequence, Tuple, Union
 
 import numpy as np
 
@@ -14,8 +15,11 @@
 from robotools.liquidhandling.exceptions import (
     VolumeOverflowError,
     VolumeUnderflowError,
+    VolumeUnderflowWarning,
 )
 
+_log = logging.getLogger(__name__)
+
 
 class Labware:
     """Represents an array of liquid cavities."""
@@ -289,6 +293,8 @@ def remove(
         wells: Union[str, Sequence[str], np.ndarray],
         volumes: Union[float, Sequence[float], np.ndarray],
         label: Optional[str] = None,
+        *,
+        on_underflow: Literal["debug", "warn", "raise"] = "raise",
     ) -> None:
         """Removes volumes from wells.
 
@@ -300,6 +306,14 @@ def remove(
             Scalar or iterable of volumes
         label : str
             Description of the operation
+        on_underflow
+            What to do about volume underflows (going below ``vmin``) in non-empty wells.
+
+            Options:
+
+            - ``"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.
         """
         wells = np.array(wells).flatten("F")
         volumes = np.array(volumes).flatten("F")
@@ -313,9 +327,19 @@ def remove(
             v_new = v_original - volume
 
             if v_new < self.min_volume:
-                raise VolumeUnderflowError(self.name, well, v_original, volume, self.min_volume, label)
-
-            self._volumes[idx] -= volume
+                msg = f'Too little volume in "{self.name}".{well}: {v_original} - {volume} < {self.min_volume} in step {label}'
+                match on_underflow:
+                    case "debug":
+                        _log.debug(msg)
+                    case "warn":
+                        warnings.warn(msg, VolumeUnderflowWarning, stacklevel=2)
+                    case "raise" | _:
+                        raise VolumeUnderflowError(
+                            self.name, well, v_original, volume, self.min_volume, label
+                        )
+                self._volumes[idx] = self.min_volume
+            else:
+                self._volumes[idx] -= volume
         self.log(label)
         return
 

robotools/liquidhandling/test_labware.py

@@ -1,3 +1,4 @@
+import logging
 import warnings
 
 import numpy as np
@@ -6,6 +7,7 @@
 from robotools.liquidhandling.exceptions import (
     VolumeOverflowError,
     VolumeUnderflowError,
+    VolumeUnderflowWarning,
 )
 from robotools.liquidhandling.labware import Labware, Trough
 
@@ -163,14 +165,40 @@ def test_remove_valid(self) -> None:
         np.testing.assert_array_equal(plate.volumes, np.array([[150, 150, 200], [200, 200, 150]]))
         return
 
-    def test_remove_too_much(self) -> None:
+    def test_raise_on_underflow_by_default(self) -> None:
         plate = Labware("TestPlate", 4, 6, min_volume=100, max_volume=250)
         wells = ["A01", "A02", "B04"]
-        with pytest.raises(VolumeUnderflowError):
+        with pytest.raises(VolumeUnderflowError, match="Too little volume"):
             plate.remove(wells, 500)
         assert len(plate.history) == 1
+
+        # Also raise when invalid on_underflow settings are passed
+        with pytest.raises(VolumeUnderflowError, match="Too little volume"):
+            plate.remove(wells, 500, on_underflow="party")
+        return
+
+    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")
+        assert plate.volumes[0, 0] == 100
+        assert len(plate.history) == 2
         return
 
+    def test_remove_debug_underflows(self, caplog) -> None:
+        plate = Labware("Pladde", 4, 6, min_volume=100, max_volume=250, initial_volumes=150)
+        wells = ["A01", "A02", "B04"]
+        caplog.clear()
+        with caplog.at_level(logging.DEBUG):
+            plate.remove(wells, 500, on_underflow="debug")
+        assert len(caplog.records) == 3
+        assert 'volume in "Pladde".A01' in caplog.records[0].message
+        assert "150.0 - 500 < 100" in caplog.records[0].message
+        assert plate.volumes[0, 0] == 100
+        assert plate.volumes[0, 1] == 100
+        assert plate.volumes[1, 3] == 100
+
 
 class TestTroughLabware:
     def test_warns_on_api(self) -> None: