Align wash_scheme=None signature and insert W; in Evo case

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",

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;;;;",

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

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

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
 
@@ -24,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)
@@ -39,7 +41,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 +62,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 +84,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 +137,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:

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`.

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
 
@@ -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,13 +118,19 @@ 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};")
         return
 
     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
 
@@ -516,7 +527,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,
     ):

robotools/worklists/test_base.py

@@ -201,12 +201,61 @@ 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])
+    @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()
+
+        # 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: