Skip to content

Commit

Permalink
refactor(api): reorganize existing transfer planner (#16724)
Browse files Browse the repository at this point in the history
Preparation for AUTH-843

# Overview

This PR moves the existing transfer planner to a new `transfers`
directory and moves some functions into a common location in
anticipation of the new `transfer_liquid` planner, which will be using
the common functions.
  • Loading branch information
sanni-t authored Nov 8, 2024
1 parent cdb909e commit 5b7be2e
Show file tree
Hide file tree
Showing 12 changed files with 218 additions and 121 deletions.
34 changes: 17 additions & 17 deletions api/src/opentrons/protocol_api/instrument_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import logging
from contextlib import ExitStack
from typing import Any, List, Optional, Sequence, Union, cast, Dict

from opentrons.protocol_engine.errors.exceptions import TipNotAttachedError
from opentrons_shared_data.errors.exceptions import (
CommandPreconditionViolated,
Expand All @@ -15,7 +16,7 @@

from opentrons.legacy_commands import publisher
from opentrons.protocols.advanced_control.mix import mix_from_kwargs
from opentrons.protocols.advanced_control import transfers
from opentrons.protocols.advanced_control.transfers import transfer as v1_transfer

from opentrons.protocols.api_support.deck_type import NoTrashDefinedError
from opentrons.protocols.api_support.types import APIVersion
Expand All @@ -38,8 +39,7 @@
from ._nozzle_layout import NozzleLayout
from . import labware, validation


AdvancedLiquidHandling = transfers.AdvancedLiquidHandling
AdvancedLiquidHandling = v1_transfer.AdvancedLiquidHandling

_DEFAULT_ASPIRATE_CLEARANCE = 1.0
_DEFAULT_DISPENSE_CLEARANCE = 1.0
Expand Down Expand Up @@ -1408,9 +1408,9 @@ def transfer( # noqa: C901
mix_strategy, mix_opts = mix_from_kwargs(kwargs)

if trash:
drop_tip = transfers.DropTipStrategy.TRASH
drop_tip = v1_transfer.DropTipStrategy.TRASH
else:
drop_tip = transfers.DropTipStrategy.RETURN
drop_tip = v1_transfer.DropTipStrategy.RETURN

new_tip = kwargs.get("new_tip")
if isinstance(new_tip, str):
Expand All @@ -1432,19 +1432,19 @@ def transfer( # noqa: C901

if blow_out and not blowout_location:
if self.current_volume:
blow_out_strategy = transfers.BlowOutStrategy.SOURCE
blow_out_strategy = v1_transfer.BlowOutStrategy.SOURCE
else:
blow_out_strategy = transfers.BlowOutStrategy.TRASH
blow_out_strategy = v1_transfer.BlowOutStrategy.TRASH
elif blow_out and blowout_location:
if blowout_location == "source well":
blow_out_strategy = transfers.BlowOutStrategy.SOURCE
blow_out_strategy = v1_transfer.BlowOutStrategy.SOURCE
elif blowout_location == "destination well":
blow_out_strategy = transfers.BlowOutStrategy.DEST
blow_out_strategy = v1_transfer.BlowOutStrategy.DEST
elif blowout_location == "trash":
blow_out_strategy = transfers.BlowOutStrategy.TRASH
blow_out_strategy = v1_transfer.BlowOutStrategy.TRASH

if new_tip != types.TransferTipPolicy.NEVER:
tr, next_tip = labware.next_available_tip(
_, next_tip = labware.next_available_tip(
self.starting_tip,
self.tip_racks,
active_channels,
Expand All @@ -1456,9 +1456,9 @@ def transfer( # noqa: C901

touch_tip = None
if kwargs.get("touch_tip"):
touch_tip = transfers.TouchTipStrategy.ALWAYS
touch_tip = v1_transfer.TouchTipStrategy.ALWAYS

default_args = transfers.Transfer()
default_args = v1_transfer.Transfer()

disposal = kwargs.get("disposal_volume")
if disposal is None:
Expand All @@ -1471,7 +1471,7 @@ def transfer( # noqa: C901
f"working volume, {max_volume}uL"
)

transfer_args = transfers.Transfer(
transfer_args = v1_transfer.Transfer(
new_tip=new_tip or default_args.new_tip,
air_gap=air_gap,
carryover=kwargs.get("carryover") or default_args.carryover,
Expand All @@ -1484,10 +1484,10 @@ def transfer( # noqa: C901
blow_out_strategy=blow_out_strategy or default_args.blow_out_strategy,
touch_tip_strategy=(touch_tip or default_args.touch_tip_strategy),
)
transfer_options = transfers.TransferOptions(
transfer_options = v1_transfer.TransferOptions(
transfer=transfer_args, mix=mix_opts
)
plan = transfers.TransferPlan(
plan = v1_transfer.TransferPlan(
volume,
source,
dest,
Expand All @@ -1500,7 +1500,7 @@ def transfer( # noqa: C901
self._execute_transfer(plan)
return self

def _execute_transfer(self, plan: transfers.TransferPlan) -> None:
def _execute_transfer(self, plan: v1_transfer.TransferPlan) -> None:
for cmd in plan:
getattr(self, cmd["method"])(*cmd["args"], **cmd["kwargs"])

Expand Down
38 changes: 38 additions & 0 deletions api/src/opentrons/protocols/advanced_control/common.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
"""Common resources for all advanced control functions."""
import enum
from typing import NamedTuple, Optional


class MixStrategy(enum.Enum):
BOTH = enum.auto()
BEFORE = enum.auto()
AFTER = enum.auto()
NEVER = enum.auto()


class MixOpts(NamedTuple):
"""
Options to customize behavior of mix.
These options will be passed to
:py:meth:`InstrumentContext.mix` when it is called during the
transfer.
"""

repetitions: Optional[int] = None
volume: Optional[float] = None
rate: Optional[float] = None


MixOpts.repetitions.__doc__ = ":py:class:`int`"
MixOpts.volume.__doc__ = ":py:class:`float`"
MixOpts.rate.__doc__ = ":py:class:`float`"


class Mix(NamedTuple):
"""
Options to control mix behavior before aspirate and after dispense.
"""

mix_before: MixOpts = MixOpts()
mix_after: MixOpts = MixOpts()
2 changes: 1 addition & 1 deletion api/src/opentrons/protocols/advanced_control/mix.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from typing import Any, Dict, Tuple

from opentrons.protocols.advanced_control.transfers import MixStrategy, Mix
from .common import MixStrategy, Mix


def mix_from_kwargs(top_kwargs: Dict[str, Any]) -> Tuple[MixStrategy, Mix]:
Expand Down
Empty file.
45 changes: 45 additions & 0 deletions api/src/opentrons/protocols/advanced_control/transfers/common.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
"""Common functions between v1 transfer and liquid-class-based transfer."""
from typing import Iterable, Generator, Tuple, TypeVar

Target = TypeVar("Target")


def check_valid_volume_parameters(
disposal_volume: float, air_gap: float, max_volume: float
) -> None:
if air_gap >= max_volume:
raise ValueError(
"The air gap must be less than the maximum volume of the pipette"
)
elif disposal_volume >= max_volume:
raise ValueError(
"The disposal volume must be less than the maximum volume of the pipette"
)
elif disposal_volume + air_gap >= max_volume:
raise ValueError(
"The sum of the air gap and disposal volume must be less than"
" the maximum volume of the pipette"
)


def expand_for_volume_constraints(
volumes: Iterable[float],
targets: Iterable[Target],
max_volume: float,
) -> Generator[Tuple[float, "Target"], None, None]:
"""Split a sequence of proposed transfers if necessary to keep each
transfer under the given max volume.
"""
# A final defense against an infinite loop.
# Raising a proper exception with a helpful message is left to calling code,
# because it has more context about what the user is trying to do.
assert max_volume > 0
for volume, target in zip(volumes, targets):
while volume > max_volume * 2:
yield max_volume, target
volume -= max_volume

if volume > max_volume:
volume /= 2
yield volume, target
yield volume, target
Original file line number Diff line number Diff line change
Expand Up @@ -9,19 +9,20 @@
Callable,
Generator,
Iterator,
Iterable,
Sequence,
Tuple,
TypedDict,
TypeAlias,
TYPE_CHECKING,
TypeVar,
)
from opentrons.protocol_api.labware import Labware, Well
from opentrons import types
from opentrons.protocols.api_support.types import APIVersion
from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType

from . import common as tx_commons
from ..common import Mix, MixOpts, MixStrategy


AdvancedLiquidHandling = Union[
Well,
Expand All @@ -44,13 +45,6 @@ class TransferStep(TypedDict):
"""The version after which partial tip support and nozzle maps were made available."""


class MixStrategy(enum.Enum):
BOTH = enum.auto()
BEFORE = enum.auto()
AFTER = enum.auto()
NEVER = enum.auto()


class DropTipStrategy(enum.Enum):
TRASH = enum.auto()
RETURN = enum.auto()
Expand Down Expand Up @@ -239,34 +233,6 @@ class PickUpTipOpts(NamedTuple):
PickUpTipOpts.increment.__doc__ = ":py:class:`int`"


class MixOpts(NamedTuple):
"""
Options to customize behavior of mix.
These options will be passed to
:py:meth:`InstrumentContext.mix` when it is called during the
transfer.
"""

repetitions: Optional[int] = None
volume: Optional[float] = None
rate: Optional[float] = None


MixOpts.repetitions.__doc__ = ":py:class:`int`"
MixOpts.volume.__doc__ = ":py:class:`float`"
MixOpts.rate.__doc__ = ":py:class:`float`"


class Mix(NamedTuple):
"""
Options to control mix behavior before aspirate and after dispense.
"""

mix_before: MixOpts = MixOpts()
mix_after: MixOpts = MixOpts()


Mix.mix_before.__doc__ = """
Options applied to mix before aspirate.
See :py:class:`.Mix.MixOpts`.
Expand Down Expand Up @@ -534,12 +500,12 @@ def _plan_transfer(self) -> Generator[TransferStep, None, None]:
"""
# reform source target lists
sources, dests = self._extend_source_target_lists(self._sources, self._dests)
self._check_valid_volume_parameters(
tx_commons.check_valid_volume_parameters(
disposal_volume=self._strategy.disposal_volume,
air_gap=self._strategy.air_gap,
max_volume=self._instr.max_volume,
)
plan_iter = self._expand_for_volume_constraints(
plan_iter = tx_commons.expand_for_volume_constraints(
self._volumes,
zip(sources, dests),
self._instr.max_volume
Expand Down Expand Up @@ -626,7 +592,7 @@ def _plan_distribute(self) -> Generator[TransferStep, None, None]:
"""

self._check_valid_volume_parameters(
tx_commons.check_valid_volume_parameters(
disposal_volume=self._strategy.disposal_volume,
air_gap=self._strategy.air_gap,
max_volume=self._instr.max_volume,
Expand All @@ -637,7 +603,7 @@ def _plan_distribute(self) -> Generator[TransferStep, None, None]:
# recommend users to specify a disposal vol when using distribute.
# First method keeps distribute consistent with current behavior while
# the other maintains consistency in default behaviors of all functions
plan_iter = self._expand_for_volume_constraints(
plan_iter = tx_commons.expand_for_volume_constraints(
self._volumes,
self._dests,
# todo(mm, 2021-03-09): Is it right for this to be
Expand Down Expand Up @@ -686,29 +652,6 @@ def _plan_distribute(self) -> Generator[TransferStep, None, None]:
)
yield from self._new_tip_action()

Target = TypeVar("Target")

@staticmethod
def _expand_for_volume_constraints(
volumes: Iterable[float], targets: Iterable[Target], max_volume: float
) -> Generator[Tuple[float, "Target"], None, None]:
"""Split a sequence of proposed transfers if necessary to keep each
transfer under the given max volume.
"""
# A final defense against an infinite loop.
# Raising a proper exception with a helpful message is left to calling code,
# because it has more context about what the user is trying to do.
assert max_volume > 0
for volume, target in zip(volumes, targets):
while volume > max_volume * 2:
yield max_volume, target
volume -= max_volume

if volume > max_volume:
volume /= 2
yield volume, target
yield volume, target

def _plan_consolidate(self) -> Generator[TransferStep, None, None]:
"""
* **Source/ Dest:** Many sources to one destination
Expand Down Expand Up @@ -752,7 +695,7 @@ def _plan_consolidate(self) -> Generator[TransferStep, None, None]:
# air_gap=self._strategy.air_gap,
# max_volume=self._instr.max_volume,
# )
plan_iter = self._expand_for_volume_constraints(
plan_iter = tx_commons.expand_for_volume_constraints(
# todo(mm, 2021-03-09): Is it right to use _instr.max_volume here?
# Why don't we account for tip max volume, disposal volume, or air
# gap?
Expand Down Expand Up @@ -929,8 +872,8 @@ def _create_volume_list(
)
return volume

@staticmethod
def _create_volume_gradient(
self,
min_v: float,
max_v: float,
total: int,
Expand All @@ -947,22 +890,6 @@ def _map_volume(i: int) -> float:

return [_map_volume(i) for i in range(total)]

def _check_valid_volume_parameters(
self, disposal_volume: float, air_gap: float, max_volume: float
) -> None:
if air_gap >= max_volume:
raise ValueError(
"The air gap must be less than the maximum volume of the pipette"
)
elif disposal_volume >= max_volume:
raise ValueError(
"The disposal volume must be less than the maximum volume of the pipette"
)
elif disposal_volume + air_gap >= max_volume:
raise ValueError(
"The sum of the air gap and disposal volume must be less than the maximum volume of the pipette"
)

def _check_valid_well_list(
self, well_list: List[Any], id: str, old_well_list: List[Any]
) -> None:
Expand Down
10 changes: 9 additions & 1 deletion api/src/opentrons/types.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,15 @@
from __future__ import annotations
import enum
from math import sqrt, isclose
from typing import TYPE_CHECKING, Any, NamedTuple, Iterator, Union, List, Optional
from typing import (
TYPE_CHECKING,
Any,
NamedTuple,
Iterator,
Union,
List,
Optional,
)

from opentrons_shared_data.robot.types import RobotType

Expand Down
Loading

0 comments on commit 5b7be2e

Please sign in to comment.