diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 446c3799189a..ff04c8bf650e 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -186,11 +186,12 @@ def get_minimum_liquid_sense_height(self) -> float: return self._core.get_minimum_liquid_sense_height() @requires_version(2, 0) - def aspirate( + def aspirate( # noqa: C901 self, volume: Optional[float] = None, location: Optional[Union[types.Location, labware.Well]] = None, rate: float = 1.0, + flow_rate: Optional[float] = None, ) -> InstrumentContext: """ Draw liquid into a pipette tip. @@ -227,6 +228,9 @@ def aspirate( `. If not specified, defaults to 1.0. See :ref:`new-plunger-flow-rates`. :type rate: float + :param flow_rate: The absolute flow rate in µL/s. If ``flow_rate`` is specified, + ``rate`` must not be set. + :type flow_rate: float :returns: This instance. .. note:: @@ -236,10 +240,25 @@ def aspirate( ``location``, specify it as a keyword argument: ``pipette.aspirate(location=plate['A1'])`` + .. versionchanged:: 2.24 + Added the ``flow_rate`` parameter. """ + if flow_rate is not None: + if self.api_version < APIVersion(2, 24): + raise APIVersionError( + api_element="flow_rate", + until_version="2.24", + current_version=f"{self.api_version}", + ) + if rate != 1.0: + raise ValueError("rate must not be set if flow_rate is specified") + rate = flow_rate / self._core.get_aspirate_flow_rate() + else: + flow_rate = self._core.get_aspirate_flow_rate(rate) + _log.debug( - "aspirate {} from {} at {}".format( - volume, location if location else "current position", rate + "aspirate {} from {} at {} µL/s".format( + volume, location if location else "current position", flow_rate ) ) @@ -276,7 +295,6 @@ def aspirate( c_vol = self._core.get_available_volume() if volume is None else volume else: c_vol = self._core.get_available_volume() if not volume else volume - flow_rate = self._core.get_aspirate_flow_rate(rate) if ( self.api_version >= APIVersion(2, 20) @@ -312,7 +330,7 @@ def aspirate( return self @requires_version(2, 0) - def dispense( + def dispense( # noqa: C901 self, volume: Optional[float] = None, location: Optional[ @@ -320,6 +338,7 @@ def dispense( ] = None, rate: float = 1.0, push_out: Optional[float] = None, + flow_rate: Optional[float] = None, ) -> InstrumentContext: """ Dispense liquid from a pipette tip. @@ -376,15 +395,19 @@ def dispense( `. If not specified, defaults to 1.0. See :ref:`new-plunger-flow-rates`. :type rate: float + :param push_out: Continue past the plunger bottom to help ensure all liquid leaves the tip. Measured in µL. The default value is ``None``. When not specified or set to ``None``, the plunger moves by a non-zero default amount. - For a table of default values, see :ref:`push-out-dispense`. :type push_out: float + :param flow_rate: The absolute flow rate in µL/s. If ``flow_rate`` is specified, + ``rate`` must not be set. + :type flow_rate: float + :returns: This instance. .. note:: @@ -400,6 +423,9 @@ def dispense( .. versionchanged:: 2.17 Behavior of the ``volume`` parameter. + .. versionchanged:: 2.24 + Added the ``flow_rate`` parameter. + .. versionchanged:: 2.24 ``location`` is no longer required if the pipette just moved to, dispensed, or blew out into a trash bin or waste chute. @@ -410,13 +436,27 @@ def dispense( until_version="2.15", current_version=f"{self.api_version}", ) + + if flow_rate is not None: + if self.api_version < APIVersion(2, 24): + raise APIVersionError( + api_element="flow_rate", + until_version="2.24", + current_version=f"{self.api_version}", + ) + if rate != 1.0: + raise ValueError("rate must not be set if flow_rate is specified") + rate = flow_rate / self._core.get_dispense_flow_rate() + else: + flow_rate = self._core.get_dispense_flow_rate(rate) + _log.debug( - "dispense {} from {} at {}".format( - volume, location if location else "current position", rate + "dispense {} from {} at {} µL/s".format( + volume, location if location else "current position", flow_rate ) ) - last_location = self._get_last_location_by_api_version() + last_location = self._get_last_location_by_api_version() try: target = validation.validate_location( location=location, last_location=last_location @@ -434,8 +474,6 @@ def dispense( else: c_vol = self._core.get_current_volume() if not volume else volume - flow_rate = self._core.get_dispense_flow_rate(rate) - if isinstance(target, validation.DisposalTarget): with publisher.publish_context( broker=self.broker, diff --git a/api/tests/opentrons/protocol_api/test_instrument_context.py b/api/tests/opentrons/protocol_api/test_instrument_context.py index 77279aa86618..e703a25173e6 100644 --- a/api/tests/opentrons/protocol_api/test_instrument_context.py +++ b/api/tests/opentrons/protocol_api/test_instrument_context.py @@ -469,6 +469,40 @@ def test_aspirate_raises_no_location( subject.aspirate(location=None) +def test_aspirate_flow_rate( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + subject: InstrumentContext, + mock_protocol_core: ProtocolCore, +) -> None: + """It should aspirate with absolute_flow_rate.""" + decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT) + last_location = Location(point=Point(9, 9, 9), labware=None) + decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return( + last_location + ) + decoy.when(mock_instrument_core.get_aspirate_flow_rate()).then_return(400) + + subject.aspirate(volume=30, flow_rate=600) + + decoy.verify( + mock_instrument_core.aspirate( + location=last_location, + well_core=None, + in_place=True, + volume=30, + rate=1.5, # requested flow_rate is 1.5 times default of 400 + flow_rate=600, + meniscus_tracking=None, + ), + times=1, + ) + + # Should raise if both `rate` and `flow_rate` are specified: + with pytest.raises(ValueError): + subject.aspirate(volume=30, rate=1.5, flow_rate=600) + + def test_blow_out_to_well( decoy: Decoy, mock_instrument_core: InstrumentCore, @@ -1126,6 +1160,41 @@ def test_dispense_push_out_on_not_allowed_version( subject.dispense(push_out=3) +def test_dispense_flow_rate( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + subject: InstrumentContext, + mock_protocol_core: ProtocolCore, +) -> None: + """It should dispense with absolute_flow_rate.""" + decoy.when(mock_instrument_core.get_mount()).then_return(Mount.RIGHT) + last_location = Location(point=Point(9, 9, 9), labware=None) + decoy.when(mock_protocol_core.get_last_location(Mount.RIGHT)).then_return( + last_location + ) + decoy.when(mock_instrument_core.get_dispense_flow_rate()).then_return(400) + + subject.dispense(volume=30, flow_rate=600) + + decoy.verify( + mock_instrument_core.dispense( + location=last_location, + well_core=None, + volume=30, + rate=1.5, # requested flow_rate is 1.5 times default of 400 + flow_rate=600, + in_place=True, + push_out=None, + meniscus_tracking=None, + ), + times=1, + ) + + # Should raise if both `rate` and `flow_rate` are specified: + with pytest.raises(ValueError): + subject.dispense(volume=30, rate=1.5, flow_rate=600) + + def test_touch_tip( decoy: Decoy, mock_instrument_core: InstrumentCore,