Add kvs_get and kvs_set method for RPC devices

aioshelly/rpc_device/device.py

@@ -5,6 +5,7 @@
 import asyncio
 import logging
 from collections.abc import Callable, Iterable
+from contextlib import suppress
 from enum import Enum, auto
 from functools import partial
 from typing import TYPE_CHECKING, Any, cast
@@ -39,6 +40,7 @@
     RpcCallError,
     ShellyError,
 )
+from ..json import JSONDecodeError, json_dumps, json_loads
 from .blerpc import BleRPC
 from .models import (
     ShellyBLEConfig,
@@ -581,6 +583,24 @@ async def wall_display_set_screen(self, value: bool) -> None:
         params = {"on": value}
         await self.call_rpc("Ui.Screen.Set", params)
 
+    async def kvs_get(self, key: str) -> dict[str, Any]:
+        """Get value from KVS."""
+        params = {"key": key}
+        response = await self.call_rpc("KVS.Get", params)
+
+        with suppress(JSONDecodeError):
+            response["value"] = json_loads(response["value"])
+
+        return response
+
+    async def kvs_set(
+        self, key: str, value: str | float | bool | dict | list | None
+    ) -> None:
+        """Set value in KVS."""
+        val = json_dumps(value) if isinstance(value, (dict, list)) else value
+        params = {"key": key, "value": val}
+        await self.call_rpc("KVS.Set", params)
+
     async def poll(self) -> None:
         """Poll device for calls that do not receive push updates."""
         calls: list[tuple[str, dict[str, Any] | None]] = [("Shelly.GetStatus", None)]

tests/rpc_device/test_device.py

@@ -1677,3 +1677,63 @@ async def test_wall_display_set_screen(
 
     assert call_args_list[0][0][0][0][0] == "Ui.Screen.Set"
     assert call_args_list[0][0][0][0][1] == {"on": True}
+
+
+@pytest.mark.parametrize(
+    ("in_value", "out_value"),
+    [
+        ("value1", "value1"),
+        ({"key1": "val1"}, '{"key1":"val1"}'),
+        ([{"key1": "val1"}, {"key2": "val2"}], '[{"key1":"val1"},{"key2":"val2"}]'),
+        (42.5, 42.5),
+        (True, True),
+        (None, None),
+    ],
+)
+@pytest.mark.asyncio
+async def test_kvs_set(
+    rpc_device: RpcDevice,
+    in_value: str | float | bool | dict | list | None,
+    out_value: str,
+) -> None:
+    """Test RpcDevice kvs_set() method."""
+    await rpc_device.kvs_set("key1", in_value)
+
+    assert rpc_device.call_rpc_multiple.call_count == 1
+    call_args_list = rpc_device.call_rpc_multiple.call_args_list
+
+    assert call_args_list[0][0][0][0][0] == "KVS.Set"
+    assert call_args_list[0][0][0][0][1] == {"key": "key1", "value": out_value}
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        ("value1", "value1"),
+        ('{"key1":"val1"}', {"key1": "val1"}),
+        ('[{"key1":"val1"},{"key2":"val2"}]', [{"key1": "val1"}, {"key2": "val2"}]),
+        (42.5, 42.5),
+        (True, True),
+        (None, None),
+    ],
+)
+@pytest.mark.asyncio
+async def test_kvs_get(
+    rpc_device: RpcDevice,
+    value: str | float | bool | None,
+    expected: str | float | bool | dict | list | None,
+) -> None:
+    """Test RpcDevice kvs_get() method."""
+    rpc_device.call_rpc_multiple.return_value = [
+        {"etag": "16mLia9TRt8lGhj9Zf5Dp6Hw==", "value": value}
+    ]
+
+    result = await rpc_device.kvs_get("key1")
+
+    assert result == {"etag": "16mLia9TRt8lGhj9Zf5Dp6Hw==", "value": expected}
+
+    assert rpc_device.call_rpc_multiple.call_count == 1
+    call_args_list = rpc_device.call_rpc_multiple.call_args_list
+
+    assert call_args_list[0][0][0][0][0] == "KVS.Get"
+    assert call_args_list[0][0][0][0][1] == {"key": "key1"}