tools: cloud: device: KV update schedule

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}"
 

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,16 +76,26 @@ 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))
 
 
 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

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"