Skip to content

Commit

Permalink
feat(api): add new InstrumentContext.distribute_liquid() method (#17355)
Browse files Browse the repository at this point in the history
# Overview

Adds a new `InstrumentContext.distribute_liquid()` method that takes in
a single source and multiple destinations for liquid distribution. This
method does argument and state verification before delegating the
implementation to the cores.

## Risk assessment

None. Adds a new feature that doesn't affect anything else.
sanni-t authored Jan 27, 2025
1 parent 75b5d48 commit 68070c0
Showing 6 changed files with 403 additions and 7 deletions.
12 changes: 12 additions & 0 deletions api/src/opentrons/protocol_api/core/engine/instrument.py
Original file line number Diff line number Diff line change
@@ -1098,6 +1098,18 @@ def _pick_up_tip() -> None:
if new_tip != TransferTipPolicyV2.NEVER:
_drop_tip()

def distribute_liquid(
self,
liquid_class: LiquidClass,
volume: float,
source: Tuple[Location, WellCore],
dest: List[Tuple[Location, WellCore]],
new_tip: TransferTipPolicyV2,
tip_racks: List[Tuple[Location, LabwareCore]],
trash_location: Union[Location, TrashBin, WasteChute],
) -> None:
pass

def _get_location_and_well_core_from_next_tip_info(
self,
tip_info: NextTipInfo,
17 changes: 17 additions & 0 deletions api/src/opentrons/protocol_api/core/instrument.py
Original file line number Diff line number Diff line change
@@ -326,6 +326,23 @@ def transfer_liquid(
"""Transfer a liquid from source to dest according to liquid class properties."""
...

@abstractmethod
def distribute_liquid(
self,
liquid_class: LiquidClass,
volume: float,
source: Tuple[types.Location, WellCoreType],
dest: List[Tuple[types.Location, WellCoreType]],
new_tip: TransferTipPolicyV2,
tip_racks: List[Tuple[types.Location, LabwareCoreType]],
trash_location: Union[types.Location, TrashBin, WasteChute],
) -> None:
"""
Distribute a liquid from single source to multiple destinations
according to liquid class properties.
"""
...

@abstractmethod
def is_tip_tracking_available(self) -> bool:
"""Return whether auto tip tracking is available for the pipette's current nozzle configuration."""
Original file line number Diff line number Diff line change
@@ -570,10 +570,22 @@ def transfer_liquid(
tip_racks: List[Tuple[types.Location, LegacyLabwareCore]],
trash_location: Union[types.Location, TrashBin, WasteChute],
) -> None:
"""This will never be called because it was added in .."""
# TODO(spp, 2024-11-20): update the docstring and error to include API version
"""This will never be called because it was added in API 2.23"""
assert False, "transfer_liquid is not supported in legacy context"

def distribute_liquid(
self,
liquid_class: LiquidClass,
volume: float,
source: Tuple[types.Location, LegacyWellCore],
dest: List[Tuple[types.Location, LegacyWellCore]],
new_tip: TransferTipPolicyV2,
tip_racks: List[Tuple[types.Location, LegacyLabwareCore]],
trash_location: Union[types.Location, TrashBin, WasteChute],
) -> None:
"""This will never be called because it was added in API 2.23"""
assert False, "distribute_liquid is not supported in legacy context"

def get_active_channels(self) -> int:
"""This will never be called because it was added in API 2.16."""
assert False, "get_active_channels only supported in API 2.16 & later"
Original file line number Diff line number Diff line change
@@ -490,10 +490,22 @@ def transfer_liquid(
tip_racks: List[Tuple[types.Location, LegacyLabwareCore]],
trash_location: Union[types.Location, TrashBin, WasteChute],
) -> None:
"""Transfer a liquid from source to dest according to liquid class properties."""
# TODO(spp, 2024-11-20): update the docstring and error to include API version
"""This will never be called because it was added in API 2.23."""
assert False, "transfer_liquid is not supported in legacy context"

def distribute_liquid(
self,
liquid_class: LiquidClass,
volume: float,
source: Tuple[types.Location, LegacyWellCore],
dest: List[Tuple[types.Location, LegacyWellCore]],
new_tip: TransferTipPolicyV2,
tip_racks: List[Tuple[types.Location, LegacyLabwareCore]],
trash_location: Union[types.Location, TrashBin, WasteChute],
) -> None:
"""This will never be called because it was added in API 2.23."""
assert False, "distribute_liquid is not supported in legacy context"

def get_active_channels(self) -> int:
"""This will never be called because it was added in API 2.16."""
assert False, "get_active_channels only supported in API 2.16 & later"
85 changes: 85 additions & 0 deletions api/src/opentrons/protocol_api/instrument_context.py
Original file line number Diff line number Diff line change
@@ -1604,6 +1604,91 @@ def transfer_liquid(
)
return self

def distribute_liquid(
self,
liquid_class: LiquidClass,
volume: float,
source: labware.Well,
dest: Union[
labware.Well, Sequence[labware.Well], Sequence[Sequence[labware.Well]]
],
new_tip: TransferTipPolicyV2Type = "once",
trash_location: Optional[
Union[types.Location, labware.Well, TrashBin, WasteChute]
] = None,
) -> InstrumentContext:
"""
Distribute liquid from a single source to multiple destinations
using the specified liquid class properties.
TODO: Add args description.
"""
if not feature_flags.allow_liquid_classes(
robot_type=RobotTypeEnum.robot_literal_to_enum(
self._protocol_core.robot_type
)
):
raise NotImplementedError("This method is not implemented.")

if not isinstance(source, labware.Well):
raise ValueError(f"Source should be a single Well but received {source}.")
flat_dests_list = validation.ensure_valid_flat_wells_list_for_transfer_v2(dest)
for well in [source] + flat_dests_list:
instrument.validate_takes_liquid(
location=well.top(),
reject_module=True,
reject_adapter=True,
)

valid_new_tip = validation.ensure_new_tip_policy(new_tip)
if valid_new_tip == TransferTipPolicyV2.NEVER:
if self._last_tip_picked_up_from is None:
raise RuntimeError(
"Pipette has no tip attached to perform transfer."
" Either do a pick_up_tip beforehand or specify a new_tip parameter"
" of 'once' or 'always'."
)
else:
tip_racks = [self._last_tip_picked_up_from.parent]
else:
tip_racks = self._tip_racks
if self.current_volume != 0:
raise RuntimeError(
"A transfer on a liquid class cannot start with liquid already in the tip."
" Ensure that all previously aspirated liquid is dispensed before starting"
" a new transfer."
)

_trash_location: Union[types.Location, labware.Well, TrashBin, WasteChute]
if trash_location is None:
saved_trash = self.trash_container
if isinstance(saved_trash, labware.Labware):
_trash_location = saved_trash.wells()[0]
else:
_trash_location = saved_trash
else:
_trash_location = trash_location

checked_trash_location = validation.ensure_valid_trash_location_for_transfer_v2(
trash_location=_trash_location
)
self._core.distribute_liquid(
liquid_class=liquid_class,
volume=volume,
source=(types.Location(types.Point(), labware=source), source._core),
dest=[
(types.Location(types.Point(), labware=well), well._core)
for well in flat_dests_list
],
new_tip=valid_new_tip,
tip_racks=[
(types.Location(types.Point(), labware=rack), rack._core)
for rack in tip_racks
],
trash_location=checked_trash_location,
)
return self

@requires_version(2, 0)
def delay(self, *args: Any, **kwargs: Any) -> None:
"""
264 changes: 261 additions & 3 deletions api/tests/opentrons/protocol_api/test_instrument_context.py
Original file line number Diff line number Diff line change
@@ -1799,7 +1799,7 @@ def test_transfer_liquid_raises_for_non_liquid_handling_locations(
robot_type: RobotType,
minimal_liquid_class_def2: LiquidClassSchemaV1,
) -> None:
"""It should raise errors if source and destination are not of same length."""
"""It should raise errors if source or dest are invalid for liquid handling."""
test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
mock_well = decoy.mock(cls=Well)
decoy.when(mock_protocol_core.robot_type).then_return(robot_type)
@@ -1894,7 +1894,7 @@ def test_transfer_liquid_raises_if_tip_has_liquid(
robot_type: RobotType,
minimal_liquid_class_def2: LiquidClassSchemaV1,
) -> None:
"""It should raise errors if there is no tip attached."""
"""It should raise errors if tip has liquid before starting transfer."""
test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
mock_well = decoy.mock(cls=Well)
tip_racks = [decoy.mock(cls=Labware)]
@@ -1943,7 +1943,7 @@ def test_transfer_liquid_delegates_to_engine_core(
robot_type: RobotType,
minimal_liquid_class_def2: LiquidClassSchemaV1,
) -> None:
"""It should load liquid class into engine and delegate the transfer execution to core."""
"""It should delegate the transfer execution to core."""
test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
mock_well = decoy.mock(cls=Well)
tip_racks = [decoy.mock(cls=Labware)]
@@ -1989,3 +1989,261 @@ def test_transfer_liquid_delegates_to_engine_core(
trash_location=trash_location.move(Point(1, 2, 3)),
)
)


@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"])
def test_distribute_liquid_raises_for_invalid_locations(
decoy: Decoy,
mock_protocol_core: ProtocolCore,
subject: InstrumentContext,
mock_feature_flags: None,
robot_type: RobotType,
minimal_liquid_class_def2: LiquidClassSchemaV1,
) -> None:
"""It should raise errors if source or destination is invalid."""
test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
mock_well = decoy.mock(cls=Well)
decoy.when(mock_protocol_core.robot_type).then_return(robot_type)
decoy.when(
ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type))
).then_return(True)
decoy.when(
mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([[mock_well]])
).then_raise(ValueError("Oh no"))
with pytest.raises(ValueError):
subject.distribute_liquid(
liquid_class=test_liq_class,
volume=10,
source=mock_well,
dest=[[mock_well]],
)
with pytest.raises(ValueError, match="Source should be a single Well"):
subject.distribute_liquid(
liquid_class=test_liq_class,
volume=10,
source="abc", # type: ignore
dest=[mock_well],
)


@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"])
def test_distribute_liquid_raises_if_more_than_one_source(
decoy: Decoy,
mock_protocol_core: ProtocolCore,
subject: InstrumentContext,
mock_feature_flags: None,
robot_type: RobotType,
minimal_liquid_class_def2: LiquidClassSchemaV1,
) -> None:
"""It should raise error if source is more than one well."""
test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
mock_well = decoy.mock(cls=Well)
decoy.when(mock_protocol_core.robot_type).then_return(robot_type)
decoy.when(
ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type))
).then_return(True)
with pytest.raises(ValueError, match="Source should be a single Well"):
subject.distribute_liquid(
liquid_class=test_liq_class, volume=10, source=[mock_well, mock_well], dest=[mock_well] # type: ignore
)


@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"])
def test_distribute_liquid_raises_for_non_liquid_handling_locations(
decoy: Decoy,
mock_protocol_core: ProtocolCore,
subject: InstrumentContext,
mock_feature_flags: None,
robot_type: RobotType,
minimal_liquid_class_def2: LiquidClassSchemaV1,
) -> None:
"""It should raise errors if source or dest are invalid for liquid handling."""
test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
mock_well = decoy.mock(cls=Well)
decoy.when(mock_protocol_core.robot_type).then_return(robot_type)
decoy.when(
ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type))
).then_return(True)
decoy.when(
mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well])
).then_return([mock_well])
decoy.when(
mock_instrument_support.validate_takes_liquid(
mock_well.top(), reject_module=True, reject_adapter=True
)
).then_raise(ValueError("Uh oh"))
with pytest.raises(ValueError, match="Uh oh"):
subject.distribute_liquid(
liquid_class=test_liq_class, volume=10, source=mock_well, dest=[mock_well]
)


@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"])
def test_distribute_liquid_raises_for_bad_tip_policy(
decoy: Decoy,
mock_protocol_core: ProtocolCore,
subject: InstrumentContext,
mock_feature_flags: None,
robot_type: RobotType,
minimal_liquid_class_def2: LiquidClassSchemaV1,
) -> None:
"""It should raise errors if new_tip is invalid."""
test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
mock_well = decoy.mock(cls=Well)
decoy.when(mock_protocol_core.robot_type).then_return(robot_type)
decoy.when(
ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type))
).then_return(True)
decoy.when(
mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well])
).then_return([mock_well])
decoy.when(mock_validation.ensure_new_tip_policy("once")).then_raise(
ValueError("Uh oh")
)
with pytest.raises(ValueError, match="Uh oh"):
subject.distribute_liquid(
liquid_class=test_liq_class,
volume=10,
source=mock_well,
dest=[mock_well],
new_tip="once",
)


@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"])
def test_distribute_liquid_raises_for_no_tip(
decoy: Decoy,
mock_protocol_core: ProtocolCore,
subject: InstrumentContext,
mock_feature_flags: None,
robot_type: RobotType,
minimal_liquid_class_def2: LiquidClassSchemaV1,
) -> None:
"""It should raise errors if there is no tip attached."""
test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
mock_well = decoy.mock(cls=Well)
decoy.when(mock_protocol_core.robot_type).then_return(robot_type)
decoy.when(
ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type))
).then_return(True)
decoy.when(
mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well])
).then_return([mock_well])
decoy.when(mock_validation.ensure_new_tip_policy("never")).then_return(
TransferTipPolicyV2.NEVER
)
with pytest.raises(RuntimeError, match="Pipette has no tip"):
subject.distribute_liquid(
liquid_class=test_liq_class,
volume=10,
source=mock_well,
dest=[mock_well],
new_tip="never",
)


@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"])
def test_distribute_liquid_raises_if_tip_has_liquid(
decoy: Decoy,
mock_protocol_core: ProtocolCore,
mock_instrument_core: InstrumentCore,
subject: InstrumentContext,
mock_feature_flags: None,
robot_type: RobotType,
minimal_liquid_class_def2: LiquidClassSchemaV1,
) -> None:
"""It should raise errors if the tip has liquid at the start of distribution."""
test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
mock_well = decoy.mock(cls=Well)
tip_racks = [decoy.mock(cls=Labware)]

subject.starting_tip = None
subject.tip_racks = tip_racks

decoy.when(mock_protocol_core.robot_type).then_return(robot_type)
decoy.when(
ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type))
).then_return(True)
decoy.when(
mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well])
).then_return([mock_well])
decoy.when(mock_validation.ensure_new_tip_policy("never")).then_return(
TransferTipPolicyV2.ONCE
)
decoy.when(mock_instrument_core.get_nozzle_map()).then_return(MOCK_MAP)
decoy.when(mock_instrument_core.get_active_channels()).then_return(2)
decoy.when(
labware.next_available_tip(
starting_tip=None,
tip_racks=tip_racks,
channels=2,
nozzle_map=MOCK_MAP,
)
).then_return((decoy.mock(cls=Labware), decoy.mock(cls=Well)))
decoy.when(mock_instrument_core.get_current_volume()).then_return(1000)
with pytest.raises(RuntimeError, match="liquid already in the tip"):
subject.distribute_liquid(
liquid_class=test_liq_class,
volume=10,
source=mock_well,
dest=[mock_well],
new_tip="never",
)


@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"])
def test_distribute_liquid_delegates_to_engine_core(
decoy: Decoy,
mock_protocol_core: ProtocolCore,
mock_instrument_core: InstrumentCore,
subject: InstrumentContext,
mock_feature_flags: None,
robot_type: RobotType,
minimal_liquid_class_def2: LiquidClassSchemaV1,
) -> None:
"""It should delegate the execution to core."""
test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
mock_well = decoy.mock(cls=Well)
tip_racks = [decoy.mock(cls=Labware)]
trash_location = Location(point=Point(1, 2, 3), labware=mock_well)
next_tiprack = decoy.mock(cls=Labware)
subject.starting_tip = None
subject._tip_racks = tip_racks

decoy.when(mock_protocol_core.robot_type).then_return(robot_type)
decoy.when(
ff.allow_liquid_classes(RobotTypeEnum.robot_literal_to_enum(robot_type))
).then_return(True)
decoy.when(
mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well])
).then_return([mock_well])
decoy.when(mock_validation.ensure_new_tip_policy("never")).then_return(
TransferTipPolicyV2.ONCE
)
decoy.when(mock_instrument_core.get_nozzle_map()).then_return(MOCK_MAP)
decoy.when(mock_instrument_core.get_active_channels()).then_return(2)
decoy.when(mock_instrument_core.get_current_volume()).then_return(0)
decoy.when(
mock_validation.ensure_valid_trash_location_for_transfer_v2(trash_location)
).then_return(trash_location.move(Point(1, 2, 3)))
decoy.when(next_tiprack.uri).then_return("tiprack-uri")
decoy.when(mock_instrument_core.get_pipette_name()).then_return("pipette-name")
subject.distribute_liquid(
liquid_class=test_liq_class,
volume=10,
source=mock_well,
dest=[mock_well],
new_tip="never",
trash_location=trash_location,
)
decoy.verify(
mock_instrument_core.distribute_liquid(
liquid_class=test_liq_class,
volume=10,
source=(Location(Point(), labware=mock_well), mock_well._core),
dest=[(Location(Point(), labware=mock_well), mock_well._core)],
new_tip=TransferTipPolicyV2.ONCE,
tip_racks=[(Location(Point(), labware=tip_racks[0]), tip_racks[0]._core)],
trash_location=trash_location.move(Point(1, 2, 3)),
)
)

0 comments on commit 68070c0

Please sign in to comment.