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 <jordan@embeint.com>

Author: JordanYates

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

@@ -11,43 +11,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 +27,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,