feat(shared-data): Add support for PEEK pipettes (#17036)

<!--
Thanks for taking the time to open a Pull Request (PR)! Please make sure
you've read the "Opening Pull Requests" section of our Contributing
Guide:


https://github.com/Opentrons/opentrons/blob/edge/CONTRIBUTING.md#opening-pull-requests

GitHub provides robust markdown to format your PR. Links, diagrams,
pictures, and videos along with text formatting make it possible to
create a rich and informative PR. For more information on GitHub
markdown, see:


https://docs.github.com/en/get-started/writing-on-github/getting-started-with-writing-and-formatting-on-github/basic-writing-and-formatting-syntax

To ensure your code is reviewed quickly and thoroughly, please fill out
the sections below to the best of your ability!
-->

# Overview

This Adds support for the new oem pipette by doing the following:

- Add support for the flashing and factory testing pipettes with the new
serial number prefix P1KP
- Add the shared data definitions with updated max flowrates
- Add ability to change the max speed when the pipette definition has
the new "highSpeed" quirk
- Disable support for the pressure sensor
  - Don't monitor for over pressure
- Throw errors if trying to enable liquid presence detection on a
pipette
  - Throw errors if trying to explicitly use LLD
  - Support UI differences for the new pipette name

<!--
Describe your PR at a high level. State acceptance criteria and how this
PR fits into other work. Link issues, PRs, and other relevant resources.
-->

## Test Plan and Hands on Testing

<!--
Describe your testing of the PR. Emphasize testing not reflected in the
code. Attach protocols, logs, screenshots and any other assets that
support your testing.
-->

## Changelog

<!--
List changes introduced by this PR considering future developers and the
end user. Give careful thought and clear documentation to breaking
changes.
-->

## Review requests

<!--
- What do you need from reviewers to feel confident this PR is ready to
merge?
- Ask questions.
-->

## Risk assessment

<!--
- Indicate the level of attention this PR needs.
- Provide context to guide reviewers.
- Discuss trade-offs, coupling, and side effects.
- Look for the possibility, even if you think it's small, that your
change may affect some other part of the system.
- For instance, changing return tip behavior may also change the
behavior of labware calibration.
- How do your unit tests and on hands on testing mitigate this PR's
risks and the risk of future regressions?
- Especially in high risk PRs, explain how you know your testing is
enough.
-->

---------

Co-authored-by: Caila Marashaj <98041399+caila-marashaj@users.noreply.github.com>
Co-authored-by: Jethary <jethary@opentrons.com>

Author: 3 people

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

@@ -60,6 +60,14 @@ def restore_system_constraints(self) -> AsyncIterator[None]:
     def grab_pressure(self, channels: int, mount: OT3Mount) -> AsyncIterator[None]:
         ...
 
+    def set_pressure_sensor_available(
+        self, pipette_axis: Axis, available: bool
+    ) -> None:
+        ...
+
+    def get_pressure_sensor_available(self, pipette_axis: Axis) -> bool:
+        ...
+
     def update_constraints_for_gantry_load(self, gantry_load: GantryLoad) -> None:
         ...
 
@@ -70,7 +78,11 @@ def update_constraints_for_calibration_with_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:
         ...
 

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

@@ -197,6 +197,7 @@
     PipetteLiquidNotFoundError,
     CommunicationError,
     PythonException,
+    UnsupportedHardwareCommand,
 )
 
 from .subsystem_manager import SubsystemManager
@@ -362,6 +363,7 @@ def __init__(
                 self._configuration.motion_settings, GantryLoad.LOW_THROUGHPUT
             )
         )
+        self._pressure_sensor_available: Dict[NodeId, bool] = {}
 
     @asynccontextmanager
     async def restore_system_constraints(self) -> AsyncIterator[None]:
