Refactor how well position numbers are calculate to be Fluent-trough compatible

robotools/evotools/__init__.py

@@ -1,3 +1,4 @@
 from robotools.evotools.types import Labwares, Tip, int_to_tip
+from robotools.evotools.utils import get_well_position
 from robotools.evotools.worklist import EvoWorklist, Worklist
 from robotools.worklists.exceptions import InvalidOperationError

robotools/evotools/test_utils.py

@@ -0,0 +1,25 @@
+import pytest
+
+from robotools import Labware, Trough
+from robotools.evotools import utils
+
+
+def test_get_well_position():
+    plate = Labware("plate", 3, 4, min_volume=0, max_volume=50)
+    assert utils.get_well_position(plate, "A01") == 1
+    assert utils.get_well_position(plate, "B01") == 2
+    assert utils.get_well_position(plate, "B04") == 11
+
+    trough = Trough("trough", 2, 3, min_volume=0, max_volume=50)
+    assert utils.get_well_position(trough, "A01") == 1
+    assert utils.get_well_position(trough, "B01") == 2
+    assert utils.get_well_position(trough, "A02") == 3
+    assert utils.get_well_position(trough, "A03") == 5
+
+    with pytest.raises(ValueError, match="not an alphanumeric well ID"):
+        utils.get_well_position(trough, "A-3")
+
+    # Currently not implemented at the Labware level:
+    # megaplate = Labware("mplate", 50, 3, min_volume=0, max_volume=50)
+    # assert utils.get_well_position(megaplate, "AA2") == 51
+    pass

robotools/evotools/test_worklist.py

@@ -590,6 +590,45 @@ def test_history_condensation_within_labware(self) -> None:
         return
 
 
+class TestTroughLabwareWorklist:
+    def test_aspirate(self) -> None:
+        source = Trough(
+            "SourceLW", virtual_rows=3, columns=3, min_volume=10, max_volume=200, initial_volumes=200
+        )
+        with EvoWorklist() as wl:
+            wl.aspirate(source, ["A01", "A02", "C02"], 50)
+            wl.aspirate(source, ["A01", "A02", "C02"], [1, 2, 3])
+            assert wl == [
+                "A;SourceLW;;;1;;50.00;;;;",
+                "A;SourceLW;;;4;;50.00;;;;",
+                "A;SourceLW;;;6;;50.00;;;;",
+                "A;SourceLW;;;1;;1.00;;;;",
+                "A;SourceLW;;;4;;2.00;;;;",
+                "A;SourceLW;;;6;;3.00;;;;",
+            ]
+            np.testing.assert_array_equal(source.volumes, [[149, 95, 200]])
+            assert len(source.history) == 3
+        return
+
+    def test_dispense(self) -> None:
+        destination = Trough("DestinationLW", virtual_rows=3, columns=3, min_volume=10, max_volume=200)
+        with EvoWorklist() as wl:
+            wl.dispense(destination, ["A01", "A02", "A03", "B01"], 50)
+            wl.dispense(destination, ["A01", "A02", "C02"], [1, 2, 3])
+            assert wl == [
+                "D;DestinationLW;;;1;;50.00;;;;",
+                "D;DestinationLW;;;4;;50.00;;;;",
+                "D;DestinationLW;;;7;;50.00;;;;",
+                "D;DestinationLW;;;2;;50.00;;;;",
+                "D;DestinationLW;;;1;;1.00;;;;",
+                "D;DestinationLW;;;4;;2.00;;;;",
+                "D;DestinationLW;;;6;;3.00;;;;",
+            ]
+            np.testing.assert_array_equal(destination.volumes, [[101, 55, 50]])
+            assert len(destination.history) == 3
+        return
+
+
 class TestEvoCommands:
     def test_evo_aspirate(self) -> None:
         lw = Labware("A", 4, 5, min_volume=10, max_volume=100)

robotools/evotools/utils.py

@@ -1,4 +1,7 @@
 """Generic utility functions."""
+import re
+
+from robotools.liquidhandling import Labware
 
 
 def to_hex(dec: int):
@@ -12,3 +15,26 @@ def to_hex(dec: int):
     if rest == 0:
         return digits[x]
     return to_hex(rest) + digits[x]
