tools: gateway: support pyocd RTT

mypy.ini

@@ -14,6 +14,9 @@ ignore_missing_imports = True
 [mypy-pynrfjprog]
 ignore_missing_imports = True
 
+[mypy-pyocd.*]
+ignore_missing_imports = True
+
 [mypy-simple_term_menu]
 ignore_missing_imports = True
 

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "tabulate",
     "typing-extensions",
     "keyring",
+    "pyocd",
     "pyyaml",
 ]
 

requirements.txt

@@ -6,6 +6,7 @@ httpx
 pylink-square
 pynrfjprog
 pyserial
+pyocd
 python-dateutil
 simple-term-menu
 tabulate

src/infuse_iot/commands.py

@@ -77,6 +77,6 @@ def data_recv_cb(self, offset: int, data: bytes) -> None:
     def data_progress_cb(self, offset: int) -> None:
         """Progress callback"""
 
-    def handle_response(self, return_code, response):
+    def handle_response(self, return_code: int, response: ctypes.LittleEndianStructure | None):
         """Handle RPC_RSP"""
         raise NotImplementedError

src/infuse_iot/generated/tdf_definitions.py

@@ -1022,6 +1022,150 @@ class device_tilt(TdfReadingBase):
             "cosine": "{:.6f}",
         }
 
+    class nrf9x_gnss_pvt(TdfReadingBase):
+        """nRF9x GNSS PVT data frame"""
+
+        name = "NRF9X_GNSS_PVT"
+        _fields_ = [
+            ("_lat", ctypes.c_int32),
+            ("_lon", ctypes.c_int32),
+            ("_height", ctypes.c_int32),
+            ("_h_acc", ctypes.c_uint32),
+            ("_v_acc", ctypes.c_uint32),
+            ("_h_speed", ctypes.c_int32),
+            ("_h_speed_acc", ctypes.c_uint32),
+            ("_v_speed", ctypes.c_int32),
+            ("_v_speed_acc", ctypes.c_uint32),
+            ("_head_mot", ctypes.c_int32),
+            ("_head_acc", ctypes.c_uint32),
+            ("year", ctypes.c_uint16),
+            ("month", ctypes.c_uint8),
+            ("day", ctypes.c_uint8),
+            ("hour", ctypes.c_uint8),
+            ("min", ctypes.c_uint8),
+            ("sec", ctypes.c_uint8),
+            ("ms", ctypes.c_uint16),
+            ("_p_dop", ctypes.c_uint16),
+            ("_h_dop", ctypes.c_uint16),
+            ("_v_dop", ctypes.c_uint16),
+            ("_t_dop", ctypes.c_uint16),
+            ("flags", ctypes.c_uint8),
+            ("num_sv", ctypes.c_uint8),
+        ]
+        _pack_ = 1
+        _postfix_ = {
+            "lat": "deg",
+            "lon": "deg",
+            "height": "m",
+            "h_acc": "m",
+            "v_acc": "m",
+            "h_speed": "m/s",
+            "h_speed_acc": "m/s",
+            "v_speed": "m/s",
+            "v_speed_acc": "m/s",
+            "head_mot": "deg",
+            "head_acc": "deg",
+            "year": "",
+            "month": "",
+            "day": "",
+            "hour": "",
+            "min": "",
+            "sec": "",
+            "ms": "",
+            "p_dop": "",
+            "h_dop": "",
+            "v_dop": "",
+            "t_dop": "",
+            "flags": "",
+            "num_sv": "",
+        }
+        _display_fmt_ = {
+            "lat": "{}",
+            "lon": "{}",
+            "height": "{}",
+            "h_acc": "{}",
+            "v_acc": "{}",
+            "h_speed": "{}",
+            "h_speed_acc": "{}",
+            "v_speed": "{}",
+            "v_speed_acc": "{}",
+            "head_mot": "{}",
+            "head_acc": "{}",
+            "year": "{}",
+            "month": "{}",
+            "day": "{}",
+            "hour": "{}",
+            "min": "{}",
+            "sec": "{}",
+            "ms": "{}",
+            "p_dop": "{}",
+            "h_dop": "{}",
+            "v_dop": "{}",
+            "t_dop": "{}",
+            "flags": "0x{:02x}",
+            "num_sv": "{}",
+        }
+
+        @property
+        def lat(self):
+            return self._lat * 1e-07
+
+        @property
+        def lon(self):
+            return self._lon * 1e-07
+
+        @property
+        def height(self):
+            return self._height * 0.001
+
+        @property
+        def h_acc(self):
+            return self._h_acc * 0.001
+
+        @property
+        def v_acc(self):
+            return self._v_acc * 0.001
+
+        @property
+        def h_speed(self):
+            return self._h_speed * 0.001
+
+        @property
+        def h_speed_acc(self):
+            return self._h_speed_acc * 0.001
+
+        @property
+        def v_speed(self):
+            return self._v_speed * 0.001
+
+        @property
+        def v_speed_acc(self):
+            return self._v_speed_acc * 0.001
+
+        @property
+        def head_mot(self):
+            return self._head_mot * 1e-05
+
+        @property
+        def head_acc(self):
+            return self._head_acc * 1e-05
+
+        @property
+        def p_dop(self):
+            return self._p_dop * 0.01
+
+        @property
+        def h_dop(self):
+            return self._h_dop * 0.01
+
+        @property
+        def v_dop(self):
+            return self._v_dop * 0.01
+
+        @property
+        def t_dop(self):
+            return self._t_dop * 0.01
+
     class array_type(TdfReadingBase):
         """Example array type"""
 
