tools: ota_upgrade: filter by board name

src/infuse_iot/commands.py

@@ -10,9 +10,25 @@
 from abc import ABCMeta, abstractmethod
 from typing import Any
 
+import infuse_iot.rpc_wrappers as wrappers
 from infuse_iot.epacket.packet import Auth
 
 
+def wrapper_from_command_id(command_id: int):
+    import importlib
+    import pkgutil
+
+    for _, name, _ in pkgutil.walk_packages(wrappers.__path__):
+        full_name = f"{wrappers.__name__}.{name}"
+        module = importlib.import_module(full_name)
+
+        # Add RPC wrapper to parser
+        cmd_cls = getattr(module, name)
+        if command_id == cmd_cls.COMMAND_ID:
+            return cmd_cls
+    return None
+
+
 class InfuseCommand(metaclass=ABCMeta):
     """Infuse-IoT SDK meta-tool command parent class"""
 
@@ -95,3 +111,8 @@ def data_progress_cb(self, offset: int) -> None:
     def handle_response(self, return_code: int, response: ctypes.LittleEndianStructure | None) -> None:
         """Handle RPC_RSP"""
         raise NotImplementedError
+
+    @classmethod
+    def handle_json_response(cls, response: dict) -> None:
+        """Handle json response from cloud"""
+        raise NotImplementedError

src/infuse_iot/rpc_wrappers/application_info.py

@@ -35,3 +35,23 @@ def handle_response(self, return_code, response):
         print(f"\t     KV CRC: 0x{r.kv_crc:08x}")
         print(f"\t   O Blocks: {r.data_blocks_internal}")
         print(f"\t   E Blocks: {r.data_blocks_external}")
+
+    @classmethod
+    def handle_json_response(cls, response: dict) -> None:
+        rsp = defs.application_info.response(
+            int(response["application_id"]),
+            defs.rpc_struct_mcuboot_img_sem_ver(
+                int(response["version"]["major"]),
+                int(response["version"]["minor"]),
+                int(response["version"]["revision"]),
+                int(response["version"]["build_num"]),
+            ),
+            int(response["network_id"]),
+            int(response["uptime"]),
+            int(response["reboots"]),
+            int(response["kv_crc"]),
+            int(response["data_blocks_internal"]),
+            int(response["data_blocks_external"]),
+        )
+        x = cls({})
+        x.handle_response(0, rsp)

src/infuse_iot/rpc_wrappers/last_reboot.py

@@ -38,3 +38,18 @@ def handle_response(self, return_code, response):
         print(f"\t     Thread: {response.thread.decode('utf-8')}")
         for idx, val in enumerate(response.esf):
             print(f"\t    ESF[{idx:2d}]: 0x{val:08x}")
+
+    @classmethod
+    def handle_json_response(cls, response: dict) -> None:
+        rsp = defs.last_reboot.response(
+            int(response["reason"]),
+            int(response["epoch_time_source"]),
+            int(response["epoch_time"]),
+            int(response["hardware_flags"]),
+            int(response["uptime"]),
+            int(response["param_1"]),
+            int(response["param_2"]),
+        )
+        rsp.esf = [int(x) for x in response["esf"]]
+        x = cls({})
+        x.handle_response(0, rsp)

src/infuse_iot/rpc_wrappers/wifi_scan.py

@@ -41,3 +41,28 @@ def handle_response(self, return_code, response):
 
         headers = ["SSID", "BSSID", "Band", "Channel", "Security", "RSSI"]
         print(tabulate.tabulate(table, headers=headers))
