From ae1d89420230efc7a97dde3cc4d8753cef461c44 Mon Sep 17 00:00:00 2001 From: Michael Osthege Date: Wed, 16 Jul 2025 14:39:26 +0200 Subject: [PATCH 1/3] Return aspiratable volumes from `Labware.remove` method Part of #102. --- robotools/liquidhandling/labware.py | 16 +++++++++++----- robotools/liquidhandling/test_labware.py | 13 ++++++++++--- 2 files changed, 21 insertions(+), 8 deletions(-) diff --git a/robotools/liquidhandling/labware.py b/robotools/liquidhandling/labware.py index d6384f9..b5cb00f 100644 --- a/robotools/liquidhandling/labware.py +++ b/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,6 +314,11 @@ 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 `_. - ``"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") @@ -321,6 +326,7 @@ def remove( 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. diff --git a/robotools/liquidhandling/test_labware.py b/robotools/liquidhandling/test_labware.py index 4549966..3e523ff 100644 --- a/robotools/liquidhandling/test_labware.py +++ b/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) From 99e02e3b2cf73d708e2616a7626218ebc7fdefda Mon Sep 17 00:00:00 2001 From: Michael Osthege Date: Wed, 16 Jul 2025 14:49:47 +0200 Subject: [PATCH 2/3] Return vaspirated from worklist's aspiration methods --- robotools/evotools/test_worklist.py | 4 +++- robotools/evotools/worklist.py | 11 ++++++++--- robotools/worklists/base.py | 12 +++++++++--- robotools/worklists/test_base.py | 4 +++- 4 files changed, 23 insertions(+), 8 deletions(-) diff --git a/robotools/evotools/test_worklist.py b/robotools/evotools/test_worklist.py index 0fc9332..2a9d00c 100644 --- a/robotools/evotools/test_worklist.py +++ b/robotools/evotools/test_worklist.py @@ -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 diff --git a/robotools/evotools/worklist.py b/robotools/evotools/worklist.py index b89e990..c70ea2b 100644 --- a/robotools/evotools/worklist.py +++ b/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 `_. - ``"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, diff --git a/robotools/worklists/base.py b/robotools/worklists/base.py index 87c373b..2ec135f 100644 --- a/robotools/worklists/base.py +++ b/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, @@ -643,6 +648,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 diff --git a/robotools/worklists/test_base.py b/robotools/worklists/test_base.py index 0eb6383..2b23d15 100644 --- a/robotools/worklists/test_base.py +++ b/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;;;;", From 11de601132b545e25dbb567cb6100f6811c204d9 Mon Sep 17 00:00:00 2001 From: Michael Osthege Date: Wed, 16 Jul 2025 15:19:46 +0200 Subject: [PATCH 3/3] Digitally transfer only the aspiratable volumes in worklist methods Closes #102 --- pyproject.toml | 2 +- robotools/evotools/test_worklist.py | 4 ++-- robotools/evotools/worklist.py | 5 ++++- robotools/fluenttools/test_worklist.py | 4 ++-- robotools/fluenttools/worklist.py | 5 ++++- robotools/worklists/base.py | 19 +++++++++++++++++-- 6 files changed, 30 insertions(+), 9 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index a2da2ef..58b24d0 100644 --- a/pyproject.toml +++ b/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" diff --git a/robotools/evotools/test_worklist.py b/robotools/evotools/test_worklist.py index 2a9d00c..c2d042f 100644 --- a/robotools/evotools/test_worklist.py +++ b/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") diff --git a/robotools/evotools/worklist.py b/robotools/evotools/worklist.py index c70ea2b..59c1234 100644 --- a/robotools/evotools/worklist.py +++ b/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 diff --git a/robotools/fluenttools/test_worklist.py b/robotools/fluenttools/test_worklist.py index bf96830..39d2e72 100644 --- a/robotools/fluenttools/test_worklist.py +++ b/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") diff --git a/robotools/fluenttools/worklist.py b/robotools/fluenttools/worklist.py index 386b680..8f4775e 100644 --- a/robotools/fluenttools/worklist.py +++ b/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 diff --git a/robotools/worklists/base.py b/robotools/worklists/base.py index 2ec135f..c06a715 100644 --- a/robotools/worklists/base.py +++ b/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: