diff --git a/src/infuse_iot/tools/cloud.py b/src/infuse_iot/tools/cloud.py index 7dc9881..78e83c3 100644 --- a/src/infuse_iot/tools/cloud.py +++ b/src/infuse_iot/tools/cloud.py @@ -33,6 +33,7 @@ from infuse_iot.api_client.api.coap import get_coap_files from infuse_iot.api_client.api.device import ( create_device_application_update_by_device_id, + create_device_kv_entry_update_by_device_id_and_key_id, get_device_application_state_by_device_id, get_device_application_updates_by_device_id, get_device_by_device_id, @@ -49,7 +50,7 @@ from infuse_iot.api_client.types import File, Unset from infuse_iot.commands import InfuseCommand from infuse_iot.credentials import get_api_key -from infuse_iot.util.argparse import InfuseDeviceId, ValidRelease +from infuse_iot.util.argparse import HexString, InfuseDeviceId, ValidRelease from infuse_iot.util.console import choose_one, user_confirm, user_response from infuse_iot.util.version import Version @@ -200,6 +201,12 @@ def add_parser(cls, parser): "--base64", action="store_true", help="Display values as base64 strings instead of decoding" ) + kv_update = tool_parser.add_parser("kv_update", help="Key-Value update") + kv_update.set_defaults(command_fn=cls.kv_update) + kv_update.add_argument("--id", type=InfuseDeviceId, required=True, help="Infuse-IoT device ID") + kv_update.add_argument("--key", "-k", type=int, required=True, help="Key ID to update") + kv_update.add_argument("--val", "-v", type=HexString, required=True, help="Key value as a hex string") + dfu_parser = tool_parser.add_parser("dfu", help="Manage device firmware upgrades") dfu_parser.set_defaults(command_fn=cls.dfu) dfu_parser.add_argument("--id", type=InfuseDeviceId, required=True, help="Infuse-IoT device ID") @@ -333,6 +340,28 @@ def kv_state(self, client: Client): print(tabulate(table)) + def kv_update(self, client: Client): + id_str = f"{self.args.id:016x}" + + val_encoded = base64.b64encode(self.args.val).decode("utf-8") + update = models.NewDeviceKVEntryUpdate(data=val_encoded) + + rsp = create_device_kv_entry_update_by_device_id_and_key_id.sync( + client=client, + device_id=id_str, + key_id=self.args.key, + body=update, + ) + if rsp is None: + sys.exit("KV update: No response") + elif isinstance(rsp, models.Error): + sys.exit(f"<{rsp.code}>: {rsp.message}") + elif isinstance(rsp, models.DeviceKVEntry): + print(f"Device {id_str} key {self.args.key} already has value {self.args.val.hex()}") + else: + assert isinstance(rsp, models.DeviceKVEntryUpdate) + print(f"Device {id_str} update scheduled with ID {rsp.id}") + def dfu(self, client: Client): id_str = f"{self.args.id:016x}" diff --git a/src/infuse_iot/util/argparse.py b/src/infuse_iot/util/argparse.py index 6590ddf..36c2d14 100644 --- a/src/infuse_iot/util/argparse.py +++ b/src/infuse_iot/util/argparse.py @@ -14,7 +14,7 @@ class ValidFile: """Filesystem file that exists""" - def __new__(cls, string) -> pathlib.Path: # type: ignore + def __new__(cls, string: str) -> pathlib.Path: # type: ignore p = pathlib.Path(string) if p.exists(): if p.is_dir(): @@ -28,7 +28,7 @@ def __new__(cls, string) -> pathlib.Path: # type: ignore class ValidDir: """Filesystem directory that exists""" - def __new__(cls, string) -> pathlib.Path: # type: ignore + def __new__(cls, string: str) -> pathlib.Path: # type: ignore p = pathlib.Path(string) if not p.exists(): raise argparse.ArgumentTypeError(f"{string} does not exist") @@ -40,7 +40,7 @@ def __new__(cls, string) -> pathlib.Path: # type: ignore class ValidRelease: """Infuse-IoT release folder""" - def __init__(self, string): + def __init__(self, string: str): p: pathlib.Path = ValidDir(string) # type: ignore metadata = p / "manifest.yaml" if not metadata.exists(): @@ -54,7 +54,7 @@ def __init__(self, string): class BtLeAddress: """Bluetooth Low-Energy address""" - def __new__(cls, string) -> int: # type: ignore + def __new__(cls, string: str) -> int: # type: ignore pattern = r"((([0-9a-fA-F]{2}):){5})([0-9a-fA-F]{2})" if re.match(pattern, string): @@ -76,7 +76,7 @@ def to_ctype(cls, addr_type: rpc_enum_bt_le_addr_type, value: int) -> rpc_struct ) @classmethod - def integer_value(cls, string) -> int: + def integer_value(cls, string: str) -> int: """Integer value from address string""" return cast(int, cls(string)) @@ -84,8 +84,18 @@ def integer_value(cls, string) -> int: class InfuseDeviceId: """Infuse-IoT Device ID""" - def __new__(cls, string) -> int: # type: ignore + def __new__(cls, string: str) -> int: # type: ignore try: return int(string, 16) except ValueError as e: raise argparse.ArgumentTypeError(f"{string} is not a valid hex ID") from e + + +class HexString: + """Hexadecimal string""" + + def __new__(cls, string: str) -> bytes: # type: ignore + try: + return bytes.fromhex(string) + except ValueError as e: + raise argparse.ArgumentTypeError(f"{string} is not a valid hex ID") from e diff --git a/tests/util/test_argparse.py b/tests/util/test_argparse.py index a3a8584..45a2d03 100644 --- a/tests/util/test_argparse.py +++ b/tests/util/test_argparse.py @@ -6,7 +6,7 @@ import pytest -from infuse_iot.util.argparse import BtLeAddress, InfuseDeviceId, ValidDir, ValidFile +from infuse_iot.util.argparse import BtLeAddress, HexString, InfuseDeviceId, ValidDir, ValidFile assert "TOXTEMPDIR" in os.environ, "you must run these tests using tox" @@ -59,3 +59,16 @@ def test_infuse_device_id(): assert InfuseDeviceId("99") == 0x99 assert InfuseDeviceId("0x1234aa43bc") == 0x1234AA43BC assert InfuseDeviceId("1234aa43bc") == 0x1234AA43BC + + +def test_hexstring(): + with pytest.raises(argparse.ArgumentTypeError): + HexString("NotHex") + with pytest.raises(argparse.ArgumentTypeError): + HexString("aa:bb::00") + with pytest.raises(argparse.ArgumentTypeError): + HexString("0xaa") + assert HexString("aabb") == b"\xaa\xbb" + assert HexString("AABB") == b"\xaa\xbb" + assert HexString("aa00bb") == b"\xaa\x00\xbb" + assert HexString("00AABB") == b"\x00\xaa\xbb"