fix(api): peek pipette bug fixes (#16985)

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

@@ -77,13 +77,12 @@ 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
+        self,
+        mount: OT3Mount,
+        acceleration: float,
+        gantry_load: GantryLoad,
+        high_speed_pipette: bool = False,
     ) -> None:
         ...
 

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

@@ -50,7 +50,6 @@
     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
 
@@ -406,28 +405,24 @@ 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)
         )
 
     def update_constraints_for_plunger_acceleration(
-        self, mount: OT3Mount, acceleration: float, gantry_load: GantryLoad
+        self,
+        mount: OT3Mount,
+        acceleration: float,
+        gantry_load: GantryLoad,
+        high_speed_pipette: bool = False,
     ) -> None:
         new_constraints = get_system_constraints_for_plunger_acceleration(
-            self._configuration.motion_settings, gantry_load, mount, acceleration
+            self._configuration.motion_settings,
+            gantry_load,
+            mount,
+            acceleration,
+            high_speed_pipette,
         )
         self._move_manager.update_constraints(new_constraints)
 

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

@@ -234,13 +234,12 @@ 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
+        self,
+        mount: OT3Mount,
+        acceleration: float,
+        gantry_load: GantryLoad,
+        high_speed_pipette: bool = False,
     ) -> None:
         self._sim_gantry_load = gantry_load
 

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

@@ -284,12 +284,22 @@ def get_system_constraints_for_plunger_acceleration(
     gantry_load: GantryLoad,
     mount: OT3Mount,
     acceleration: float,
+    high_speed_pipette: bool = False,
 ) -> "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()])
+
+    def _get_axis_max_speed(ax: Axis) -> float:
+        if ax == Axis.of_main_tool_actuator(mount) and high_speed_pipette:
+            _max_speed = float(DEFAULT_EMULSIFYING_PIPETTE_AXIS_MAX_SPEED)
+        else:
+            _max_speed = old_constraints["default_max_speed"][axis_kind]
+        return _max_speed
+
     for axis_kind in axis_kinds:
         for axis in Axis.of_kind(axis_kind):
+            _default_max_speed = _get_axis_max_speed(axis)
             if axis == Axis.of_main_tool_actuator(mount):
                 _accel = acceleration
             else:
@@ -298,7 +308,7 @@ def get_system_constraints_for_plunger_acceleration(
                 _accel,
                 old_constraints["max_speed_discontinuity"][axis_kind],
                 old_constraints["direction_change_speed_discontinuity"][axis_kind],
-                old_constraints["default_max_speed"][axis_kind],
+                _default_max_speed,
             )
     return new_constraints
 

api/src/opentrons/hardware_control/ot3api.py

@@ -299,8 +299,11 @@ async def set_system_constraints_for_calibration(self) -> None:
     async def set_system_constraints_for_plunger_acceleration(
         self, mount: OT3Mount, acceleration: float
     ) -> None:
+        high_speed_pipette = self._pipette_handler.get_pipette(
+            mount
+        ).is_high_speed_pipette()
         self._backend.update_constraints_for_plunger_acceleration(
-            mount, acceleration, self._gantry_load
+            mount, acceleration, self._gantry_load, high_speed_pipette
         )
 
     @contextlib.asynccontextmanager
@@ -636,25 +639,13 @@ async def cache_pipette(
         )
         self._pipette_handler.hardware_instruments[mount] = p
 
-        if self._pipette_handler.has_pipette(mount):
-            self._confirm_pipette_motion_constraints(mount)
-
         if config is not None:
             self._set_pressure_sensor_available(mount, instrument_config=config)
 
         # 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
-            )
-
     def get_pressure_sensor_available(self, mount: OT3Mount) -> bool:
         pip_axis = Axis.of_main_tool_actuator(mount)
         return self._backend.get_pressure_sensor_available(pip_axis)

shared-data/python/opentrons_shared_data/pipette/model_constants.py

@@ -20,6 +20,7 @@
     PipetteGenerationType.FLEX: {
         PipetteChannelType.SINGLE_CHANNEL: RobotMountConfigs(2133.33, 230.15, 80),
         PipetteChannelType.EIGHT_CHANNEL: RobotMountConfigs(2133.33, 230.15, 80),
+        PipetteChannelType.EIGHT_CHANNEL_EM: RobotMountConfigs(2133.33, 230.15, 80),
         PipetteChannelType.NINETY_SIX_CHANNEL: RobotMountConfigs(2133.33, 230.15, 80),
     },
 }

shared-data/python/opentrons_shared_data/pipette/pipette_load_name_conversions.py

@@ -81,6 +81,8 @@ def channels_from_string(channels: str) -> PipetteChannelType:
     if channels == "96":
         return PipetteChannelType.NINETY_SIX_CHANNEL
     elif "multi" in channels:
+        if "em" in channels:
+            return PipetteChannelType.EIGHT_CHANNEL_EM
         return PipetteChannelType.EIGHT_CHANNEL
     elif channels == "single":
         return PipetteChannelType.SINGLE_CHANNEL
@@ -115,6 +117,8 @@ def get_channel_from_pipette_name(pipette_name_tuple: Tuple[str, ...]) -> str:
     elif "96" in pipette_name_tuple:
         return "ninety_six_channel"
     else:
+        if "em" in pipette_name_tuple:
+            return "eight_channel_em"
         return "eight_channel"
 
 
@@ -154,7 +158,6 @@ def version_from_generation(pipette_name_tuple: Tuple[str, ...]) -> PipetteVersi
     )
     model_from_pipette_name = pipette_name_tuple[0]
     channel_from_pipette_name = get_channel_from_pipette_name(pipette_name_tuple)
-
     paths_to_validate = (
         get_shared_data_root() / "pipette" / "definitions" / "2" / "general"
     )
@@ -246,7 +249,12 @@ def convert_pipette_name(
 
     """
     split_pipette_name = name.split("_")
-    channels = channels_from_string(split_pipette_name[1])
+    channels_type = split_pipette_name[1]
+    if len(split_pipette_name) > 2:
+        if split_pipette_name[2] == "em":
+            channels_type = "multi_em"
+
+    channels = channels_from_string(channels_type)
     if provided_version:
         version = version_from_string(provided_version)
     else:

shared-data/python/opentrons_shared_data/pipette/types.py

@@ -49,13 +49,16 @@ def check_and_return_type(
 class PipetteChannelType(int, enum.Enum):
     SINGLE_CHANNEL = 1
     EIGHT_CHANNEL = 8
+    EIGHT_CHANNEL_EM = 82
     NINETY_SIX_CHANNEL = 96
 
     def __str__(self) -> str:
         if self.value == 96:
             return "96"
         elif self.value == 8:
             return "multi"
+        elif self.value == 82:
+            return "multi_em"
         else:
             return "single"