+
+    @classmethod
+    def handle_json_response(cls, response: dict) -> None:
+        table = []
+        for network in response["networks"]:
+            bssid = ":".join([f"{int(b):02x}" for b in network["bssid"]])
+            try:
+                security = str(z_wifi.SecurityType(int(network["security"])))
+            except ValueError:
+                security = f"Unknown ({network['security']})"
+
+            table.append(
+                [
+                    network["ssid"],
+                    bssid,
+                    str(z_wifi.FrequencyBand(int(network["band"]))),
+                    network["channel"],
+                    security,
+                    f"{network['rssi']} dBm",
+                ]
+            )
+
+        headers = ["SSID", "BSSID", "Band", "Channel", "Security", "RSSI"]
+        print(f"Total Networks: {response['network_count']}")
+        print(tabulate.tabulate(table, headers=headers))

src/infuse_iot/tools/ota_upgrade.py

@@ -23,13 +23,15 @@
 from infuse_iot.common import InfuseID
 from infuse_iot.definitions.rpc import bt_file_copy_basic, file_write_basic, rpc_enum_file_action
 from infuse_iot.epacket.packet import Auth, HopReceived
+from infuse_iot.generated.tdf_definitions import readings
 from infuse_iot.rpc_client import RpcClient
 from infuse_iot.socket_comms import (
     GatewayRequestConnectionRequest,
     LocalClient,
     default_multicast_address,
 )
 from infuse_iot.util.argparse import ValidFile, ValidRelease
+from infuse_iot.util.crc import crc16_ccitt
 from infuse_iot.zephyr.errno import errno
 
 
@@ -59,9 +61,11 @@ def __init__(self, args):
             self._single_diff = args.single
         else:
             raise NotImplementedError("Unknow upgrade type")
-        self._app_name = self._release.metadata["application"]["primary"]
-        self._app_id = self._release.metadata["application"]["id"]
-        self._new_ver = self._release.metadata["application"]["version"]
+        app_meta = self._release.metadata["application"]
+        self._app_name = app_meta["primary"]
+        self._app_id = app_meta["id"]
+        self._new_ver = app_meta["version"]
+        self._board_crc = crc16_ccitt(app_meta["board"].encode("utf-8"))
         self._handled: list[int] = []
         self._pending: dict[int, float] = {}
         self._missing_diffs: set[str] = set()
@@ -225,6 +229,8 @@ def run(self):
                         continue
                 if source.infuse_id in self._handled:
                     continue
+                if isinstance(announce, readings.announce_v2) and announce.board_crc != self._board_crc:
+                    continue
                 v = announce.version
                 v_str = f"{v.major}.{v.minor}.{v.revision}+{v.build_num:08x}"
 

src/infuse_iot/tools/rpc_cloud.py

@@ -19,7 +19,7 @@
 from infuse_iot.api_client.api.rpc import get_rpc_by_id, send_rpc
 from infuse_iot.api_client.models import Error, NewRPCMessage, NewRPCReq, RPCParams, RPCReqDataHeader, RpcRsp
 from infuse_iot.api_client.models.downlink_message_status import DownlinkMessageStatus
-from infuse_iot.commands import InfuseCommand, InfuseRpcCommand
+from infuse_iot.commands import InfuseCommand, InfuseRpcCommand, wrapper_from_command_id
 from infuse_iot.credentials import get_api_key
 from infuse_iot.definitions.rpc import id_type_mapping
 from infuse_iot.zephyr.errno import errno
@@ -111,6 +111,11 @@ def query(self, client: Client):
             command_name = id_type_mapping[rpc_req.command_id].NAME
         except KeyError:
             command_name = "Unknown"
+        try:
+            command_wrapper = wrapper_from_command_id(rpc_req.command_id)
+        except Exception:
+            command_wrapper = None
+
         print(f"   RPC ID: {rpc_req.command_id} ({command_name})")
         print(f"       To: {rsp.device.device_id}")
         # Manually detect downlink expiry, as the API doesn't do it
@@ -133,6 +138,12 @@ def query(self, client: Client):
             extra = f" ({errno(-rpc_rsp.return_code).name})" if rpc_rsp.return_code < 0 else ""
             print(f"   Result: {rpc_rsp.return_code}{extra}")
             if rpc_rsp.params:
