From a1fa954620b51dca286cb14bbb11a91195db7d49 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Wed, 8 Oct 2025 11:36:56 +1000 Subject: [PATCH 1/7] generated: regenerate RPC definitions Regenerate RPC definitions to get new information. Signed-off-by: Jordan Yates --- src/infuse_iot/generated/rpc_definitions.py | 157 +++++++++++++++----- src/infuse_iot/util/ctypes.py | 1 + 2 files changed, 122 insertions(+), 36 deletions(-) diff --git a/src/infuse_iot/generated/rpc_definitions.py b/src/infuse_iot/generated/rpc_definitions.py index cf5ca4f..e56643d 100644 --- a/src/infuse_iot/generated/rpc_definitions.py +++ b/src/infuse_iot/generated/rpc_definitions.py @@ -29,6 +29,7 @@ class rpc_struct_kv_store_value(VLACompatLittleEndianStruct): ("len", ctypes.c_int16), ] vla_field = ("data", 0 * ctypes.c_uint8) + vla_counted_by = "len" _pack_ = 1 @@ -137,6 +138,7 @@ class rpc_struct_wifi_scan_result(VLACompatLittleEndianStruct): ("ssid_len", ctypes.c_uint8), ] vla_field = ("ssid", 0 * ctypes.c_char) + vla_counted_by = "ssid_len" _pack_ = 1 @@ -230,9 +232,17 @@ class rpc_enum_zperf_data_source(enum.IntEnum): ENCRYPT = 128 -class reboot: +class RPCDefinitionBase: + NAME: str + HELP: str + DESCRIPTION: str + COMMAND_ID: int + + +class reboot(RPCDefinitionBase): """Reboot the device after a delay""" + NAME = "reboot" HELP = "Reboot the device after a delay" DESCRIPTION = "Reboot the device after a delay" COMMAND_ID = 1 @@ -250,9 +260,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class fault: +class fault(RPCDefinitionBase): """Immediately trigger an exception on the device""" + NAME = "fault" HELP = "Immediately trigger an exception on the device" DESCRIPTION = "Immediately trigger an exception on the device" COMMAND_ID = 2 @@ -269,9 +280,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class time_get: +class time_get(RPCDefinitionBase): """Get the current time knowledge of the device""" + NAME = "time_get" HELP = "Get the current time knowledge of the device" DESCRIPTION = "Get the current time knowledge of the device" COMMAND_ID = 3 @@ -289,9 +301,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class time_set: +class time_set(RPCDefinitionBase): """Set the current time of the device""" + NAME = "time_set" HELP = "Set the current time of the device" DESCRIPTION = "Set the current time of the device" COMMAND_ID = 4 @@ -307,9 +320,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class kv_write: +class kv_write(RPCDefinitionBase): """Write values to the KV store""" + NAME = "kv_write" HELP = "Write values to the KV store" DESCRIPTION = "Write values to the KV store" COMMAND_ID = 5 @@ -327,9 +341,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class kv_read: +class kv_read(RPCDefinitionBase): """Read values from the KV store""" + NAME = "kv_read" HELP = "Read values from the KV store" DESCRIPTION = "Read values from the KV store" COMMAND_ID = 6 @@ -347,9 +362,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class kv_reflect_crcs: +class kv_reflect_crcs(RPCDefinitionBase): """Read KV store CRC's""" + NAME = "kv_reflect_crcs" HELP = "Read KV store CRC's" DESCRIPTION = "Read KV store CRC's" COMMAND_ID = 7 @@ -369,9 +385,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class zbus_channel_state: +class zbus_channel_state(RPCDefinitionBase): """Query current state of zbus channel""" + NAME = "zbus_channel_state" HELP = "Query current state of zbus channel" DESCRIPTION = "Query current state of zbus channel" COMMAND_ID = 8 @@ -392,9 +409,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class application_info: +class application_info(RPCDefinitionBase): """Query basic application versions and state""" + NAME = "application_info" HELP = "Query basic application versions and state" DESCRIPTION = "Query basic application versions and state" COMMAND_ID = 9 @@ -417,9 +435,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class wifi_scan: +class wifi_scan(RPCDefinitionBase): """Scan for WiFi networks""" + NAME = "wifi_scan" HELP = "Scan for WiFi networks" DESCRIPTION = "Scan for WiFi networks" COMMAND_ID = 10 @@ -436,9 +455,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class wifi_state: +class wifi_state(RPCDefinitionBase): """Get current WiFi interface state""" + NAME = "wifi_state" HELP = "Get current WiFi interface state" DESCRIPTION = "Get current WiFi interface state" COMMAND_ID = 11 @@ -455,9 +475,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class last_reboot: +class last_reboot(RPCDefinitionBase): """Retrieve information pertaining to the previous reboot""" + NAME = "last_reboot" HELP = "Retrieve information pertaining to the previous reboot" DESCRIPTION = "Retrieve information pertaining to the previous reboot" COMMAND_ID = 12 @@ -481,9 +502,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class data_logger_state: +class data_logger_state(RPCDefinitionBase): """Get state of a data logger""" + NAME = "data_logger_state" HELP = "Get state of a data logger" DESCRIPTION = "Get state of a data logger" COMMAND_ID = 13 @@ -510,9 +532,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class data_logger_read: +class data_logger_read(RPCDefinitionBase): """Read data from data logger""" + NAME = "data_logger_read" HELP = "Read data from data logger" DESCRIPTION = "Read data from data logger" COMMAND_ID = 14 @@ -533,9 +556,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class mem_read: +class mem_read(RPCDefinitionBase): """Read arbitrary memory (NO ADDRESS VALIDATION PERFORMED)""" + NAME = "mem_read" HELP = "Read arbitrary memory (NO ADDRESS VALIDATION PERFORMED)" DESCRIPTION = "Read arbitrary memory (NO ADDRESS VALIDATION PERFORMED)" COMMAND_ID = 15 @@ -554,9 +578,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class infuse_states_query: +class infuse_states_query(RPCDefinitionBase): """Read current Infuse-IoT application states""" + NAME = "infuse_states_query" HELP = "Read current Infuse-IoT application states" DESCRIPTION = "Read current Infuse-IoT application states" COMMAND_ID = 16 @@ -575,9 +600,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class infuse_states_update: +class infuse_states_update(RPCDefinitionBase): """Update Infuse-IoT application states""" + NAME = "infuse_states_update" HELP = "Update Infuse-IoT application states" DESCRIPTION = "Update Infuse-IoT application states" COMMAND_ID = 17 @@ -594,9 +620,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class data_logger_erase: +class data_logger_erase(RPCDefinitionBase): """Erase all data from a data logger""" + NAME = "data_logger_erase" HELP = "Erase all data from a data logger" DESCRIPTION = "Erase all data from a data logger" COMMAND_ID = 18 @@ -613,9 +640,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class heap_stats: +class heap_stats(RPCDefinitionBase): """Query stats of heaps""" + NAME = "heap_stats" HELP = "Query stats of heaps" DESCRIPTION = "Query stats of heaps" COMMAND_ID = 19 @@ -630,9 +658,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class lte_at_cmd: +class lte_at_cmd(RPCDefinitionBase): """Run AT command against LTE modem""" + NAME = "lte_at_cmd" HELP = "Run AT command against LTE modem" DESCRIPTION = "Run AT command against LTE modem" COMMAND_ID = 20 @@ -648,9 +677,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class lte_state: +class lte_state(RPCDefinitionBase): """Get current LTE interface state""" + NAME = "lte_state" HELP = "Get current LTE interface state" DESCRIPTION = "Get current LTE interface state" COMMAND_ID = 21 @@ -667,9 +697,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class data_logger_read_available: +class data_logger_read_available(RPCDefinitionBase): """Read data from data logger, with auto-updating start_block""" + NAME = "data_logger_read_available" HELP = "Read data from data logger, with auto-updating start_block" DESCRIPTION = "Read data from data logger, with auto-updating start_block" COMMAND_ID = 22 @@ -693,9 +724,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class coap_download: +class coap_download(RPCDefinitionBase): """Download a file from a COAP server (Infuse-IoT DTLS protected)""" + NAME = "coap_download" HELP = "Download a file from a COAP server (Infuse-IoT DTLS protected)" DESCRIPTION = "Download a file from a COAP server (Infuse-IoT DTLS protected)" COMMAND_ID = 30 @@ -720,9 +752,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class zperf_upload: +class zperf_upload(RPCDefinitionBase): """Network upload bandwidth testing using zperf/iperf""" + NAME = "zperf_upload" HELP = "Network upload bandwidth testing using zperf/iperf" DESCRIPTION = "Network upload bandwidth testing using zperf/iperf" COMMAND_ID = 31 @@ -754,9 +787,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class file_write_basic: +class file_write_basic(RPCDefinitionBase): """Write a file to the device""" + NAME = "file_write_basic" HELP = "Write a file to the device" DESCRIPTION = "Write a file to the device" COMMAND_ID = 40 @@ -776,9 +810,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class annotate: +class annotate(RPCDefinitionBase): """Write an annotation to the device""" + NAME = "annotate" HELP = "Write an annotation to the device" DESCRIPTION = "Write an annotation to the device" COMMAND_ID = 41 @@ -796,9 +831,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class bt_connect_infuse: +class bt_connect_infuse(RPCDefinitionBase): """Connect to an Infuse-IoT Bluetooth device""" + NAME = "bt_connect_infuse" HELP = "Connect to an Infuse-IoT Bluetooth device" DESCRIPTION = "Connect to an Infuse-IoT Bluetooth device" COMMAND_ID = 50 @@ -822,9 +858,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class bt_disconnect: +class bt_disconnect(RPCDefinitionBase): """Disconnect from a Bluetooth device""" + NAME = "bt_disconnect" HELP = "Disconnect from a Bluetooth device" DESCRIPTION = "Disconnect from a Bluetooth device" COMMAND_ID = 51 @@ -840,9 +877,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class bt_file_copy_basic: +class bt_file_copy_basic(RPCDefinitionBase): """Copy a local file to a remote device over Bluetooth""" + NAME = "bt_file_copy_basic" HELP = "Copy a local file to a remote device over Bluetooth" DESCRIPTION = "Copy a local file to a remote device over Bluetooth" COMMAND_ID = 52 @@ -864,9 +902,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class bt_file_copy_coap: +class bt_file_copy_coap(RPCDefinitionBase): """Copy a file fetched from COAP to a remote device over Bluetooth""" + NAME = "bt_file_copy_coap" HELP = "Copy a file fetched from COAP to a remote device over Bluetooth" DESCRIPTION = "Copy a file fetched from COAP to a remote device over Bluetooth" COMMAND_ID = 53 @@ -896,9 +935,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class bt_mcumgr_reboot: +class bt_mcumgr_reboot(RPCDefinitionBase): """Connect to a Bluetooth device and run the MCUMGR reboot command""" + NAME = "bt_mcumgr_reboot" HELP = "Connect to a Bluetooth device and run the MCUMGR reboot command" DESCRIPTION = "Connect to a Bluetooth device and run the MCUMGR reboot command" COMMAND_ID = 54 @@ -915,9 +955,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class gravity_reference_update: +class gravity_reference_update(RPCDefinitionBase): """Store the current accelerometer vector as the gravity reference""" + NAME = "gravity_reference_update" HELP = "Store the current accelerometer vector as the gravity reference" DESCRIPTION = "Store the current accelerometer vector as the gravity reference" COMMAND_ID = 60 @@ -938,9 +979,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class security_state: +class security_state(RPCDefinitionBase): """Query current security state and validate identity""" + NAME = "security_state" HELP = "Query current security state and validate identity" DESCRIPTION = "Query current security state and validate identity" COMMAND_ID = 30000 @@ -962,9 +1004,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class data_sender: +class data_sender(RPCDefinitionBase): """Send multiple INFUSE_RPC_DATA packets""" + NAME = "data_sender" HELP = "Send multiple INFUSE_RPC_DATA packets" DESCRIPTION = "Send multiple INFUSE_RPC_DATA packets" COMMAND_ID = 32765 @@ -978,9 +1021,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class data_receiver: +class data_receiver(RPCDefinitionBase): """Receive multiple INFUSE_RPC_DATA packets""" + NAME = "data_receiver" HELP = "Receive multiple INFUSE_RPC_DATA packets" DESCRIPTION = "Receive multiple INFUSE_RPC_DATA packets" COMMAND_ID = 32766 @@ -999,9 +1043,10 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 -class echo: +class echo(RPCDefinitionBase): """Echo any input data in the response""" + NAME = "echo" HELP = "Echo any input data in the response" DESCRIPTION = "Echo any input data in the response" COMMAND_ID = 32767 @@ -1017,7 +1062,47 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 +id_type_mapping: dict[int, type[RPCDefinitionBase]] = { + reboot.COMMAND_ID: reboot, + fault.COMMAND_ID: fault, + time_get.COMMAND_ID: time_get, + time_set.COMMAND_ID: time_set, + kv_write.COMMAND_ID: kv_write, + kv_read.COMMAND_ID: kv_read, + kv_reflect_crcs.COMMAND_ID: kv_reflect_crcs, + zbus_channel_state.COMMAND_ID: zbus_channel_state, + application_info.COMMAND_ID: application_info, + wifi_scan.COMMAND_ID: wifi_scan, + wifi_state.COMMAND_ID: wifi_state, + last_reboot.COMMAND_ID: last_reboot, + data_logger_state.COMMAND_ID: data_logger_state, + data_logger_read.COMMAND_ID: data_logger_read, + mem_read.COMMAND_ID: mem_read, + infuse_states_query.COMMAND_ID: infuse_states_query, + infuse_states_update.COMMAND_ID: infuse_states_update, + data_logger_erase.COMMAND_ID: data_logger_erase, + heap_stats.COMMAND_ID: heap_stats, + lte_at_cmd.COMMAND_ID: lte_at_cmd, + lte_state.COMMAND_ID: lte_state, + data_logger_read_available.COMMAND_ID: data_logger_read_available, + coap_download.COMMAND_ID: coap_download, + zperf_upload.COMMAND_ID: zperf_upload, + file_write_basic.COMMAND_ID: file_write_basic, + annotate.COMMAND_ID: annotate, + bt_connect_infuse.COMMAND_ID: bt_connect_infuse, + bt_disconnect.COMMAND_ID: bt_disconnect, + bt_file_copy_basic.COMMAND_ID: bt_file_copy_basic, + bt_file_copy_coap.COMMAND_ID: bt_file_copy_coap, + bt_mcumgr_reboot.COMMAND_ID: bt_mcumgr_reboot, + gravity_reference_update.COMMAND_ID: gravity_reference_update, + security_state.COMMAND_ID: security_state, + data_sender.COMMAND_ID: data_sender, + data_receiver.COMMAND_ID: data_receiver, + echo.COMMAND_ID: echo, +} + __all__ = [ + "id_type_mapping", "rpc_struct_mcuboot_img_sem_ver", "rpc_struct_kv_store_value", "rpc_struct_kv_store_crc", diff --git a/src/infuse_iot/util/ctypes.py b/src/infuse_iot/util/ctypes.py index 96708df..fe30dd5 100644 --- a/src/infuse_iot/util/ctypes.py +++ b/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: From 200da533108dc08575cbbdf31a2a150a35488434 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Wed, 8 Oct 2025 11:33:44 +1000 Subject: [PATCH 2/7] tools: rpc_cloud: print command ID on query Print the command ID (and name) when displaying RPC results. Signed-off-by: Jordan Yates --- src/infuse_iot/tools/rpc_cloud.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/infuse_iot/tools/rpc_cloud.py b/src/infuse_iot/tools/rpc_cloud.py index d92b44f..f6cbbe6 100644 --- a/src/infuse_iot/tools/rpc_cloud.py +++ b/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)) From eb4d17963ec91bb2c398fa604a351f75604ddfe6 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Sat, 11 Oct 2025 16:49:08 +1000 Subject: [PATCH 3/7] rpc_wrappers: data_logger_read: handle unordered data Properly handle data that is received out of order. Signed-off-by: Jordan Yates --- .../rpc_wrappers/data_logger_read.py | 22 ++++++++++++------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/src/infuse_iot/rpc_wrappers/data_logger_read.py b/src/infuse_iot/rpc_wrappers/data_logger_read.py index 6d719eb..75ae4f8 100644 --- a/src/infuse_iot/rpc_wrappers/data_logger_read.py +++ b/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() From c5d98e90fe0e06c45bd88ef65441b7c910798e4b Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Mon, 13 Oct 2025 18:03:46 +1000 Subject: [PATCH 4/7] tools: provision: `--dry-run` Add the `--dry-run` option so that users can see what parameters would be used without actually creating the board. Signed-off-by: Jordan Yates --- src/infuse_iot/tools/provision.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/src/infuse_iot/tools/provision.py b/src/infuse_iot/tools/provision.py index 0e56d3b..c3d2506 100644 --- a/src/infuse_iot/tools/provision.py +++ b/src/infuse_iot/tools/provision.py @@ -66,6 +66,9 @@ 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 @@ -73,6 +76,7 @@ def __init__(self, args): 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 From 6e7af7a48b7384455cfe45b222231b0b7e8f0fd3 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Mon, 13 Oct 2025 18:04:26 +1000 Subject: [PATCH 5/7] zephyr: lte: align constants with 3GPP TS 127.007 Align the LTE constants with 3GPP TS 127.007, as was done in the embedded repository. Signed-off-by: Jordan Yates --- src/infuse_iot/zephyr/lte.py | 46 +++++++++++++++++++++++++++++++----- 1 file changed, 40 insertions(+), 6 deletions(-) diff --git a/src/infuse_iot/zephyr/lte.py b/src/infuse_iot/zephyr/lte.py index 3d24089..0362652 100644 --- a/src/infuse_iot/zephyr/lte.py +++ b/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] From ff24fc490d6486bff844cb2cf4ce855e9ac3be2a Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Mon, 13 Oct 2025 22:09:45 +1000 Subject: [PATCH 6/7] util: ctypes: handle `vla_counted_by` Handle automatically constructing VLA structs with the `vla_counted_by` attribute. Signed-off-by: Jordan Yates --- src/infuse_iot/util/ctypes.py | 41 +++++++++++++++++++++++++++-------- tests/util/test_ctypes.py | 39 +++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 9 deletions(-) diff --git a/src/infuse_iot/util/ctypes.py b/src/infuse_iot/util/ctypes.py index fe30dd5..e77602c 100644 --- a/src/infuse_iot/util/ctypes.py +++ b/src/infuse_iot/util/ctypes.py @@ -43,6 +43,7 @@ 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 @@ -50,16 +51,38 @@ class property is not `None`, it will consume the remainder of 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: diff --git a/tests/util/test_ctypes.py b/tests/util/test_ctypes.py index 110a203..21c6292 100644 --- a/tests/util/test_ctypes.py +++ b/tests/util/test_ctypes.py @@ -11,6 +11,16 @@ class VLABase(VLACompatLittleEndianStruct): ("first", ctypes.c_uint32), ] vla_field = ("vla", 0 * ctypes.c_uint32) + _pack_ = 1 + + +class VLACountedBy(VLACompatLittleEndianStruct): + _fields_ = [ + ("count", ctypes.c_uint8), + ] + vla_field = ("data", 0 * ctypes.c_uint8) + vla_counted_by = "count" + _pack_ = 1 class VLANested(VLACompatLittleEndianStruct): @@ -18,6 +28,15 @@ class VLANested(VLACompatLittleEndianStruct): ("first", ctypes.c_uint32), ] vla_field = ("vla", VLABase) + _pack_ = 1 + + +class VLADualNested(VLACompatLittleEndianStruct): + _fields_ = [ + ("num", ctypes.c_uint8), + ] + vla_field = ("values", 0 * VLACountedBy) + _pack_ = 1 class VLANone(VLACompatLittleEndianStruct): @@ -25,6 +44,7 @@ class VLANone(VLACompatLittleEndianStruct): ("first", ctypes.c_uint32), ("second", ctypes.c_uint32), ] + _pack_ = 1 def test_vla_compat_struct(): @@ -47,6 +67,25 @@ def test_vla_compat_struct(): assert none.first == 2 assert none.second == 3 + one_element = b"\x01\x01\xff" + two_element = b"\x02\x01\x33\x02\xaa\x55" + dual_nested_one = VLADualNested.vla_from_buffer_copy(one_element) + assert dual_nested_one.num == 1 + assert len(dual_nested_one.values) == 1 + assert dual_nested_one.values[0].count == 1 + assert len(dual_nested_one.values[0].data) == 1 + assert dual_nested_one.values[0].data[0] == 0xFF + dual_nested_two = VLADualNested.vla_from_buffer_copy(two_element) + assert dual_nested_two.num == 2 + assert len(dual_nested_two.values) == 2 + assert dual_nested_two.values[0].count == 1 + assert len(dual_nested_one.values[0].data) == 1 + assert dual_nested_two.values[0].data[0] == 0x33 + assert dual_nested_two.values[1].count == 2 + assert len(dual_nested_two.values[1].data) == 2 + assert dual_nested_two.values[1].data[0] == 0xAA + assert dual_nested_two.values[1].data[1] == 0x55 + unaligned = b"\x00" * 31 try: VLABase.vla_from_buffer_copy(unaligned) From 66b540034aecf05c756857fbbe574abc4443eb48 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Mon, 13 Oct 2025 22:10:36 +1000 Subject: [PATCH 7/7] rpc_wrappers: remove custom VLA parsing logic Remove the custom VLA parsing logic from commands since the builtin classes now handle them from the definitions. Signed-off-by: Jordan Yates --- src/infuse_iot/rpc_wrappers/kv_read.py | 44 +------------------ src/infuse_iot/rpc_wrappers/lte_modem_info.py | 20 +++------ src/infuse_iot/rpc_wrappers/wifi_scan.py | 43 +----------------- 3 files changed, 10 insertions(+), 97 deletions(-) diff --git a/src/infuse_iot/rpc_wrappers/kv_read.py b/src/infuse_iot/rpc_wrappers/kv_read.py index 8bc2eb2..1ab4ed8 100644 --- a/src/infuse_iot/rpc_wrappers/kv_read.py +++ b/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: diff --git a/src/infuse_iot/rpc_wrappers/lte_modem_info.py b/src/infuse_iot/rpc_wrappers/lte_modem_info.py index 2cb8dcf..c71f206 100644 --- a/src/infuse_iot/rpc_wrappers/lte_modem_info.py +++ b/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}") diff --git a/src/infuse_iot/rpc_wrappers/wifi_scan.py b/src/infuse_iot/rpc_wrappers/wifi_scan.py index 7d28507..8d90d84 100644 --- a/src/infuse_iot/rpc_wrappers/wifi_scan.py +++ b/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,