fix(api,hardware): update devices in bootloader (#12383)

Currently, if we have a device in its bootloader during our network
probe, we probably don't update its firmware properly. That's for a
couple reasons:
- If it's a pipette, we don't actually know what pipette type it is,
since it won't be in attached pipettes and the bootloader can't give us
the type
- Even if it's not, we might not handle the node id properly

This commit fixes both problems.

For pipettes, we take advantage of the device subidentifier recently
introduced to the device info response payload. That subidentifier is
device-specific; for the pipettes it will be the pipette type, and for
others it isn't specified and is 0. That means that we can now find the
right pipette type for a pipette that is stuck in its bootloader. It
also means we can simplify the hardware interface by no longer requiring
the attached pipettes dict, though we only want that once we're
confident all pipettes have the new bootloader.

As a fallback for everything else, we should now explicitly handle
bootloader nodes at every stage.

One problem this doesn't fix is that independent of firmware update, the robot
server complains if it doesn't have a certain core set of node ids responding
on the network. In that case, where for instance gantry x or y or the head are
stuck in the bootloader, we'll still have a bad time. But it's separate enough to
require a followup.

Closes RQA-584

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

@@ -163,7 +163,7 @@ def node_id_to_subsystem(node_id: NodeId) -> "OT3SubSystem":
     node_to_subsystem = {
         node: subsystem for subsystem, node in SUBSYSTEM_NODEID.items()
     }
-    return node_to_subsystem[node_id]
+    return node_to_subsystem[node_id.application_for()]
 
 
 def get_current_settings(

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

@@ -202,9 +202,11 @@ def fw_update_info() -> Dict[NodeId, str]:
 
 @pytest.fixture
 def fw_node_info() -> Dict[NodeId, DeviceInfoCache]:
-    node_cache1 = DeviceInfoCache(NodeId.head, 1, "12345678", None, PCBARevision(None))
+    node_cache1 = DeviceInfoCache(
+        NodeId.head, 1, "12345678", None, PCBARevision(None), subidentifier=0
+    )
     node_cache2 = DeviceInfoCache(
-        NodeId.gantry_x, 1, "12345678", None, PCBARevision(None)
+        NodeId.gantry_x, 1, "12345678", None, PCBARevision(None), subidentifier=0
     )
     return {NodeId.head: node_cache1, NodeId.gantry_x: node_cache2}
 

hardware/opentrons_hardware/firmware_bindings/constants.py

@@ -31,6 +31,27 @@ class NodeId(int, Enum):
     head_bootloader = head | 0xF
     gripper_bootloader = gripper | 0xF
 
+    def is_bootloader(self) -> bool:
+        """Whether this node ID is a bootloader."""
+        return bool(self.value & 0xF == 0xF)
+
+    def bootloader_for(self) -> "NodeId":
+        """The associated bootloader node ID for the node.
+
+        This is safe to call on any node id, including ones that are already bootloaders.
+        """
+        return NodeId(self.value | 0xF)
+
+    def application_for(self) -> "NodeId":
+        """The associated core node ID for the node (i.e. head, not head_l).
+
+        This is safe to call on any node ID, including non-core application node IDs like
+        head_l. It will always give the code node ID.
+        """
+        # in c this would be & ~0xf but in python that gives 0x10 for some reason
+        # so let's write out the whole byte
+        return NodeId(self.value & 0xF0)
+
 
 # make these negative numbers so there is no chance they overlap with NodeId
 @unique
@@ -39,6 +60,14 @@ class USBTarget(int, Enum):
 
     rear_panel = -1
 
+    def is_bootloader(self) -> bool:
+        """Whether this is a bootloader id (always false)."""
+        return False
+
+    def application_for(self) -> "USBTarget":
+        """The corresponding application id."""
+        return self
+
 
 FirmwareTarget = Union[NodeId, USBTarget]
 

hardware/opentrons_hardware/firmware_bindings/messages/binary_message_definitions.py

@@ -65,6 +65,7 @@ class DeviceInfoResponse(utils.BinarySerializable):
     flags: VersionFlagsField = VersionFlagsField(0)
     shortsha: FirmwareShortSHADataField = FirmwareShortSHADataField(bytes())
     revision: OptionalRevisionField = OptionalRevisionField("", "", "")
+    subidentifier: utils.UInt8Field = utils.UInt8Field(0)
 
     @classmethod
     def build(cls, data: bytes) -> "DeviceInfoResponse":
@@ -102,8 +103,16 @@ def build(cls, data: bytes) -> "DeviceInfoResponse":
 
         revision = OptionalRevisionField.build(data[data_iter:])
 
+        data_iter = data_iter + revision.NUM_BYTES
+        try:
+            subidentifier = utils.UInt8Field.build(
+                int.from_bytes(data[data_iter : data_iter + 1], "big")
+            )
+        except IndexError:
+            subidentifier = utils.UInt8Field.build(0)
+
         return DeviceInfoResponse(
-            message_id, length, version, flags, shortsha, revision
+            message_id, length, version, flags, shortsha, revision, subidentifier
         )
 
 

hardware/opentrons_hardware/firmware_bindings/messages/payloads.py

@@ -4,6 +4,7 @@
 #  from __future__ import annotations
 from dataclasses import dataclass, field, asdict
 from . import message_definitions
+from typing import Iterator
 
 from .fields import (
     FirmwareShortSHADataField,
@@ -83,16 +84,29 @@ def build(cls, data: bytes) -> "DeviceInfoResponsePayload":
         consumed_by_super = _DeviceInfoResponsePayloadBase.get_size()
         superdict = asdict(_DeviceInfoResponsePayloadBase.build(data))
         message_index = superdict.pop("message_index")
+
+        # we want to parse this by adding extra 0s that may not be necessary,
+        # which is annoying and complex, so let's wrap it in an iterator
+        def _data_for_optionals(consumed: int, buf: bytes) -> Iterator[bytes]:
+            extended = buf + b"\x00\x00\x00\x00"
+            yield extended[consumed:]
+            consumed += 4
+            extended = extended + b"\x00"
+            yield extended[consumed : consumed + 1]
+
+        optionals_yielder = _data_for_optionals(consumed_by_super, data)
         inst = cls(
             **superdict,
-            revision=OptionalRevisionField.build(
-                (data + b"\x00\x00\x00\x00")[consumed_by_super:]
+            revision=OptionalRevisionField.build(next(optionals_yielder)),
+            subidentifier=utils.UInt8Field.build(
+                int.from_bytes(next(optionals_yielder), "big")
             ),
         )
         inst.message_index = message_index
         return inst
 
     revision: OptionalRevisionField
+    subidentifier: utils.UInt8Field
 
 
 @dataclass(eq=False)

hardware/opentrons_hardware/firmware_update/run.py

@@ -254,7 +254,7 @@ async def _run_can_update(
         initiator = FirmwareUpdateInitiator(messenger)
         downloader = FirmwareUpdateDownloader(messenger)
 
-        target = Target(system_node=node_id)
+        target = Target.from_single_node(node_id)
 
         logger.info(f"Initiating FW Update on {target}.")
         await self._status_queue.put((node_id, (FirmwareUpdateStatus.updating, 0)))
@@ -297,7 +297,7 @@ async def _run_can_update(
         else:
             logger.info("Skipping erase step.")
 
-        logger.info(f"Downloading FW to {target.bootloader_node}.")
+        logger.info(f"Downloading {filepath} to {target.bootloader_node}.")
         with open(filepath) as f:
             hex_processor = HexRecordProcessor.from_file(f)
             async for download_progress in downloader.run(

hardware/opentrons_hardware/firmware_update/target.py

@@ -1,29 +1,23 @@
 """Firmware update target."""
-from dataclasses import dataclass, field
-from typing_extensions import Final
+from dataclasses import dataclass
 
 from opentrons_hardware.firmware_bindings import NodeId
 
 
-BootloaderNodeIdMap: Final = {
-    NodeId.head: NodeId.head_bootloader,
-    NodeId.pipette_left: NodeId.pipette_left_bootloader,
-    NodeId.pipette_right: NodeId.pipette_right_bootloader,
-    NodeId.gantry_x: NodeId.gantry_x_bootloader,
-    NodeId.gantry_y: NodeId.gantry_y_bootloader,
-    NodeId.gripper: NodeId.gripper_bootloader,
-}
-
-
 @dataclass
 class Target:
     """Pair of a sub-system's node id with its bootloader's node id."""
 
     system_node: NodeId
-    bootloader_node: NodeId = field(init=False)
+    bootloader_node: NodeId
 
-    def __post_init__(self) -> None:
-        """Assign computed values."""
-        bn = BootloaderNodeIdMap.get(self.system_node)
-        assert bn, f"'{self.system_node}' is not valid for firmware update."
-        self.bootloader_node = bn
+    @classmethod
+    def from_single_node(cls, node: NodeId) -> "Target":
+        """Build a Target from just one of its node ids."""
+        assert node not in (
+            NodeId.broadcast,
+            NodeId.host,
+        ), f"Invalid update target {node}"
+        return cls(
+            system_node=node.application_for(), bootloader_node=node.bootloader_for()
+        )

hardware/opentrons_hardware/firmware_update/utils.py

@@ -8,16 +8,7 @@
 import logging
 import os
 from pathlib import Path
-from typing import (
-    Any,
-    Dict,
-    Optional,
-    Set,
-    Union,
-    Tuple,
-    Iterable,
-    Iterator,
-)
+from typing import Any, Dict, Optional, Set, Union, Tuple, Iterable, Iterator, cast
 from opentrons_hardware.firmware_bindings.constants import (
     FirmwareTarget,
     NodeId,
@@ -33,13 +24,9 @@
 
 _DEFAULT_PCBA_REVS: Final[Dict[FirmwareTarget, str]] = {
     NodeId.head: "c2",
-    NodeId.head_bootloader: "c2",
     NodeId.gantry_x: "c1",
-    NodeId.gantry_x_bootloader: "c1",
     NodeId.gantry_y: "c1",
-    NodeId.gantry_y_bootloader: "c1",
     NodeId.gripper: "c1",
-    NodeId.gripper_bootloader: "c1",
     USBTarget.rear_panel: "b1",
 }
 
@@ -64,6 +51,8 @@ class FirmwareUpdateType(Enum):
     pipettes_96 = 6
     rear_panel = 7
     unknown = -1
+    unknown_no_subtype = -2
+    unknown_no_revision = -3
 
     def __str__(self) -> str:
         """Name of enum."""
@@ -101,7 +90,14 @@ def from_node(cls, node: NodeId) -> "FirmwareUpdateType":
             NodeId.gantry_y: cls.gantry_y,
             NodeId.gripper: cls.gripper,
         }
-        return lookup.get(node, cls.unknown)
+        return lookup[node.application_for()]
+
+    @classmethod
+    def from_node_info(cls, node: NodeId, subid: int) -> "FirmwareUpdateType":
+        """Build an update type from a node and subid."""
+        if node.application_for() in (NodeId.pipette_left, NodeId.pipette_right):
+            return cls.from_pipette(PipetteType(subid))
+        return cls.from_node(node)
 
     @classmethod
     def from_pipette(cls, pipette: PipetteType) -> "FirmwareUpdateType":
@@ -194,36 +190,48 @@ def _revision_for_core_or_gripper(device_info: DeviceInfoCache) -> str:
     because PCBAs of the default revision were built before revision handling was
     introduced, and cannot be updated because too many were made.
     """
-    return device_info.revision.main or _DEFAULT_PCBA_REVS[device_info.target]
+    return (
+        device_info.revision.main
+        or _DEFAULT_PCBA_REVS[device_info.target.application_for()]
+    )
 
 
 def _revision_for_pipette(
-    pipette_type: PipetteType, device_info: DeviceInfoCache
+    device_info: DeviceInfoCache, fallback_pipette_type: PipetteType
 ) -> str:
     """Returns the appropriate defaulted revision for a pipette.
 
     The default revision can be determined solely from the pipette type. This is
     needed because PCBAs of the default revision were built before revision handling
     was introduced, and cannot be updated because too many were made.
     """
+    try:
+        pipette_type = PipetteType(device_info.subidentifier)
+    except ValueError:
+        pipette_type = fallback_pipette_type
     return device_info.revision.main or _DEFAULT_PCBA_REVS_PIPETTE[pipette_type]
 
 
+def _revision(version_cache: DeviceInfoCache) -> str:
+    if version_cache.target.application_for() in (
+        NodeId.pipette_left,
+        NodeId.pipette_right,
+    ):
+        return _revision_for_pipette(
+            version_cache, PipetteType(version_cache.subidentifier)
+        )
+    else:
+        return _revision_for_core_or_gripper(version_cache)
+
+
 def _update_details_for_device(
-    attached_pipettes: Dict[NodeId, PipetteType],
     version_cache: DeviceInfoCache,
 ) -> Tuple[FirmwareUpdateType, str]:
     if version_cache.target in NodeId:
         node = NodeId(version_cache.target)
-        if node in attached_pipettes:
-            pipette_type = attached_pipettes[node]
-            return FirmwareUpdateType.from_pipette(pipette_type), _revision_for_pipette(
-                pipette_type, version_cache
-            )
-        else:
-            return FirmwareUpdateType.from_node(node), _revision_for_core_or_gripper(
-                version_cache
-            )
+        return FirmwareUpdateType.from_node_info(
+            node, version_cache.subidentifier
+        ), _revision(version_cache)
     else:
         return FirmwareUpdateType.from_usb_target(
             USBTarget(version_cache.target)
@@ -233,22 +241,27 @@ def _update_details_for_device(
 def _update_type_for_device(
     attached_pipettes: Dict[NodeId, PipetteType],
     version_cache: DeviceInfoCache,
-) -> Union[Tuple[FirmwareUpdateType, str], Tuple[None, None]]:
+) -> Tuple[FirmwareUpdateType, str]:
     try:
-        update_type, revision = _update_details_for_device(
-            attached_pipettes, version_cache
-        )
+        update_type, revision = _update_details_for_device(version_cache)
     except KeyError:
-        pipette_type = (
-            attached_pipettes.get(NodeId(version_cache.target), None)
-            if version_cache.target in NodeId
-            else None
+        log.error(
+            f"Node {version_cache.target.name} has no revision or default revision and cannot be updated"
         )
+        return (FirmwareUpdateType.unknown_no_revision, "")
+    except ValueError:
+        # This means we did not have a valid pipette type in the version cache, so fall back to the passed-in
+        # attached_pipettes data.
+        # TODO: Delete this fallback when all pipettes have subidentifiers in their bootloaders.
+        if version_cache.target in attached_pipettes:
+            pipette_type = attached_pipettes[cast(NodeId, version_cache.target)]
+            return FirmwareUpdateType.from_pipette(pipette_type), _revision_for_pipette(
+                version_cache, pipette_type
+            )
         log.error(
-            f"Node {version_cache.target.name} (pipette {pipette_type}) "
-            "has no revision or default revision"
+            f"Target {version_cache.target.name} has no known subtype and cannot be updated"
         )
-        return None, None
+        return (FirmwareUpdateType.unknown_no_subtype, "")
     return update_type, revision
 
 
@@ -259,15 +272,7 @@ def _update_types_from_devices(
     for version_cache in devices:
         log.debug(f"Checking firmware update for {version_cache.target.name}")
         # skip pipettes that dont have a PipetteType
-        node_id = version_cache.target
-        if node_id in [
-            NodeId.pipette_left,
-            NodeId.pipette_right,
-        ] and not attached_pipettes.get(NodeId(node_id)):
-            continue
         update_type, rev = _update_type_for_device(attached_pipettes, version_cache)
-        if not rev or not update_type:
-            continue
         yield (version_cache, update_type, rev)
 
 
@@ -284,6 +289,9 @@ def _should_update(
     update_info: UpdateInfo,
     force: bool,
 ) -> bool:
+    if version_cache.target.is_bootloader():
+        log.info(f"Update required for {version_cache.target} (in bootloader)")
+        return True
     if force:
         log.info(f"Update required for {version_cache.target} (forced)")
         return True
@@ -303,7 +311,7 @@ def _update_info_for_type(
     update_type: Optional[FirmwareUpdateType],
 ) -> Optional[UpdateInfo]:
     # Given the update_type find the corresponding updateInfo, monadically on update_info
-    if not update_type:
+    if not update_type or update_type.value <= FirmwareUpdateType.unknown.value:
         return None
     try:
         update_info = known_firmware[update_type]
@@ -337,6 +345,8 @@ def _info_for_required_updates(
             yield version_cache.target, update_info.version, update_info.files_by_revision, rev
 
 
+# TODO: When we're confident that all pipettes report their type in the device info subidentifier
+# field, both in application and bootloader, we can remove the attached_pipettes from this codepath.
 def check_firmware_updates(
     device_info: Dict[FirmwareTarget, DeviceInfoCache],
     attached_pipettes: Dict[NodeId, PipetteType],