diff --git a/api/src/opentrons/hardware_control/dev_types.py b/api/src/opentrons/hardware_control/dev_types.py index a6773cb9184..575a5e612d9 100644 --- a/api/src/opentrons/hardware_control/dev_types.py +++ b/api/src/opentrons/hardware_control/dev_types.py @@ -100,6 +100,8 @@ class PipetteDict(InstrumentDict): pipette_bounding_box_offsets: PipetteBoundingBoxOffsetDefinition current_nozzle_map: NozzleMap lld_settings: Optional[Dict[str, Dict[str, float]]] + plunger_positions: Dict[str, float] + shaft_ul_per_mm: float class PipetteStateDict(TypedDict): diff --git a/api/src/opentrons/hardware_control/instruments/ot2/pipette.py b/api/src/opentrons/hardware_control/instruments/ot2/pipette.py index 7fc15c4c2d3..2d63342cf19 100644 --- a/api/src/opentrons/hardware_control/instruments/ot2/pipette.py +++ b/api/src/opentrons/hardware_control/instruments/ot2/pipette.py @@ -28,7 +28,7 @@ CommandPreconditionViolated, ) from opentrons_shared_data.pipette.ul_per_mm import ( - piecewise_volume_conversion, + calculate_ul_per_mm, PIPETTING_FUNCTION_FALLBACK_VERSION, PIPETTING_FUNCTION_LATEST_VERSION, ) @@ -584,21 +584,9 @@ def get_nominal_tip_overlap_dictionary_by_configuration( # want this to unbounded. @functools.lru_cache(maxsize=100) def ul_per_mm(self, ul: float, action: UlPerMmAction) -> float: - if action == "aspirate": - fallback = self._active_tip_settings.aspirate.default[ - PIPETTING_FUNCTION_FALLBACK_VERSION - ] - sequence = self._active_tip_settings.aspirate.default.get( - self._pipetting_function_version, fallback - ) - else: - fallback = self._active_tip_settings.dispense.default[ - PIPETTING_FUNCTION_FALLBACK_VERSION - ] - sequence = self._active_tip_settings.dispense.default.get( - self._pipetting_function_version, fallback - ) - return piecewise_volume_conversion(ul, sequence) + return calculate_ul_per_mm( + ul, action, self._active_tip_settings, self._pipetting_function_version + ) def __str__(self) -> str: return "{} current volume {}ul critical point: {} at {}".format( diff --git a/api/src/opentrons/hardware_control/instruments/ot2/pipette_handler.py b/api/src/opentrons/hardware_control/instruments/ot2/pipette_handler.py index 931c99fd4c6..7bd41e02e74 100644 --- a/api/src/opentrons/hardware_control/instruments/ot2/pipette_handler.py +++ b/api/src/opentrons/hardware_control/instruments/ot2/pipette_handler.py @@ -260,6 +260,13 @@ def get_attached_instrument(self, mount: MountType) -> PipetteDict: "pipette_bounding_box_offsets" ] = instr.config.pipette_bounding_box_offsets result["lld_settings"] = instr.config.lld_settings + result["plunger_positions"] = { + "top": instr.plunger_positions.top, + "bottom": instr.plunger_positions.bottom, + "blow_out": instr.plunger_positions.blow_out, + "drop_tip": instr.plunger_positions.drop_tip, + } + result["shaft_ul_per_mm"] = instr.config.shaft_ul_per_mm return cast(PipetteDict, result) @property diff --git a/api/src/opentrons/hardware_control/instruments/ot3/pipette.py b/api/src/opentrons/hardware_control/instruments/ot3/pipette.py index 109747ea1b9..5a4d9261bfd 100644 --- a/api/src/opentrons/hardware_control/instruments/ot3/pipette.py +++ b/api/src/opentrons/hardware_control/instruments/ot3/pipette.py @@ -27,7 +27,7 @@ InvalidInstrumentData, ) from opentrons_shared_data.pipette.ul_per_mm import ( - piecewise_volume_conversion, + calculate_ul_per_mm, PIPETTING_FUNCTION_FALLBACK_VERSION, PIPETTING_FUNCTION_LATEST_VERSION, ) @@ -529,23 +529,13 @@ def tip_presence_responses(self) -> int: # want this to unbounded. @functools.lru_cache(maxsize=100) def ul_per_mm(self, ul: float, action: UlPerMmAction) -> float: - if action == "aspirate": - fallback = self._active_tip_settings.aspirate.default[ - PIPETTING_FUNCTION_FALLBACK_VERSION - ] - sequence = self._active_tip_settings.aspirate.default.get( - self._pipetting_function_version, fallback - ) - elif action == "blowout": - return self._config.shaft_ul_per_mm - else: - fallback = self._active_tip_settings.dispense.default[ - PIPETTING_FUNCTION_FALLBACK_VERSION - ] - sequence = self._active_tip_settings.dispense.default.get( - self._pipetting_function_version, fallback - ) - return piecewise_volume_conversion(ul, sequence) + return calculate_ul_per_mm( + ul, + action, + self._active_tip_settings, + self._pipetting_function_version, + self._config.shaft_ul_per_mm, + ) def __str__(self) -> str: return "{} current volume {}ul critical point: {} at {}".format( @@ -585,6 +575,7 @@ def as_dict(self) -> "Pipette.DictType": "versioned_tip_overlap": self.tip_overlap, "back_compat_names": self._config.pipette_backcompat_names, "supported_tips": self.liquid_class.supported_tips, + "shaft_ul_per_mm": self._config.shaft_ul_per_mm, } ) return self._config_as_dict diff --git a/api/src/opentrons/hardware_control/instruments/ot3/pipette_handler.py b/api/src/opentrons/hardware_control/instruments/ot3/pipette_handler.py index f64078fcbff..dda5031a8a3 100644 --- a/api/src/opentrons/hardware_control/instruments/ot3/pipette_handler.py +++ b/api/src/opentrons/hardware_control/instruments/ot3/pipette_handler.py @@ -282,6 +282,13 @@ def get_attached_instrument(self, mount: OT3Mount) -> PipetteDict: "pipette_bounding_box_offsets" ] = instr.config.pipette_bounding_box_offsets result["lld_settings"] = instr.config.lld_settings + result["plunger_positions"] = { + "top": instr.plunger_positions.top, + "bottom": instr.plunger_positions.bottom, + "blow_out": instr.plunger_positions.blow_out, + "drop_tip": instr.plunger_positions.drop_tip, + } + result["shaft_ul_per_mm"] = instr.config.shaft_ul_per_mm return cast(PipetteDict, result) @property diff --git a/api/src/opentrons/protocol_api/__init__.py b/api/src/opentrons/protocol_api/__init__.py index 2f35bb46764..41a061f5a94 100644 --- a/api/src/opentrons/protocol_api/__init__.py +++ b/api/src/opentrons/protocol_api/__init__.py @@ -30,7 +30,16 @@ ) from .disposal_locations import TrashBin, WasteChute from ._liquid import Liquid, LiquidClass -from ._types import OFF_DECK +from ._types import ( + OFF_DECK, + PLUNGER_BLOWOUT, + PLUNGER_TOP, + PLUNGER_BOTTOM, + PLUNGER_DROPTIP, + ASPIRATE_ACTION, + DISPENSE_ACTION, + BLOWOUT_ACTION, +) from ._nozzle_layout import ( COLUMN, PARTIAL_COLUMN, @@ -69,12 +78,22 @@ "Liquid", "LiquidClass", "Parameters", + # Partial Tip types "COLUMN", "PARTIAL_COLUMN", "SINGLE", "ROW", "ALL", + # Deck location types "OFF_DECK", + # Pipette plunger types + "PLUNGER_BLOWOUT", + "PLUNGER_TOP", + "PLUNGER_BOTTOM", + "PLUNGER_DROPTIP", + "ASPIRATE_ACTION", + "DISPENSE_ACTION", + "BLOWOUT_ACTION", "RuntimeParameterRequiredError", "CSVParameter", # For internal Opentrons use only: diff --git a/api/src/opentrons/protocol_api/_types.py b/api/src/opentrons/protocol_api/_types.py index 9890e29c2bc..0e73405b3b7 100644 --- a/api/src/opentrons/protocol_api/_types.py +++ b/api/src/opentrons/protocol_api/_types.py @@ -17,3 +17,27 @@ class OffDeckType(enum.Enum): See :ref:`off-deck-location` for details on using ``OFF_DECK`` with :py:obj:`ProtocolContext.move_labware()`. """ + + +class PlungerPositionTypes(enum.Enum): + PLUNGER_TOP = "top" + PLUNGER_BOTTOM = "bottom" + PLUNGER_BLOWOUT = "blow_out" + PLUNGER_DROPTIP = "drop_tip" + + +PLUNGER_TOP: Final = PlungerPositionTypes.PLUNGER_TOP +PLUNGER_BOTTOM: Final = PlungerPositionTypes.PLUNGER_BOTTOM +PLUNGER_BLOWOUT: Final = PlungerPositionTypes.PLUNGER_BLOWOUT +PLUNGER_DROPTIP: Final = PlungerPositionTypes.PLUNGER_DROPTIP + + +class PipetteActionTypes(enum.Enum): + ASPIRATE_ACTION = "aspirate" + DISPENSE_ACTION = "dispense" + BLOWOUT_ACTION = "blowout" + + +ASPIRATE_ACTION: Final = PipetteActionTypes.ASPIRATE_ACTION +DISPENSE_ACTION: Final = PipetteActionTypes.DISPENSE_ACTION +BLOWOUT_ACTION: Final = PipetteActionTypes.BLOWOUT_ACTION diff --git a/api/src/opentrons/protocol_api/core/engine/robot.py b/api/src/opentrons/protocol_api/core/engine/robot.py index 477f1968c5a..df80917e091 100644 --- a/api/src/opentrons/protocol_api/core/engine/robot.py +++ b/api/src/opentrons/protocol_api/core/engine/robot.py @@ -1,13 +1,16 @@ -from typing import Optional, Dict +from typing import Optional, Dict, Union from opentrons.hardware_control import SyncHardwareAPI from opentrons.types import Mount, MountType, Point, AxisType, AxisMapType +from opentrons_shared_data.pipette import types as pip_types +from opentrons.protocol_api._types import PipetteActionTypes, PlungerPositionTypes from opentrons.protocol_engine import commands as cmd from opentrons.protocol_engine.clients import SyncClient as EngineClient from opentrons.protocol_engine.types import DeckPoint, MotorAxis from opentrons.protocol_api.core.robot import AbstractRobot + _AXIS_TYPE_TO_MOTOR_AXIS = { AxisType.X: MotorAxis.X, AxisType.Y: MotorAxis.Y, @@ -39,12 +42,57 @@ def __init__( def _convert_to_engine_mount(self, axis_map: AxisMapType) -> Dict[MotorAxis, float]: return {_AXIS_TYPE_TO_MOTOR_AXIS[ax]: dist for ax, dist in axis_map.items()} - def get_pipette_type_from_engine(self, mount: Mount) -> Optional[str]: + def get_pipette_type_from_engine( + self, mount: Union[Mount, str] + ) -> Optional[pip_types.PipetteNameType]: """Get the pipette attached to the given mount.""" - engine_mount = MountType[mount.name] + if isinstance(mount, Mount): + engine_mount = MountType[mount.name] + else: + if mount.lower() == "right": + engine_mount = MountType.RIGHT + else: + engine_mount = MountType.LEFT maybe_pipette = self._engine_client.state.pipettes.get_by_mount(engine_mount) return maybe_pipette.pipetteName if maybe_pipette else None + def get_plunger_position_from_name( + self, mount: Mount, position_name: PlungerPositionTypes + ) -> float: + engine_mount = MountType[mount.name] + maybe_pipette = self._engine_client.state.pipettes.get_by_mount(engine_mount) + if not maybe_pipette: + return 0.0 + return self._engine_client.state.pipettes.lookup_plunger_position_name( + maybe_pipette.id, position_name.value + ) + + def get_plunger_position_from_volume( + self, mount: Mount, volume: float, action: PipetteActionTypes, robot_type: str + ) -> float: + engine_mount = MountType[mount.name] + maybe_pipette = self._engine_client.state.pipettes.get_by_mount(engine_mount) + if not maybe_pipette: + raise RuntimeError( + f"Cannot load plunger position as no pipette is attached to {mount}" + ) + convert_volume = ( + self._engine_client.state.pipettes.lookup_volume_to_mm_conversion( + maybe_pipette.id, volume, action.value + ) + ) + plunger_bottom = ( + self._engine_client.state.pipettes.lookup_plunger_position_name( + maybe_pipette.id, "bottom" + ) + ) + mm = volume / convert_volume + if robot_type == "OT-2 Standard": + position = plunger_bottom + mm + else: + position = plunger_bottom - mm + return round(position, 6) + def move_to(self, mount: Mount, destination: Point, speed: Optional[float]) -> None: engine_mount = MountType[mount.name] engine_destination = DeckPoint( diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_protocol_core.py b/api/src/opentrons/protocol_api/core/legacy/legacy_protocol_core.py index e672a6fe839..d0b95ed82ca 100644 --- a/api/src/opentrons/protocol_api/core/legacy/legacy_protocol_core.py +++ b/api/src/opentrons/protocol_api/core/legacy/legacy_protocol_core.py @@ -37,7 +37,6 @@ class LegacyProtocolCore( LegacyInstrumentCore, LegacyLabwareCore, legacy_module_core.LegacyModuleCore, - # None, ] ): def __init__( diff --git a/api/src/opentrons/protocol_api/core/robot.py b/api/src/opentrons/protocol_api/core/robot.py index 7eade528413..95def3e17f3 100644 --- a/api/src/opentrons/protocol_api/core/robot.py +++ b/api/src/opentrons/protocol_api/core/robot.py @@ -1,12 +1,28 @@ from abc import abstractmethod, ABC -from typing import Optional +from typing import Optional, Union from opentrons.types import AxisMapType, Mount, Point +from opentrons_shared_data.pipette.types import PipetteNameType +from opentrons.protocol_api._types import PlungerPositionTypes, PipetteActionTypes class AbstractRobot(ABC): @abstractmethod - def get_pipette_type_from_engine(self, mount: Mount) -> Optional[str]: + def get_pipette_type_from_engine( + self, mount: Union[Mount, str] + ) -> Optional[PipetteNameType]: + ... + + @abstractmethod + def get_plunger_position_from_volume( + self, mount: Mount, volume: float, action: PipetteActionTypes, robot_type: str + ) -> float: + ... + + @abstractmethod + def get_plunger_position_from_name( + self, mount: Mount, position_name: PlungerPositionTypes + ) -> float: ... @abstractmethod diff --git a/api/src/opentrons/protocol_api/robot_context.py b/api/src/opentrons/protocol_api/robot_context.py index 272330e1664..5b0e578f9bb 100644 --- a/api/src/opentrons/protocol_api/robot_context.py +++ b/api/src/opentrons/protocol_api/robot_context.py @@ -19,6 +19,7 @@ from .core.common import ProtocolCore, RobotCore from .module_contexts import ModuleContext from .labware import Labware +from ._types import PipetteActionTypes, PlungerPositionTypes class HardwareManager(NamedTuple): @@ -200,14 +201,43 @@ def axis_coordinates_for( raise TypeError("You must specify a location to move to.") def plunger_coordinates_for_volume( - self, mount: Union[Mount, str], volume: float - ) -> None: - raise NotImplementedError() + self, mount: Union[Mount, str], volume: float, action: PipetteActionTypes + ) -> AxisMapType: + """ + Build a :py:class:`.types.AxisMapType` for a pipette plunger motor from volume. + + """ + pipette_name = self._core.get_pipette_type_from_engine(mount) + if not pipette_name: + raise ValueError( + f"Expected a pipette to be attached to provided mount {mount}" + ) + mount = validation.ensure_mount_for_pipette(mount, pipette_name) + pipette_axis = AxisType.plunger_axis_for_mount(mount) + + pipette_position = self._core.get_plunger_position_from_volume( + mount, volume, action, self._protocol_core.robot_type + ) + return {pipette_axis: pipette_position} def plunger_coordinates_for_named_position( - self, mount: Union[Mount, str], position_name: str - ) -> None: - raise NotImplementedError() + self, mount: Union[Mount, str], position_name: PlungerPositionTypes + ) -> AxisMapType: + """ + Build a :py:class:`.types.AxisMapType` for a pipette plunger motor from position_name. + + """ + pipette_name = self._core.get_pipette_type_from_engine(mount) + if not pipette_name: + raise ValueError( + f"Expected a pipette to be attached to provided mount {mount}" + ) + mount = validation.ensure_mount_for_pipette(mount, pipette_name) + pipette_axis = AxisType.plunger_axis_for_mount(mount) + pipette_position = self._core.get_plunger_position_from_name( + mount, position_name + ) + return {pipette_axis: pipette_position} def build_axis_map(self, axis_map: StringAxisMap) -> AxisMapType: """Take in a :py:class:`.types.StringAxisMap` and output a :py:class:`.types.AxisMapType`. diff --git a/api/src/opentrons/protocol_engine/execution/gantry_mover.py b/api/src/opentrons/protocol_engine/execution/gantry_mover.py index 7306bc4e4d1..c77a9e1bad2 100644 --- a/api/src/opentrons/protocol_engine/execution/gantry_mover.py +++ b/api/src/opentrons/protocol_engine/execution/gantry_mover.py @@ -64,6 +64,7 @@ HardwareAxis.Q: MotorAxis.AXIS_96_CHANNEL_CAM, } + # The height of the bottom of the pipette nozzle at home position without any tips. # We rely on this being the same for every OT-3 pipette. # @@ -305,7 +306,6 @@ async def move_mount_to( ) -> Point: """Move the given hardware mount to a waypoint.""" assert len(waypoints) > 0, "Must have at least one waypoint" - log.info(f"Moving mount {mount}") for waypoint in waypoints: log.info(f"The current waypoint moving is {waypoint}") await self._hardware_api.move_to( @@ -340,6 +340,10 @@ async def move_axes( mount, refresh=True ) log.info(f"The current position of the robot is: {current_position}.") + converted_current_position_deck = ( + self._hardware_api.get_deck_from_machine(current_position) + ) + log.info(f"The current position of the robot is: {current_position}.") pos_hw = target_axis_map_from_relative(pos_hw, current_position) log.info( diff --git a/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py b/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py index d3998c69bd1..6387bf5dcf1 100644 --- a/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py +++ b/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py @@ -67,6 +67,8 @@ class LoadedStaticPipetteData: back_left_corner_offset: Point front_right_corner_offset: Point pipette_lld_settings: Optional[Dict[str, Dict[str, float]]] + plunger_positions: Dict[str, float] + shaft_ul_per_mm: float class VirtualPipetteDataProvider: @@ -252,6 +254,7 @@ def _get_virtual_pipette_static_config_by_model( # noqa: C901 pip_back_left = config.pipette_bounding_box_offsets.back_left_corner pip_front_right = config.pipette_bounding_box_offsets.front_right_corner + plunger_positions = config.plunger_positions_configurations[liquid_class] return LoadedStaticPipetteData( model=str(pipette_model), display_name=config.display_name, @@ -280,6 +283,13 @@ def _get_virtual_pipette_static_config_by_model( # noqa: C901 pip_front_right[0], pip_front_right[1], pip_front_right[2] ), pipette_lld_settings=config.lld_settings, + plunger_positions={ + "top": plunger_positions.top, + "bottom": plunger_positions.bottom, + "blow_out": plunger_positions.blow_out, + "drop_tip": plunger_positions.drop_tip, + }, + shaft_ul_per_mm=config.shaft_ul_per_mm, ) def get_virtual_pipette_static_config( @@ -327,6 +337,8 @@ def get_pipette_static_config( front_right_offset[0], front_right_offset[1], front_right_offset[2] ), pipette_lld_settings=pipette_dict["lld_settings"], + plunger_positions=pipette_dict["plunger_positions"], + shaft_ul_per_mm=pipette_dict["shaft_ul_per_mm"], ) diff --git a/api/src/opentrons/protocol_engine/state/pipettes.py b/api/src/opentrons/protocol_engine/state/pipettes.py index e0f2cef1155..d20b8665318 100644 --- a/api/src/opentrons/protocol_engine/state/pipettes.py +++ b/api/src/opentrons/protocol_engine/state/pipettes.py @@ -10,11 +10,15 @@ Mapping, Optional, Tuple, + cast, ) from typing_extensions import assert_never from opentrons_shared_data.pipette import pipette_definition +from opentrons_shared_data.pipette.ul_per_mm import calculate_ul_per_mm +from opentrons_shared_data.pipette.types import UlPerMmAction + from opentrons.config.defaults_ot2 import Z_RETRACT_DISTANCE from opentrons.hardware_control.dev_types import PipetteDict from opentrons.hardware_control import CriticalPoint @@ -99,6 +103,8 @@ class StaticPipetteConfig: bounding_nozzle_offsets: BoundingNozzlesOffsets default_nozzle_map: NozzleMap # todo(mm, 2024-10-14): unused, remove? lld_settings: Optional[Dict[str, Dict[str, float]]] + plunger_positions: Dict[str, float] + shaft_ul_per_mm: float @dataclasses.dataclass @@ -288,6 +294,8 @@ def _update_pipette_config(self, state_update: update_types.StateUpdate) -> None ), default_nozzle_map=config.nozzle_map, lld_settings=config.pipette_lld_settings, + plunger_positions=config.plunger_positions, + shaft_ul_per_mm=config.shaft_ul_per_mm, ) self._state.flow_rates_by_id[ state_update.pipette_config.pipette_id @@ -772,3 +780,31 @@ def get_nozzle_configuration_supports_lld(self, pipette_id: str) -> bool: ): return False return True + + def lookup_volume_to_mm_conversion( + self, pipette_id: str, volume: float, action: str + ) -> float: + """Get the volumn to mm conversion for a pipette.""" + try: + lookup_volume = self.get_working_volume(pipette_id) + except errors.TipNotAttachedError: + lookup_volume = self.get_maximum_volume(pipette_id) + + pipette_config = self.get_config(pipette_id) + lookup_table_from_config = pipette_config.tip_configuration_lookup_table + try: + tip_settings = lookup_table_from_config[lookup_volume] + except KeyError: + tip_settings = list(lookup_table_from_config.values())[0] + return calculate_ul_per_mm( + volume, + cast(UlPerMmAction, action), + tip_settings, + shaft_ul_per_mm=pipette_config.shaft_ul_per_mm, + ) + + def lookup_plunger_position_name( + self, pipette_id: str, position_name: str + ) -> float: + """Get the plunger position provided for the given pipette id.""" + return self.get_config(pipette_id).plunger_positions[position_name] diff --git a/api/src/opentrons/types.py b/api/src/opentrons/types.py index fa57ce0dcd5..1f73d63c8c6 100644 --- a/api/src/opentrons/types.py +++ b/api/src/opentrons/types.py @@ -292,6 +292,11 @@ def mount_for_axis(cls, axis: "AxisType") -> Mount: } return map_mount_to_axis[axis] + @classmethod + def plunger_axis_for_mount(cls, mount: Mount) -> "AxisType": + map_plunger_axis_mount = {Mount.LEFT: cls.P_L, Mount.RIGHT: cls.P_R} + return map_plunger_axis_mount[mount] + @classmethod def ot2_axes(cls) -> List["AxisType"]: return [ diff --git a/api/tests/opentrons/protocol_api/test_robot_context.py b/api/tests/opentrons/protocol_api/test_robot_context.py index c1bdfe48c3f..36b94c52b15 100644 --- a/api/tests/opentrons/protocol_api/test_robot_context.py +++ b/api/tests/opentrons/protocol_api/test_robot_context.py @@ -17,6 +17,9 @@ from opentrons.protocol_api.core.common import ProtocolCore, RobotCore from opentrons.protocol_api import RobotContext, ModuleContext from opentrons.protocol_api.deck import Deck +from opentrons_shared_data.pipette.types import PipetteNameType + +from opentrons.protocol_api._types import PipetteActionTypes, PlungerPositionTypes @pytest.fixture @@ -58,7 +61,12 @@ def subject( api_version: APIVersion, ) -> RobotContext: """Get a RobotContext test subject with its dependencies mocked out.""" - decoy.when(mock_core.get_pipette_type_from_engine(Mount.LEFT)).then_return(None) + decoy.when(mock_core.get_pipette_type_from_engine(Mount.LEFT)).then_return( + PipetteNameType.P1000_SINGLE_FLEX + ) + decoy.when(mock_core.get_pipette_type_from_engine(Mount.RIGHT)).then_return( + PipetteNameType.P1000_SINGLE_FLEX + ) return RobotContext( core=mock_core, api_version=api_version, protocol_core=mock_protocol ) @@ -176,3 +184,73 @@ def test_get_axes_coordinates_for( """Test `RobotContext.get_axis_coordinates_for`.""" res = subject.axis_coordinates_for(mount, location_to_move) assert res == expected_axis_map + + +@pytest.mark.parametrize( + argnames=["mount", "volume", "action", "expected_axis_map"], + argvalues=[ + (Mount.RIGHT, 200, PipetteActionTypes.ASPIRATE_ACTION, {AxisType.P_R: 100}), + (Mount.LEFT, 100, PipetteActionTypes.DISPENSE_ACTION, {AxisType.P_L: 100}), + ], +) +def test_plunger_coordinates_for_volume( + decoy: Decoy, + subject: RobotContext, + mount: Mount, + volume: float, + action: PipetteActionTypes, + expected_axis_map: AxisMapType, +) -> None: + """Test `RobotContext.plunger_coordinates_for_volume`.""" + decoy.when( + subject._core.get_plunger_position_from_volume( + mount, volume, action, "OT-3 Standard" + ) + ).then_return(100) + + result = subject.plunger_coordinates_for_volume(mount, volume, action) + assert result == expected_axis_map + + +@pytest.mark.parametrize( + argnames=["mount", "position_name", "expected_axis_map"], + argvalues=[ + (Mount.RIGHT, PlungerPositionTypes.PLUNGER_TOP, {AxisType.P_R: 3}), + ( + Mount.RIGHT, + PlungerPositionTypes.PLUNGER_BOTTOM, + {AxisType.P_R: 3}, + ), + ], +) +def test_plunger_coordinates_for_named_position( + decoy: Decoy, + subject: RobotContext, + mount: Mount, + position_name: PlungerPositionTypes, + expected_axis_map: AxisMapType, +) -> None: + """Test `RobotContext.plunger_coordinates_for_named_position`.""" + decoy.when( + subject._core.get_plunger_position_from_name(mount, position_name) + ).then_return(3) + result = subject.plunger_coordinates_for_named_position(mount, position_name) + assert result == expected_axis_map + + +def test_plunger_methods_raise_without_pipette( + mock_core: RobotCore, mock_protocol: ProtocolCore, api_version: APIVersion +) -> None: + """Test that `RobotContext` plunger functions raise without pipette attached.""" + subject = RobotContext( + core=mock_core, api_version=api_version, protocol_core=mock_protocol + ) + with pytest.raises(ValueError): + subject.plunger_coordinates_for_named_position( + Mount.LEFT, PlungerPositionTypes.PLUNGER_TOP + ) + + with pytest.raises(ValueError): + subject.plunger_coordinates_for_volume( + Mount.LEFT, 200, PipetteActionTypes.ASPIRATE_ACTION + ) diff --git a/api/tests/opentrons/protocol_engine/commands/test_configure_for_volume.py b/api/tests/opentrons/protocol_engine/commands/test_configure_for_volume.py index d237c9e6090..9be08a0a71b 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_configure_for_volume.py +++ b/api/tests/opentrons/protocol_engine/commands/test_configure_for_volume.py @@ -63,6 +63,13 @@ async def test_configure_for_volume_implementation( back_left_corner_offset=Point(10, 20, 30), front_right_corner_offset=Point(40, 50, 60), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) decoy.when( diff --git a/api/tests/opentrons/protocol_engine/commands/test_load_pipette.py b/api/tests/opentrons/protocol_engine/commands/test_load_pipette.py index a42bbc4e4d9..570666e9c98 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_load_pipette.py +++ b/api/tests/opentrons/protocol_engine/commands/test_load_pipette.py @@ -69,6 +69,13 @@ async def test_load_pipette_implementation( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) decoy.when( @@ -137,6 +144,13 @@ async def test_load_pipette_implementation_96_channel( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) decoy.when( diff --git a/api/tests/opentrons/protocol_engine/execution/test_equipment_handler.py b/api/tests/opentrons/protocol_engine/execution/test_equipment_handler.py index b7a020c2d35..3ee027c24c1 100644 --- a/api/tests/opentrons/protocol_engine/execution/test_equipment_handler.py +++ b/api/tests/opentrons/protocol_engine/execution/test_equipment_handler.py @@ -154,6 +154,13 @@ def loaded_static_pipette_data( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) diff --git a/api/tests/opentrons/protocol_engine/resources/test_pipette_data_provider.py b/api/tests/opentrons/protocol_engine/resources/test_pipette_data_provider.py index 086b3ec297b..cbf7fa6174e 100644 --- a/api/tests/opentrons/protocol_engine/resources/test_pipette_data_provider.py +++ b/api/tests/opentrons/protocol_engine/resources/test_pipette_data_provider.py @@ -65,6 +65,13 @@ def test_get_virtual_pipette_static_config( back_left_corner_offset=Point(0, 0, 10.45), front_right_corner_offset=Point(0, 0, 10.45), pipette_lld_settings={}, + plunger_positions={ + "top": 19.5, + "bottom": -8.5, + "blow_out": -13.0, + "drop_tip": -27.0, + }, + shaft_ul_per_mm=0.785, ) @@ -94,6 +101,13 @@ def test_configure_virtual_pipette_for_volume( back_left_corner_offset=Point(-8.0, -22.0, -259.15), front_right_corner_offset=Point(-8.0, -22.0, -259.15), pipette_lld_settings={"t50": {"minHeight": 1.0, "minVolume": 0.0}}, + plunger_positions={ + "top": 0.0, + "bottom": 71.5, + "blow_out": 76.5, + "drop_tip": 90.5, + }, + shaft_ul_per_mm=0.785, ) subject_instance.configure_virtual_pipette_for_volume( "my-pipette", 1, result1.model @@ -120,6 +134,13 @@ def test_configure_virtual_pipette_for_volume( back_left_corner_offset=Point(-8.0, -22.0, -259.15), front_right_corner_offset=Point(-8.0, -22.0, -259.15), pipette_lld_settings={"t50": {"minHeight": 1.0, "minVolume": 0.0}}, + plunger_positions={ + "top": 0.0, + "bottom": 61.5, + "blow_out": 76.5, + "drop_tip": 90.5, + }, + shaft_ul_per_mm=0.785, ) @@ -149,6 +170,13 @@ def test_load_virtual_pipette_by_model_string( back_left_corner_offset=Point(-16.0, 43.15, 35.52), front_right_corner_offset=Point(16.0, -43.15, 35.52), pipette_lld_settings={}, + plunger_positions={ + "top": 19.5, + "bottom": -14.5, + "blow_out": -19.0, + "drop_tip": -33.4, + }, + shaft_ul_per_mm=9.621, ) @@ -246,6 +274,8 @@ def pipette_dict( "t200": {"minHeight": 0.5, "minVolume": 0}, "t1000": {"minHeight": 0.5, "minVolume": 0}, }, + "plunger_positions": {"top": 100, "bottom": 20, "blow_out": 10, "drop_tip": 0}, + "shaft_ul_per_mm": 5.0, } @@ -292,6 +322,8 @@ def test_get_pipette_static_config( "t200": {"minHeight": 0.5, "minVolume": 0}, "t1000": {"minHeight": 0.5, "minVolume": 0}, }, + plunger_positions={"top": 100, "bottom": 20, "blow_out": 10, "drop_tip": 0}, + shaft_ul_per_mm=5.0, ) 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 42ee037c1ce..abfb31f5f2a 100644 --- a/api/tests/opentrons/protocol_engine/state/test_geometry_view.py +++ b/api/tests/opentrons/protocol_engine/state/test_geometry_view.py @@ -2611,6 +2611,13 @@ def test_get_next_drop_tip_location( back_right_corner=Point(x=40, y=20, z=60), ), lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) ) decoy.when(mock_pipette_view.get_mount("pip-123")).then_return(pipette_mount) diff --git a/api/tests/opentrons/protocol_engine/state/test_pipette_store.py b/api/tests/opentrons/protocol_engine/state/test_pipette_store.py index 31b1a7f3a2c..60c857e4911 100644 --- a/api/tests/opentrons/protocol_engine/state/test_pipette_store.py +++ b/api/tests/opentrons/protocol_engine/state/test_pipette_store.py @@ -220,6 +220,13 @@ def test_handles_load_pipette( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) config_update = update_types.PipetteConfigUpdate( pipette_id="pipette-id", @@ -772,6 +779,13 @@ def test_add_pipette_config( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) subject.handle_action( @@ -810,6 +824,13 @@ def test_add_pipette_config( back_right_corner=Point(x=4, y=2, z=3), ), lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) assert subject.state.flow_rates_by_id["pipette-id"].default_aspirate == {"a": 1.0} assert subject.state.flow_rates_by_id["pipette-id"].default_dispense == {"b": 2.0} diff --git a/api/tests/opentrons/protocol_engine/state/test_pipette_view.py b/api/tests/opentrons/protocol_engine/state/test_pipette_view.py index 64e663a24e5..14c43bf70f6 100644 --- a/api/tests/opentrons/protocol_engine/state/test_pipette_view.py +++ b/api/tests/opentrons/protocol_engine/state/test_pipette_view.py @@ -291,6 +291,13 @@ def test_get_pipette_working_volume( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) }, ) @@ -322,6 +329,13 @@ def test_get_pipette_working_volume_raises_if_tip_volume_is_none( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) }, ) @@ -364,6 +378,13 @@ def test_get_pipette_available_volume( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), "pipette-id-none": StaticPipetteConfig( min_volume=1, @@ -380,6 +401,13 @@ def test_get_pipette_available_volume( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), }, ) @@ -492,6 +520,13 @@ def test_get_static_config( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) subject = get_pipette_view( @@ -543,6 +578,13 @@ def test_get_nominal_tip_overlap( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) subject = get_pipette_view(static_config_by_id={"pipette-id": config}) @@ -967,6 +1009,13 @@ def test_get_pipette_bounds_at_location( bounding_nozzle_offsets=_SAMPLE_NOZZLE_BOUNDS_OFFSETS, pipette_bounding_box_offsets=bounding_box_offsets, lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ) }, ) diff --git a/api/tests/opentrons/protocol_engine/state/test_tip_state.py b/api/tests/opentrons/protocol_engine/state/test_tip_state.py index abb408d7418..8abcc6a24e2 100644 --- a/api/tests/opentrons/protocol_engine/state/test_tip_state.py +++ b/api/tests/opentrons/protocol_engine/state/test_tip_state.py @@ -119,6 +119,13 @@ def test_get_next_tip_returns_none( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -177,6 +184,13 @@ def test_get_next_tip_returns_first_tip( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -229,6 +243,13 @@ def test_get_next_tip_used_starting_tip( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -314,6 +335,13 @@ def test_get_next_tip_skips_picked_up_tip( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -377,6 +405,13 @@ def test_get_next_tip_with_starting_tip( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -444,6 +479,13 @@ def test_get_next_tip_with_starting_tip_8_channel( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -514,6 +556,13 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -545,6 +594,13 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -615,6 +671,13 @@ def test_get_next_tip_with_starting_tip_out_of_tips( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -685,6 +748,13 @@ def test_get_next_tip_with_column_and_starting_tip( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -734,6 +804,13 @@ def test_reset_tips( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) @@ -796,6 +873,13 @@ def test_handle_pipette_config_action( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -929,6 +1013,13 @@ def test_active_channels( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -989,6 +1080,13 @@ def test_next_tip_uses_active_channels( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -1087,6 +1185,13 @@ def test_next_tip_automatic_tip_tracking_with_partial_configurations( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( @@ -1239,6 +1344,13 @@ def test_next_tip_automatic_tip_tracking_tiprack_limits( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, ), ) subject.handle_action( diff --git a/shared-data/python/opentrons_shared_data/pipette/ul_per_mm.py b/shared-data/python/opentrons_shared_data/pipette/ul_per_mm.py index 3423f0f49e5..774231ac40d 100644 --- a/shared-data/python/opentrons_shared_data/pipette/ul_per_mm.py +++ b/shared-data/python/opentrons_shared_data/pipette/ul_per_mm.py @@ -1,11 +1,46 @@ -from typing import List, Tuple +from typing import List, Tuple, Optional -from opentrons_shared_data.pipette.pipette_definition import PipetteFunctionKeyType +from opentrons_shared_data.pipette.pipette_definition import ( + PipetteFunctionKeyType, + SupportedTipsDefinition, +) +from opentrons_shared_data.pipette.types import UlPerMmAction PIPETTING_FUNCTION_FALLBACK_VERSION: PipetteFunctionKeyType = "1" PIPETTING_FUNCTION_LATEST_VERSION: PipetteFunctionKeyType = "2" +def calculate_ul_per_mm( + ul: float, + action: UlPerMmAction, + active_tip_settings: SupportedTipsDefinition, + requested_pipetting_version: Optional[PipetteFunctionKeyType] = None, + shaft_ul_per_mm: Optional[float] = None, +) -> float: + assumed_requested_pipetting_version = ( + requested_pipetting_version + if requested_pipetting_version + else PIPETTING_FUNCTION_LATEST_VERSION + ) + if action == "aspirate": + fallback = active_tip_settings.aspirate.default[ + PIPETTING_FUNCTION_FALLBACK_VERSION + ] + sequence = active_tip_settings.aspirate.default.get( + assumed_requested_pipetting_version, fallback + ) + elif action == "blowout" and shaft_ul_per_mm: + return shaft_ul_per_mm + else: + fallback = active_tip_settings.dispense.default[ + PIPETTING_FUNCTION_FALLBACK_VERSION + ] + sequence = active_tip_settings.dispense.default.get( + assumed_requested_pipetting_version, fallback + ) + return piecewise_volume_conversion(ul, sequence) + + def piecewise_volume_conversion( ul: float, sequence: List[Tuple[float, float, float]] ) -> float: