util: ctypes: handle vla_counted_by

src/infuse_iot/rpc_wrappers/data_logger_read.py

@@ -31,7 +31,7 @@ def __init__(self, args):
         else:
             raise NotImplementedError
         self.expected_offset = 0
-        self.output = b""
+        self.output = bytearray()
         self.start_time = time.time()
 
     def request_struct(self):
@@ -43,14 +43,20 @@ def request_json(self):
     def data_recv_cb(self, offset: int, data: bytes) -> None:
         if self.expected_offset == 0:
             self.start_time = time.time()
-        if offset != self.expected_offset:
+        if offset == self.expected_offset:
+            self.output += data
+            # Next expected offset
+            self.expected_offset = offset + len(data)
+        else:
             missing = offset - self.expected_offset
-            print(f"Missed {missing:d} bytes from offset 0x{self.expected_offset:08x}")
-            self.output += b"\x00" * missing
-
-        self.output += data
-        # Next expected offset
-        self.expected_offset = offset + len(data)
+            if missing > 0:
+                print(f"Missed {missing:d} bytes from offset 0x{self.expected_offset:08x}")
+                self.output += b"\x00" * missing
+                self.output += data
+                self.expected_offset = offset + len(data)
+            else:
+                print(f"Received missing bytes from offset 0x{self.expected_offset:08x}")
+                self.output[offset : offset + len(data)] = data
 
     def handle_response(self, return_code, response):
         end_time = time.time()

src/infuse_iot/rpc_wrappers/kv_read.py

@@ -12,48 +12,6 @@
 
 
 class kv_read(InfuseRpcCommand, defs.kv_read):
-    class request(ctypes.LittleEndianStructure):
-        _fields_ = [
-            ("num", ctypes.c_uint8),
-        ]
-        _pack_ = 1
-
-    class response:
-        @classmethod
-        def from_buffer_copy(cls, source: bytes, _offset: int = 0) -> list:
-            values = []
-            while len(source) > 0:
-
-                class kv_store_header(ctypes.LittleEndianStructure):
-                    _fields_ = [
-                        ("id", ctypes.c_uint16),
-                        ("len", ctypes.c_int16),
-                    ]
-                    _pack_ = 1
-
-                header = kv_store_header.from_buffer_copy(source)
-                struct: ctypes.LittleEndianStructure
-                if header.len > 0:
-
-                    class kv_store_value(ctypes.LittleEndianStructure):
-                        _fields_ = [
-                            ("id", ctypes.c_uint16),
-                            ("len", ctypes.c_int16),
-                            ("data", ctypes.c_ubyte * header.len),
-                        ]
-                        _pack_ = 1
-
-                    struct = kv_store_value.from_buffer_copy(source)
-                else:
-                    struct = header
-                values.append(struct)
-                source = source[ctypes.sizeof(struct) :]
-            return values
-
-        @classmethod
-        def vla_from_buffer_copy(cls, source: bytes, offset: int = 0) -> list:
-            return cls.from_buffer_copy(source, offset)
-
     @classmethod
     def add_parser(cls, parser):
         parser.add_argument("--keys", "-k", required=True, type=int, nargs="+", help="Keys to read")
@@ -73,7 +31,7 @@ def handle_response(self, return_code, response):
             print(f"Invalid data buffer ({errno.strerror(-return_code)})")
             return
 
-        for r in response:
+        for r in response.values:
             if r.len > 0:
                 b = bytes(r.data)
                 try:

src/infuse_iot/rpc_wrappers/lte_modem_info.py

@@ -15,12 +15,6 @@ class lte_modem_info(kv_read.kv_read):
     HELP = "Get LTE modem information"
     DESCRIPTION = "Get LTE modem information"
 
-    class request(kv_read.kv_read.request):
-        pass
-
-    class response(kv_read.kv_read.response):
-        pass
-
     @classmethod
     def add_parser(cls, parser):
         return
@@ -43,9 +37,9 @@ def str_decode(r):
                 return "Unknown"
             return str(kv_structs.kv_string.vla_from_buffer_copy(bytes(r.data)))
 
-        modem_imei = struct_decode(kv_slots.lte_modem_imei, response[3])
-        pdp_ctx = struct_decode(kv_slots.lte_pdp_config, response[5])
-        system_modes = struct_decode(kv_slots.lte_networking_modes, response[6])
+        modem_imei = struct_decode(kv_slots.lte_modem_imei, response.values[3])
+        pdp_ctx = struct_decode(kv_slots.lte_pdp_config, response.values[5])
+        system_modes = struct_decode(kv_slots.lte_networking_modes, response.values[6])
 
         if pdp_ctx:
             pdp_str = f'"{str(pdp_ctx.apn)}" ({lte_pdp_ctx.lte_pdp_ctx.PDPFamily(pdp_ctx.family).name})'
@@ -58,10 +52,10 @@ def str_decode(r):
         else:
             modes_str = "default"
 
-        print(f"\t   Model: {str_decode(response[0])}")
-        print(f"\tFirmware: {str_decode(response[1])}")
-        print(f"\t     ESN: {str_decode(response[2])}")
+        print(f"\t   Model: {str_decode(response.values[0])}")
+        print(f"\tFirmware: {str_decode(response.values[1])}")
+        print(f"\t     ESN: {str_decode(response.values[2])}")
         print(f"\t    IMEI: {modem_imei.imei}")
-        print(f"\t     SIM: {str_decode(response[4])}")
+        print(f"\t     SIM: {str_decode(response.values[4])}")
         print(f"\t     APN: {pdp_str}")
         print(f"\t    Mode: {modes_str}")

src/infuse_iot/rpc_wrappers/wifi_scan.py

@@ -1,7 +1,5 @@
 #!/usr/bin/env python3
 
-import ctypes
-
 import tabulate
 
 import infuse_iot.definitions.rpc as defs
@@ -11,43 +9,6 @@
 
 
 class wifi_scan(InfuseRpcCommand, defs.wifi_scan):
-    class response(ctypes.LittleEndianStructure):
-        @classmethod
-        def from_buffer_copy(cls, source, offset=0):
-            values = []
-            source = source[1:]
-            while len(source) > 0:
-
-                class scan_rsp_header(ctypes.LittleEndianStructure):
-                    _fields_ = [
-                        ("band", ctypes.c_uint8),
-                        ("channel", ctypes.c_uint8),
-                        ("security", ctypes.c_uint8),
-                        ("rssi", ctypes.c_int8),
-                        ("bssid", 6 * ctypes.c_char),
-                        ("ssid_length", ctypes.c_uint8),
-                    ]
-                    _pack_ = 1
-
-                header = scan_rsp_header.from_buffer_copy(source)
-
-                class scan_result(ctypes.LittleEndianStructure):
-                    _fields_ = [
-                        ("band", ctypes.c_uint8),
-                        ("channel", ctypes.c_uint8),
-                        ("security", ctypes.c_uint8),
-                        ("rssi", ctypes.c_int8),
-                        ("bssid", 6 * ctypes.c_char),
-                        ("ssid_length", ctypes.c_uint8),
-                        ("ssid", header.ssid_length * ctypes.c_char),
-                    ]
-                    _pack_ = 1
-
-                struct = scan_result.from_buffer_copy(source)
-                values.append(struct)
-                source = source[ctypes.sizeof(struct) :]
-            return values
-
     @classmethod
     def add_parser(cls, parser):
         return
@@ -64,12 +25,12 @@ def handle_response(self, return_code, response):
             return
 
         table = []