+
+
+_WELLID_MATCHER = re.compile(r"^([a-zA-Z]+?)(\d+?)$")
+"""Compiled RegEx for matching well row & column from alphanumeric IDs."""
+
+
+def get_well_position(labware: Labware, well: str) -> int:
+    """Calculate the EVO-style well position from the alphanumeric ID."""
+    # Extract row & column number from the alphanumeric ID
+    m = _WELLID_MATCHER.match(well)
+    if m is None:
+        raise ValueError(f"This is not an alphanumeric well ID: '{well}'.")
+    row = m.group(1)
+    column = int(m.group(2))
+
+    r = labware.row_ids.index(row)
+    c = labware.column_ids.index(column)
+
+    # Calculate the position from the row & column number.
+    # The EVO counts virtual rows in troughs too.
+    if labware.virtual_rows is not None:
+        return 1 + c * labware.virtual_rows + r
+    return 1 + c * labware.n_rows + r

robotools/evotools/worklist.py

@@ -10,6 +10,7 @@
 from robotools import liquidhandling
 from robotools.evotools import commands
 from robotools.evotools.types import Tip
+from robotools.evotools.utils import get_well_position
 from robotools.worklists.base import BaseWorklist
 from robotools.worklists.utils import (
     optimize_partition_by,
@@ -25,6 +26,9 @@
 class EvoWorklist(BaseWorklist):
     """Context manager for the creation of Tecan EVO worklists."""
 
+    def _get_well_position(self, labware: liquidhandling.Labware, well: str) -> int:
+        return get_well_position(labware, well)
+
     def evo_aspirate(
         self,
         labware: liquidhandling.Labware,

robotools/fluenttools/__init__.py

@@ -1 +1,2 @@
+from robotools.fluenttools.utils import get_well_position
 from robotools.fluenttools.worklist import FluentWorklist

robotools/fluenttools/test_utils.py

@@ -0,0 +1,25 @@
+import pytest
+
+from robotools import Labware, Trough
+from robotools.fluenttools import utils
+
+
+def test_get_well_position():
+    plate = Labware("plate", 3, 4, min_volume=0, max_volume=50)
+    assert utils.get_well_position(plate, "A01") == 1
+    assert utils.get_well_position(plate, "B01") == 2
+    assert utils.get_well_position(plate, "B04") == 11
+
+    trough = Trough("trough", 2, 3, min_volume=0, max_volume=50)
+    assert utils.get_well_position(trough, "A01") == 1
+    assert utils.get_well_position(trough, "B01") == 1
+    assert utils.get_well_position(trough, "A02") == 2
+    assert utils.get_well_position(trough, "A03") == 3
+
+    with pytest.raises(ValueError, match="not an alphanumeric well ID"):
+        utils.get_well_position(trough, "🧨")
+
+    # Currently not implemented at the Labware level:
+    # megaplate = Labware("mplate", 50, 3, min_volume=0, max_volume=50)
+    # assert utils.get_well_position(megaplate, "AA2") == 51
+    pass

robotools/fluenttools/utils.py

@@ -0,0 +1,28 @@
+"""Generic utility functions."""
+import re
+
+from robotools.liquidhandling import Labware
+
+_WELLID_MATCHER = re.compile(r"^([a-zA-Z]+?)(\d+?)$")
+"""Compiled RegEx for matching well row & column from alphanumeric IDs."""
+
+
+def get_well_position(labware: Labware, well: str) -> int:
+    """Calculate the EVO-style well position from the alphanumeric ID."""
+    # Extract row & column number from the alphanumeric ID
+    m = _WELLID_MATCHER.match(well)
+    if m is None:
+        raise ValueError(f"This is not an alphanumeric well ID: '{well}'.")
+    row = m.group(1)
+    column = int(m.group(2))
+
+    c = labware.column_ids.index(column)
+
+    # The Fluent does NOT count rows inside troughs!
+    if labware.is_trough:
+        return 1 + c
+
+    # Therefore the row number is only relevant for non-trough labware.
+    row = well[0]
+    r = labware.row_ids.index(row)
+    return 1 + c * labware.n_rows + r

robotools/fluenttools/worklist.py

@@ -4,6 +4,7 @@
 
 import numpy as np
 
+from robotools.fluenttools.utils import get_well_position
 from robotools.liquidhandling.labware import Labware
 from robotools.worklists.base import BaseWorklist
 from robotools.worklists.utils import (
@@ -26,6 +27,9 @@ def __init__(
     ) -> None:
         super().__init__(filepath, max_volume, auto_split)
 
+    def _get_well_position(self, labware: Labware, well: str) -> int:
+        return get_well_position(labware, well)
+
     def transfer(
         self,
         source: Labware,

robotools/liquidhandling/labware.py

@@ -54,6 +54,12 @@ def indices(self) -> Dict[str, Tuple[int, int]]:
     @property
     def positions(self) -> Dict[str, int]:
         """Mapping of well-ids to EVOware-compatible position numbers."""
+        warnings.warn(
+            "`Labware.positions` is deprecated in favor of model-specific implementations."
+            "Use `robotools.evotools.get_well_positions()` or `robotools.fluenttools.get_well_positions()`.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         return self._positions
 
     @property

robotools/liquidhandling/test_labware.py

@@ -32,14 +32,15 @@ def test_init(self) -> None:
             "B03": (1, 2),
         }
         assert plate.indices == exp
-        assert plate.positions == {
-            "A01": 1,
-            "A02": 3,
-            "A03": 5,
-            "B01": 2,
-            "B02": 4,
-            "B03": 6,
-        }
+        with pytest.warns(DeprecationWarning, match="in favor of model-specific"):
+            assert plate.positions == {
+                "A01": 1,
+                "A02": 3,
+                "A03": 5,
+                "B01": 2,
+                "B02": 4,
+                "B03": 6,
+            }
         return
 
     def test_invalid_init(self) -> None:
@@ -215,28 +216,29 @@ def test_init_trough(self) -> None:
             "E03": (0, 2),
             "E04": (0, 3),
         }
-        assert trough.positions == {
-            "A01": 1,
-            "A02": 6,
-            "A03": 11,
-            "A04": 16,
-            "B01": 2,
-            "B02": 7,
-            "B03": 12,
-            "B04": 17,
-            "C01": 3,
-            "C02": 8,
-            "C03": 13,
-            "C04": 18,
-            "D01": 4,
-            "D02": 9,
-            "D03": 14,
-            "D04": 19,
-            "E01": 5,
-            "E02": 10,
-            "E03": 15,
-            "E04": 20,
-        }
+        with pytest.warns(DeprecationWarning, match="in favor of model-specific"):
+            assert trough.positions == {
+                "A01": 1,
+                "A02": 6,
+                "A03": 11,
+                "A04": 16,
+                "B01": 2,
+                "B02": 7,
+                "B03": 12,
+                "B04": 17,
+                "C01": 3,
+                "C02": 8,
+                "C03": 13,
+                "C04": 18,
+                "D01": 4,
+                "D02": 9,
+                "D03": 14,
+                "D04": 19,
+                "E01": 5,
+                "E02": 10,
+                "E03": 15,
+                "E04": 20,
+            }
         return
 
     def test_initial_volumes(self) -> None:

robotools/worklists/base.py

@@ -9,6 +9,7 @@
 
 from robotools import liquidhandling
 from robotools.evotools.types import Tip
+from robotools.liquidhandling import Labware
 from robotools.worklists.exceptions import CompatibilityError, InvalidOperationError
 from robotools.worklists.utils import prepare_aspirate_dispense_parameters
 
@@ -63,6 +64,12 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None:
             self.save(self._filepath)
         return
 
+    def _get_well_position(self, labware: Labware, well: str) -> int:
+        """Internal method to resolve the well number for a given labware well."""
+        raise TypeError(
+            "The use of a specific worklist type (typically EvoWorklist or FluentWorklist) is required for this operation."
+        )
+
     def save(self, filepath: Union[str, Path]) -> None:
         """Writes the worklist to the filepath.
 
@@ -457,7 +464,7 @@ def aspirate(
         self.comment(label)
         for well, volume in zip(wells, volumes):
             if volume > 0:
-                self.aspirate_well(labware.name, labware.positions[well], volume, **kwargs)
+                self.aspirate_well(labware.name, self._get_well_position(labware, well), volume, **kwargs)
         return
 
     def dispense(
@@ -497,7 +504,7 @@ def dispense(
         self.comment(label)
         for well, volume in zip(wells, volumes):
             if volume > 0:
-                self.dispense_well(labware.name, labware.positions[well], volume, **kwargs)
+                self.dispense_well(labware.name, self._get_well_position(labware, well), volume, **kwargs)
         return
 
     def transfer(
@@ -587,7 +594,7 @@ def distribute(
 
         # transform destination wells into range + mask
         destination_wells = numpy.array(destination_wells).flatten("F")
-        dst_wells = list(sorted([destination.positions[w] for w in destination_wells]))
+        dst_wells = list(sorted([self._get_well_position(destination, w) for w in destination_wells]))
         dst_start, dst_end = dst_wells[0], dst_wells[-1]
         excluded_dst_wells = set(range(dst_start, dst_end + 1)).difference(dst_wells)