Implement FluentWorklist.transfer method

With ``wash_scheme=None`` option.

robotools/fluenttools/test_worklist.py

@@ -0,0 +1,44 @@
+import pytest
+
+from robotools.fluenttools.worklist import FluentWorklist
+from robotools.liquidhandling.labware import Labware
+
+
+class TestFluentWorklist:
+    def test_transfer(self):
+        A = Labware("A", 3, 4, min_volume=10, max_volume=200)
+        A.add("A01", 100)
+        with FluentWorklist() as wl:
+            wl.transfer(
+                A,
+                "A01",
+                A,
+                "B01",
+                50,
+            )
+        assert len(wl) == 3
+        a, d, w = wl
+        assert a.startswith("A;")
+        assert d.startswith("D;")
+        assert w == "W1;"
+        assert A.volumes[0, 0] == 50
+        pass
+
+    def test_input_checks(self):
+        A = Labware("A", 3, 4, min_volume=10, max_volume=200, initial_volumes=150)
+        with FluentWorklist() as wl:
+            with pytest.raises(ValueError, match="must be equal"):
+                wl.transfer(A, ["A01", "B01", "C01"], A, ["A01", "B01"], 20)
+            with pytest.raises(ValueError, match="must be equal"):
+                wl.transfer(A, ["A01", "B01"], A, ["A01", "B01", "C01"], 20)
+            with pytest.raises(ValueError, match="must be equal"):
+                wl.transfer(A, ["A01", "B01"], A, "A01", [30, 40, 25])
+        pass
+
+    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)
+        assert len(wl) == 3
+        assert wl[-1] == "F;"
+        pass

robotools/fluenttools/worklist.py

@@ -1,7 +1,16 @@
+from collections.abc import Sequence
 from pathlib import Path
 from typing import Optional, Union
 
+import numpy as np
+
+from robotools.liquidhandling.labware import Labware
 from robotools.worklists.base import BaseWorklist
+from robotools.worklists.utils import (
+    optimize_partition_by,
+    partition_by_column,
+    partition_volume,
+)
 
 __all__ = ("FluentWorklist",)
 
@@ -15,5 +24,125 @@ def __init__(
         max_volume: Union[int, float] = 950,
         auto_split: bool = True,
     ) -> None:
-        raise NotImplementedError("Be patient.")
         super().__init__(filepath, max_volume, auto_split)
+
+    def transfer(
+        self,
+        source: Labware,
+        source_wells: Union[str, Sequence[str], np.ndarray],
+        destination: Labware,
+        destination_wells: Union[str, Sequence[str], np.ndarray],
+        volumes: Union[float, Sequence[float], np.ndarray],
+        *,
+        label: Optional[str] = None,
+        wash_scheme: Optional[int] = 1,
+        partition_by: str = "auto",
+        **kwargs,
+    ) -> None:
+        """Transfer operation between two labwares.
+
+        Parameters
+        ----------
+        source
+            Source labware
+        source_wells
+            List of source well ids
+        destination
+            Destination labware
+        destination_wells
+            List of destination well ids
+        volumes
+            Volume(s) to 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.
+        partition_by : str
+            one of 'auto' (default), 'source' or 'destination'
+                'auto': partitioning by source unless the source is a Trough
+                'source': partitioning by source columns
+                'destination': partitioning by destination columns
+        kwargs
+            Additional keyword arguments to pass to aspirate and dispense.
+            Most prominent example: `liquid_class`.
+            Take a look at `Worklist.aspirate_well` for the full list of options.
+        """
+        # reformat the convenience parameters
+        source_wells = np.array(source_wells).flatten("F")
+        destination_wells = np.array(destination_wells).flatten("F")
+        volumes = np.array(volumes).flatten("F")
+        nmax = max((len(source_wells), len(destination_wells), len(volumes)))
+
+        if len(source_wells) == 1:
+            source_wells = np.repeat(source_wells, nmax)
+        if len(destination_wells) == 1:
+            destination_wells = np.repeat(destination_wells, nmax)
+        if len(volumes) == 1:
+            volumes = np.repeat(volumes, nmax)
+        lengths = (len(source_wells), len(destination_wells), len(volumes))
+        if len(set(lengths)) != 1:
+            raise ValueError(f"Number of source/destination/volumes must be equal. They were {lengths}")
+
+        # automatic partitioning
+        partition_by = optimize_partition_by(source, destination, partition_by, label)
+
+        # the label applies to the entire transfer operation and is not logged at individual aspirate/dispense steps
+        self.comment(label)
+        nsteps = 0
+        lvh_extra = 0
+
+        for srcs, dsts, vols in partition_by_column(source_wells, destination_wells, volumes, partition_by):
+            # make vector of volumes into vector of volume-lists
+            vol_lists = [
+                partition_volume(float(v), max_volume=self.max_volume) if self.auto_split else [v]
+                for v in vols
+            ]
+            # transfer from this source column until all wells are done
+            npartitions = max(map(len, vol_lists))
+            # Count only the extra steps created by LVH
+            lvh_extra += sum([len(vs) - 1 for vs in vol_lists])
+            for p in range(npartitions):
+                naccessed = 0
+                # iterate the rows
+                for s, d, vs in zip(srcs, dsts, vol_lists):
+                    # transfer the next volume-fraction for this well
+                    if len(vs) > p:
+                        v = vs[p]
+                        if v > 0:
+                            self.aspirate(source, s, v, label=None, **kwargs)
+                            self.dispense(
+                                destination,
+                                d,
+                                v,
+                                label=None,
+                                compositions=[source.get_well_composition(s)],
+                                **kwargs,
+                            )
+                            nsteps += 1
+                            if wash_scheme is not None:
+                                self.wash(scheme=wash_scheme)
+                            else:
+                                self.flush()
+                            naccessed += 1
+                # LVH: if multiple wells are accessed, don't group across partitions
+                if npartitions > 1 and naccessed > 1 and not p == npartitions - 1:
+                    self.commit()
+            # LVH: don't group across columns
+            if npartitions > 1:
+                self.commit()
+
+        # Condense the labware logs into one operation
+        # after the transfer operation completed to facilitate debugging.
+        # Also include the number of extra steps because of LVH if applicable.
+        if lvh_extra:
+            if label:
+                label = f"{label} ({lvh_extra} LVH steps)"
+            else:
+                label = f"{lvh_extra} LVH steps"
+        if destination == source:
+            source.condense_log(nsteps * 2, label=label)
+        else:
+            source.condense_log(nsteps, label=label)
+            destination.condense_log(nsteps, label=label)
+        return

robotools/test_worklists.py

@@ -30,7 +30,6 @@ def test_recommended_instantiation():
         warnings.simplefilter("error")
         BaseWorklist()
         EvoWorklist()
-    with pytest.raises(NotImplementedError):
         FluentWorklist()
     pass
 

robotools/worklists/test_lvh.py

@@ -1,14 +1,17 @@
 import numpy as np
+import pytest
 
 from robotools.evotools.worklist import EvoWorklist
+from robotools.fluenttools.worklist import FluentWorklist
 from robotools.liquidhandling.labware import Labware
 
 
 class TestLargeVolumeHandling:
-    def test_single_split(self) -> None:
+    @pytest.mark.parametrize("cls", [EvoWorklist, FluentWorklist])
+    def test_single_split(self, cls) -> None:
         src = Labware("A", 3, 2, min_volume=1000, max_volume=25000, initial_volumes=12000)
         dst = Labware("B", 3, 2, min_volume=1000, max_volume=25000)
-        with EvoWorklist(auto_split=True) as wl:
+        with cls(auto_split=True) as wl:
             wl.transfer(src, "A01", dst, "A01", 2000, label="Transfer more than 2x the max")
             assert wl == [
                 "C;Transfer more than 2x the max",
@@ -46,10 +49,11 @@ def test_single_split(self) -> None:
         )
         return
 
-    def test_column_split(self) -> None:
+    @pytest.mark.parametrize("cls", [EvoWorklist, FluentWorklist])
+    def test_column_split(self, cls) -> None:
         src = Labware("A", 4, 2, min_volume=1000, max_volume=25000, initial_volumes=12000)
         dst = Labware("B", 4, 2, min_volume=1000, max_volume=25000)
-        with EvoWorklist(auto_split=True) as wl:
+        with cls(auto_split=True) as wl:
             wl.transfer(
                 src, ["A01", "B01", "D01", "C01"], dst, ["A01", "B01", "D01", "C01"], [1500, 250, 0, 1200]
             )
@@ -93,10 +97,11 @@ def test_column_split(self) -> None:
         )
         return
 
-    def test_block_split(self) -> None:
+    @pytest.mark.parametrize("cls", [EvoWorklist, FluentWorklist])
+    def test_block_split(self, cls) -> None:
         src = Labware("A", 3, 2, min_volume=1000, max_volume=25000, initial_volumes=12000)
         dst = Labware("B", 3, 2, min_volume=1000, max_volume=25000)
-        with EvoWorklist(auto_split=True) as wl:
+        with cls(auto_split=True) as wl:
             wl.transfer(
                 # A01, B01, A02, B02
                 src,