From 68070c07fa3573b3df62a422d65e259942b5d0ad Mon Sep 17 00:00:00 2001 From: Sanniti Pimpley Date: Mon, 27 Jan 2025 15:04:25 -0500 Subject: [PATCH] feat(api): add new InstrumentContext.distribute_liquid() method (#17355) # 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. --- .../protocol_api/core/engine/instrument.py | 12 + .../opentrons/protocol_api/core/instrument.py | 17 ++ .../core/legacy/legacy_instrument_core.py | 16 +- .../legacy_instrument_core.py | 16 +- .../protocol_api/instrument_context.py | 85 ++++++ .../protocol_api/test_instrument_context.py | 264 +++++++++++++++++- 6 files changed, 403 insertions(+), 7 deletions(-) diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index 2ed46bd4b91..9ae9b349789 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/api/src/opentrons/protocol_api/core/engine/instrument.py @@ -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, diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py index 04f069e75f4..aa653dccaea 100644 --- a/api/src/opentrons/protocol_api/core/instrument.py +++ b/api/src/opentrons/protocol_api/core/instrument.py @@ -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.""" diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py index ff25a20d77c..46cf36de2e9 100644 --- a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py +++ b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py @@ -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" diff --git a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py index 4c8f68ecb86..93445f94f05 100644 --- a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py +++ b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py @@ -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" diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index b0f0f666e74..11552ef9ec3 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -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: """ diff --git a/api/tests/opentrons/protocol_api/test_instrument_context.py b/api/tests/opentrons/protocol_api/test_instrument_context.py index c73a154c937..2a279ca1ad7 100644 --- a/api/tests/opentrons/protocol_api/test_instrument_context.py +++ b/api/tests/opentrons/protocol_api/test_instrument_context.py @@ -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)), + ) + )