feat(api): add new speed for emulsifying pipette plunger (#16917)

Author: caila-marashaj

api/src/opentrons/config/defaults_ot3.py

@@ -75,6 +75,7 @@
 DEFAULT_GRIPPER_MOUNT_OFFSET: Final[Offset] = (84.55, -12.75, 93.85)
 DEFAULT_SAFE_HOME_DISTANCE: Final = 5
 DEFAULT_CALIBRATION_AXIS_MAX_SPEED: Final = 30
+DEFAULT_EMULSIFYING_PIPETTE_AXIS_MAX_SPEED: Final = 90
 
 DEFAULT_MAX_SPEEDS: Final[ByGantryLoad[Dict[OT3AxisKind, float]]] = ByGantryLoad(
     high_throughput={

api/src/opentrons/hardware_control/backends/flex_protocol.py

@@ -69,6 +69,11 @@ def update_constraints_for_calibration_with_gantry_load(
     ) -> None:
         ...
 
+    def update_constraints_for_emulsifying_pipette(
+        self, mount: OT3Mount, gantry_load: GantryLoad
+    ) -> None:
+        ...
+
     def update_constraints_for_plunger_acceleration(
         self, mount: OT3Mount, acceleration: float, gantry_load: GantryLoad
     ) -> None:

api/src/opentrons/hardware_control/backends/ot3controller.py

@@ -50,6 +50,7 @@
     get_system_constraints,
     get_system_constraints_for_calibration,
     get_system_constraints_for_plunger_acceleration,
+    get_system_constraints_for_emulsifying_pipette,
 )
 from .tip_presence_manager import TipPresenceManager
 
@@ -393,6 +394,18 @@ def update_constraints_for_calibration_with_gantry_load(
             f"Set system constraints for calibration: {self._move_manager.get_constraints()}"
         )
 
+    def update_constraints_for_emulsifying_pipette(
+        self, mount: OT3Mount, gantry_load: GantryLoad
+    ) -> None:
+        self._move_manager.update_constraints(
+            get_system_constraints_for_emulsifying_pipette(
+                self._configuration.motion_settings, gantry_load, mount
+            )
+        )
+        log.debug(
+            f"Set system constraints for emulsifying pipette: {self._move_manager.get_constraints()}"
+        )
+
     def update_constraints_for_gantry_load(self, gantry_load: GantryLoad) -> None:
         self._move_manager.update_constraints(
             get_system_constraints(self._configuration.motion_settings, gantry_load)

api/src/opentrons/hardware_control/backends/ot3simulator.py

@@ -234,6 +234,11 @@ def update_constraints_for_calibration_with_gantry_load(
     ) -> None:
         self._sim_gantry_load = gantry_load
 
+    def update_constraints_for_emulsifying_pipette(
+        self, mount: OT3Mount, gantry_load: GantryLoad
+    ) -> None:
+        pass
+
     def update_constraints_for_plunger_acceleration(
         self, mount: OT3Mount, acceleration: float, gantry_load: GantryLoad
     ) -> None:

api/src/opentrons/hardware_control/backends/ot3utils.py

@@ -2,7 +2,10 @@
 from typing import Dict, Iterable, List, Set, Tuple, TypeVar, cast, Sequence, Optional
 from typing_extensions import Literal
 from logging import getLogger
-from opentrons.config.defaults_ot3 import DEFAULT_CALIBRATION_AXIS_MAX_SPEED
+from opentrons.config.defaults_ot3 import (
+    DEFAULT_CALIBRATION_AXIS_MAX_SPEED,
+    DEFAULT_EMULSIFYING_PIPETTE_AXIS_MAX_SPEED,
+)
 from opentrons.config.types import OT3MotionSettings, OT3CurrentSettings, GantryLoad
 from opentrons.hardware_control.types import (
     Axis,
@@ -300,6 +303,31 @@ def get_system_constraints_for_plunger_acceleration(
     return new_constraints
 
 
+def get_system_constraints_for_emulsifying_pipette(
+    config: OT3MotionSettings,
+    gantry_load: GantryLoad,
+    mount: OT3Mount,
+) -> "SystemConstraints[Axis]":
+    old_constraints = config.by_gantry_load(gantry_load)
+    new_constraints = {}
+    axis_kinds = set([k for _, v in old_constraints.items() for k in v.keys()])
+    for axis_kind in axis_kinds:
+        for axis in Axis.of_kind(axis_kind):
+            if axis == Axis.of_main_tool_actuator(mount):
+                _max_speed = float(DEFAULT_EMULSIFYING_PIPETTE_AXIS_MAX_SPEED)
+            else:
+                _max_speed = old_constraints["default_max_speed"][axis_kind]
+            new_constraints[axis] = AxisConstraints.build(
+                max_acceleration=old_constraints["acceleration"][axis_kind],
+                max_speed_discont=old_constraints["max_speed_discontinuity"][axis_kind],
+                max_direction_change_speed_discont=old_constraints[
+                    "direction_change_speed_discontinuity"
+                ][axis_kind],
+                max_speed=_max_speed,
+            )
+    return new_constraints
+
+
 def _convert_to_node_id_dict(
     axis_pos: Coordinates[Axis, CoordinateValue],
 ) -> NodeIdMotionValues:

api/src/opentrons/hardware_control/instruments/ot3/pipette.py

@@ -41,6 +41,7 @@
     UlPerMmAction,
     PipetteName,
     PipetteModel,
+    Quirks,
 )
 from opentrons_shared_data.pipette import (
     load_data as load_pipette_data,
@@ -225,6 +226,9 @@ def active_tip_settings(self) -> SupportedTipsDefinition:
     def push_out_volume(self) -> float:
         return self._active_tip_settings.default_push_out_volume
 
+    def is_high_speed_pipette(self) -> bool:
+        return Quirks.highSpeed in self._config.quirks
+
     def act_as(self, name: PipetteName) -> None:
         """Reconfigure to act as ``name``. ``name`` must be either the
         actual name of the pipette, or a name in its back-compatibility

api/src/opentrons/hardware_control/ot3api.py

@@ -634,10 +634,21 @@ async def cache_pipette(
             self._feature_flags.use_old_aspiration_functions,
         )
         self._pipette_handler.hardware_instruments[mount] = p
+        if self._pipette_handler.has_pipette(mount):
+            self._confirm_pipette_motion_constraints(mount)
         # TODO (lc 12-5-2022) Properly support backwards compatibility
         # when applicable
         return skipped
 
+    def _confirm_pipette_motion_constraints(
+        self,
+        mount: OT3Mount,
+    ) -> None:
+        if self._pipette_handler.get_pipette(mount).is_high_speed_pipette():
+            self._backend.update_constraints_for_emulsifying_pipette(
+                mount, self.gantry_load
+            )
+
     async def cache_gripper(self, instrument_data: AttachedGripper) -> bool:
         """Set up gripper based on scanned information."""
         grip_cal = load_gripper_calibration_offset(instrument_data.get("id"))

api/tests/opentrons/hardware_control/backends/test_ot3_utils.py

@@ -3,7 +3,7 @@
 from opentrons_hardware.hardware_control.motion_planning import Move
 from opentrons.hardware_control.backends import ot3utils
 from opentrons_hardware.firmware_bindings.constants import NodeId
-from opentrons.hardware_control.types import Axis, OT3Mount
+from opentrons.hardware_control.types import Axis, OT3Mount, OT3AxisKind
 from numpy import float64 as f64
 
 from opentrons.config import defaults_ot3, types as conf_types
@@ -95,6 +95,22 @@ def test_get_system_contraints_for_plunger() -> None:
     assert updated_contraints[axis].max_acceleration == set_acceleration
 
 
+@pytest.mark.parametrize(["mount"], [[OT3Mount.LEFT], [OT3Mount.RIGHT]])
+def test_get_system_constraints_for_emulsifying_pipette(mount: OT3Mount) -> None:
+    set_max_speed = 90
+    config = defaults_ot3.build_with_defaults({})
+    pipette_ax = Axis.of_main_tool_actuator(mount)
+    default_pip_max_speed = config.motion_settings.default_max_speed[
+        conf_types.GantryLoad.LOW_THROUGHPUT
+    ][OT3AxisKind.P]
+    updated_constraints = ot3utils.get_system_constraints_for_emulsifying_pipette(
+        config.motion_settings, conf_types.GantryLoad.LOW_THROUGHPUT, mount
+    )
+    other_pipette = list(set(Axis.pipette_axes()) - {pipette_ax})[0]
+    assert updated_constraints[pipette_ax].max_speed == set_max_speed
+    assert updated_constraints[other_pipette].max_speed == default_pip_max_speed
+
+
 @pytest.mark.parametrize(
     ["moving", "expected"],
     [

hardware/tests/opentrons_hardware/hardware_control/test_motion_plan.py

@@ -2,7 +2,7 @@
 import numpy as np
 from hypothesis import given, assume, strategies as st
 from hypothesis.extra import numpy as hynp
-from typing import Iterator, List, Tuple
+from typing import Iterator, List, Tuple, Dict
 
 from opentrons_hardware.hardware_control.motion_planning import move_manager
 from opentrons_hardware.hardware_control.motion_planning.types import (
@@ -210,3 +210,60 @@ def test_close_move_plan(
     )
 
     assert converged, f"Failed to converge: {blend_log}"
+
+
+def test_pipette_high_speed_motion() -> None:
+    """Test that updated motion constraint doesn't get overridden by motion planning."""
+    origin: Dict[str, int] = {
+        "X": 499,
+        "Y": 499,
+        "Z": 499,
+        "A": 499,
+        "B": 499,
+        "C": 499,
+    }
+    target_list = []
+    axis_kinds = ["X", "Y", "Z", "A", "B", "C"]
+    constraints: SystemConstraints[str] = {}
+    for axis_kind in axis_kinds:
+        constraints[axis_kind] = AxisConstraints.build(
+            max_acceleration=500,
+            max_speed_discont=500,
+            max_direction_change_speed_discont=500,
+            max_speed=500,
+        )
+        origin_mapping: Dict[str, float] = {axis_kind: float(origin[axis_kind])}
+        target_list.append(MoveTarget.build(origin_mapping, 500))
+
+    set_axis_kind = "A"
+    dummy_em_pipette_max_speed = 90.0
+    manager = move_manager.MoveManager(constraints=constraints)
+
+    new_axis_constraint = AxisConstraints.build(
+        max_acceleration=float(constraints[set_axis_kind].max_acceleration),
+        max_speed_discont=float(constraints[set_axis_kind].max_speed_discont),
+        max_direction_change_speed_discont=float(
+            constraints[set_axis_kind].max_direction_change_speed_discont
+        ),
+        max_speed=90.0,
+    )
+    new_constraints = {}
+
+    for axis_kind in constraints.keys():
+        if axis_kind == set_axis_kind:
+            new_constraints[axis_kind] = new_axis_constraint
+        else:
+            new_constraints[axis_kind] = constraints[axis_kind]
+
+    manager.update_constraints(constraints=new_constraints)
+    converged, blend_log = manager.plan_motion(
+        origin=origin,
+        target_list=target_list,
+        iteration_limit=20,
+    )
+    for move in blend_log[0]:
+        unit_vector = move.unit_vector
+        for block in move.blocks:
+            top_set_axis_speed = unit_vector[set_axis_kind] * block.final_speed
+            if top_set_axis_speed != 0:
+                assert abs(top_set_axis_speed) == dummy_em_pipette_max_speed

shared-data/pipette/definitions/2/general/eight_channel_em/p1000/3_0.json

@@ -312,7 +312,7 @@
   "shaftDiameter": 4.5,
   "shaftULperMM": 15.904,
   "backlashDistance": 0.1,
-  "quirks": [],
+  "quirks": ["highSpeed"],
   "plungerHomingConfigurations": {
     "current": 1.0,
     "speed": 30