fix(api): flex: do not home non-present axes (#12604)

We weren't checking whether a given axis was present or not before
homing it, which is suboptimal. Now we only home the present ones.

This fixes an issue where when the HTTP API for home was called with a
mount, we'd try and home the pipette axis of the mount. This is probably
wrong, but it also should be allowed.

Did some little refactors along with this.

Closes RET-1354

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

@@ -45,6 +45,8 @@
     sub_system_to_node_id,
     NODEID_SUBSYSTEM,
     USBTARGET_SUBSYSTEM,
+    filter_probed_core_nodes,
+    motor_nodes,
 )
 
 try:
@@ -305,21 +307,7 @@ def _attached_pipettes_to_nodes(
         return pipette_nodes
 
     def _motor_nodes(self) -> Set[NodeId]:
-        # do the replacement of head and gripper devices
-        motor_nodes = self._replace_gripper_node(self._present_devices)
-        motor_nodes = self._replace_head_node(motor_nodes)
-        bootloader_nodes = {
-            NodeId.pipette_left_bootloader,
-            NodeId.pipette_right_bootloader,
-            NodeId.gantry_x_bootloader,
-            NodeId.gantry_y_bootloader,
-            NodeId.head_bootloader,
-            NodeId.gripper_bootloader,
-        }
-        # remove any bootloader nodes
-        motor_nodes -= bootloader_nodes
-        # filter out usb nodes
-        return {NodeId(target) for target in motor_nodes if target in NodeId}
+        return motor_nodes(self._present_devices)
 
     def get_instrument_update(
         self, mount: OT3Mount, pipette_subtype: Optional[PipetteSubType] = None
@@ -635,7 +623,7 @@ async def home(self, axes: Sequence[OT3Axis]) -> OT3AxisMap[float]:
         Returns:
             A dictionary containing the new positions of each axis
         """
-        checked_axes = [axis for axis in axes if self._axis_is_present(axis)]
+        checked_axes = [axis for axis in axes if self.axis_is_present(axis)]
         assert (
             OT3Axis.G not in checked_axes
         ), "Please home G axis using gripper_home_jaw()"
@@ -1020,49 +1008,6 @@ def home_position() -> OT3AxisMap[float]:
             node_to_axis(k): v for k, v in OT3Controller._get_home_position().items()
         }
 
-    @staticmethod
-    def _replace_head_node(targets: Set[FirmwareTarget]) -> Set[FirmwareTarget]:
-        """Replace the head core node with its two sides.
-
-        The node ID for the head central controller is what shows up in a network probe,
-        but what we actually send commands to an overwhelming majority of the time is
-        the head_l and head_r synthetic node IDs, and those are what we want in the
-        network map.
-        """
-        if NodeId.head in targets:
-            targets.remove(NodeId.head)
-            targets.add(NodeId.head_r)
-            targets.add(NodeId.head_l)
-        return targets
-
-    @staticmethod
-    def _replace_gripper_node(targets: Set[FirmwareTarget]) -> Set[FirmwareTarget]:
-        """Replace the gripper core node with its two axes.
-
-        The node ID for the gripper controller is what shows up in a network probe,
-        but what we actually send most commands to is the gripper_z and gripper_g
-        synthetic nodes, so we should have them in the network map instead.
-        """
-        if NodeId.gripper in targets:
-            targets.remove(NodeId.gripper)
-            targets.add(NodeId.gripper_z)
-            targets.add(NodeId.gripper_g)
-        return targets
-
-    @staticmethod
-    def _filter_probed_core_nodes(
-        current_set: Set[FirmwareTarget], probed_set: Set[FirmwareTarget]
-    ) -> Set[FirmwareTarget]:
-        core_replaced: Set[FirmwareTarget] = {
-            NodeId.gantry_x,
-            NodeId.gantry_y,
-            NodeId.head,
-            USBTarget.rear_panel,
-        }
-        current_set -= core_replaced
-        current_set |= probed_set
-        return current_set
-
     async def _probe_core(self, timeout: float = 5.0) -> None:
         """Update the list of core nodes present on the network.
 
@@ -1077,7 +1022,7 @@ async def _probe_core(self, timeout: float = 5.0) -> None:
         if self._usb_messenger is not None:
             core_nodes.add(USBTarget.rear_panel)
         core_present = set(await self._network_info.probe(core_nodes, timeout))
-        self._present_devices = self._filter_probed_core_nodes(
+        self._present_devices = filter_probed_core_nodes(
             self._present_devices, core_present
         )
 
@@ -1111,7 +1056,7 @@ async def probe_network(self, timeout: float = 5.0) -> None:
         self._present_devices = core_present
         log.info(f"The present devices are now {self._present_devices}")
 
-    def _axis_is_present(self, axis: OT3Axis) -> bool:
+    def axis_is_present(self, axis: OT3Axis) -> bool:
         try:
             return axis_to_node(axis) in self._motor_nodes()
         except KeyError:

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

@@ -32,6 +32,7 @@
     create_tip_action_group,
     PipetteAction,
     NODEID_SUBSYSTEM,
+    motor_nodes,
 )
 
 from opentrons_hardware.firmware_bindings.constants import (
@@ -496,6 +497,15 @@ def fw_version(self) -> Dict[OT3SubSystem, int]:
             NODEID_SUBSYSTEM[node.application_for()]: 0 for node in self._present_nodes
         }
 
+    def axis_is_present(self, axis: OT3Axis) -> bool:
+        try:
+            return axis_to_node(axis) in motor_nodes(
+                cast(Set[FirmwareTarget], self._present_nodes)
+            )
+        except KeyError:
+            # Currently unhandled axis
+            return False
+
     @property
     def update_required(self) -> bool:
         return self._update_required

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

@@ -234,6 +234,67 @@ def _convert_to_node_id_dict(
     return target
 
 
+def replace_head_node(targets: Set[FirmwareTarget]) -> Set[FirmwareTarget]:
+    """Replace the head core node with its two sides.
+
+    The node ID for the head central controller is what shows up in a network probe,
+    but what we actually send commands to an overwhelming majority of the time is
+    the head_l and head_r synthetic node IDs, and those are what we want in the
+    network map.
+    """
+    if NodeId.head in targets:
+        targets.remove(NodeId.head)
+        targets.add(NodeId.head_r)
+        targets.add(NodeId.head_l)
+    return targets
+
+
+def replace_gripper_node(targets: Set[FirmwareTarget]) -> Set[FirmwareTarget]:
+    """Replace the gripper core node with its two axes.
+
+    The node ID for the gripper controller is what shows up in a network probe,
+    but what we actually send most commands to is the gripper_z and gripper_g
+    synthetic nodes, so we should have them in the network map instead.
+    """
+    if NodeId.gripper in targets:
+        targets.remove(NodeId.gripper)
+        targets.add(NodeId.gripper_z)
+        targets.add(NodeId.gripper_g)
+    return targets
+
+
+def filter_probed_core_nodes(
+    current_set: Set[FirmwareTarget], probed_set: Set[FirmwareTarget]
+) -> Set[FirmwareTarget]:
+    core_replaced: Set[FirmwareTarget] = {
+        NodeId.gantry_x,
+        NodeId.gantry_y,
+        NodeId.head,
+        USBTarget.rear_panel,
+    }
+    current_set -= core_replaced
+    current_set |= probed_set
+    return current_set
+
+
+def motor_nodes(devices: Set[FirmwareTarget]) -> Set[NodeId]:
+    # do the replacement of head and gripper devices
+    motor_nodes = replace_gripper_node(devices)
+    motor_nodes = replace_head_node(motor_nodes)
+    bootloader_nodes = {
+        NodeId.pipette_left_bootloader,
+        NodeId.pipette_right_bootloader,
+        NodeId.gantry_x_bootloader,
+        NodeId.gantry_y_bootloader,
+        NodeId.head_bootloader,
+        NodeId.gripper_bootloader,
+    }
+    # remove any bootloader nodes
+    motor_nodes -= bootloader_nodes
+    # filter out usb nodes
+    return {NodeId(target) for target in motor_nodes if target in NodeId}
+
+
 def create_move_group(
     origin: Coordinates[OT3Axis, CoordinateValue],
     moves: List[Move[OT3Axis]],

api/src/opentrons/hardware_control/ot3api.py

@@ -1248,7 +1248,12 @@ async def home(
             checked_axes.append(OT3Axis.Q)
         self._log.info(f"Homing {axes}")
 
-        home_seq = [ax for ax in OT3Axis.home_order() if ax in checked_axes]
+        home_seq = [
+            ax
+            for ax in OT3Axis.home_order()
+            if (ax in checked_axes and self._backend.axis_is_present(ax))
+        ]
+        self._log.info(f"home was called with {axes} generating sequence {home_seq}")
         await self._home(home_seq)
 
     def get_engaged_axes(self) -> Dict[Axis, bool]:

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

@@ -514,34 +514,6 @@ async def fake_probe(expected, timeout):
             await controller.get_attached_instruments({})
 
 
-def test_nodeid_replace_head():
-    assert OT3Controller._replace_head_node(set([NodeId.head, NodeId.gantry_x])) == set(
-        [NodeId.head_l, NodeId.head_r, NodeId.gantry_x]
-    )
-    assert OT3Controller._replace_head_node(set([NodeId.gantry_x])) == set(
-        [NodeId.gantry_x]
-    )
-    assert OT3Controller._replace_head_node(set([NodeId.head_l])) == set(
-        [NodeId.head_l]
-    )
-
-
-def test_nodeid_replace_gripper():
-    assert OT3Controller._replace_gripper_node(
-        set([NodeId.gripper, NodeId.head])
-    ) == set([NodeId.gripper_g, NodeId.gripper_z, NodeId.head])
-    assert OT3Controller._replace_gripper_node(set([NodeId.head])) == set([NodeId.head])
-    assert OT3Controller._replace_gripper_node(set([NodeId.gripper_g])) == set(
-        [NodeId.gripper_g]
-    )
-
-
-def test_nodeid_filter_probed_core():
-    assert OT3Controller._filter_probed_core_nodes(
-        set([NodeId.gantry_x, NodeId.pipette_left]), set([NodeId.gantry_y])
-    ) == set([NodeId.gantry_y, NodeId.pipette_left])
-
-
 async def test_gripper_home_jaw(controller: OT3Controller, mock_move_group_run):
     await controller.gripper_home_jaw(25)
     for call in mock_move_group_run.call_args_list:

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

@@ -25,3 +25,27 @@ def test_create_step():
     assert len(move_group) == 3
     for step in move_group:
         assert set(present_nodes) == set(step.keys())
+
+
+def test_nodeid_filter_probed_core():
+    assert ot3utils.filter_probed_core_nodes(
+        set([NodeId.gantry_x, NodeId.pipette_left]), set([NodeId.gantry_y])
+    ) == set([NodeId.gantry_y, NodeId.pipette_left])
+
+
+def test_nodeid_replace_head():
+    assert ot3utils.replace_head_node(set([NodeId.head, NodeId.gantry_x])) == set(
+        [NodeId.head_l, NodeId.head_r, NodeId.gantry_x]
+    )
+    assert ot3utils.replace_head_node(set([NodeId.gantry_x])) == set([NodeId.gantry_x])
+    assert ot3utils.replace_head_node(set([NodeId.head_l])) == set([NodeId.head_l])
+
+
+def test_nodeid_replace_gripper():
+    assert ot3utils.replace_gripper_node(set([NodeId.gripper, NodeId.head])) == set(
+        [NodeId.gripper_g, NodeId.gripper_z, NodeId.head]
+    )
+    assert ot3utils.replace_gripper_node(set([NodeId.head])) == set([NodeId.head])
+    assert ot3utils.replace_gripper_node(set([NodeId.gripper_g])) == set(
+        [NodeId.gripper_g]
+    )

api/tests/opentrons/hardware_control/test_ot3_api.py

@@ -241,6 +241,19 @@ async def mock_instrument_handlers(
         yield mock_gripper_handler, mock_pipette_handler
 
 
+@pytest.fixture
+async def gripper_present(ot3_hardware: ThreadManager[OT3API]) -> None:
+    # attach a gripper if we're testing the gripper mount
+    gripper_config = gc.load(GripperModel.v1)
+    instr_data = AttachedGripper(config=gripper_config, id="test")
+    ot3_hardware._backend._attached_instruments[OT3Mount.GRIPPER] = {
+        "model": GripperModel.v1,
+        "id": "test",
+    }
+    ot3_hardware._backend._present_nodes.add(NodeId.gripper)
+    await ot3_hardware.cache_gripper(instr_data)
+
+
 @pytest.mark.parametrize(
     "load_configs,load",
     (
@@ -399,6 +412,7 @@ async def test_move_to_without_homing_first(
     homed_axis: List[OT3Axis],
 ) -> None:
     """Before a mount can be moved, XY and the corresponding Z  must be homed first"""
+    await ot3_hardware.cache_instruments()
     if mount == OT3Mount.GRIPPER:
         # attach a gripper if we're testing the gripper mount
         gripper_config = gc.load(GripperModel.v1)
@@ -413,7 +427,6 @@ async def test_move_to_without_homing_first(
         Point(0.001, 0.001, 0.001),
     )
     mock_home.assert_called_once()
-    assert ot3_hardware._backend.check_motor_status(homed_axis)
 
 
 @pytest.mark.parametrize(
@@ -707,10 +720,8 @@ async def test_gripper_capacitive_sweep(
     ot3_hardware: ThreadManager[OT3API],
     mock_move_to: AsyncMock,
     mock_backend_capacitive_pass: AsyncMock,
+    gripper_present: None,
 ) -> None:
-    gripper_config = gc.load(GripperModel.v1)
-    instr_data = AttachedGripper(config=gripper_config, id="g12345")
-    await ot3_hardware.cache_gripper(instr_data)
     await ot3_hardware.home()
     await ot3_hardware.grip(5)
     ot3_hardware._gripper_handler.get_gripper().current_jaw_displacement = 5
@@ -772,11 +783,10 @@ async def test_has_gripper(
     assert ot3_hardware.has_gripper() is True
 
 
-async def test_gripper_action(
+async def test_gripper_action_fails_with_no_gripper(
     ot3_hardware: ThreadManager[OT3API],
     mock_grip: AsyncMock,
     mock_ungrip: AsyncMock,
-    mock_hold_jaw_width: AsyncMock,
 ) -> None:
     with pytest.raises(
         GripperNotAttachedError, match="Cannot perform action without gripper attached"
@@ -790,9 +800,21 @@ async def test_gripper_action(
         await ot3_hardware.ungrip()
     mock_ungrip.assert_not_called()
 
-    # cache gripper
+
+async def test_gripper_action_works_with_gripper(
+    ot3_hardware: ThreadManager[OT3API],
+    mock_grip: AsyncMock,
+    mock_ungrip: AsyncMock,
+    mock_hold_jaw_width: AsyncMock,
+    gripper_present: None,
+) -> None:
+
     gripper_config = gc.load(GripperModel.v1)
-    instr_data = AttachedGripper(config=gripper_config, id="g12345")
+    instr_data = AttachedGripper(config=gripper_config, id="test")
+    ot3_hardware._backend._attached_instruments[OT3Mount.GRIPPER] = {
+        "model": GripperModel.v1,
+        "id": "test",
+    }
     await ot3_hardware.cache_gripper(instr_data)
 
     with pytest.raises(GripError, match="Gripper jaw must be homed before moving"):