@@ -1072,5 +1216,6 @@ class array_type(TdfReadingBase):
     34: readings.lte_tac_cells,
     35: readings.wifi_ap_info,
     36: readings.device_tilt,
+    37: readings.nrf9x_gnss_pvt,
     100: readings.array_type,
 }

src/infuse_iot/rpc_client.py

@@ -32,11 +32,14 @@ def __init__(
 
     def _finalise_command(
         self, rpc_rsp: PacketReceived, rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure]
-    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure]:
+    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
         # Convert response bytes back to struct form
         rsp_header = rpc.ResponseHeader.from_buffer_copy(rpc_rsp.payload)
         rsp_payload = rpc_rsp.payload[ctypes.sizeof(rpc.ResponseHeader) :]
-        rsp_data = rsp_decoder(rsp_payload)
+        try:
+            rsp_data = rsp_decoder(rsp_payload)
+        except ValueError:
+            rsp_data = None
         return (rsp_header, rsp_data)
 
     def _client_recv(self) -> ClientNotification | None:
@@ -89,7 +92,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]:
+    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
         self._request_id += 1
         ack_period = 1
         header = rpc.RequestHeader(self._request_id, cmd_id)  # type: ignore
@@ -151,7 +154,7 @@ def run_data_recv_cmd(
         params: bytes,
         recv_cb: Callable[[int, bytes], None],
         rsp_decoder: Callable[[bytes], ctypes.LittleEndianStructure],
-    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure]:
+    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
         self._request_id += 1
         header = rpc.RequestHeader(self._request_id, cmd_id)
         data_hdr = rpc.RequestDataHeader(0xFFFFFFFF, 0)
@@ -195,7 +198,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]:
+    ) -> tuple[rpc.ResponseHeader, ctypes.LittleEndianStructure | None]:
         self._request_id += 1
         header = rpc.RequestHeader(self._request_id, cmd_id)  # type: ignore
 

src/infuse_iot/rpc_wrappers/application_info.py

@@ -1,9 +1,9 @@
 #!/usr/bin/env python3
 
-import os
 
 import infuse_iot.generated.rpc_definitions as defs
 from infuse_iot.commands import InfuseRpcCommand
+from infuse_iot.zephyr.errno import errno
 
 
 class application_info(InfuseRpcCommand, defs.application_info):
@@ -22,7 +22,7 @@ def request_json(self):
 
     def handle_response(self, return_code, response):
         if return_code != 0:
-            print(f"Failed to query current time ({os.strerror(-return_code)})")
+            print(f"Failed to query current time ({errno.strerror(-return_code)})")
             return
 
         r = response

src/infuse_iot/rpc_wrappers/bt_connect_infuse.py

@@ -1,6 +1,5 @@
 #!/usr/bin/env python3
 
-import os
 
 import infuse_iot.generated.rpc_definitions as defs
 from infuse_iot.commands import InfuseRpcCommand
@@ -11,6 +10,7 @@
 )
 from infuse_iot.util.argparse import BtLeAddress
 from infuse_iot.util.ctypes import bytes_to_uint8
+from infuse_iot.zephyr.errno import errno
 
 
 class bt_connect_infuse(InfuseRpcCommand, defs.bt_connect_infuse):
@@ -59,7 +59,7 @@ def request_struct(self) -> defs.bt_connect_infuse.request:
 
     def handle_response(self, return_code, response):
         if return_code < 0:
