Implement on_underflow in worklist transfer methods

Author: michaelosthege

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:

robotools/evotools/worklist.py

@@ -230,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.
@@ -260,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`.
@@ -317,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,