rpc_wrappers: generic: added

src/infuse_iot/commands.py

@@ -45,12 +45,12 @@ def DESCRIPTION(self) -> str:
 
 
 class InfuseRpcCommand:
-    RPC_DATA_SEND = False
-    RPC_DATA_SEND_CHUNKED = False
-    RPC_DATA_RECEIVE = False
+    RPC_DATA_SEND: bool = False
+    RPC_DATA_SEND_CHUNKED: bool = False
+    RPC_DATA_RECEIVE: bool = False
 
     @classmethod
-    def add_parser(cls, parser: argparse.ArgumentParser):
+    def add_parser(cls, parser: argparse.ArgumentParser) -> None:
         raise NotImplementedError
 
     def __init__(self, **kwargs):
@@ -82,10 +82,12 @@ def data_payload_recv_len(self) -> int:
 
     def data_recv_cb(self, offset: int, data: bytes) -> None:
         """Data received callback"""
+        raise NotImplementedError
 
     def data_progress_cb(self, offset: int) -> None:
         """Progress callback"""
+        raise NotImplementedError
 
-    def handle_response(self, return_code: int, response: ctypes.LittleEndianStructure | None):
+    def handle_response(self, return_code: int, response: ctypes.LittleEndianStructure | None) -> None:
         """Handle RPC_RSP"""
         raise NotImplementedError

src/infuse_iot/generated/kv_definitions.py

@@ -60,6 +60,16 @@ class kv_range_u8(VLACompatLittleEndianStruct):
         ]
         _pack_ = 1
 
+    class kv_utc_hms(VLACompatLittleEndianStruct):
+        """UTC Hour-Minute-Second"""
+
+        _fields_ = [
+            ("hour", ctypes.c_uint8),
+            ("minute", ctypes.c_uint8),
+            ("second", ctypes.c_uint8),
+        ]
+        _pack_ = 1
+
 
 class slots:
     class reboots(VLACompatLittleEndianStruct):
@@ -328,6 +338,18 @@ class bluetooth_throughput_limit(VLACompatLittleEndianStruct):
         ]
         _pack_ = 1
 
+    class led_disable_daily_time_range(VLACompatLittleEndianStruct):
+        """Disable LEDs between two UTC times daily"""
+
+        NAME = "LED_DISABLE_DAILY_TIME_RANGE"
+        BASE_ID = 53
+        RANGE = 1
+        _fields_ = [
+            ("disable_start", structs.kv_utc_hms),
+            ("disable_end", structs.kv_utc_hms),
+        ]
+        _pack_ = 1
+
     class gravity_reference(VLACompatLittleEndianStruct):
         """Reference gravity vector for tilt calculations"""
 
@@ -419,6 +441,7 @@ class secure_storage_reserved(VLACompatLittleEndianStruct):
         50: bluetooth_peer,
         51: lora_config,
         52: bluetooth_throughput_limit,
+        53: led_disable_daily_time_range,
         60: gravity_reference,
         100: geofence,
         101: geofence,

src/infuse_iot/rpc_wrappers/generic.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+
+import ctypes
+
+from infuse_iot.commands import InfuseRpcCommand
+from infuse_iot.util.ctypes import VLACompatLittleEndianStruct, bytes_to_uint8
+from infuse_iot.zephyr.errno import errno
+
+
+class generic(InfuseRpcCommand):
+    NAME = "generic"
+    HELP = "Generic RPC sender"
+    DESCRIPTION = "Generic RPC sender"
+    COMMAND_ID = -1
+
+    class response(VLACompatLittleEndianStruct):
+        _fields_ = []
+        vla_field = ("val", 0 * ctypes.c_uint8)
+        _pack_ = 1
+
+    @classmethod
+    def add_parser(cls, parser):
+        parser.add_argument("--rpc", type=int, required=True, help="RPC command ID")
+        parser.add_argument("--args", type=str, default="", help="Arguments as byte string")
+
+    def __init__(self, args):
+        self.COMMAND_ID = args.rpc
+        self._arg_bytes = bytes.fromhex(args.args)
+
+    def request_struct(self):
+        class request(ctypes.LittleEndianStructure):
+            _fields_ = [
+                ("val", len(self._arg_bytes) * ctypes.c_uint8),
+            ]
+            _pack_ = 1
+
+        return request(bytes_to_uint8(self._arg_bytes))
+
+    def handle_response(self, return_code, response):
+        print(f"Return Code: {return_code} ({errno.strerror(-return_code)})")
+        if len(response.val) > 0:
+            print(f"   Response: {bytes(response.val).hex()}")
+        else:
+            print("   Response: None")