-        for network in response:
+        for network in response.networks:
             bssid = ":".join([f"{b:02x}" for b in network.bssid])
 
             table.append(
                 [
-                    network.ssid.decode("utf-8"),
+                    bytes(network.ssid).decode("utf-8"),
                     bssid,
                     str(z_wifi.FrequencyBand(network.band)),
                     network.channel,

src/infuse_iot/tools/provision.py

@@ -66,13 +66,17 @@ def add_parser(cls, parser):
             type=str,
             help="Define a number of key-value pairs for metadata",
         )
+        parser.add_argument(
+            "--dry-run", action="store_true", help="Generate the request that would be sent, but do not send it"
+        )
 
     def __init__(self, args):
         self._vendor = args.vendor
         self._snr = args.snr
         self._board = args.board
         self._org = args.organisation
         self._id = args.id
+        self._dry_run = args.dry_run
         self._metadata = {}
         if args.metadata:
             for meta in args.metadata:
@@ -112,6 +116,10 @@ def create_device(self, client: Client, soc_name: str, hardware_id_str: str):
         if self._id:
             new_board.device_id = f"{self._id:016x}"
 
+        if self._dry_run:
+            print(new_board)
+            return
+
         response = create_device.sync_detailed(client=client, body=new_board)
         if response.status_code != HTTPStatus.CREATED:
             sys.exit(f"Failed to create device:\n\t<{response.status_code}> {response.content.decode('utf-8')}")
@@ -141,6 +149,9 @@ def run(self):
             elif response.status_code == HTTPStatus.NOT_FOUND:
                 # Create new device here
                 self.create_device(client, interface.soc_name, hardware_id_str)
+                # Exit if dry run only
+                if self._dry_run:
+                    return
                 # Query information back out
                 response = get_device_by_soc_and_mcu_id.sync_detailed(
                     client=client, soc=interface.soc_name, mcu_id=hardware_id_str

src/infuse_iot/tools/rpc_cloud.py

@@ -20,6 +20,7 @@
 from infuse_iot.api_client.models.downlink_message_status import DownlinkMessageStatus
 from infuse_iot.commands import InfuseCommand, InfuseRpcCommand
 from infuse_iot.credentials import get_api_key
+from infuse_iot.definitions.rpc import id_type_mapping
 
 
 class SubCommand(InfuseCommand):
@@ -99,8 +100,14 @@ def query(self, client: Client):
                 else:
                     print(f"  Through: Direct ({route.interface.upper()})")
         if downlink.status == DownlinkMessageStatus.COMPLETED:
+            rpc_req = downlink.rpc_req
             rpc_rsp = downlink.rpc_rsp
             assert isinstance(rpc_rsp, RpcRsp)
+            try:
+                command_name = id_type_mapping[rpc_req.command_id].NAME
+            except KeyError:
+                command_name = "Unknown"
+            print(f"   RPC ID: {rpc_req.command_id} ({command_name})")
             print(f"   Result: {rpc_rsp.return_code}")
             if rpc_rsp.params:
                 print(json.dumps(rpc_rsp.params.additional_properties, indent=4))

src/infuse_iot/util/ctypes.py

@@ -32,6 +32,7 @@ class VLACompatLittleEndianStruct(ctypes.LittleEndianStructure):
     """
 
     vla_field: tuple[str, type[Any]] | None = None
+    vla_counted_by: str | None = None
 
     @classmethod
     def vla_from_buffer_copy(cls, source, offset=0) -> Self:
@@ -42,23 +43,46 @@ class property is not `None`, it will consume the remainder of
         """
 
         base = cls.from_buffer_copy(source, offset)
+        vla_val: list | VLACompatLittleEndianStruct
         if cls.vla_field is None:
             return base
 
         remainder = source[ctypes.sizeof(cls) :]
         vla_field_name, vla_field_type = cls.vla_field  # type: ignore
 
         if issubclass(vla_field_type, ctypes.Array):
-            array_base: ctypes._CData = vla_field_type._type_  # type: ignore
+            array_base: ctypes._PyCSimpleType = vla_field_type._type_  # type: ignore
+            if hasattr(array_base, "vla_counted_by"):
+                # This is an array of VLA arrays where the sub-arrys define their own length
+                vla_val = []
+                # Consume all remaining buffer bytes
+                while len(remainder) > 0:
+                    sub_vla_field_name, sub_vla_field_type = array_base.vla_field  # type: ignore
+                    sub_array_base: ctypes._CData = sub_vla_field_type._type_  # type: ignore
+                    sub_base = array_base.from_buffer_copy(remainder)
+                    sub_base_size = ctypes.sizeof(sub_base)
+                    sub_count = getattr(sub_base, array_base.vla_counted_by)
+                    if sub_count < 0:
+                        # Assume that negative length is an error code and use 0
+                        vla_val.append(sub_base)
+                    else:
+                        sub_vla_type = sub_count * sub_array_base
+                        # Don't use ctypes.sizeof on constructed type, it returns the wrong value
+                        sub_vla_size = sub_count * ctypes.sizeof(sub_array_base)
+                        sub_vla_val = sub_vla_type.from_buffer_copy(remainder[sub_base_size:])
+                        setattr(sub_base, sub_vla_field_name, sub_vla_val)
+                        vla_val.append(sub_base)
+                    remainder = remainder[sub_base_size + sub_vla_size :]
+            else:
+                # Determine the number of VLA elements on "source"
+                vla_byte_len = (len(source) - offset) - ctypes.sizeof(cls)
+                vla_element_size = ctypes.sizeof(array_base)
+                if vla_byte_len % vla_element_size != 0:
+                    raise TypeError(f"Unaligned VLA buffer for {cls} (len {len(source)})")
+                vla_num = vla_byte_len // vla_element_size
+                vla_type = vla_num * array_base
+                vla_val = vla_type.from_buffer_copy(remainder)
 
-            # Determine the number of VLA elements on "source"
-            vla_byte_len = (len(source) - offset) - ctypes.sizeof(cls)
-            vla_element_size = ctypes.sizeof(array_base)
-            if vla_byte_len % vla_element_size != 0:
-                raise TypeError(f"Unaligned VLA buffer for {cls} (len {len(source)})")
-            vla_num = vla_byte_len // vla_element_size
-            vla_type = vla_num * array_base
-            vla_val = vla_type.from_buffer_copy(remainder)
         elif issubclass(vla_field_type, VLACompatLittleEndianStruct):
             vla_val = vla_field_type.vla_from_buffer_copy(remainder)
         else:

src/infuse_iot/zephyr/lte.py

@@ -4,6 +4,8 @@
 
 
 class RegistrationState(enum.IntEnum):
+    """Network registration state (3GPP TS 127.007)"""
+
     NOT_REGISTERED = 0
     REGISTERED_HOME = 1
     SEARCHING = 2
@@ -26,15 +28,47 @@ def __str__(self):
 
 
 class AccessTechnology(enum.IntEnum):
-    NONE = 0
-    LTE_M = 7
-    NB_IOT = 9
+    """Access Technology (3GPP TS 127.007)"""
+
+    GSM = 0
+    GSM_COMPACT = 1
+    UTRAN = 2
+    GSM_EGPRS = 3
+    UTRAN_HSDPA = 4
+    UTRAN_HSUPA = 5
+    UTRAN_HSDPA_HSUPA = 6
+    E_UTRAN = 7
+    EC_GSM_IOT = 8
+    E_UTRAN_NB_S1 = 9
+    E_UTRA_5G_CN = 10
+    NR_5G_CN = 11
+    NG_RAN = 12
+    E_UTRA_NR_DUAL = 13
+    E_UTRAN_NB_S1_SAT = 14
+    E_UTRAN_WB_S1_SAT = 15
+    NG_RAN_SAT = 16
+    UNKNOWN = 255
 
     def __str__(self):
         pretty_names = {
-            self.NONE: "None",
-            self.LTE_M: "LTE-M",
-            self.NB_IOT: "NB-IoT",
+            self.GSM: "GSM (2G, 3GPP Rel 99)",
+            self.GSM_COMPACT: "GSM Compact (2G, 3GPP Rel 99)",
+            self.UTRAN: "UTRAN (3G, 3GPP Rel 99)",
+            self.GSM_EGPRS: "GSM Enhanced (2.5G, 3GPP Rel 99)",
+            self.UTRAN_HSDPA: "UTRAN High Speed Downlink (3.5G, 3GPP Rel 5)",
+            self.UTRAN_HSUPA: "UTRAN High Speed Uplink (3.75G, 3GPP Rel 6)",
+            self.UTRAN_HSDPA_HSUPA: "UTRAN High Speed Uplink/Downlink (3.75G, 3GPP Rel 6)",
+            self.E_UTRAN: "LTE/Evolved UTRAN (4G, 3GPP Rel 8)",
+            self.EC_GSM_IOT: "Extended Coverage GSM for IoT (2G, 3GPP Rel 13)",
+            self.E_UTRAN_NB_S1: "EUTRAN Narrowband-IoT (4G, 3GPP Rel 13)",
+            self.E_UTRA_5G_CN: "LTE/E-UTRA connected to 5G Core Network (5G, 3GPP Rel 15)",
+            self.NR_5G_CN: "New Radio with 5G Core Network (5G, 3GPP Rel 15)",
+            self.NG_RAN: "Next Generation RAN (5G, 3GPP Rel 15)",
+            self.E_UTRA_NR_DUAL: "LTE/E-UTRA & NR dual connectivity (5G, 3GPP Rel 15)",
+            self.E_UTRAN_NB_S1_SAT: "Narrowband-IoT over Satellite (4G, 3GPP Rel 17)",
+            self.E_UTRAN_WB_S1_SAT: "LTE (wideband) over Satellite (4G, 3GPP Rel 17)",
+            self.NG_RAN_SAT: "Next Generation RAN over Satellite (5G, 3GPP Rel 17)",
+            self.UNKNOWN: "Unknown",
         }
         return pretty_names[self]