From 8640aca535ea893c4c242206cc10c76c64cc3076 Mon Sep 17 00:00:00 2001 From: Michael Osthege Date: Mon, 23 Jun 2025 11:03:12 +0200 Subject: [PATCH 1/5] Improve type hints on `BaseWorklist` --- pyproject.toml | 1 + robotools/worklists/base.py | 16 +++++++++++++--- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e9ed339..d30b86e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ classifiers = [ ] dependencies = [ "numpy", + "typing_extensions", ] [project.urls] diff --git a/robotools/worklists/base.py b/robotools/worklists/base.py index 8d11d4b..87c373b 100644 --- a/robotools/worklists/base.py +++ b/robotools/worklists/base.py @@ -6,6 +6,7 @@ from typing import Dict, Iterable, List, Literal, Optional, Sequence, Union import numpy +from typing_extensions import Self from robotools import liquidhandling from robotools.evotools.types import Tip @@ -18,7 +19,7 @@ logger = logging.getLogger(__name__) -class BaseWorklist(list): +class BaseWorklist(list[str]): """Context manager for the creation of Worklists.""" def __init__( @@ -60,7 +61,7 @@ def filepath(self) -> Optional[Path]: return Path(self._filepath) return None - def __enter__(self) -> "BaseWorklist": + def __enter__(self) -> Self: self.clear() return self @@ -448,6 +449,7 @@ def aspirate( volumes: Union[float, Sequence[float], numpy.ndarray], *, label: Optional[str] = None, + on_underflow: Literal["debug", "warn", "raise"] = "raise", **kwargs, ) -> None: """Performs aspiration from the provided labware. @@ -462,6 +464,14 @@ def aspirate( Volume(s) to aspirate 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 `_. + - ``"raise"`` raises a :class:`~robotools.liquidhandling.exceptions.VolumeUnderflowError` about underflowing wells. kwargs Additional keyword arguments to pass to `aspirate_well`. Most prominent example: `liquid_class`. @@ -471,7 +481,7 @@ def aspirate( volumes = numpy.array(volumes).flatten("F") if len(volumes) == 1: volumes = numpy.repeat(volumes, len(wells)) - labware.remove(wells, volumes, label) + labware.remove(wells, volumes, label, on_underflow=on_underflow) self.comment(label) for well, volume in zip(wells, volumes): if volume > 0: From 09db50b24fbede747a2d3d12e46ff0d67790cf32 Mon Sep 17 00:00:00 2001 From: Michael Osthege Date: Mon, 23 Jun 2025 10:50:27 +0200 Subject: [PATCH 2/5] Introduce warning types --- robotools/__init__.py | 2 ++ robotools/liquidhandling/__init__.py | 2 ++ robotools/liquidhandling/exceptions.py | 10 ++++++++++ 3 files changed, 14 insertions(+) diff --git a/robotools/__init__.py b/robotools/__init__.py index d4d6e45..a94644e 100644 --- a/robotools/__init__.py +++ b/robotools/__init__.py @@ -10,7 +10,9 @@ Trough, VolumeOverflowError, VolumeUnderflowError, + VolumeUnderflowWarning, VolumeViolationException, + VolumeViolationWarning, ) from .transform import ( WellRandomizer, diff --git a/robotools/liquidhandling/__init__.py b/robotools/liquidhandling/__init__.py index b57e07c..cda82d2 100644 --- a/robotools/liquidhandling/__init__.py +++ b/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 diff --git a/robotools/liquidhandling/exceptions.py b/robotools/liquidhandling/exceptions.py index 5874f59..8dc83d1 100644 --- a/robotools/liquidhandling/exceptions.py +++ b/robotools/liquidhandling/exceptions.py @@ -5,7 +5,9 @@ __all__ = ( "VolumeOverflowError", "VolumeUnderflowError", + "VolumeUnderflowWarning", "VolumeViolationException", + "VolumeViolationWarning", ) @@ -13,6 +15,10 @@ 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.""" From 31e3889dd3ea3bbafbff271e89d193bad2e08291 Mon Sep 17 00:00:00 2001 From: Michael Osthege Date: Mon, 23 Jun 2025 10:53:25 +0200 Subject: [PATCH 3/5] Introduce `Labware.remove(..., on_underflow)` setting This makes it easier to ignore underflows that are done on purpose, for example when aspirating supernatant. --- pyproject.toml | 4 +-- robotools/liquidhandling/labware.py | 32 +++++++++++++++++++++--- robotools/liquidhandling/test_labware.py | 32 ++++++++++++++++++++++-- 3 files changed, 59 insertions(+), 9 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d30b86e..a2da2ef 100644 --- a/pyproject.toml +++ b/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", diff --git a/robotools/liquidhandling/labware.py b/robotools/liquidhandling/labware.py index de6bdcd..d6384f9 100644 --- a/robotools/liquidhandling/labware.py +++ b/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 `_. + - ``"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 diff --git a/robotools/liquidhandling/test_labware.py b/robotools/liquidhandling/test_labware.py index ab6ab60..4549966 100644 --- a/robotools/liquidhandling/test_labware.py +++ b/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: From e2757a3f31c58ee785d0a9e3ca668bd98e0d65a9 Mon Sep 17 00:00:00 2001 From: Michael Osthege Date: Mon, 23 Jun 2025 11:04:46 +0200 Subject: [PATCH 4/5] Implement `on_underflow` in `EvoWorklist.evo_aspirate` --- robotools/evotools/test_worklist.py | 6 ++++-- robotools/evotools/worklist.py | 11 ++++++++++- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/robotools/evotools/test_worklist.py b/robotools/evotools/test_worklist.py index 21dd8eb..bdf2000 100644 --- a/robotools/evotools/test_worklist.py +++ b/robotools/evotools/test_worklist.py @@ -639,12 +639,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: diff --git a/robotools/evotools/worklist.py b/robotools/evotools/worklist.py index 993e3d2..240d06c 100644 --- a/robotools/evotools/worklist.py +++ b/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 `_. + - ``"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, From 54b4161f8c0540801d005a21e2311cf8c8865b52 Mon Sep 17 00:00:00 2001 From: Michael Osthege Date: Mon, 23 Jun 2025 11:14:00 +0200 Subject: [PATCH 5/5] Implement `on_underflow` in worklist `transfer` methods --- robotools/evotools/test_worklist.py | 17 +++++++++++++++++ robotools/evotools/worklist.py | 11 ++++++++++- robotools/fluenttools/test_worklist.py | 17 +++++++++++++++++ robotools/fluenttools/worklist.py | 11 ++++++++++- 4 files changed, 54 insertions(+), 2 deletions(-) diff --git a/robotools/evotools/test_worklist.py b/robotools/evotools/test_worklist.py index bdf2000..0fc9332 100644 --- a/robotools/evotools/test_worklist.py +++ b/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: diff --git a/robotools/evotools/worklist.py b/robotools/evotools/worklist.py index 240d06c..b89e990 100644 --- a/robotools/evotools/worklist.py +++ b/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 `_. + - ``"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, diff --git a/robotools/fluenttools/test_worklist.py b/robotools/fluenttools/test_worklist.py index d21598b..bf96830 100644 --- a/robotools/fluenttools/test_worklist.py +++ b/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") diff --git a/robotools/fluenttools/worklist.py b/robotools/fluenttools/worklist.py index 48adaeb..386b680 100644 --- a/robotools/fluenttools/worklist.py +++ b/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 `_. + - ``"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,