rpc_wrappers: infuse_states_*: added

src/infuse_iot/generated/tdf_definitions.py

@@ -1199,6 +1199,51 @@ class infuse_bluetooth_rssi(TdfReadingBase):
             "rssi": "{}",
         }
 
+    class adc_raw_8(TdfReadingBase):
+        """Generic 8bit raw ADC reading"""
+
+        name = "ADC_RAW_8"
+        _fields_ = [
+            ("val", ctypes.c_int8),
+        ]
+        _pack_ = 1
+        _postfix_ = {
+            "val": "",
+        }
+        _display_fmt_ = {
+            "val": "{}",
+        }
+
+    class adc_raw_16(TdfReadingBase):
+        """Generic 16bit raw ADC reading"""
+
+        name = "ADC_RAW_16"
+        _fields_ = [
+            ("val", ctypes.c_int16),
+        ]
+        _pack_ = 1
+        _postfix_ = {
+            "val": "",
+        }
+        _display_fmt_ = {
+            "val": "{}",
+        }
+
+    class adc_raw_32(TdfReadingBase):
+        """Generic 32bit raw ADC reading"""
+
+        name = "ADC_RAW_32"
+        _fields_ = [
+            ("val", ctypes.c_int32),
+        ]
+        _pack_ = 1
+        _postfix_ = {
+            "val": "",
+        }
+        _display_fmt_ = {
+            "val": "{}",
+        }
+
     class array_type(TdfReadingBase):
         """Example array type"""
 
@@ -1252,5 +1297,8 @@ class array_type(TdfReadingBase):
     37: readings.nrf9x_gnss_pvt,
     38: readings.battery_charge_accumulated,
     39: readings.infuse_bluetooth_rssi,
+    40: readings.adc_raw_8,
+    41: readings.adc_raw_16,
+    42: readings.adc_raw_32,
     100: readings.array_type,
 }

src/infuse_iot/rpc_client.py

@@ -2,7 +2,7 @@
 
 import ctypes
 import random
-from typing import Callable
+from collections.abc import Callable
 
 from infuse_iot import rpc
 from infuse_iot.common import InfuseType

src/infuse_iot/rpc_wrappers/infuse_states_query.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+
+import tabulate
+
+import infuse_iot.generated.rpc_definitions as defs
+from infuse_iot.commands import InfuseRpcCommand
+from infuse_iot.zephyr.errno import errno
+
+
+class infuse_states_query(InfuseRpcCommand, defs.infuse_states_query):
+    @classmethod
+    def add_parser(cls, parser):
+        parser.add_argument("--offset", type=int, default=0)
+
+    def __init__(self, args):
+        self.args = args
+
+    def request_struct(self):
+        return self.request(self.args.offset)
+
+    def request_json(self):
+        return {}
+
+    def handle_response(self, return_code, response):
+        if return_code != 0:
+            print(f"Failed to query states ({errno.strerror(-return_code)})")
+            return
+
+        states = []
+        for state in response.states:
+            states.append([state.state, "Permanent" if state.timeout == 0 else f"{state.timeout} seconds"])
+        print(tabulate.tabulate(states, headers=["State", "Duration"]))

src/infuse_iot/rpc_wrappers/infuse_states_update.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+
+import ctypes
+
+import infuse_iot.generated.rpc_definitions as defs
+from infuse_iot.commands import InfuseRpcCommand
+from infuse_iot.util.ctypes import UINT16_MAX
+from infuse_iot.zephyr.errno import errno
+
+
+class infuse_states_update(InfuseRpcCommand, defs.infuse_states_update):
+    @classmethod
+    def add_parser(cls, parser):
+        parser.add_argument("--state", "-s", type=int, required=True, help="State ID to update")
+        option = parser.add_mutually_exclusive_group(required=True)
+        option.add_argument("--set", action="store_true", help="Enable the state permanently")
+        option.add_argument("--clear", action="store_true", help="Disable the state")
+        option.add_argument("--timeout", type=int, help="Enable the state for a duration (seconds)")
+
+    def __init__(self, args):
+        self.args = args
+
+    def request_struct(self):
+        class request(ctypes.LittleEndianStructure):
+            _fields_ = [
+                ("num", ctypes.c_uint8),
+                ("state", defs.rpc_struct_infuse_state),
+            ]
+
+        timeout = 0 if self.args.set else (UINT16_MAX if self.args.clear else self.args.timeout)
+        state = defs.rpc_struct_infuse_state(self.args.state, timeout)
+
+        return request(1, state)
+
+    def request_json(self):
+        return {}
+
+    def handle_response(self, return_code, response):
+        if return_code != 0:
+            print(f"Failed to update state {self.args.state} ({errno.strerror(-return_code)})")
+        else:
+            print(f"Updated state {self.args.state}")

src/infuse_iot/rpc_wrappers/reboot.py

@@ -17,6 +17,9 @@ def __init__(self, args):
     def request_struct(self):
         return self.request(self._delay_ms)
 
+    def request_json(self):
+        return {"delay_ms": str(self._delay_ms)}
+
     def handle_response(self, return_code, response):
         if return_code == 0:
             print(f"Rebooting in {response.delay_ms} ms")

src/infuse_iot/rpc_wrappers/sym_read.py

@@ -2,7 +2,6 @@
 
 import binascii
 import sys
-from typing import Optional
 
 import tabulate
 from elftools.dwarf.die import DIE
@@ -29,7 +28,7 @@ def __init__(self, args):
         # Ignore context-manager warning since ELFFile requires the file to remain opened
         self.elf_file = open(args.elf, "rb")  # noqa: SIM115
         self.elf = ELFFile(self.elf_file)
-        self.symbol_die: Optional[DIE]
+        self.symbol_die: DIE | None
 
         if args.sym:
             symbols = elftools.symbols_from_name(self.elf, args.sym)

src/infuse_iot/tools/cloud.py

@@ -8,17 +8,23 @@
 import sys
 from http import HTTPStatus
 from json import loads
+from typing import Any
 
 from tabulate import tabulate
 
 from infuse_iot.api_client import Client
 from infuse_iot.api_client.api.board import (
     create_board,
+    get_board_by_id,
     get_boards,
 )
+from infuse_iot.api_client.api.device import (
+    get_device_by_device_id,
+)
 from infuse_iot.api_client.api.organisation import (
     create_organisation,
     get_all_organisations,
+    get_organisation_by_id,
 )
 from infuse_iot.api_client.models import Error, NewBoard, NewOrganisation
 from infuse_iot.commands import InfuseCommand
@@ -147,6 +153,47 @@ def create(self, client):
             print(f"<{rsp.status_code}>: {c['message']}")
 
 
+class Device(CloudSubCommand):
+    @classmethod
+    def add_parser(cls, parser):
+        parser_boards = parser.add_parser("device", help="Infuse-IoT devices")
+        parser_boards.set_defaults(command_class=cls)
+
+        tool_parser = parser_boards.add_subparsers(title="commands", metavar="<command>", required=True)
+
+        info_parser = tool_parser.add_parser("info")
+        info_parser.set_defaults(command_fn=cls.info)
+        info_parser.add_argument("--id", type=str, help="Infuse-IoT device ID")
+
+    def run(self):
+        with self.client() as client:
+            self.args.command_fn(self, client)
+
+    def info(self, client):
+        id_int = int(self.args.id, 0)
+        id_str = f"{id_int:016x}"
+        info = get_device_by_device_id.sync(client=client, device_id=id_str)
+        if info is None:
+            sys.exit(f"No device with Infuse-IoT ID {id_str} found")
+        metadata: list[tuple[str, Any]] = []
+        if info.metadata:
+            metadata = [(f"Metadata.{k}", v) for k, v in info.metadata.additional_properties.items()]
+
+        org = get_organisation_by_id.sync(client=client, id=info.organisation_id)
+        board = get_board_by_id.sync(client=client, id=info.board_id)
+
+        table: list[tuple[str, Any]] = [
+            ("UUID", info.id),
+            ("MCU ID", info.mcu_id),
+            ("Organisation", f"{info.organisation_id} ({org.name if org else 'Unknown'})"),
+            ("Board", f"{info.board_id} ({board.name if board else 'Unknown'})"),
+            ("Created", info.created_at),
+            ("Updated", info.updated_at),
+            *metadata,
+        ]
+        print(tabulate(table))
+
+
 class SubCommand(InfuseCommand):
     NAME = "cloud"
     HELP = "Infuse-IoT cloud interaction"
