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/epacket/packet.py

@@ -159,7 +159,7 @@ def from_serial(cls, database: DeviceDatabase, serial_frame: bytes) -> list[Self
             frame_type = decode_mapping[common_header.interface]
 
             # Extract interface address (Only Bluetooth supported)
-            addr = Address.from_bytes(common_header.interface, packet_bytes)
+            addr = Address.from_bytes(common_header.interface, bytes(packet_bytes))
             del packet_bytes[: addr.len()]
 
             # Decrypting packet

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/gateway.py

@@ -252,7 +252,7 @@ def _handle_serial_frame(self, frame: bytearray):
         try:
             # Decode the serial packet
             try:
-                decoded = PacketReceived.from_serial(self._common.ddb, frame)
+                decoded = PacketReceived.from_serial(self._common.ddb, bytes(frame))
             except NoKeyError:
                 assert self._common.ddb.gateway is not None
                 if not self._common.ddb.has_network_id(self._common.ddb.gateway):

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/tools/serial_throughput.py

@@ -8,16 +8,20 @@
 import random
 import sys
 import time
+from datetime import datetime
 
 from infuse_iot.commands import InfuseCommand
 from infuse_iot.common import InfuseID, InfuseType
+from infuse_iot.definitions.rpc import time_set
 from infuse_iot.epacket.packet import Auth, PacketOutput
+from infuse_iot.rpc_client import RpcClient
 from infuse_iot.socket_comms import (
     ClientNotificationEpacketReceived,
     GatewayRequestEpacketSend,
     LocalClient,
     default_multicast_address,
 )
+from infuse_iot.time import InfuseTime
 
 
 class SubCommand(InfuseCommand):
@@ -38,6 +42,11 @@ def __init__(self, args):
         self._client = LocalClient(default_multicast_address(), 1.0)
         self._iterations = args.iterations
 
+    def run_time_set(self):
+        rpc_client = RpcClient(self._client, 128, InfuseID.GATEWAY)
+        params = time_set.request(InfuseTime.epoch_time_from_unix(datetime.now().timestamp()))
+        rpc_client.run_standard_cmd(time_set.COMMAND_ID, Auth.DEVICE, bytes(params), time_set.response.from_buffer_copy)
+
     def run_send_test(self, num, size, queue_size):
         assert size >= 4
         self._client.set_rx_timeout(0.2)
@@ -85,6 +94,9 @@ def run(self):
         if not self._client.comms_check():
             sys.exit("No communications gateway detected (infuse gateway/bt_native)")
 
+        # Set remote time to avoid key regeneration on every packet
+        self.run_time_set()
+
         # No queuing
         print(f"Averaged across {self._iterations} packets with no queuing:")
         self.run_send_test(self._iterations, 4, 1)

src/infuse_iot/util/crc.py

@@ -0,0 +1,22 @@
+#!/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
+    """
+    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)

src/infuse_iot/util/elftools.py

@@ -2,6 +2,7 @@
 
 import ctypes
 
+from elftools.dwarf.compileunit import CompileUnit
 from elftools.dwarf.die import DIE
 from elftools.dwarf.dwarf_expr import DW_OP_name2opcode
 from elftools.dwarf.dwarfinfo import DWARFInfo
@@ -97,13 +98,16 @@ def dwarf_die_from_symbol(elf: ELFFile, symbol: Symbol) -> DIE | None:
             candidate_cu = CU
             candidate_die = die
             if "DW_AT_location" in die.attributes:
-                die_location = die.attributes.get("DW_AT_location").value
+                die_location = die.attributes.get("DW_AT_location")
+                if die_location is None:
+                    continue
+                die_location_val = die_location.value
                 # Constant addresses are in a list of form [0x03, addr_bytes]
-                if not isinstance(die_location, list):
+                if not isinstance(die_location_val, list):
                     continue
-                if die_location[0] != DW_OP_name2opcode["DW_OP_addr"]:
+                if die_location_val[0] != DW_OP_name2opcode["DW_OP_addr"]:
                     continue
-                address = int.from_bytes(die_location[1:], "little")
+                address = int.from_bytes(die_location_val[1:], "little")
                 if address == symbol.entry["st_value"]:
                     return die
 
@@ -128,11 +132,11 @@ def dwarf_die_file_info(elf: ELFFile, die: DIE) -> tuple[str | None, int]:
     line_attr = die.attributes["DW_AT_decl_line"]
 
     dwarfinfo = elf.get_dwarf_info()
-    lineprogram = dwarfinfo.line_program_for_CU(die.cu)
-    if lineprogram is None:
-        cu_filename = None
-    else:
-        cu_filename = lineprogram["file_entry"][file_attr.value - 1].name.decode("latin-1")
+    cu_filename: str | None = None
+    if isinstance(die.cu, CompileUnit):
+        lineprogram = dwarfinfo.line_program_for_CU(die.cu)
+        if lineprogram is not None:
+            cu_filename = lineprogram["file_entry"][file_attr.value - 1].name.decode("latin-1")
 
     return cu_filename, line_attr.value
 
@@ -158,7 +162,10 @@ def __init__(
 
 def _type_from_dwarf_info(dwarfinfo: DWARFInfo, die: DIE):
     refaddr = die.attributes["DW_AT_type"].value + die.cu.cu_offset
-    return dwarfinfo.get_DIE_from_refaddr(refaddr, die.cu)
+    cu: CompileUnit | None = None
+    if isinstance(die.cu, CompileUnit):
+        cu = die.cu
+    return dwarfinfo.get_DIE_from_refaddr(refaddr, cu)
 
 
 def dwarf_die_variable_inf(

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