From 29f3bf4ce815d292de99499d5071008679a7b5ad Mon Sep 17 00:00:00 2001 From: Michael Osthege Date: Wed, 17 Jul 2024 22:54:26 +0200 Subject: [PATCH 1/2] Align `transfer(wash_scheme=...)` options New options are limited to: * 1, 2, 3, 4 for wash scheme numbers * "flush" for inserting a `F;` * "reuse" for not inserting anything --- robotools/__init__.py | 2 +- robotools/evotools/test_worklist.py | 2 +- robotools/evotools/worklist.py | 28 ++++++++++++++++++---- robotools/fluenttools/test_worklist.py | 2 +- robotools/fluenttools/worklist.py | 28 ++++++++++++++++------ robotools/utils.py | 10 ++++---- robotools/worklists/base.py | 4 ++-- robotools/worklists/test_base.py | 32 ++++++++++++++++++++++++++ 8 files changed, 86 insertions(+), 22 deletions(-) diff --git a/robotools/__init__.py b/robotools/__init__.py index c3d83bb..9122a86 100644 --- a/robotools/__init__.py +++ b/robotools/__init__.py @@ -20,7 +20,7 @@ from .utils import DilutionPlan, get_trough_wells from .worklists import BaseWorklist, CompatibilityError -__version__ = "1.10.1" +__version__ = "1.11.0" __all__ = ( "BaseWorklist", "CompatibilityError", diff --git a/robotools/evotools/test_worklist.py b/robotools/evotools/test_worklist.py index cefaa97..21dd8eb 100644 --- a/robotools/evotools/test_worklist.py +++ b/robotools/evotools/test_worklist.py @@ -104,7 +104,7 @@ def test_transfer_2d_volumes_no_wash(self) -> None: [15.3, 17.53], ] ), - wash_scheme=None, + wash_scheme="reuse", ) assert wl == [ "A;A;;;1;;20.00;;;;", diff --git a/robotools/evotools/worklist.py b/robotools/evotools/worklist.py index d85773e..993e3d2 100644 --- a/robotools/evotools/worklist.py +++ b/robotools/evotools/worklist.py @@ -3,7 +3,7 @@ import logging import textwrap import warnings -from typing import Dict, List, Optional, Sequence, Tuple, Union +from typing import Dict, List, Literal, Optional, Sequence, Tuple, Union import numpy as np @@ -219,7 +219,7 @@ def transfer( volumes: Union[float, Sequence[float], np.ndarray], *, label: Optional[str] = None, - wash_scheme: int = 1, + wash_scheme: Literal[1, 2, 3, 4, "flush", "reuse"] = 1, partition_by: str = "auto", **kwargs, ) -> None: @@ -239,8 +239,13 @@ def transfer( Volume(s) to transfer label : str Label of the operation to log into labware history - wash_scheme : int - Wash scheme to apply after every tip use + wash_scheme + - One of ``{1, 2, 3, 4}`` to select a wash scheme for fixed tips, + or drop tips when using DiTis. + - ``"flush"`` blows out tips, but does not drop DiTis, and only does a short wash with fixed tips. + - ``"reuse"`` continues pipetting without flushing, dropping or washing. + Passing ``None`` is deprecated, results in ``"reuse"`` behavior and emits a warning. + partition_by : str one of 'auto' (default), 'source' or 'destination' 'auto': partitioning by source unless the source is a Trough @@ -257,6 +262,15 @@ def transfer( volumes = np.array(volumes).flatten("F") nmax = max((len(source_wells), len(destination_wells), len(volumes))) + # Deal with deprecated behavior + if wash_scheme is None: + warnings.warn( + "wash_scheme=None is deprecated. For tip reuse pass 'reuse'.", + DeprecationWarning, + stacklevel=2, + ) + wash_scheme = "reuse" + if len(source_wells) == 1: source_wells = np.repeat(source_wells, nmax) if len(destination_wells) == 1: @@ -304,7 +318,11 @@ def transfer( **kwargs, ) nsteps += 1 - if wash_scheme is not None: + if wash_scheme == "flush": + self.flush() + elif wash_scheme == "reuse": + pass + else: self.wash(scheme=wash_scheme) naccessed += 1 # LVH: if multiple wells are accessed, don't group across partitions diff --git a/robotools/fluenttools/test_worklist.py b/robotools/fluenttools/test_worklist.py index 23112f3..d21598b 100644 --- a/robotools/fluenttools/test_worklist.py +++ b/robotools/fluenttools/test_worklist.py @@ -38,7 +38,7 @@ def test_input_checks(self): def test_transfer_flush(self): A = Labware("A", 3, 4, min_volume=10, max_volume=200, initial_volumes=150) with FluentWorklist() as wl: - wl.transfer(A, "A01", A, "B01", 20, wash_scheme=None) + wl.transfer(A, "A01", A, "B01", 20, wash_scheme="flush") assert len(wl) == 3 assert wl[-1] == "F;" pass diff --git a/robotools/fluenttools/worklist.py b/robotools/fluenttools/worklist.py index e07fd35..76ff6f9 100644 --- a/robotools/fluenttools/worklist.py +++ b/robotools/fluenttools/worklist.py @@ -1,6 +1,7 @@ +import warnings from collections.abc import Sequence from pathlib import Path -from typing import Optional, Union +from typing import Literal, Optional, Union import numpy as np @@ -39,7 +40,7 @@ def transfer( volumes: Union[float, Sequence[float], np.ndarray], *, label: Optional[str] = None, - wash_scheme: Optional[int] = 1, + wash_scheme: Literal[1, 2, 3, 4, "flush", "reuse"] = 1, partition_by: str = "auto", **kwargs, ) -> None: @@ -60,8 +61,12 @@ def transfer( label Label of the operation to log into labware history wash_scheme - Wash scheme to apply after every tip use. - If ``None``, only a flush is inserted instead of a wash. + - One of ``{1, 2, 3, 4}`` to select a wash scheme for fixed tips, + or drop tips when using DiTis. + - ``"flush"`` blows out tips, but does not drop DiTis, and only does a short wash with fixed tips. + - ``"reuse"`` continues pipetting without flushing, dropping or washing. + Passing ``None`` is deprecated, results in ``"flush"`` behavior and emits a warning. + partition_by : str one of 'auto' (default), 'source' or 'destination' 'auto': partitioning by source unless the source is a Trough @@ -78,6 +83,13 @@ def transfer( volumes = np.array(volumes).flatten("F") nmax = max((len(source_wells), len(destination_wells), len(volumes))) + # Deal with deprecated behavior + if wash_scheme is None: + warnings.warn( + "wash_scheme=None is deprecated. For flushing pass 'flush'.", DeprecationWarning, stacklevel=2 + ) + wash_scheme = "flush" + if len(source_wells) == 1: source_wells = np.repeat(source_wells, nmax) if len(destination_wells) == 1: @@ -124,10 +136,12 @@ def transfer( **kwargs, ) nsteps += 1 - if wash_scheme is not None: - self.wash(scheme=wash_scheme) - else: + if wash_scheme == "flush": self.flush() + elif wash_scheme == "reuse": + pass + else: + self.wash(scheme=wash_scheme) naccessed += 1 # LVH: if multiple wells are accessed, don't group across partitions if npartitions > 1 and naccessed > 1 and not p == npartitions - 1: diff --git a/robotools/utils.py b/robotools/utils.py index 6c2270f..0cfbafe 100644 --- a/robotools/utils.py +++ b/robotools/utils.py @@ -1,6 +1,6 @@ """Module with robot-agnostic utilities.""" import collections -from typing import Callable, Iterable, List, Optional, Sequence, Tuple, Union +from typing import Callable, Iterable, List, Literal, Optional, Sequence, Tuple, Union import numpy @@ -178,7 +178,7 @@ def to_worklist( pre_mix_hook: Optional[Callable[[int, BaseWorklist], Optional[BaseWorklist]]] = None, post_mix_hook: Optional[Callable[[int, BaseWorklist], Optional[BaseWorklist]]] = None, mix_threshold: float = 0.05, - mix_wash: int = 2, + mix_wash: Literal[1, 2, 3, 4, "flush", "reuse"] = 2, mix_repeat: int = 2, mix_volume: float = 0.8, lc_stock_trough: str = "Trough_Water_FD_AspLLT", @@ -235,9 +235,9 @@ def to_worklist( to multiple destinations. mix_threshold : float Maximum fraction of total dilution volume (self.vmax) that may be diluted without subsequent mixing (defaults to 0.05 or 5%) - mix_wash : int - Number of the wash scheme inbetween mixing steps - The recommended wash scheme is 0 mL + 1 mL with fast wash. + mix_wash + Transfer wash scheme inbetween mixing steps. + The recommended wash scheme is 0 mL + 1 mL with fast wash, or ``"flush"``. mix_repeat : int How often to mix after diluting. May be set to 0, particularly when combined with a `pre_mix_hook`. diff --git a/robotools/worklists/base.py b/robotools/worklists/base.py index ba5bc60..fcff899 100644 --- a/robotools/worklists/base.py +++ b/robotools/worklists/base.py @@ -3,7 +3,7 @@ import logging import math from pathlib import Path -from typing import Dict, Iterable, List, Optional, Sequence, Union +from typing import Dict, Iterable, List, Literal, Optional, Sequence, Union import numpy @@ -516,7 +516,7 @@ def transfer( volumes: Union[float, Sequence[float], numpy.ndarray], *, label: Optional[str] = None, - wash_scheme: int = 1, + wash_scheme: Literal[1, 2, 3, 4, "flush", "reuse"] = 1, partition_by: str = "auto", **kwargs, ): diff --git a/robotools/worklists/test_base.py b/robotools/worklists/test_base.py index bc389d3..222f57a 100644 --- a/robotools/worklists/test_base.py +++ b/robotools/worklists/test_base.py @@ -203,6 +203,38 @@ def test_wash(self) -> None: assert wl == exp return + @pytest.mark.parametrize("cls", [EvoWorklist, FluentWorklist]) + @pytest.mark.parametrize( + "scheme,exp", + [ + (1, "W1;"), + (2, "W2;"), + (3, "W3;"), + (4, "W4;"), + ("flush", "F;"), + ("reuse", None), + ], + ) + def test_wash_schemes(self, cls, scheme, exp): + A = Labware("A", 2, 4, min_volume=50, max_volume=250, initial_volumes=200) + with cls() as wl: + wl.transfer(A, "A01", A, "A01", 100, wash_scheme=scheme) + if exp is None: + assert wl[-1].startswith("D;") + else: + assert wl[-1] == exp + + # Test deprecated None setting that had different behavior on EVO/Fluent + with pytest.warns(DeprecationWarning, match="wash_scheme=None is deprecated"): + wl.transfer(A, "B01", A, "B01", 50, wash_scheme=None) + if cls is EvoWorklist: + assert wl[-1].startswith("D;") + elif cls is FluentWorklist: + assert wl[-1] == "F;" + else: + raise NotImplementedError() + pass + def test_decontaminate(self) -> None: with BaseWorklist() as wl: wl.decontaminate() From c9e9e5acea6ded33b28417b27287cdedb71e499c Mon Sep 17 00:00:00 2001 From: Michael Osthege Date: Thu, 18 Jul 2024 19:00:17 +0200 Subject: [PATCH 2/2] Introduce `Worklist(diti_mode=...)` for DiTi-compatible wash schemes Closes #77 --- robotools/fluenttools/worklist.py | 3 ++- robotools/worklists/base.py | 11 +++++++++++ robotools/worklists/test_base.py | 17 +++++++++++++++++ 3 files changed, 30 insertions(+), 1 deletion(-) diff --git a/robotools/fluenttools/worklist.py b/robotools/fluenttools/worklist.py index 76ff6f9..48adaeb 100644 --- a/robotools/fluenttools/worklist.py +++ b/robotools/fluenttools/worklist.py @@ -25,8 +25,9 @@ def __init__( filepath: Optional[Union[str, Path]] = None, max_volume: Union[int, float] = 950, auto_split: bool = True, + diti_mode: bool = False, ) -> None: - super().__init__(filepath, max_volume, auto_split) + super().__init__(filepath, max_volume, auto_split, diti_mode) def _get_well_position(self, labware: Labware, well: str) -> int: return get_well_position(labware, well) diff --git a/robotools/worklists/base.py b/robotools/worklists/base.py index fcff899..8d11d4b 100644 --- a/robotools/worklists/base.py +++ b/robotools/worklists/base.py @@ -26,6 +26,7 @@ def __init__( filepath: Optional[Union[str, Path]] = None, max_volume: Union[int, float] = 950, auto_split: bool = True, + diti_mode: bool = False, ) -> None: """Creates a worklist writer. @@ -38,6 +39,9 @@ def __init__( auto_split : bool If `True`, large volumes in transfer operations are automatically splitted. If set to `False`, `InvalidOperationError` is raised when a pipetting volume exceeds `max_volume`. + diti_mode + Activate this when using DiTis. + Uses ``W;`` for all wash schemes and raises errors when using commands that are only for fixed tips. """ self._filepath: Optional[Path] = None if filepath is not None: @@ -46,6 +50,7 @@ def __init__( raise ValueError("The `max_volume` parameter is required.") self.max_volume = max_volume self.auto_split = auto_split + self.diti_mode = diti_mode super().__init__() @property @@ -113,6 +118,10 @@ def wash(self, scheme: int = 1) -> None: scheme : int Number indicating the wash scheme (default: 1) """ + if self.diti_mode: + self.append("W;") + return + if not scheme in {1, 2, 3, 4}: raise ValueError("scheme must be either 1, 2, 3 or 4") self.append(f"W{scheme};") @@ -120,6 +129,8 @@ def wash(self, scheme: int = 1) -> None: def decontaminate(self) -> None: """Decontamination wash consists of a decontamination wash followed by a normal wash.""" + if self.diti_mode: + raise InvalidOperationError("Decontamination wash is not available with DiTis.") self.append("WD;") return diff --git a/robotools/worklists/test_base.py b/robotools/worklists/test_base.py index 222f57a..0eb6383 100644 --- a/robotools/worklists/test_base.py +++ b/robotools/worklists/test_base.py @@ -201,6 +201,14 @@ def test_wash(self) -> None: "W4;", ] assert wl == exp + + with BaseWorklist(diti_mode=True) as wl: + wl.wash() + wl.wash(3) + assert wl == [ + "W;", + "W;", + ] return @pytest.mark.parametrize("cls", [EvoWorklist, FluentWorklist]) @@ -233,12 +241,21 @@ def test_wash_schemes(self, cls, scheme, exp): assert wl[-1] == "F;" else: raise NotImplementedError() + + # In DiTi mode, any numeric wash scheme results in "W;" + with cls(diti_mode=True) as wl: + wl.transfer(A, "A01", A, "A01", 25, wash_scheme=2) + assert wl[-1] == "W;" pass def test_decontaminate(self) -> None: with BaseWorklist() as wl: wl.decontaminate() assert wl == ["WD;"] + + with BaseWorklist(diti_mode=True) as wl: + with pytest.raises(InvalidOperationError, match="not available"): + wl.decontaminate() return def test_flush(self) -> None: