Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Implement FluentWorklist.transfer method #57

Merged
merged 3 commits into from
Jan 16, 2024
Merged

Implement FluentWorklist.transfer method #57

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8f9a7a9
Move LVH tests to a common place
michaelosthege Jan 12, 2024
9fcfb29
Implement `FluentWorklist.transfer` method
michaelosthege Jan 12, 2024
112cb67
Bump to 1.9.0
michaelosthege Jan 16, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion robotools/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,4 +14,4 @@
from .utils import DilutionPlan, get_trough_wells
from .worklists import BaseWorklist, CompatibilityError

__version__ = "1.8.0"
__version__ = "1.9.0"
164 changes: 0 additions & 164 deletions robotools/evotools/test_worklist.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -590,170 +590,6 @@ def test_history_condensation_within_labware(self) -> None:
return


class TestLargeVolumeHandling:
def test_single_split(self) -> 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:
wl.transfer(src, "A01", dst, "A01", 2000, label="Transfer more than 2x the max")
assert wl == [
"C;Transfer more than 2x the max",
"A;A;;;1;;667.00;;;;",
"D;B;;;1;;667.00;;;;",
"W1;",
# no breaks when pipetting single wells
"A;A;;;1;;667.00;;;;",
"D;B;;;1;;667.00;;;;",
"W1;",
# no breaks when pipetting single wells
"A;A;;;1;;666.00;;;;",
"D;B;;;1;;666.00;;;;",
"W1;",
"B;", # always break after partitioning
]
# Two extra steps were necessary because of LVH
assert "Transfer more than 2x the max (2 LVH steps)" in src.report
assert "Transfer more than 2x the max (2 LVH steps)" in dst.report
np.testing.assert_array_equal(
src.volumes,
[
[12000 - 2000, 12000],
[12000, 12000],
[12000, 12000],
],
)
np.testing.assert_array_equal(
dst.volumes,
[
[2000, 0],
[0, 0],
[0, 0],
],
)
return

def test_column_split(self) -> 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:
wl.transfer(
src, ["A01", "B01", "D01", "C01"], dst, ["A01", "B01", "D01", "C01"], [1500, 250, 0, 1200]
)
assert wl == [
"A;A;;;1;;750.00;;;;",
"D;B;;;1;;750.00;;;;",
"W1;",
"A;A;;;2;;250.00;;;;",
"D;B;;;2;;250.00;;;;",
"W1;",
# D01 is ignored because the volume is 0
"A;A;;;3;;600.00;;;;",
"D;B;;;3;;600.00;;;;",
"W1;",
"B;", # within-column break
"A;A;;;1;;750.00;;;;",
"D;B;;;1;;750.00;;;;",
"W1;",
"A;A;;;3;;600.00;;;;",
"D;B;;;3;;600.00;;;;",
"W1;",
"B;", # tailing break after partitioning
]
np.testing.assert_array_equal(
src.volumes,
[
[12000 - 1500, 12000],
[12000 - 250, 12000],
[12000 - 1200, 12000],
[12000, 12000],
],
)
np.testing.assert_array_equal(
dst.volumes,
[
[1500, 0],
[250, 0],
[1200, 0],
[0, 0],
],
)
return

def test_block_split(self) -> 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:
wl.transfer(
# A01, B01, A02, B02
src,
src.wells[:2, :],
dst,
["A01", "B01", "C01", "A02"],
[1500, 250, 1200, 3000],
)
assert wl == [
"A;A;;;1;;750.00;;;;",
"D;B;;;1;;750.00;;;;",
"W1;",
"A;A;;;2;;250.00;;;;",
"D;B;;;2;;250.00;;;;",
"W1;",
"B;", # within-column 1 break
"A;A;;;1;;750.00;;;;",
"D;B;;;1;;750.00;;;;",
"W1;",
"B;", # between-column 1/2 break
"A;A;;;4;;600.00;;;;",
"D;B;;;3;;600.00;;;;",
"W1;",
"A;A;;;5;;750.00;;;;",
"D;B;;;4;;750.00;;;;",
"W1;",
"B;", # within-column 2 break
"A;A;;;4;;600.00;;;;",
"D;B;;;3;;600.00;;;;",
"W1;",
"A;A;;;5;;750.00;;;;",
"D;B;;;4;;750.00;;;;",
"W1;",
"B;", # within-column 2 break
"A;A;;;5;;750.00;;;;",
"D;B;;;4;;750.00;;;;",
"W1;",
# no break because only one well is accessed in this partition
"A;A;;;5;;750.00;;;;",
"D;B;;;4;;750.00;;;;",
"W1;",
"B;", # tailing break after partitioning
]

# How the number of splits is calculated:
# 1500 is split 2x → 1 extra
# 250 is not split
# 1200 is split 2x → 1 extra
# 3000 is split 4x → 3 extra
# Sum of extra steps: 5
assert "5 LVH steps" in src.report
assert "5 LVH steps" in dst.report
np.testing.assert_array_equal(
src.volumes,
[
[12000 - 1500, 12000 - 1200],
[12000 - 250, 12000 - 3000],
[12000, 12000],
],
)
np.testing.assert_array_equal(
dst.volumes,
[
[1500, 3000],
[250, 0],
[1200, 0],
],
)
return


class TestEvoCommands:
def test_evo_aspirate(self) -> None:
lw = Labware("A", 4, 5, min_volume=10, max_volume=100)
Expand Down
44 changes: 44 additions & 0 deletions robotools/fluenttools/test_worklist.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
michaelosthege marked this conversation as resolved.
Show resolved Hide resolved

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
131 changes: 130 additions & 1 deletion robotools/fluenttools/worklist.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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",)

Expand All @@ -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
1 change: 0 additions & 1 deletion robotools/test_worklists.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@ def test_recommended_instantiation():
warnings.simplefilter("error")
BaseWorklist()
EvoWorklist()
with pytest.raises(NotImplementedError):
FluentWorklist()
pass

Expand Down
Loading
Loading