fix(api): monitor the pressure sensor more

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

@@ -45,7 +45,7 @@
     motor_nodes,
     LIMIT_SWITCH_OVERTRAVEL_DISTANCE,
     map_pipette_type_to_sensor_id,
-    moving_axes_in_move_group,
+    moving_pipettes_in_move_group,
     gripper_jaw_state_from_fw,
     get_system_constraints,
     get_system_constraints_for_calibration,
@@ -667,12 +667,9 @@ async def move(
             else False,
         )
 
-        mounts_moving = [
-            k
-            for k in moving_axes_in_move_group(move_group)
-            if k in [NodeId.pipette_left, NodeId.pipette_right]
-        ]
-        async with self._monitor_overpressure(mounts_moving):
+        pipettes_moving = moving_pipettes_in_move_group(move_group)
+
+        async with self._monitor_overpressure(pipettes_moving):
             positions = await runner.run(can_messenger=self._messenger)
         self._handle_motor_status_response(positions)
 

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

@@ -498,6 +498,20 @@ def create_gripper_jaw_hold_group(encoder_position_um: int) -> MoveGroup:
     return move_group
 
 
+def moving_pipettes_in_move_group(group: MoveGroup) -> List[NodeId]:
+    """Utility function to get which pipette nodes are moving either in z or their plunger."""
+    all_nodes = [node for step in group for node, _ in step.items()]
+    moving_nodes = moving_axes_in_move_group(group)
+    pipettes_moving: List[NodeId] = [
+        k for k in moving_nodes if k in [NodeId.pipette_left, NodeId.pipette_right]
+    ]
+    if NodeId.head_l in moving_nodes and NodeId.pipette_left in all_nodes:
+        pipettes_moving.append(NodeId.pipette_left)
+    if NodeId.head_r in moving_nodes and NodeId.pipette_right in all_nodes:
+        pipettes_moving.append(NodeId.pipette_right)
+    return pipettes_moving
+
+
 def moving_axes_in_move_group(group: MoveGroup) -> Set[NodeId]:
     """Utility function to get only the moving nodes in a move group."""
     ret: Set[NodeId] = set()

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

@@ -1,3 +1,5 @@
+import pytest
+from typing import List
 from opentrons_hardware.hardware_control.motion_planning import Move
 from opentrons_hardware.hardware_control.motion import (
     create_step,
@@ -119,3 +121,48 @@ def test_get_system_contraints_for_plunger() -> None:
     )
 
     assert updated_contraints[axis].max_acceleration == set_acceleration
+
+
+@pytest.mark.parametrize(
+    ["moving", "expected"],
+    [
+        [
+            [NodeId.gantry_x, NodeId.gantry_y, NodeId.gripper_g, NodeId.gripper_z],
+            [],
+        ],
+        [
+            [NodeId.head_l],
+            [NodeId.pipette_left],
+        ],
+        [
+            [NodeId.head_r],
+            [NodeId.pipette_right],
+        ],
+    ],
+)
+def test_moving_pipettes_in_move_group(
+    moving: List[NodeId], expected: List[NodeId]
+) -> None:
+    """Test that we can filter out the nonmoving nodes."""
+    present_nodes = [
+        NodeId.gantry_x,
+        NodeId.gantry_y,
+        NodeId.head_l,
+        NodeId.head_r,
+        NodeId.pipette_left,
+        NodeId.pipette_right,
+        NodeId.gripper_g,
+        NodeId.gripper_z,
+    ]
+    move_group = [
+        create_step(
+            distance={node: f64(100) for node in moving},
+            velocity={node: f64(100) for node in moving},
+            acceleration={node: f64(0) for node in moving},
+            duration=f64(1),
+            present_nodes=present_nodes,
+        )
+    ]
+
+    moving_pipettes = ot3utils.moving_pipettes_in_move_group(move_group)
+    assert set(moving_pipettes) == set(expected)