rpc_client: internal timeouts

scripts/apn_set.py

@@ -78,6 +78,8 @@ def announce_observed(self, live: Live, infuse_id: int, pkt: readings.announce |
                 hdr, rsp = rpc_client.run_standard_cmd(
                     rpc.kv_write.COMMAND_ID, Auth.DEVICE, params, self.response.vla_from_buffer_copy
                 )
+                if hdr is None:
+                    return
                 if hdr.return_code == 0:
                     assert rsp is not None and hasattr(rsp, "rc")
                     if rsp.rc[0] == 0:

scripts/reboot_count_reset.py

@@ -67,6 +67,8 @@ def announce_observed(self, live: Live, infuse_id: int, pkt: readings.announce |
                 hdr, _ = rpc_client.run_standard_cmd(
                     rpc.kv_write.COMMAND_ID, Auth.DEVICE, params, rpc.kv_write.response.from_buffer_copy
                 )
+                if hdr is None:
+                    return
                 if hdr.return_code == 0:
                     self.updated.append(infuse_id)
 

src/infuse_iot/rpc_client.py

@@ -2,6 +2,7 @@
 
 import ctypes
 import random
+import time
 from collections.abc import Callable
 
 from infuse_iot import rpc
@@ -25,11 +26,15 @@ def __init__(
         rx_cb: Callable[[ClientNotification], None] | None = None,
     ):
         self._request_id = random.randint(0, 2**31 - 1)
+        self._timeout = 10.0
         self._client = client
         self._id = infuse_id
         self._max_payload = max_payload
         self._rx_cb = rx_cb
 
+    def set_timeout(self, timeout: float):
+        self._timeout = timeout
+
     def _finalise_command(
         self, rpc_rsp: PacketReceived, rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure]
     ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
@@ -48,8 +53,9 @@ def _client_recv(self) -> ClientNotification | None:
             self._rx_cb(rsp)
         return rsp
 
-    def _wait_data_ack(self) -> PacketReceived:
-        while True:
+    def _wait_data_ack(self) -> PacketReceived | None:
+        timeout = time.time() + self._timeout
+        while time.time() < timeout:
             rsp = self._client_recv()
             if rsp is None:
                 continue
@@ -66,10 +72,12 @@ def _wait_data_ack(self) -> PacketReceived:
             if data_ack.request_id != self._request_id:
                 continue
             return rsp.epacket
+        return None
 
-    def _wait_rpc_rsp(self) -> PacketReceived:
+    def _wait_rpc_rsp(self) -> PacketReceived | None:
+        timeout = time.time() + self._timeout
         # Wait for responses
-        while True:
+        while time.time() < timeout:
             rsp = self._client_recv()
             if rsp is None:
                 continue
@@ -83,6 +91,7 @@ def _wait_rpc_rsp(self) -> PacketReceived:
             if rsp_header.request_id != self._request_id:
                 continue
             return rsp.epacket
+        return None
 
     def _run_data_send_core(
         self,
@@ -94,7 +103,7 @@ def _run_data_send_core(
         packet_idx: bool,
         progress_cb: Callable[[int], None] | None,
         rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure],
-    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
+    ) -> tuple[rpc.ResponseHeader | None, ctypes.LittleEndianStructure | None]:
         self._request_id += 1
         ack_period = 2
         header = rpc.RequestHeader(self._request_id, cmd_id)  # type: ignore
@@ -112,6 +121,8 @@ def _run_data_send_core(
 
         # Wait for initial ACK
         recv = self._wait_data_ack()
+        if recv is None:
+            return None, None
         if recv.ptype == InfuseType.RPC_RSP:
             return self._finalise_command(recv, rsp_decoder)
 
@@ -133,6 +144,8 @@ def _run_data_send_core(
             # Wait for ACKs at the period
             if ack_cnt == ack_period:
                 recv = self._wait_data_ack()
+                if recv is None:
+                    return None, None
                 if recv.ptype == InfuseType.RPC_RSP:
                     return self._finalise_command(recv, rsp_decoder)
                 ack_cnt = 0
@@ -142,6 +155,9 @@ def _run_data_send_core(
                 progress_cb(chunk_id + 1 if packet_idx else offset)
 
         recv = self._wait_rpc_rsp()
+        if recv is None:
+            return None, None
+
         return self._finalise_command(recv, rsp_decoder)
 
     def run_data_send_cmd(
@@ -152,7 +168,7 @@ def run_data_send_cmd(
         data: bytes,
         progress_cb: Callable[[int], None] | None,
         rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure],
-    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
+    ) -> tuple[rpc.ResponseHeader | None, ctypes.LittleEndianStructure | None]:
         # Maxmimum payload size of interface
         size = self._max_payload - ctypes.sizeof(rpc.DataHeader)
         # Round payload down to multiple of 4 bytes
@@ -170,7 +186,7 @@ def run_data_send_cmd_chunked(
         data: list[bytes],
         progress_cb: Callable[[int], None] | None,
         rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure],
-    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
+    ) -> tuple[rpc.ResponseHeader | None, ctypes.LittleEndianStructure | None]:
         return self._run_data_send_core(cmd_id, auth, params, data, len(data), True, progress_cb, rsp_decoder)
 
     def run_data_recv_cmd(
@@ -225,7 +241,7 @@ def run_data_recv_cmd(
 
     def run_standard_cmd(
         self, cmd_id: int, auth: Auth, params: bytes, rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure]
-    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
+    ) -> tuple[rpc.ResponseHeader | None, ctypes.LittleEndianStructure | None]:
         self._request_id += 1
         header = rpc.RequestHeader(self._request_id, cmd_id)  # type: ignore
 
@@ -239,4 +255,6 @@ def run_standard_cmd(
         req = GatewayRequestEpacketSend(pkt)
         self._client.send(req)
         recv = self._wait_rpc_rsp()
+        if recv is None:
+            return None, None
         return self._finalise_command(recv, rsp_decoder)

src/infuse_iot/tools/bt_log.py

@@ -35,7 +35,7 @@ def __init__(self, args):
 
     @classmethod
     def add_parser(cls, parser):
-        parser.add_argument("--id", type=lambda x: int(x, 0), help="Infuse ID to receive logs for")
+        parser.add_argument("--id", type=lambda x: int(x, 0), required=True, help="Infuse ID to receive logs for")
         parser.add_argument("--data", action="store_true", help="Subscribe to the data characteristic as well")
         parser.add_argument(
             "--conn-timeout", type=int, default=10000, help="Timeout to wait for a connection to the device (ms)"

src/infuse_iot/tools/gateway.py

@@ -114,7 +114,7 @@ def handle(self, pkt: PacketReceived):
             infuse_id = self._ddb.infuse_id_from_bluetooth(if_addr)
             if infuse_id is None:
                 Console.log_error(f"Infuse ID of {if_addr} not known")
-            else:
+            elif header.return_code == 0:
                 self._ddb.observe_security_state(
                     infuse_id,
                     bytes(resp.cloud_public_key),
@@ -369,7 +369,7 @@ def _bt_connect_cb(self, pkt: PacketReceived, rc: int, response: bytes):
         assert infuse_id is not None, "ID was required to initiate connection?"
         assert self._common.server is not None
 
-        if rc < 0:
+        if rc != 0:
             rsp = ClientNotificationConnectionFailed(infuse_id)
             self._common.notification_broadcast(rsp)
             return

src/infuse_iot/tools/ota_upgrade.py

@@ -153,8 +153,9 @@ def gateway_diff_load(self):
                 None,
                 file_write_basic.response.from_buffer_copy,
             )
-            if hdr.return_code != 0:
-                sys.exit(f"Failed to save diff file to gateway (({errno.strerror(-hdr.return_code)}))")
+            return_code = hdr.return_code if hdr else -1
+            if return_code != 0:
+                sys.exit(f"Failed to save diff file to gateway (({errno.strerror(-return_code)}))")
             print(f"'{self._single_diff}' written to gateway")
 
     def run_file_upload(self, live: Live, mtu: int, source: HopReceived):
@@ -172,7 +173,9 @@ def run_file_upload(self, live: Live, mtu: int, source: HopReceived):
             file_write_basic.response.from_buffer_copy,
         )
 
-        if hdr.return_code == 0:
+        if hdr is None:
+            self._failed += 1
+        elif hdr.return_code == 0:
             self._pending[source.infuse_id] = time.time() + 60
 
     def run_file_copy(self, live: Live, mtu: int, source: HopReceived):
@@ -195,7 +198,9 @@ def run_file_copy(self, live: Live, mtu: int, source: HopReceived):
             bytes(params),
             bt_file_copy_basic.response.from_buffer_copy,
         )
-        if hdr.return_code == 0:
+        if hdr is None:
+            self._failed += 1
+        elif hdr.return_code == 0:
             self._pending[source.infuse_id] = time.time() + 60
 
     def run(self):

src/infuse_iot/tools/rpc.py

@@ -126,8 +126,9 @@ def run(self):
                         params,
                         decode_fn,
                     )
+                return_code = hdr.return_code if hdr else -1
                 # Handle response
-                self._command.handle_response(hdr.return_code, rsp)
+                self._command.handle_response(return_code, rsp)
 
                 if self._args.conn_log:
                     while True:

src/infuse_iot/tools/tdf_list.py

@@ -30,11 +30,17 @@ class SubCommand(InfuseCommand):
     @classmethod
     def add_parser(cls, parser):
         parser.add_argument("--array-all", action="store_true", help="Display all array values, not just the last")
+        parser.add_argument(
+            "--id", type=lambda x: int(x, 0), action="append", default=[], help="Limit displayed TDFs by device ID"
+        )
+        parser.add_argument("--min-rssi", type=int, help="Minimum RSSI to display TDF")
 
     def __init__(self, args):
         self._client = LocalClient(default_multicast_address(), 1.0)
         self._decoder = TDF()
         self._array_all = args.array_all
+        self._ids = args.id
+        self._min_rssi = args.min_rssi
 
     def append_tdf(
         self,
@@ -111,13 +117,18 @@ def run(self) -> None:
                 continue
             source = msg.epacket.route[0]
 
+            if len(self._ids) > 0 and source.infuse_id not in self._ids:
+                continue
+            if self._min_rssi is not None and source.rssi < self._min_rssi:
+                continue
+
             table: list[tuple[str | None, str | None, str, str, str]] = []
 
             tdf: TDF.Reading
             for tdf in self._decoder.decode(msg.epacket.payload):
                 self.append_readings(table, tdf)
 
-            print(f"Infuse ID: {source.infuse_id:016x}")
+            print(f"Infuse ID: 0x{source.infuse_id:016x}")
             print(f"Interface: {source.interface.name}")
             print(f"  Address: {source.interface_address}")
             print(f"     RSSI: {source.rssi} dBm")