From 279a91340f2242ce8176b9e86e938438572429df Mon Sep 17 00:00:00 2001 From: Sanniti Date: Thu, 2 Jan 2025 01:02:34 -0500 Subject: [PATCH] added tests --- .../protocol_api/core/engine/instrument.py | 6 +- .../engine/transfer_components_executor.py | 128 ++++---- api/tests/opentrons/conftest.py | 2 +- .../core/engine/test_instrument_core.py | 77 ++++- .../test_transfer_components_executor.py | 300 +++++++++++++++++- 5 files changed, 447 insertions(+), 66 deletions(-) diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index 5bec152b857..b67ecf17bac 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/api/src/opentrons/protocol_api/core/engine/instrument.py @@ -1013,7 +1013,7 @@ def aspirate_liquid_class( liquid=0, air_gap=0, ) - components_executer = tx_comps_executor.TransferComponentsExecutor( + components_executor = tx_comps_executor.TransferComponentsExecutor( instrument_core=self, transfer_properties=transfer_properties, target_location=aspirate_location, @@ -1040,10 +1040,10 @@ def dispense_liquid_class( self, volume: float, dest: Tuple[Location, WellCore], + source: Optional[Tuple[Location, WellCore]], transfer_properties: TransferProperties, transfer_type: tx_comps_executor.TransferType, tip_contents: List[tx_comps_executor.LiquidAndAirGapPair], - source: Optional[Tuple[Location, WellCore]], trash_location: Union[Location, TrashBin, WasteChute], ) -> tx_comps_executor.LiquidAndAirGapPair: """Execute single-dispense steps. @@ -1092,7 +1092,7 @@ def dispense_liquid_class( liquid=0, air_gap=0, ) - components_executor = tx_comps_executor.get_transfer_components_executor( + components_executor = tx_comps_executor.TransferComponentsExecutor( instrument_core=self, transfer_properties=transfer_properties, target_location=dispense_location, diff --git a/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py b/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py index e564fb7df4a..209599510fc 100644 --- a/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py +++ b/api/src/opentrons/protocol_api/core/engine/transfer_components_executor.py @@ -49,6 +49,7 @@ class TipState: """ ready_to_aspirate: bool = True + # TODO: maybe use the tip contents from engine state instead. last_liquid_and_air_gap_in_tip: LiquidAndAirGapPair = LiquidAndAirGapPair( liquid=0, air_gap=0, @@ -84,20 +85,6 @@ def remove_air_gap(self, volume: float) -> None: ), "Last air gap volume doe not match the volume being removed" self.last_liquid_and_air_gap_in_tip.air_gap = 0 - def update( - self, - liquid: Optional[float] = None, - air_gap: Optional[float] = None, - ready_to_aspirate: Optional[bool] = None, - ) -> None: - """Update the tip state contents with given values.""" - if liquid is not None: - self.last_liquid_and_air_gap_in_tip.liquid = liquid - if air_gap is not None: - self.last_liquid_and_air_gap_in_tip.air_gap = air_gap - if ready_to_aspirate is not None: - self.ready_to_aspirate = ready_to_aspirate - class TransferType(Enum): ONE_TO_ONE = "one_to_one" @@ -333,7 +320,7 @@ def retract_after_aspiration(self, volume: float) -> None: ) ) - def retract_after_dispensing( # noqa: C901 + def retract_after_dispensing( self, trash_location: Union[Location, TrashBin, WasteChute], source_location: Optional[Location], @@ -385,42 +372,6 @@ def retract_after_dispensing( # noqa: C901 assert retract_delay.duration is not None self._instrument.delay(retract_delay.duration) - def _do_touch_tip_and_air_gap() -> None: - touch_tip_props = retract_props.touch_tip - if touch_tip_props.enabled: - assert ( - touch_tip_props.speed is not None - and touch_tip_props.z_offset is not None - and touch_tip_props.mm_to_edge is not None - ) - # TODO: update this once touch tip has mmToEdge - self._instrument.touch_tip( - location=retract_location, - well_core=self._target_well, - radius=1, - z_offset=touch_tip_props.z_offset, - speed=touch_tip_props.speed, - ) - self._instrument.move_to( - location=retract_location, - well_core=self._target_well, - force_direct=True, - minimum_z_height=None, - # Full speed because the tip will already be out of the liquid - speed=None, - ) - - if self._transfer_type != TransferType.ONE_TO_MANY: - # TODO: check if it is okay to just do `prepare_to_aspirate` unconditionally - if not self._tip_state.ready_to_aspirate: - self._instrument.prepare_to_aspirate() - self._tip_state.ready_to_aspirate = True - self._add_air_gap( - air_gap_volume=self._transfer_properties.aspirate.retract.air_gap_by_volume.get_for_volume( - 0 - ) - ) - blowout_props = retract_props.blowout if ( blowout_props.enabled @@ -434,32 +385,97 @@ def _do_touch_tip_and_air_gap() -> None: in_place=True, ) self._tip_state.ready_to_aspirate = False - _do_touch_tip_and_air_gap() + self._do_touch_tip_and_air_gap( + location=retract_location, well=self._target_well + ) if ( blowout_props.enabled - and not blowout_props.location == BlowoutLocation.DESTINATION + and blowout_props.location != BlowoutLocation.DESTINATION ): assert blowout_props.flow_rate is not None self._instrument.set_flow_rate(blow_out=blowout_props.flow_rate) + touch_tip_and_air_gap_location: Optional[Location] if blowout_props.location == BlowoutLocation.SOURCE: if source_location is None: raise RuntimeError( - "Blowout location is source but source location is not provided." + "Blowout location is 'source' but source location is not provided." ) self._instrument.blow_out( location=source_location, well_core=source_well, in_place=False, ) + touch_tip_and_air_gap_location = source_location + touch_tip_and_air_gap_well = source_well else: self._instrument.blow_out( location=trash_location, - well_core=None, # Update this to correct well core + well_core=None, # TODO: Update this to correct well core in_place=False, ) + touch_tip_and_air_gap_location = ( + trash_location if isinstance(trash_location, Location) else None + ) + touch_tip_and_air_gap_well = ( + None # TODO: Update this to correct well core + ) + last_air_gap = self._tip_state.last_liquid_and_air_gap_in_tip.air_gap + self._tip_state.remove_air_gap(last_air_gap) self._tip_state.ready_to_aspirate = False - _do_touch_tip_and_air_gap() + self._do_touch_tip_and_air_gap( + location=touch_tip_and_air_gap_location, + well=touch_tip_and_air_gap_well, + ) + + def _do_touch_tip_and_air_gap( + self, + location: Optional[Location], + well: Optional[WellCore], + ) -> None: + """Perform touch tip and air gap as part of post-dispense retract.""" + touch_tip_props = self._transfer_properties.dispense.retract.touch_tip + if touch_tip_props.enabled: + assert ( + touch_tip_props.speed is not None + and touch_tip_props.z_offset is not None + and touch_tip_props.mm_to_edge is not None + ) + # TODO: update this once touch tip has mmToEdge + # Also, check that when blow out is a non-dest-well, + # whether the touch tip params from transfer props should be used for + # both dest-well touch tip and non-dest-well touch tip. + if well is not None and location is not None: + self._instrument.touch_tip( + location=location, + well_core=well, + radius=1, + z_offset=touch_tip_props.z_offset, + speed=touch_tip_props.speed, + ) + else: + raise RuntimeError( + "Invalid touch tip location for post-dispense retraction." + ) + self._instrument.move_to( + location=location, + well_core=well, + force_direct=True, + minimum_z_height=None, + # Full speed because the tip will already be out of the liquid + speed=None, + ) + + if self._transfer_type != TransferType.ONE_TO_MANY: + # TODO: check if it is okay to just do `prepare_to_aspirate` unconditionally + if not self._tip_state.ready_to_aspirate: + self._instrument.prepare_to_aspirate() + self._tip_state.ready_to_aspirate = True + self._add_air_gap( + air_gap_volume=self._transfer_properties.aspirate.retract.air_gap_by_volume.get_for_volume( + 0 + ) + ) def _add_air_gap(self, air_gap_volume: float) -> None: """Add an air gap.""" diff --git a/api/tests/opentrons/conftest.py b/api/tests/opentrons/conftest.py index 13c1c88b47d..581060c1f23 100755 --- a/api/tests/opentrons/conftest.py +++ b/api/tests/opentrons/conftest.py @@ -950,7 +950,7 @@ def maximal_liquid_class_def() -> LiquidClassSchemaV1: ), ), delay=DelayProperties( - enable=False, params=DelayParams(duration=0) + enable=True, params=DelayParams(duration=10) ), ), positionReference=PositionReference.WELL_BOTTOM, diff --git a/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py b/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py index b2fc7a9cccd..632b07d3288 100644 --- a/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py +++ b/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py @@ -21,6 +21,7 @@ TransferComponentsExecutor, TransferType, TipState, + LiquidAndAirGapPair, ) from opentrons.protocol_engine import ( DeckPoint, @@ -1627,9 +1628,9 @@ def test_aspirate_liquid_class( ) ).then_return(mock_transfer_components_executor) decoy.when( - mock_engine_client.state.pipettes.get_aspirated_volume("abc123") - ).then_return(111) - subject.aspirate_liquid_class( + mock_transfer_components_executor.tip_state.last_liquid_and_air_gap_in_tip + ).then_return(LiquidAndAirGapPair(liquid=111, air_gap=222)) + result = subject.aspirate_liquid_class( volume=123, source=(source_location, source_well), transfer_properties=test_transfer_properties, @@ -1648,3 +1649,73 @@ def test_aspirate_liquid_class( mock_transfer_components_executor.aspirate_and_wait(volume=123), mock_transfer_components_executor.retract_after_aspiration(volume=123), ) + assert result == LiquidAndAirGapPair(air_gap=222, liquid=111) + + +def test_dispense_liquid_class( + decoy: Decoy, + mock_engine_client: EngineClient, + subject: InstrumentCore, + minimal_liquid_class_def2: LiquidClassSchemaV1, + mock_transfer_components_executor: TransferComponentsExecutor, +) -> None: + """It should call dispense sub-steps execution based on liquid class.""" + source_well = decoy.mock(cls=WellCore) + source_location = Location(Point(1, 2, 3), labware=None) + dest_well = decoy.mock(cls=WellCore) + dest_location = Location(Point(3, 2, 1), labware=None) + test_liquid_class = LiquidClass.create(minimal_liquid_class_def2) + test_transfer_properties = test_liquid_class.get_for( + "flex_1channel_50", "opentrons_flex_96_tiprack_50ul" + ) + push_out_vol = test_transfer_properties.dispense.push_out_by_volume.get_for_volume( + 123 + ) + decoy.when( + transfer_components_executor.absolute_point_from_position_reference_and_offset( + well=dest_well, + position_reference=PositionReference.WELL_BOTTOM, + offset=Coordinate(x=0, y=0, z=-5), + ) + ).then_return(Point(1, 2, 3)) + decoy.when( + transfer_components_executor.TransferComponentsExecutor( + instrument_core=subject, + transfer_properties=test_transfer_properties, + target_location=Location(Point(1, 2, 3), labware=None), + target_well=dest_well, + transfer_type=TransferType.ONE_TO_ONE, + tip_state=TipState(), + ) + ).then_return(mock_transfer_components_executor) + decoy.when( + mock_transfer_components_executor.tip_state.last_liquid_and_air_gap_in_tip + ).then_return(LiquidAndAirGapPair(liquid=333, air_gap=444)) + result = subject.dispense_liquid_class( + volume=123, + dest=(dest_location, dest_well), + source=(source_location, source_well), + transfer_properties=test_transfer_properties, + transfer_type=TransferType.ONE_TO_ONE, + tip_contents=[], + trash_location=Location(Point(1, 2, 3), labware=None), + ) + decoy.verify( + mock_transfer_components_executor.submerge( + submerge_properties=test_transfer_properties.dispense.submerge, + ), + mock_transfer_components_executor.dispense_and_wait( + volume=123, + push_out_override=push_out_vol, + ), + mock_transfer_components_executor.mix( + mix_properties=test_transfer_properties.dispense.mix, + last_dispense_push_out=True, + ), + mock_transfer_components_executor.retract_after_dispensing( + trash_location=Location(Point(1, 2, 3), labware=None), + source_location=source_location, + source_well=source_well, + ), + ) + assert result == LiquidAndAirGapPair(air_gap=444, liquid=333) diff --git a/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py b/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py index 7f9baa6d111..1691dcfce4a 100644 --- a/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py +++ b/api/tests/opentrons/protocol_api/core/engine/test_transfer_components_executor.py @@ -5,6 +5,7 @@ LiquidClassSchemaV1, PositionReference, Coordinate, + BlowoutLocation, ) from opentrons.protocol_api._liquid import LiquidClass @@ -16,6 +17,7 @@ absolute_point_from_position_reference_and_offset, TipState, TransferType, + LiquidAndAirGapPair, ) from opentrons.types import Location, Point @@ -77,7 +79,10 @@ def test_submerge( transfer_properties=sample_transfer_props, target_location=Location(Point(), labware=None), target_well=source_well, - tip_state=TipState(), + tip_state=TipState( + ready_to_aspirate=True, + last_liquid_and_air_gap_in_tip=LiquidAndAirGapPair(liquid=0, air_gap=123), + ), transfer_type=TransferType.ONE_TO_ONE, ) decoy.when(source_well.get_bottom(0)).then_return(well_bottom_point) @@ -442,7 +447,7 @@ def test_retract_after_aspiration( mock_instrument_core.touch_tip( location=Location(Point(x=4, y=4, z=4), labware=None), well_core=source_well, - radius=0, # Update this to use mmToEdge once implemented + radius=1, # Update this to use mmToEdge once implemented z_offset=-1, speed=30, ), @@ -483,7 +488,13 @@ def test_retract_after_aspiration_without_touch_tip_and_delay( transfer_properties=sample_transfer_props, target_location=Location(Point(1, 1, 1), labware=None), target_well=source_well, - tip_state=TipState(), + tip_state=TipState( + ready_to_aspirate=True, + last_liquid_and_air_gap_in_tip=LiquidAndAirGapPair( + liquid=10, + air_gap=0, + ), + ), transfer_type=TransferType.ONE_TO_ONE, ) decoy.when(source_well.get_bottom(0)).then_return(well_bottom_point) @@ -507,6 +518,289 @@ def test_retract_after_aspiration_without_touch_tip_and_delay( ) +""" +Single dispense properties: + +"singleDispense": { + "submerge": { + "positionReference": "well-top", + "offset": {"x": 30, "y": 20, "z": 10}, + "speed": 100, + "delay": {"enable": true, "params": { "duration": 0.0 }} + }, + "retract": { + "positionReference": "well-top", + "offset": {"x": 11, "y": 22, "z": 33}, + "speed": 50, + "airGapByVolume": [[1.0, 0.1], [49.9, 0.1], [50.0, 0.0]], + "blowout": { "enable": true , "params": {"location": "source", "flowRate": 100}}, + "touchTip": { "enable": true, "params": { "zOffset": -1, "mmToEdge": 0.5, "speed": 30}}, + "delay": {"enable": true, "params": { "duration": 10 }} + }, + "positionReference": "well-bottom", + "offset": {"x": 33, "y": 22, "z": 11}, + "flowRateByVolume": [[1.0, 50.0]], + "correctionByVolume": [[0.0, 0.0]], + "mix": { "enable": true, "params": { "repetitions": 1, "volume": 50 }}, + "pushOutByVolume": [[1.0, 7.0], [4.999, 7.0], [5.0, 2.0], [10.0, 2.0], [50.0, 2.0]], + "delay": { "enable": true, "params": { "duration": 0.2 }}, +} +""" + + +def test_retract_after_dispense_with_blowout_in_source( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should execute steps to retract from well after a dispense.""" + source_location = Location(Point(1, 2, 3), labware=None) + source_well = decoy.mock(cls=WellCore) + dest_well = decoy.mock(cls=WellCore) + well_top_point = Point(1, 2, 3) + well_bottom_point = Point(4, 5, 6) + air_gap_volume = ( + sample_transfer_props.aspirate.retract.air_gap_by_volume.get_for_volume(0) + ) + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 1, 1), labware=None), + target_well=dest_well, + tip_state=TipState( + ready_to_aspirate=True, + last_liquid_and_air_gap_in_tip=LiquidAndAirGapPair( + liquid=0, + air_gap=0, + ), + ), + transfer_type=TransferType.ONE_TO_ONE, + ) + decoy.when(dest_well.get_bottom(0)).then_return(well_bottom_point) + decoy.when(dest_well.get_top(0)).then_return(well_top_point) + + subject.retract_after_dispensing( + trash_location=Location(Point(), labware=None), + source_location=source_location, + source_well=source_well, + ) + decoy.verify( + mock_instrument_core.move_to( + location=Location(Point(12, 24, 36), labware=None), + well_core=dest_well, + force_direct=True, + minimum_z_height=None, + speed=50, + ), + mock_instrument_core.delay(10), + mock_instrument_core.touch_tip( + location=Location(Point(12, 24, 36), labware=None), + well_core=dest_well, + radius=1, + z_offset=-1, + speed=30, + ), + mock_instrument_core.move_to( + location=Location(Point(12, 24, 36), labware=None), + well_core=dest_well, + force_direct=True, + minimum_z_height=None, + speed=None, + ), + mock_instrument_core.air_gap_in_place( + volume=air_gap_volume, flow_rate=air_gap_volume + ), + mock_instrument_core.delay(0.2), + mock_instrument_core.set_flow_rate(blow_out=100), + mock_instrument_core.blow_out( + location=source_location, + well_core=source_well, + in_place=False, + ), + mock_instrument_core.touch_tip( + location=source_location, + well_core=source_well, + radius=1, + z_offset=-1, + speed=30, + ), + mock_instrument_core.move_to( + location=source_location, + well_core=source_well, + force_direct=True, + minimum_z_height=None, + speed=None, + ), + mock_instrument_core.air_gap_in_place( + volume=air_gap_volume, flow_rate=air_gap_volume + ), + mock_instrument_core.delay(0.2), + ) + + +def test_retract_after_dispense_with_blowout_in_destination( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should execute steps to retract from well after a dispense.""" + source_well = decoy.mock(cls=WellCore) + dest_well = decoy.mock(cls=WellCore) + well_top_point = Point(1, 2, 3) + well_bottom_point = Point(4, 5, 6) + air_gap_volume = ( + sample_transfer_props.aspirate.retract.air_gap_by_volume.get_for_volume(0) + ) + sample_transfer_props.dispense.retract.blowout.location = ( + BlowoutLocation.DESTINATION + ) + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 1, 1), labware=None), + target_well=dest_well, + tip_state=TipState( + ready_to_aspirate=True, + last_liquid_and_air_gap_in_tip=LiquidAndAirGapPair( + liquid=10, + air_gap=0, + ), + ), + transfer_type=TransferType.ONE_TO_ONE, + ) + decoy.when(dest_well.get_bottom(0)).then_return(well_bottom_point) + decoy.when(dest_well.get_top(0)).then_return(well_top_point) + + subject.retract_after_dispensing( + trash_location=Location(Point(), labware=None), + source_location=Location(Point(1, 2, 3), labware=None), + source_well=source_well, + ) + decoy.verify( + mock_instrument_core.move_to( + location=Location(Point(12, 24, 36), labware=None), + well_core=dest_well, + force_direct=True, + minimum_z_height=None, + speed=50, + ), + mock_instrument_core.delay(10), + mock_instrument_core.set_flow_rate(blow_out=100), + mock_instrument_core.blow_out( + location=Location(Point(12, 24, 36), labware=None), + well_core=None, + in_place=True, + ), + mock_instrument_core.touch_tip( + location=Location(Point(12, 24, 36), labware=None), + well_core=dest_well, + radius=1, + z_offset=-1, + speed=30, + ), + mock_instrument_core.move_to( + location=Location(Point(12, 24, 36), labware=None), + well_core=dest_well, + force_direct=True, + minimum_z_height=None, + speed=None, + ), + mock_instrument_core.air_gap_in_place( + volume=air_gap_volume, flow_rate=air_gap_volume + ), + mock_instrument_core.delay(0.2), + ) + + +@pytest.mark.xfail +def test_retract_after_dispense_with_blowout_in_trash( + decoy: Decoy, + mock_instrument_core: InstrumentCore, + sample_transfer_props: TransferProperties, +) -> None: + """It should execute steps to retract from well after a dispense.""" + source_location = Location(Point(1, 2, 3), labware=None) + source_well = decoy.mock(cls=WellCore) + dest_well = decoy.mock(cls=WellCore) + well_top_point = Point(1, 2, 3) + well_bottom_point = Point(4, 5, 6) + air_gap_volume = ( + sample_transfer_props.aspirate.retract.air_gap_by_volume.get_for_volume(0) + ) + sample_transfer_props.dispense.retract.blowout.location = BlowoutLocation.TRASH + + subject = TransferComponentsExecutor( + instrument_core=mock_instrument_core, + transfer_properties=sample_transfer_props, + target_location=Location(Point(1, 1, 1), labware=None), + target_well=dest_well, + tip_state=TipState(), + transfer_type=TransferType.ONE_TO_ONE, + ) + decoy.when(dest_well.get_bottom(0)).then_return(well_bottom_point) + decoy.when(dest_well.get_top(0)).then_return(well_top_point) + + subject.retract_after_dispensing( + trash_location=Location(Point(7, 8, 9), labware=None), + source_location=source_location, + source_well=source_well, + ) + decoy.verify( + mock_instrument_core.move_to( + location=Location(Point(12, 24, 36), labware=None), + well_core=dest_well, + force_direct=True, + minimum_z_height=None, + speed=50, + ), + mock_instrument_core.delay(10), + mock_instrument_core.touch_tip( + location=Location(Point(12, 24, 36), labware=None), + well_core=dest_well, + radius=1, + z_offset=-1, + speed=30, + ), + mock_instrument_core.move_to( + location=Location(Point(12, 24, 36), labware=None), + well_core=dest_well, + force_direct=True, + minimum_z_height=None, + speed=None, + ), + mock_instrument_core.air_gap_in_place( + volume=air_gap_volume, flow_rate=air_gap_volume + ), + mock_instrument_core.delay(0.2), + mock_instrument_core.set_flow_rate(blow_out=100), + mock_instrument_core.blow_out( + location=Location(Point(7, 8, 9), labware=None), + well_core=None, + in_place=False, + ), + # TODO: update with trash well as well_core arg + mock_instrument_core.touch_tip( + location=Location(Point(7, 8, 9), labware=None), + well_core=source_well, + radius=1, + z_offset=-1, + speed=30, + ), + mock_instrument_core.move_to( + location=Location(Point(7, 8, 9), labware=None), + well_core=None, + force_direct=True, + minimum_z_height=None, + speed=None, + ), + mock_instrument_core.air_gap_in_place( + volume=air_gap_volume, flow_rate=air_gap_volume + ), + mock_instrument_core.delay(0.2), + ) + + @pytest.mark.parametrize( argnames=["position_reference", "offset", "expected_result"], argvalues=[