@@ -380,6 +382,16 @@ async def grab_pressure(
         async with grab_pressure(channels, tool, self._messenger):
             yield
 
+    def set_pressure_sensor_available(
+        self, pipette_axis: Axis, available: bool
+    ) -> None:
+        pip_node = axis_to_node(pipette_axis)
+        self._pressure_sensor_available[pip_node] = available
+
+    def get_pressure_sensor_available(self, pipette_axis: Axis) -> bool:
+        pip_node = axis_to_node(pipette_axis)
+        return self._pressure_sensor_available[pip_node]
+
     def update_constraints_for_calibration_with_gantry_load(
         self,
         gantry_load: GantryLoad,
@@ -399,10 +411,18 @@ def update_constraints_for_gantry_load(self, 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:
         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)
 
@@ -679,7 +699,8 @@ async def move(
 
         pipettes_moving = moving_pipettes_in_move_group(move_group)
 
-        async with self._monitor_overpressure(pipettes_moving):
+        checked_moving_pipettes = self._pipettes_to_monitor_pressure(pipettes_moving)
+        async with self._monitor_overpressure(checked_moving_pipettes):
             positions = await runner.run(can_messenger=self._messenger)
         self._handle_motor_status_response(positions)
 
@@ -786,7 +807,8 @@ async def home(
         moving_pipettes = [
             axis_to_node(ax) for ax in checked_axes if ax in Axis.pipette_axes()
         ]
-        async with self._monitor_overpressure(moving_pipettes):
+        checked_moving_pipettes = self._pipettes_to_monitor_pressure(moving_pipettes)
+        async with self._monitor_overpressure(checked_moving_pipettes):
             positions = await asyncio.gather(*coros)
         # TODO(CM): default gear motor homing routine to have some acceleration
         if Axis.Q in checked_axes:
@@ -800,6 +822,9 @@ async def home(
             self._handle_motor_status_response(position)
         return axis_convert(self._position, 0.0)
 
+    def _pipettes_to_monitor_pressure(self, pipettes: List[NodeId]) -> List[NodeId]:
+        return [pip for pip in pipettes if self._pressure_sensor_available[pip]]
+
     def _filter_move_group(self, move_group: MoveGroup) -> MoveGroup:
         new_group: MoveGroup = []
         for step in move_group:
@@ -915,6 +940,7 @@ def _lookup_serial_key(pipette_name: FirmwarePipetteName) -> str:
         lookup_name = {
             FirmwarePipetteName.p1000_single: "P1KS",
             FirmwarePipetteName.p1000_multi: "P1KM",
+            FirmwarePipetteName.p1000_multi_em: "P1KP",
             FirmwarePipetteName.p50_single: "P50S",
             FirmwarePipetteName.p50_multi: "P50M",
             FirmwarePipetteName.p1000_96: "P1KH",
@@ -949,6 +975,7 @@ def _build_attached_pip(
                     converted_name.pipette_type,
                     converted_name.pipette_channels,
                     converted_name.pipette_version,
+                    converted_name.oem_type,
                 ),
                 "id": OT3Controller._combine_serial_number(attached),
             }
@@ -1378,6 +1405,11 @@ async def liquid_probe(
     ) -> float:
         head_node = axis_to_node(Axis.by_mount(mount))
         tool = sensor_node_for_pipette(OT3Mount(mount.value))
+        if tool not in self._pipettes_to_monitor_pressure([tool]):
+            raise UnsupportedHardwareCommand(
+                "Liquid Presence Detection not available on this pipette."
+            )
+
         positions = await liquid_probe(
             messenger=self._messenger,
             tool=tool,

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

@@ -235,7 +235,11 @@ def update_constraints_for_calibration_with_gantry_load(
         self._sim_gantry_load = 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:
         self._sim_gantry_load = gantry_load
 
@@ -505,6 +509,7 @@ def _attached_pipette_to_mount(
                         converted_name.pipette_type,
                         converted_name.pipette_channels,
                         converted_name.pipette_version,
+                        converted_name.oem_type,
                     ),
                     "id": None,
                 }
@@ -527,6 +532,7 @@ def _attached_pipette_to_mount(
                     converted_name.pipette_type,
                     converted_name.pipette_channels,
                     converted_name.pipette_version,
+                    converted_name.oem_type,
                 ),
                 "id": init_instr["id"],
             }
@@ -538,6 +544,7 @@ def _attached_pipette_to_mount(
                     converted_name.pipette_type,
                     converted_name.pipette_channels,
                     converted_name.pipette_version,
+                    converted_name.oem_type,
                 ),
                 "id": 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,
@@ -281,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:
@@ -295,7 +308,32 @@ 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
+
+
+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
 

api/src/opentrons/hardware_control/dev_types.py

@@ -20,6 +20,7 @@
     PipetteConfigurations,
     SupportedTipsDefinition,
     PipetteBoundingBoxOffsetDefinition,
+    AvailableSensorDefinition,
 )
 from opentrons_shared_data.gripper import (
     GripperModel,
@@ -100,6 +101,7 @@ class PipetteDict(InstrumentDict):
     pipette_bounding_box_offsets: PipetteBoundingBoxOffsetDefinition
     current_nozzle_map: NozzleMap
     lld_settings: Optional[Dict[str, Dict[str, float]]]
+    available_sensors: AvailableSensorDefinition
 
 
 class PipetteStateDict(TypedDict):

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

@@ -56,6 +56,7 @@
     UlPerMmAction,
     PipetteName,
     PipetteModel,
+    PipetteOEMType,
 )
 from opentrons.hardware_control.dev_types import InstrumentHardwareConfigs
 
@@ -112,17 +113,20 @@ def __init__(
             pipette_type=config.pipette_type,
             pipette_channels=config.channels,
             pipette_generation=config.display_category,
+            oem_type=PipetteOEMType.OT,
         )
         self._acting_as = self._pipette_name
         self._pipette_model = PipetteModelVersionType(
             pipette_type=config.pipette_type,
             pipette_channels=config.channels,
             pipette_version=config.version,
+            oem_type=PipetteOEMType.OT,
         )
         self._valid_nozzle_maps = load_pipette_data.load_valid_nozzle_maps(
             self._pipette_model.pipette_type,
             self._pipette_model.pipette_channels,
             self._pipette_model.pipette_version,
+            PipetteOEMType.OT,
         )
         self._nozzle_offset = self._config.nozzle_offset
         self._nozzle_manager = (
@@ -189,7 +193,7 @@ def act_as(self, name: PipetteNameType) -> None:
         ], f"{self.name} is not back-compatible with {name}"
 
         liquid_model = load_pipette_data.load_liquid_model(
-            name.pipette_type, name.pipette_channels, name.get_version()
+            name.pipette_type, name.pipette_channels, name.get_version(), name.oem_type
         )
         # TODO need to grab name config here to deal with act as test
         self._liquid_class.max_volume = liquid_model["default"].max_volume
@@ -280,6 +284,7 @@ def reload_configurations(self) -> None:
             self._pipette_model.pipette_type,
             self._pipette_model.pipette_channels,
             self._pipette_model.pipette_version,
+            self._pipette_model.oem_type,
         )
         self._config_as_dict = self._config.dict()
 

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

@@ -41,6 +41,8 @@
     UlPerMmAction,
     PipetteName,
     PipetteModel,
+    Quirks,
+    PipetteOEMType,
 )
 from opentrons_shared_data.pipette import (
     load_data as load_pipette_data,
@@ -92,22 +94,26 @@ def __init__(
         self._liquid_class_name = pip_types.LiquidClasses.default
         self._liquid_class = self._config.liquid_properties[self._liquid_class_name]
 
+        oem = PipetteOEMType.get_oem_from_quirks(config.quirks)
         # TODO (lc 12-05-2022) figure out how we can safely deprecate "name" and "model"
         self._pipette_name = PipetteNameType(
             pipette_type=config.pipette_type,
             pipette_channels=config.channels,
             pipette_generation=config.display_category,
+            oem_type=oem,
         )
         self._acting_as = self._pipette_name
         self._pipette_model = PipetteModelVersionType(
             pipette_type=config.pipette_type,
             pipette_channels=config.channels,
             pipette_version=config.version,
+            oem_type=oem,
         )
         self._valid_nozzle_maps = load_pipette_data.load_valid_nozzle_maps(
             self._pipette_model.pipette_type,
             self._pipette_model.pipette_channels,
             self._pipette_model.pipette_version,
+            self._pipette_model.oem_type,
         )
         self._nozzle_offset = self._config.nozzle_offset
         self._nozzle_manager = (
@@ -225,6 +231,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
@@ -246,6 +255,7 @@ def reload_configurations(self) -> None:
             self._pipette_model.pipette_type,
             self._pipette_model.pipette_channels,
             self._pipette_model.pipette_version,
+            self._pipette_model.oem_type,
         )
         self._config_as_dict = self._config.dict()