diff --git a/api/src/opentrons/protocol_api/core/engine/well.py b/api/src/opentrons/protocol_api/core/engine/well.py index e18bdb8ecf6a..964e1f7576c1 100644 --- a/api/src/opentrons/protocol_api/core/engine/well.py +++ b/api/src/opentrons/protocol_api/core/engine/well.py @@ -3,7 +3,7 @@ from opentrons_shared_data.labware.constants import WELL_NAME_PATTERN -from opentrons.types import Point +from opentrons.types import Point, Mount, MountType from opentrons.protocol_engine import WellLocation, WellOrigin, WellOffset from opentrons.protocol_engine import commands as cmd @@ -13,6 +13,7 @@ SimulatedProbeResult, LiquidTrackingType, ) +from opentrons.protocol_engine.errors import PipetteNotAttachedError from . import point_calculations from . import stringify @@ -181,15 +182,25 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point: def estimate_liquid_height_after_pipetting( self, + mount: Mount | str, operation_volume: float, ) -> LiquidTrackingType: """Return an estimate of liquid height after pipetting without raising an error.""" labware_id = self.labware_id well_name = self._name + if isinstance(mount, Mount): + mount_type = MountType.from_hw_mount(mount) + else: + mount_type = MountType(mount) + pipette_from_mount = self._engine_client.state.pipettes.get_by_mount(mount_type) + if pipette_from_mount is None: + raise PipetteNotAttachedError(f"No pipette present on mount {mount}") + pipette_id = pipette_from_mount.id starting_liquid_height = self.current_liquid_height() projected_final_height = self._engine_client.state.geometry.get_well_height_after_liquid_handling_no_error( labware_id=labware_id, well_name=well_name, + pipette_id=pipette_id, initial_height=starting_liquid_height, volume=operation_volume, ) diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_well_core.py b/api/src/opentrons/protocol_api/core/legacy/legacy_well_core.py index e0a12039c518..671409c9587a 100644 --- a/api/src/opentrons/protocol_api/core/legacy/legacy_well_core.py +++ b/api/src/opentrons/protocol_api/core/legacy/legacy_well_core.py @@ -5,7 +5,7 @@ from opentrons.protocols.api_support.util import APIVersionError -from opentrons.types import Point +from opentrons.types import Point, Mount from opentrons.protocol_engine.types.liquid_level_detection import ( SimulatedProbeResult, @@ -129,6 +129,7 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point: def estimate_liquid_height_after_pipetting( self, + mount: Mount | str, operation_volume: float, ) -> LiquidTrackingType: """Estimate what the liquid height will be after pipetting, without raising an error.""" diff --git a/api/src/opentrons/protocol_api/core/well.py b/api/src/opentrons/protocol_api/core/well.py index d02cea5cb24b..02fe1f47f424 100644 --- a/api/src/opentrons/protocol_api/core/well.py +++ b/api/src/opentrons/protocol_api/core/well.py @@ -3,7 +3,7 @@ from abc import ABC, abstractmethod from typing import TypeVar, Optional, Union -from opentrons.types import Point +from opentrons.types import Point, Mount from opentrons.protocol_engine.types import LiquidTrackingType from .._liquid import Liquid @@ -91,6 +91,7 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point: @abstractmethod def estimate_liquid_height_after_pipetting( self, + mount: Mount | str, operation_volume: float, ) -> LiquidTrackingType: """Estimate what the liquid height will be after pipetting, without raising an error.""" diff --git a/api/src/opentrons/protocol_api/labware.py b/api/src/opentrons/protocol_api/labware.py index 51f1301bf7b3..6580c004fa33 100644 --- a/api/src/opentrons/protocol_api/labware.py +++ b/api/src/opentrons/protocol_api/labware.py @@ -39,6 +39,7 @@ Point, NozzleMapInterface, MeniscusTrackingTarget, + Mount, ) from opentrons.protocols.api_support.types import APIVersion from opentrons.protocols.api_support.util import ( @@ -348,6 +349,7 @@ def current_liquid_volume(self) -> LiquidTrackingType: @requires_version(2, 21) def estimate_liquid_height_after_pipetting( self, + mount: Mount | str, operation_volume: float, ) -> LiquidTrackingType: """Check the height of the liquid within a well. @@ -360,7 +362,7 @@ def estimate_liquid_height_after_pipetting( """ projected_final_height = self._core.estimate_liquid_height_after_pipetting( - operation_volume=operation_volume, + operation_volume=operation_volume, mount=mount ) return projected_final_height diff --git a/api/src/opentrons/protocol_engine/execution/pipetting.py b/api/src/opentrons/protocol_engine/execution/pipetting.py index b503f4ae69f4..1d21542411dc 100644 --- a/api/src/opentrons/protocol_engine/execution/pipetting.py +++ b/api/src/opentrons/protocol_engine/execution/pipetting.py @@ -199,6 +199,7 @@ async def aspirate_while_tracking( labware_id=labware_id, well_name=well_name, operation_volume=volume * -1, + pipette_id=pipette_id, ) if isinstance(aspirate_z_distance, SimulatedProbeResult): raise InvalidLiquidHeightFound( @@ -234,6 +235,7 @@ async def dispense_while_tracking( labware_id=labware_id, well_name=well_name, operation_volume=volume, + pipette_id=pipette_id, ) if isinstance(dispense_z_distance, SimulatedProbeResult): raise InvalidLiquidHeightFound( diff --git a/api/src/opentrons/protocol_engine/state/geometry.py b/api/src/opentrons/protocol_engine/state/geometry.py index 4ba4c35a32f2..84f66844c318 100644 --- a/api/src/opentrons/protocol_engine/state/geometry.py +++ b/api/src/opentrons/protocol_engine/state/geometry.py @@ -520,6 +520,7 @@ def get_well_position( well_location=well_location, well_depth=well_depth, operation_volume=operation_volume, + pipette_id=pipette_id, ) if not isinstance(offset_adjustment, SimulatedProbeResult): offset = offset.model_copy(update={"z": offset.z + offset_adjustment}) @@ -1844,6 +1845,7 @@ def get_liquid_handling_z_change( self, labware_id: str, well_name: str, + pipette_id: str, operation_volume: float, ) -> float: """Get the change in height from a liquid handling operation.""" @@ -1853,6 +1855,7 @@ def get_liquid_handling_z_change( final_height = self.get_well_height_after_liquid_handling( labware_id=labware_id, well_name=well_name, + pipette_id=pipette_id, initial_height=initial_handling_height, volume=operation_volume, ) @@ -1873,6 +1876,7 @@ def get_well_offset_adjustment( well_name: str, well_location: WellLocationType, well_depth: float, + pipette_id: Optional[str] = None, operation_volume: Optional[float] = None, ) -> LiquidTrackingType: """Return a z-axis distance that accounts for well handling height and operation volume. @@ -1906,9 +1910,14 @@ def get_well_offset_adjustment( volume = well_location.volumeOffset if volume: + if pipette_id is None: + raise ValueError( + "cannot get liquid handling offset without pipette id." + ) liquid_height_after = self.get_well_height_after_liquid_handling( labware_id=labware_id, well_name=well_name, + pipette_id=pipette_id, initial_height=initial_handling_height, volume=volume, ) @@ -2030,6 +2039,7 @@ def get_well_height_after_liquid_handling( self, labware_id: str, well_name: str, + pipette_id: str, initial_height: LiquidTrackingType, volume: float, ) -> LiquidTrackingType: @@ -2044,7 +2054,14 @@ def get_well_height_after_liquid_handling( initial_volume = find_volume_at_well_height( target_height=initial_height, well_geometry=well_geometry ) - final_volume = initial_volume + volume + final_volume = initial_volume + ( + volume + * self.get_nozzles_per_well( + labware_id=labware_id, + target_well_name=well_name, + pipette_id=pipette_id, + ) + ) return find_height_at_well_volume( target_volume=final_volume, well_geometry=well_geometry ) @@ -2058,6 +2075,7 @@ def get_well_height_after_liquid_handling_no_error( self, labware_id: str, well_name: str, + pipette_id: str, initial_height: LiquidTrackingType, volume: float, ) -> LiquidTrackingType: @@ -2073,7 +2091,14 @@ def get_well_height_after_liquid_handling_no_error( initial_volume = find_volume_at_well_height( target_height=initial_height, well_geometry=well_geometry ) - final_volume = initial_volume + volume + final_volume = initial_volume + ( + volume + * self.get_nozzles_per_well( + labware_id=labware_id, + target_well_name=well_name, + pipette_id=pipette_id, + ) + ) well_volume = find_height_at_well_volume( target_volume=final_volume, well_geometry=well_geometry, diff --git a/api/tests/opentrons/protocol_api/core/engine/test_well_core.py b/api/tests/opentrons/protocol_api/core/engine/test_well_core.py index a45ba03aac22..cad3a1fff8c9 100644 --- a/api/tests/opentrons/protocol_api/core/engine/test_well_core.py +++ b/api/tests/opentrons/protocol_api/core/engine/test_well_core.py @@ -10,9 +10,15 @@ RectangularWellDefinition2, CircularWellDefinition2, ) +from opentrons_shared_data.pipette.types import PipetteNameType from opentrons.protocol_api import MAX_SUPPORTED_VERSION -from opentrons.protocol_engine import WellLocation, WellOrigin, WellOffset +from opentrons.protocol_engine import ( + WellLocation, + WellOrigin, + WellOffset, + LoadedPipette, +) from opentrons.protocol_engine import commands as cmd from opentrons.protocol_engine.clients import SyncClient as EngineClient from opentrons.protocol_engine.errors.exceptions import ( @@ -21,7 +27,7 @@ ) from opentrons.protocols.api_support.types import APIVersion from opentrons.protocols.api_support.util import UnsupportedAPIError -from opentrons.types import Point +from opentrons.types import Point, Mount, MountType from opentrons_shared_data.labware.labware_definition import ( InnerWellGeometry, ConicalFrustum, @@ -312,11 +318,13 @@ def test_current_liquid_volume( @pytest.mark.parametrize("operation_volume", [0.0, 100, -100, 2, -4, 5]) +@pytest.mark.parametrize("mount", [Mount.LEFT, "left"]) def test_estimate_liquid_height_after_pipetting( decoy: Decoy, subject: WellCore, mock_engine_client: EngineClient, operation_volume: float, + mount: Mount | str, ) -> None: """Make sure estimate_liquid_height_after_pipetting returns the correct value and does not raise an error.""" fake_well_geometry = InnerWellGeometry( @@ -355,14 +363,24 @@ def test_estimate_liquid_height_after_pipetting( mock_engine_client.state.geometry.get_well_height_after_liquid_handling_no_error( labware_id="labware-id", well_name="well-name", + pipette_id="pipette-id", initial_height=initial_liquid_height, volume=operation_volume, ) ).then_return(fake_final_height) + decoy.when( + mock_engine_client.state.pipettes.get_by_mount(MountType.LEFT) + ).then_return( + LoadedPipette( + id="pipette-id", + pipetteName=PipetteNameType.P300_SINGLE, + mount=MountType.LEFT, + ) + ) # make sure that no error was raised final_height = subject.estimate_liquid_height_after_pipetting( - operation_volume=operation_volume, + operation_volume=operation_volume, mount=mount ) assert final_height == fake_final_height diff --git a/api/tests/opentrons/protocol_engine/execution/test_pipetting_handler.py b/api/tests/opentrons/protocol_engine/execution/test_pipetting_handler.py index e94dc074d5d7..8fd017ec194a 100644 --- a/api/tests/opentrons/protocol_engine/execution/test_pipetting_handler.py +++ b/api/tests/opentrons/protocol_engine/execution/test_pipetting_handler.py @@ -332,7 +332,10 @@ async def test_hw_aspirate_while_tracking( decoy.when( mock_state_view.geometry.get_liquid_handling_z_change( - labware_id="labware-id", well_name="A1", operation_volume=-25.0 + labware_id="labware-id", + well_name="A1", + pipette_id="pipette_id", + operation_volume=-25.0, ) ).then_return(4.544) diff --git a/api/tests/opentrons/protocol_engine/state/test_geometry_view.py b/api/tests/opentrons/protocol_engine/state/test_geometry_view.py index c29cb066e563..4e301b3a007a 100644 --- a/api/tests/opentrons/protocol_engine/state/test_geometry_view.py +++ b/api/tests/opentrons/protocol_engine/state/test_geometry_view.py @@ -1773,6 +1773,10 @@ def test_get_well_position_with_meniscus_and_literal_volume_offset( slot_pos = Point(4, 5, 6) well_def = well_plate_def.wells["B2"] + pip_type = PipetteNameType.P300_SINGLE + decoy.when(mock_pipette_view.get_nozzle_configuration("pipette-id")).then_return( + get_default_nozzle_map(pip_type) + ) decoy.when(mock_labware_view.get("labware-id")).then_return(labware_data) decoy.when(mock_labware_view.get_definition("labware-id")).then_return( well_plate_def @@ -1843,7 +1847,10 @@ def test_get_well_position_with_meniscus_and_float_volume_offset( calibration_offset = LabwareOffsetVector(x=1, y=-2, z=3) slot_pos = Point(4, 5, 6) well_def = well_plate_def.wells["B2"] - + pip_type = PipetteNameType.P300_SINGLE + decoy.when(mock_pipette_view.get_nozzle_configuration("pipette-id")).then_return( + get_default_nozzle_map(pip_type) + ) decoy.when(mock_labware_view.get("labware-id")).then_return(labware_data) decoy.when(mock_labware_view.get_definition("labware-id")).then_return( well_plate_def @@ -1941,6 +1948,10 @@ def test_get_well_position_raises_validation_error( decoy.when(mock_labware_view.get_well_geometry("labware-id", "B2")).then_return( _TEST_INNER_WELL_GEOMETRY ) + pip_type = PipetteNameType.P300_SINGLE + decoy.when(mock_pipette_view.get_nozzle_configuration("pipette-id")).then_return( + get_default_nozzle_map(pip_type) + ) decoy.when( mock_pipette_view.get_current_tip_lld_settings(pipette_id="pipette-id") ).then_return(0.5) @@ -4184,17 +4195,27 @@ def test_virtual_get_well_height_after_liquid_handling_no_error( decoy: Decoy, subject: GeometryView, mock_labware_view: LabwareView, + mock_pipette_view: PipetteView, + well_plate_def: LabwareDefinition, initial_liquid_height: LiquidTrackingType, ) -> None: """Make sure SimulatedLiquidProbe doesn't change geometry behavior.""" + pip_type = PipetteNameType.P300_SINGLE + decoy.when(mock_pipette_view.get_nozzle_configuration("pipette-id")).then_return( + get_default_nozzle_map(pip_type) + ) + decoy.when(mock_labware_view.get_definition("labware-id")).then_return( + well_plate_def + ) + decoy.when(mock_labware_view.get_well_geometry("labware-id", "B2")).then_return( _TEST_INNER_WELL_GEOMETRY ) operation_volume = 1000.0 - result_estimate = subject.get_well_height_after_liquid_handling_no_error( labware_id="labware-id", well_name="B2", + pipette_id="pipette-id", initial_height=initial_liquid_height, volume=operation_volume, ) @@ -4228,13 +4249,23 @@ def test_virtual_find_height_and_volume( def test_get_liquid_handling_z_change( decoy: Decoy, subject: GeometryView, + well_plate_def: LabwareDefinition, mock_labware_view: LabwareView, + mock_pipette_view: PipetteView, mock_well_view: WellView, ) -> None: """Test for get_liquid_handling_z_change math.""" + pip_type = PipetteNameType.P300_SINGLE + decoy.when(mock_pipette_view.get_nozzle_configuration("pipette-id")).then_return( + get_default_nozzle_map(pip_type) + ) + decoy.when(mock_labware_view.get_well_definition("labware-id", "A1")).then_return( RectangularWellDefinition3.model_construct(totalLiquidVolume=1100000) # type: ignore[call-arg] ) + decoy.when(mock_labware_view.get_definition("labware-id")).then_return( + well_plate_def + ) decoy.when(mock_labware_view.get_well_geometry("labware-id", "A1")).then_return( _TEST_INNER_WELL_GEOMETRY ) @@ -4251,7 +4282,10 @@ def test_get_liquid_handling_z_change( ) # make sure that liquid handling z change math stays the same change = subject.get_liquid_handling_z_change( - labware_id="labware-id", well_name="A1", operation_volume=199.0 + labware_id="labware-id", + well_name="A1", + pipette_id="pipette-id", + operation_volume=199.0, ) expected_change = 3.2968 assert isclose(change, expected_change)