Refine and fix bugs about firmware update tool and watchdog issue

kas/opt/module.yml

@@ -0,0 +1,16 @@
+#
+# Copyright (c) Siemens AG, 2024
+#
+# Authors:
+#  Li Hua Qian <[email protected]>
+#
+# This file is subject to the terms and conditions of the MIT License.  See
+# COPYING.MIT file in the top-level directory.
+#
+
+header:
+  version: 14
+
+local_conf_header:
+  no-module: |
+    IOT2050_MODULE_FWU = "1"

recipes-app/iot2050-eio-manager/files/gRPC/EIOManager/iot2050-eio.proto

@@ -76,11 +76,13 @@ message SyncTimeReply {
 
 /* ----------------- Update EIO firmware ----------------- */
 /* UpdateFirmwareRequest
- * - entity: 0 means checking for EIO controller
+ * - entity: 0 means updating for EIO controller
+ *           1 means updating for module
+ * - firmware_type: a customized string to specify the firmware type
  */
 message UpdateFirmwareRequest {
   int32 entity = 1;
-  string firmware = 2;
+  bytes firmware = 2;
 }
 
 /* UpdateFirmwareReply

recipes-app/iot2050-eio-manager/files/gRPC/EIOManager/iot2050_eio_pb2.pyi

@@ -53,8 +53,8 @@ class UpdateFirmwareRequest(_message.Message):
     ENTITY_FIELD_NUMBER: _ClassVar[int]
     FIRMWARE_FIELD_NUMBER: _ClassVar[int]
     entity: int
-    firmware: str
-    def __init__(self, entity: _Optional[int] = ..., firmware: _Optional[str] = ...) -> None: ...
+    firmware: bytes
+    def __init__(self, entity: _Optional[int] = ..., firmware: _Optional[bytes] = ...) -> None: ...
 
 class UpdateFirmwareReply(_message.Message):
     __slots__ = ["status", "message"]

recipes-app/iot2050-eio-manager/files/iot2050-eio-cli.py

@@ -56,19 +56,21 @@ def show_progress_bar(task, interval):
             time.sleep(interval)
 
 
-def do_update_firmware(firmware):
+def do_update_firmware(firmwares):
     with grpc.insecure_channel(iot2050_eio_api_server) as channel:
         stub = EIOManagerStub(channel)
         print("===================================================")
         print("EIO firmware update started - DO NOT INTERRUPT!")
         print("===================================================")
-        with ThreadPoolExecutor(max_workers=1) as pool:
-            future = pool.submit(
-                stub.UpdateFirmware,
-                UpdateFirmwareRequest(firmware=firmware)
-            )
-            show_progress_bar(future, 0.3)
-        print()
+        for entity in firmwares:
+            with ThreadPoolExecutor(max_workers=1) as pool:
+                future = pool.submit(
+                    stub.UpdateFirmware,
+                    UpdateFirmwareRequest(firmware=firmwares[entity],
+                                          firmware_type=entity)
+                )
+                show_progress_bar(future, 0.3)
+            print()
 
     response = future.result()
     print(f"Extended IO firmware update result: {response.status}")
@@ -86,7 +88,7 @@ def do_update_firmware(firmware):
         2. %(prog)s config retrieve config.yaml
             Retrieve the config from Extended IO Controller and store into config.yaml
         3. %(prog)s fwu controller [firmware.bin]
-            update firmware for Extended IO Controller, using firmware.bin if provided,
+            Update firmware for Extended IO Controller, using firmware.bin if provided,
             otherwise using the stock firmware file.
 
         Example Configuration File:
@@ -111,11 +113,13 @@ def do_update_firmware(firmware):
                         help='Config file in yaml format')
 
     fwu_parser = subparsers.add_parser("fwu", help='firmware update help')
-    fwu_parser.add_argument('action', metavar='ACTION',
-                        choices=['controller', 'module'],
-                        help='Specify the firmware update type')
-    fwu_parser.add_argument('firmware', nargs='?', metavar='FIRMWARE', type=str,
-                        help='Firmware file')
+    fwu_subparsers = fwu_parser.add_subparsers(help='sub-command help',
+                                       title='fwu-commands',
+                                       dest="fwu_command")
+    controller_parser = fwu_subparsers.add_parser("controller", help='controller help')
+    controller_parser.add_argument('firmware', nargs='?', metavar='FIRMWARE',
+                                   type=argparse.FileType('rb'),
+                                   help='Firmware file')
 
     args = parser.parse_args()
 
@@ -141,21 +145,24 @@ def do_update_firmware(firmware):
             with open(args.config, 'w', encoding='ascii') as f_config:
                 f_config.write(config_returned)
     elif args.command == 'fwu':
-        if args.firmware:
-            firmware = args.firmware
-        else:
-            firmware = EIO_FWU_MAP3_FW_BIN
-            status, message = do_fwu_check()
-            if status == 0:
-                # no need to update!
-                print(message)
-                sys.exit(0)
-
-        response = do_update_firmware(firmware)
-        if response.status:
-            print(f"ERROR: {response.message}")
-            print("EIO firmware update failed, please try again!")
-            sys.exit(1)
+        if args.fwu_command == 'controller':
+            firmwares = {}
+            # Map3 firmware
+            if args.firmware:
+                firmwares[0] = args.firmware.read()
+            else:
+                firmwares[0] = open(EIO_FWU_MAP3_FW_BIN, "rb").read()
+                status, message = do_fwu_check()
+                if status == 0:
+                    # no need to update!
+                    print(message)
+                    sys.exit(0)
+
+            response = do_update_firmware(firmwares)
+            if response.status:
+                print(f"ERROR: {response.message}")
+                print("EIO firmware update failed, please try again!")
+                sys.exit(1)
 
         print("EIO firmware update completed. Please reboot the device.")
     else:

recipes-app/iot2050-eio-manager/files/iot2050-eio-service.py

@@ -71,7 +71,7 @@ def SyncTime(self, request: SyncTimeRequest, context):
         return SyncTimeReply(status=0, message=f'{time}')
 
     def UpdateFirmware(self, request: UpdateFirmwareRequest, context):
-        status, message = update_firmware(request.firmware)
+        status, message = update_firmware(request.firmware, request.entity)
         return UpdateFirmwareReply(status=status, message=f'{message}')
 
     def CheckFWU(self, request: CheckFWURequest, context):

recipes-app/iot2050-eio-manager/files/iot2050_eio_fwu.py

@@ -74,7 +74,7 @@ class EIOFirmware():
     MAP3_CERTIFICATE_OFFSET = MAP3_FLASH_SIZE_1_MB - 32 * 1024
 
     def __init__(self, firmware):
-        self.firmware = open(firmware, "rb")
+        self.firmware = firmware
 
         self.chip = gpiod.Chip("/dev/gpiochip2")
         self.spi_mux_pin = self.chip.get_line(86)
@@ -95,8 +95,7 @@ def __get_write_firmware(self):
         read_buffer = ctypes.create_string_buffer(self.MAP3_FLASH_SIZE_1_MB)
         self.flash_prog.read(read_buffer, self.MAP3_FLASH_SIZE_1_MB)
 
-        self.firmware.seek(0)
-        write_firmware = self.firmware.read()
+        write_firmware = self.firmware
         # Firmware and certificate partitions are required, other
         # partitions are reserved.
         write_firmware = write_firmware[:self.MAP3_FIRMWARE_SIZE] \
@@ -128,17 +127,20 @@ def __del__(self):
         self.spi_mux_pin.release()
         if hasattr(self, "flash_prog") and self.flash_prog:
             self.flash_prog.release()
-        self.firmware.close()
 
 
 class FirmwareUpdate():
     """
     The FirmwareUpdate models the firmware updating behavior for all
     firmware update.
     """
-    def __init__(self, firmware):
+    def __init__(self, firmware, firmware_type):
         self.firmwares = {}
-        self.firmwares["map3"] = EIOFirmware(firmware)
+        if 0 == firmware_type:
+            self.firmwares[firmware_type] = EIOFirmware(firmware)
+
+        if not self.firmwares:
+            raise UpgradeError("No valid firmware!")
 
     def update(self):
         """Update the firmware to the specified flash"""
@@ -180,7 +182,7 @@ def collect_fwu_info(self) -> tuple[int, str]:
         Returns:
             tuple[int, str]:
                 The 1st int element indicates the firmware status:
-                
+
                 - 0: means no need to update
                 - 1: means firmware need update.
                 - 2: means firmware need update, however, the firmware
@@ -208,9 +210,9 @@ def collect_fwu_info(self) -> tuple[int, str]:
         return status, message
 
 
-def update_firmware(firmware):
+def update_firmware(firmware, entity):
     try:
-        FirmwareUpdate(firmware).update()
+        FirmwareUpdate(firmware, entity).update()
         return 0, "Firmware upgrade successfully!"
     except UpgradeError as e:
         return 1, e