-            print(f"Failed to connect ({os.strerror(-return_code)})")
+            print(f"Failed to connect ({errno.strerror(-return_code)})")
             return
 
         if return_code == 1:

src/infuse_iot/rpc_wrappers/bt_disconnect.py

@@ -1,6 +1,5 @@
 #!/usr/bin/env python3
 
-import os
 
 import infuse_iot.generated.rpc_definitions as defs
 from infuse_iot.commands import InfuseRpcCommand
@@ -10,6 +9,7 @@
 )
 from infuse_iot.util.argparse import BtLeAddress
 from infuse_iot.util.ctypes import bytes_to_uint8
+from infuse_iot.zephyr.errno import errno
 
 
 class bt_disconnect(InfuseRpcCommand, defs.bt_disconnect):
@@ -40,7 +40,7 @@ def request_struct(self):
 
     def handle_response(self, return_code, response):
         if return_code != 0:
-            print(f"Failed to disconnect ({os.strerror(-return_code)})")
+            print(f"Failed to disconnect ({errno.strerror(-return_code)})")
             return
         else:
             print("Disconnected")

src/infuse_iot/rpc_wrappers/coap_download.py

@@ -1,12 +1,12 @@
 #!/usr/bin/env python3
 
 import ctypes
-import os
 
 import infuse_iot.generated.rpc_definitions as defs
 from infuse_iot.commands import InfuseRpcCommand
 from infuse_iot.generated.rpc_definitions import rpc_enum_file_action
 from infuse_iot.util.ctypes import UINT32_MAX
+from infuse_iot.zephyr.errno import errno
 
 
 class coap_download(InfuseRpcCommand, defs.coap_download):
@@ -103,7 +103,7 @@ def request_json(self):
 
     def handle_response(self, return_code, response):
         if return_code != 0:
-            print(f"Failed to download file ({os.strerror(-return_code)})")
+            print(f"Failed to download file ({errno.strerror(-return_code)})")
             return
         else:
             print("File downloaded")

src/infuse_iot/rpc_wrappers/data_logger_read.py

@@ -1,11 +1,11 @@
 #!/usr/bin/env python3
 
 import binascii
-import os
 
 import infuse_iot.generated.rpc_definitions as defs
 from infuse_iot.commands import InfuseRpcCommand
 from infuse_iot.util.ctypes import UINT32_MAX
+from infuse_iot.zephyr.errno import errno
 
 
 class data_logger_read(InfuseRpcCommand, defs.data_logger_read):
@@ -50,7 +50,7 @@ def data_recv_cb(self, offset: int, data: bytes) -> None:
 
     def handle_response(self, return_code, response):
         if return_code != 0:
-            print(f"Failed to read data logger ({os.strerror(-return_code)})")
+            print(f"Failed to read data logger ({errno.strerror(-return_code)})")
             return
 
         if response.sent_len != len(self.output):

src/infuse_iot/rpc_wrappers/data_logger_state.py

@@ -1,10 +1,10 @@
 #!/usr/bin/env python3
 
-import os
 
 import infuse_iot.generated.rpc_definitions as defs
 from infuse_iot.commands import InfuseRpcCommand
 from infuse_iot.util.time import humanised_seconds
+from infuse_iot.zephyr.errno import errno
 
 
 class data_logger_state(InfuseRpcCommand, defs.data_logger_state):
@@ -30,7 +30,7 @@ def request_json(self):
 
     def handle_response(self, return_code, response):
         if return_code != 0:
-            print(f"Failed to query data logger state ({os.strerror(-return_code)})")
+            print(f"Failed to query data logger state ({errno.strerror(-return_code)})")
             return
 
         def sizeof_fmt(num, suffix="B"):
@@ -53,8 +53,8 @@ def sizeof_fmt(num, suffix="B"):
             print("\t Block Rate: N/A")
             print("\t  Byte Rate: N/A")
         elif byte_rate < 0.1:
-            print(f"\t Block Rate: {1/block_rate:.2f} sec/block")
-            print(f"\t  Byte Rate: {1/byte_rate:.2f} sec/byte")
+            print(f"\t Block Rate: {1 / block_rate:.2f} sec/block")
+            print(f"\t  Byte Rate: {1 / byte_rate:.2f} sec/byte")
         else:
             print(f"\t Block Rate: {block_rate:.2f} blocks/sec")
             print(f"\t  Byte Rate: {sizeof_fmt(byte_rate)}/sec")