diff --git a/src/infuse_iot/commands.py b/src/infuse_iot/commands.py index c9bf8d7..f82c828 100644 --- a/src/infuse_iot/commands.py +++ b/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 diff --git a/src/infuse_iot/generated/kv_definitions.py b/src/infuse_iot/generated/kv_definitions.py index 5f67682..b8aa575 100644 --- a/src/infuse_iot/generated/kv_definitions.py +++ b/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, diff --git a/src/infuse_iot/rpc_wrappers/generic.py b/src/infuse_iot/rpc_wrappers/generic.py new file mode 100644 index 0000000..ea00a64 --- /dev/null +++ b/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")