feat(hardware): add write-eeprom script (#21438)

# Overview
Add a script that's able to take the results of a dump_eeprom.py run and
re-write it to an eeprom

Very helpful in debugging eeprom migration because we can re-write the
old data for testing.

Author: ryanthecoder

hardware/opentrons_hardware/scripts/write_eeprom.py

@@ -0,0 +1,141 @@
+#!/usr/bin/env python3
+"""blow away and reset EEPROM file systems."""
+
+import argparse
+import asyncio
+import logging
+from logging.config import dictConfig
+from typing import Any, Dict, List
+
+from typing_extensions import Final
+
+from opentrons_hardware.drivers.can_bus import CanMessenger, build
+from opentrons_hardware.firmware_bindings import utils
+from opentrons_hardware.firmware_bindings.constants import NodeId
+from opentrons_hardware.firmware_bindings.messages import (
+    fields,
+    message_definitions,
+    payloads,
+)
+from opentrons_hardware.scripts.can_args import add_can_args, build_settings
+
+logger = logging.getLogger(__name__)
+
+LOG_CONFIG = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "formatters": {
+        "basic": {"format": "%(asctime)s %(name)s %(levelname)s %(message)s"}
+    },
+    "handlers": {
+        "stream_handler": {
+            "class": "logging.StreamHandler",
+            "formatter": "basic",
+            "level": logging.DEBUG,
+        },
+    },
+    "loggers": {
+        "": {
+            "handlers": ["stream_handler"],
+            "level": logging.DEBUG,
+        },
+    },
+}
+
+TARGETS: Final[Dict[str, NodeId]] = {
+    "head": NodeId.head,
+    "gantry-x": NodeId.gantry_x,
+    "gantry-y": NodeId.gantry_y,
+    "pipette-left": NodeId.pipette_left,
+    "pipette-right": NodeId.pipette_right,
+    "gripper": NodeId.gripper,
+}
+
+
+async def run(args: argparse.Namespace) -> None:
+    """Script entrypoint."""
+    eeprom_data = []
+    with open(args.file, "r") as f:
+        eeprom_data = ["".join(line.split()) for line in f]
+
+    async with (
+        build.driver(build_settings(args)) as driver,
+        CanMessenger(driver) as messenger,
+    ):
+        await rewrite_eeprom(
+            messenger,
+            TARGETS[args.target],
+            256 if args.old_version else 16384,
+            eeprom_data,
+        )
+        await asyncio.sleep(3)
+
+
+async def rewrite_eeprom(
+    messenger: CanMessenger, node: NodeId, limit: int, eeprom_data: List[str]
+) -> None:
+    """Wipe out all of the data used for the general purpose file system."""
+    start = 0
+    max_write = 8
+    for line in eeprom_data:
+        write_len = min(max_write, limit - start)
+        write_msg = message_definitions.WriteToEEPromRequest(
+            payload=payloads.EEPromDataPayload(
+                address=utils.UInt16Field(start),
+                data_length=utils.UInt16Field(write_len),
+                data=fields.EepromDataField.from_string(line),
+            )
+        )
+        start += write_len
+        await messenger.send(node, write_msg)
+
+
+def main() -> None:
+    """Entry point."""
+    parser = argparse.ArgumentParser(description=__doc__)
+
+    add_can_args(parser)
+    parser.add_argument(
+        "--target",
+        help="The FW subsystem to be cleared.",
+        type=str,
+        required=True,
+        choices=TARGETS.keys(),
+    )
+    parser.add_argument(
+        "--file",
+        help="dump file to write.",
+        type=str,
+        default="/var/log/eeprom_dump.hex",
+        required=False,
+    )
+    parser.add_argument(
+        "--less-logs",
+        help="Set log level to INFO, so we see less logs.",
+        action="store_true",
+        default=False,
+    )
+    parser.add_argument(
+        "--old-version",
+        help="Enable this flag to clear eeprom on the older 256 Byte eeproms.",
+        action="store_true",
+        default=False,
+    )
+
+    args = parser.parse_args()
+
+    def _set_log_lvl_warn(d: Dict[str, Any]) -> None:
+        for k in d.keys():
+            if isinstance(d[k], dict):
+                _set_log_lvl_warn(d[k])
+            elif k == "level":
+                d[k] = logging.WARNING
+
+    if args.less_logs:
+        _set_log_lvl_warn(LOG_CONFIG)
+    dictConfig(LOG_CONFIG)
+    asyncio.run(run(args))
+
+
+if __name__ == "__main__":
+    main()