diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 3239a6dac33..07d5c1b939e 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -1,7 +1,7 @@ from __future__ import annotations import logging from contextlib import ExitStack -from typing import Any, List, Optional, Sequence, Union, cast, Dict +from typing import Any, List, Optional, Sequence, Union, cast from opentrons_shared_data.errors.exceptions import ( CommandPreconditionViolated, CommandParameterLimitViolated, @@ -12,7 +12,10 @@ from opentrons.legacy_broker import LegacyBroker from opentrons.hardware_control.dev_types import PipetteDict from opentrons import types -from opentrons.legacy_commands import commands as cmds +from opentrons.legacy_commands import ( + commands as cmds, + protocol_commands as protocol_cmds, +) from opentrons.legacy_commands import publisher from opentrons.protocols.advanced_control.mix import mix_from_kwargs @@ -477,12 +480,14 @@ def dispense( return self @requires_version(2, 0) - def mix( + def mix( # noqa: C901 self, repetitions: int = 1, volume: Optional[float] = None, location: Optional[Union[types.Location, labware.Well]] = None, rate: float = 1.0, + aspirate_delay: Optional[float] = None, + dispense_delay: Optional[float] = None, ) -> InstrumentContext: """ Mix a volume of liquid by repeatedly aspirating and dispensing it in a single location. @@ -507,6 +512,8 @@ def mix( dispensing flow rate is calculated as ``rate`` multiplied by :py:attr:`flow_rate.dispense `. See :ref:`new-plunger-flow-rates`. + :param aspirate_delay: How long to wait after each aspirate in the mix, in seconds. + :param dispense_delay: How long to wait after each dispense in the mix, in seconds. :raises: ``UnexpectedTipRemovalError`` -- If no tip is attached to the pipette. :returns: This instance. @@ -519,6 +526,8 @@ def mix( .. versionchanged:: 2.21 Does not repeatedly check for liquid presence. + .. versionchanged:: 2.24 + Adds the ``aspirate_delay`` and ``dispense_delay`` parameters. """ _log.debug( "mixing {}uL with {} repetitions in {} at rate={}".format( @@ -533,9 +542,43 @@ def mix( else: c_vol = self._core.get_available_volume() if not volume else volume - dispense_kwargs: Dict[str, Any] = {} - if self.api_version >= APIVersion(2, 16): - dispense_kwargs["push_out"] = 0.0 + if aspirate_delay and self.api_version < APIVersion(2, 24): + raise APIVersionError( + api_element="aspirate_delay", + until_version="2.24", + current_version=f"{self._api_version}", + ) + if dispense_delay and self.api_version < APIVersion(2, 24): + raise APIVersionError( + api_element="dispense_delay", + until_version="2.24", + current_version=f"{self._api_version}", + ) + + def delay_with_publish(seconds: float) -> None: + # We don't have access to ProtocolContext.delay() which would automatically + # publish a message to the broker, so we have to do it manually: + with publisher.publish_context( + broker=self.broker, + command=protocol_cmds.delay(seconds=seconds, minutes=0, msg=None), + ): + self._protocol_core.delay(seconds=seconds, msg=None) + + def aspirate_with_delay( + location: Optional[types.Location | labware.Well], + ) -> None: + self.aspirate(volume, location, rate) + if aspirate_delay: + delay_with_publish(aspirate_delay) + + def dispense_with_delay(push_out: Optional[float]) -> None: + # protocol_api_old/test_context.py does not allow push_out at all, even if + # it's set to None, so we have to hide the argument to make the test pass. + # I don't know if the test is even valid, but I'm afraid to change the test. + dispense_kwargs = {"push_out": push_out} if push_out is not None else {} + self.dispense(volume, None, rate, **dispense_kwargs) + if dispense_delay: + delay_with_publish(dispense_delay) with publisher.publish_context( broker=self.broker, @@ -546,13 +589,19 @@ def mix( location=location, ), ): - self.aspirate(volume, location, rate) + aspirate_with_delay(location=location) with AutoProbeDisable(self): while repetitions - 1 > 0: - self.dispense(volume, rate=rate, **dispense_kwargs) - self.aspirate(volume, rate=rate) + # starting in 2.16, we disable push_out on all but the last + # dispense() to prevent the tip from jumping out of the liquid + # during the mix (PR #14004): + dispense_with_delay( + push_out=0 if self.api_version >= APIVersion(2, 16) else None + ) + # aspirate location was set above, do subsequent aspirates in-place: + aspirate_with_delay(location=None) repetitions -= 1 - self.dispense(volume, rate=rate) + dispense_with_delay(push_out=None) return self @requires_version(2, 0) diff --git a/api/src/opentrons/protocols/api_support/definitions.py b/api/src/opentrons/protocols/api_support/definitions.py index a353e1d49fe..b96440a3295 100644 --- a/api/src/opentrons/protocols/api_support/definitions.py +++ b/api/src/opentrons/protocols/api_support/definitions.py @@ -1,6 +1,6 @@ from .types import APIVersion -MAX_SUPPORTED_VERSION = APIVersion(2, 23) +MAX_SUPPORTED_VERSION = APIVersion(2, 24) """The maximum supported protocol API version in this release.""" MIN_SUPPORTED_VERSION = APIVersion(2, 0) diff --git a/api/tests/opentrons/protocol_api/test_instrument_context.py b/api/tests/opentrons/protocol_api/test_instrument_context.py index 78e1101a1ee..ab9c7b97297 100644 --- a/api/tests/opentrons/protocol_api/test_instrument_context.py +++ b/api/tests/opentrons/protocol_api/test_instrument_context.py @@ -1583,6 +1583,76 @@ def test_mix_with_lpd( ) +def test_mix_with_delay( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + subject: InstrumentContext, + mock_protocol_core: ProtocolCore, +) -> None: + """It should delay after the aspirate/dispense in a mix.""" + mock_well = decoy.mock(cls=Well) + input_location = Location(point=Point(2, 2, 2), labware=mock_well) + decoy.when(mock_protocol_core.get_last_location(Mount.LEFT)).then_return( + input_location, + ) # last location same as input_location, so in_place should be true + decoy.when(mock_instrument_core.get_aspirate_flow_rate(1)).then_return(4.56) + decoy.when(mock_instrument_core.get_dispense_flow_rate(1)).then_return(5.67) + decoy.when(mock_instrument_core.has_tip()).then_return(True) + decoy.when(mock_instrument_core.get_current_volume()).then_return(0.0) + + subject.mix( + repetitions=2, + volume=10.0, + location=input_location, + aspirate_delay=3, + dispense_delay=4, + ) + decoy.verify( + mock_instrument_core.aspirate( + location=input_location, + well_core=mock_well._core, + volume=10.0, + rate=1, + flow_rate=4.56, + in_place=True, + meniscus_tracking=None, + ), + mock_protocol_core.delay(3, msg=None), # aspirate delay + mock_instrument_core.dispense( + location=input_location, + well_core=mock_well._core, + volume=10.0, + rate=1, + flow_rate=5.67, + in_place=True, + push_out=0.0, + meniscus_tracking=None, + ), + mock_protocol_core.delay(4, msg=None), # dispense delay + mock_instrument_core.aspirate( + location=input_location, + well_core=mock_well._core, + volume=10.0, + rate=1, + flow_rate=4.56, + in_place=True, + meniscus_tracking=None, + ), + mock_protocol_core.delay(3, msg=None), # aspirate delay + mock_instrument_core.dispense( + location=input_location, + well_core=mock_well._core, + volume=10.0, + rate=1, + flow_rate=5.67, + in_place=True, + push_out=None, + meniscus_tracking=None, + ), + mock_protocol_core.delay(4, msg=None), # dispense delay + ) + + @pytest.mark.ot3_only @pytest.mark.parametrize("clean,expected", [(True, 1), (False, 0)]) def test_aspirate_with_lpd(