+                try:
+                    if command_wrapper:
+                        command_wrapper.handle_json_response(rpc_rsp.params.additional_properties)
+                        return
+                except NotImplementedError:
+                    pass
                 print(json.dumps(rpc_rsp.params.additional_properties, indent=4))
             elif rpc_rsp.params_encoded:
                 raw_rsp = base64.b64decode(rpc_rsp.params_encoded)

src/infuse_iot/util/crc.py

@@ -0,0 +1,23 @@
+#!/usr/bin/env python3
+
+# Source of truth: https://reveng.sourceforge.io/crc-catalogue/all.htm
+
+
+def crc16_kermit(data: bytes) -> int:
+    """
+    CRC-16-KERMIT Algorithm
+    """
+    data = bytearray(data)
+    crc = 0x0000
+    for b in data:
+        e = (crc ^ b) & 0xFF
+        f = e ^ ((e << 4) & 0xFF)
+        crc = (crc >> 8) ^ (f << 8) ^ (f << 3) ^ (f >> 4)
+    return crc
+
+
+def crc16_ccitt(data: bytes) -> int:
+    """
+    CRC-16-CCITT Algorithm (Alias of KERMIT)
+    """
+    return crc16_kermit(data)

tests/test_commands.py

@@ -0,0 +1,19 @@
+#!/usr/bin/env python3
+
+import os
+
+from infuse_iot.commands import wrapper_from_command_id
+from infuse_iot.rpc_wrappers import application_info, security_public_keys, wifi_scan
+
+assert "TOXTEMPDIR" in os.environ, "you must run these tests using tox"
+
+
+def test_wrapper_from_command_id():
+    def class_test(wrapper_class):
+        assert wrapper_class == wrapper_from_command_id(wrapper_class.COMMAND_ID)
+
+    class_test(application_info.application_info)
+    class_test(wifi_scan.wifi_scan)
+    class_test(security_public_keys.security_public_keys)
+
+    assert wrapper_from_command_id(123456789) is None

tests/test_help.py

@@ -1,3 +1,5 @@
+#!/usr/bin/env python3
+
 import os
 import subprocess
 import sys

tests/test_main.py

@@ -1,3 +1,5 @@
+#!/usr/bin/env python3
+
 import os
 import subprocess
 import sys

tests/test_socket_comms.py

@@ -1,3 +1,5 @@
+#!/usr/bin/env python3
+
 import os
 
 import infuse_iot.socket_comms as comms

tests/util/test_argparse.py

@@ -1,3 +1,5 @@
+#!/usr/bin/env python3
+
 import argparse
 import os
 import pathlib

tests/util/test_crc.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python3
+
+import os
+
+import infuse_iot.util.crc as crc
+
+assert "TOXTEMPDIR" in os.environ, "you must run these tests using tox"
+
+test_string = "123456789"
+test_bytes = test_string.encode("utf-8")
+
+
+def test_crc16_kermit():
+    # Check bytes from https://reveng.sourceforge.io/crc-catalogue/all.htm
+    # Algorithm: CRC-16/KERMIT
+    assert crc.crc16_kermit(test_bytes) == 0x2189
+
+
+def test_crc16_ccitt():
+    # Check bytes from https://reveng.sourceforge.io/crc-catalogue/all.htm
+    # Algorithm: CRC-16/KERMIT
+    assert crc.crc16_ccitt(test_bytes) == 0x2189

tests/util/test_ctypes.py

@@ -1,3 +1,5 @@
+#!/usr/bin/env python3
+
 import ctypes
 import os
 

tests/util/test_threading.py

@@ -1,3 +1,5 @@
+#!/usr/bin/env python3
+
 import os
 import time
 

tests/util/test_time.py

@@ -1,3 +1,5 @@
+#!/usr/bin/env python3
+
 import os
 
 from infuse_iot.util.time import humanised_seconds