@@ -158,6 +205,7 @@ def add_parser(cls, parser):
 
         Organisations.add_parser(subparser)
         Boards.add_parser(subparser)
+        Device.add_parser(subparser)
 
     def __init__(self, args):
         self.tool = args.command_class(args)

src/infuse_iot/tools/gateway.py

@@ -13,7 +13,7 @@
 import random
 import threading
 import time
-from typing import Callable
+from collections.abc import Callable
 
 import cryptography
 import cryptography.exceptions

src/infuse_iot/util/elftools.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 
 import ctypes
-from typing import Optional
 
 from elftools.dwarf.die import DIE
 from elftools.dwarf.dwarf_expr import DW_OP_name2opcode
@@ -58,7 +57,7 @@ def symbols_from_name(elf: ELFFile, name: str) -> list[Symbol]:
     return symbols
 
 
-def symbol_from_address(elf: ELFFile, address: int) -> Optional[Symbol]:
+def symbol_from_address(elf: ELFFile, address: int) -> Symbol | None:
     """Get a list of symbols from an ELF file with names matching the provided string"""
     symtab = None
 
@@ -104,7 +103,7 @@ def dwarf_die_from_symbol(elf: ELFFile, symbol: Symbol) -> DIE | None:
     return None
 
 
-def dwarf_die_file_info(elf: ELFFile, die: DIE) -> tuple[Optional[str], int]:
+def dwarf_die_file_info(elf: ELFFile, die: DIE) -> tuple[str | None, int]:
     file_attr = die.attributes["DW_AT_decl_file"]
     line_attr = die.attributes["DW_AT_decl_line"]
 
@@ -123,8 +122,8 @@ def __init__(
         self,
         name: str,
         tag: str,
-        ctype: Optional[type[ctypes._SimpleCData]],
-        children: Optional[list[Self]],
+        ctype: type[ctypes._SimpleCData] | None,
+        children: list[Self] | None,
         offset: int,
     ):
         self.name = name
@@ -143,7 +142,7 @@ def _type_from_dwarf_info(dwarfinfo: DWARFInfo, die: DIE):
 
 
 def dwarf_die_variable_inf(
-    dwarfinfo: DWARFInfo, die: DIE, offset: int = 0, name_override: Optional[str] = None
+    dwarfinfo: DWARFInfo, die: DIE, offset: int = 0, name_override: str | None = None
 ) -> dwarf_field:
     type_die = _type_from_dwarf_info(dwarfinfo, die)
 

src/infuse_iot/util/threading.py

@@ -1,13 +1,13 @@
 #!/usr/bin/env python3
 
 import threading
-from typing import Callable, Optional
+from collections.abc import Callable
 
 
 class SignaledThread(threading.Thread):
     """Thread that can be signaled to terminate"""
 
-    def __init__(self, fn: Callable[[], None], sig: Optional[threading.Event] = None):
+    def __init__(self, fn: Callable[[], None], sig: threading.Event | None = None):
         self._fn = fn
         self._sig = sig if sig is not None else threading.Event()
         super().__init__(target=self.run_loop)