Add support for checking if RPC device supports scripts

aioshelly/rpc_device/device.py

@@ -50,6 +50,10 @@
 
 MAX_ITERATIONS = 10
 
+RPC_CALL_ERR_METHOD_NOT_FOUND = -114
+RPC_CALL_ERR_INVALID_ARG = -105
+RPC_CALL_ERR_NO_HANDLER = 404
+
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -596,3 +600,36 @@ async def _retrieve_blutrv_components(self, components: dict[str, Any]) -> None:
             # if there are no errors, the response does not contain an errors object
             status.setdefault("errors", [])
             self._status.update({component["key"]: status})
+
+    async def supports_scripts(self) -> bool:
+        """Check if the device supports scripts.
+
+        Try to read a script to check if the device supports scripts,
+        if it supports scripts, it should return the script
+        or a specific error code if the script does not exist.
+        {"code":-105,"message":"Argument 'id', value 1 not found!"}
+
+        Errors by devices that do not support scripts:
+
+        Shelly Wall display:
+        {"code":-114,"message":"Method Script.GetCode failed: Method not found!"}
+
+        Shelly X MOD1
+        {"code":404,"message":"No handler for Script.GetCode"}
+        """
+        try:
+            await self.script_getcode(1)
+        except RpcCallError as err:
+            # The device supports scripts, but the script does not exist
+            if err.code == RPC_CALL_ERR_INVALID_ARG:
+                return True
+            # The device does not support scripts
+            if err.code in [
+                RPC_CALL_ERR_METHOD_NOT_FOUND,
+                RPC_CALL_ERR_NO_HANDLER,
+            ]:
+                return False
+            raise
+
+        # The device returned a script, it supports scripts
+        return True

tests/rpc_device/test_device.py

@@ -735,3 +735,49 @@ async def test_incorrect_shutdown(
     await rpc_device1.shutdown()
 
     assert "error during shutdown: KeyError('AABBCCDDEEFF')" in caplog.text
+
+
+@pytest.mark.parametrize(
+    ("side_effect", "supports_scripts"),
+    [
+        (RpcCallError(-105, "Argument 'id', value 1 not found!"), True),
+        (RpcCallError(-114, "Method Script.GetCode failed: Method not found!"), False),
+        (RpcCallError(404, "No handler for Script.GetCode"), False),
+        (
+            [
+                {
+                    "id": 5,
+                    "src": "shellyplus2pm-a8032ab720ac",
+                    "dst": "aios-2293750469632",
+                    "result": {"data": "script"},
+                }
+            ],
+            True,
+        ),
+    ],
+)
+@pytest.mark.asyncio
+async def test_supports_scripts(
+    rpc_device: RpcDevice,
+    side_effect: Exception | dict[str, Any],
+    supports_scripts: bool,
+) -> None:
+    """Test supports_scripts method."""
+    rpc_device.call_rpc_multiple.side_effect = [side_effect]
+
+    result = await rpc_device.supports_scripts()
+
+    assert result == supports_scripts
+    assert rpc_device.call_rpc_multiple.call_count == 1
+    assert rpc_device.call_rpc_multiple.call_args[0][0][0][0] == "Script.GetCode"
+    assert rpc_device.call_rpc_multiple.call_args[0][0][0][1] == {"id": 1}
+
+
+@pytest.mark.asyncio
+async def test_supports_scripts_raises_unkown_errors(rpc_device: RpcDevice) -> None:
+    """Test supports_scripts raises for unknown errors."""
+    message = "Missing required argument 'id'!"
+    rpc_device.call_rpc_multiple.side_effect = [RpcCallError(-103, message)]
+
+    with pytest.raises(RpcCallError, match=message):
+        await rpc_device.supports_scripts()

tools/common/__init__.py

@@ -194,3 +194,14 @@ async def wait_for_keyboard_interrupt() -> None:
     sig_event = asyncio.Event()
     signal.signal(signal.SIGINT, lambda _exit_code, _frame: sig_event.set())
     await sig_event.wait()
+
+
+async def check_rpc_device_supports_scripts(
+    options: ConnectionOptions, gen: int | None
+) -> None:
+    """Check if RPC device supports scripts."""
+    async with ClientSession() as aiohttp_session:
+        device: RpcDevice = await create_device(aiohttp_session, options, gen)
+        await device.initialize()
+        print(f"Supports scripts: {await device.supports_scripts()}")
+        await device.shutdown()

tools/example.py

@@ -13,6 +13,7 @@
 
 from aiohttp import ClientSession
 from common import (
+    check_rpc_device_supports_scripts,
     close_connections,
     coap_context,
     connect_and_print_device,
@@ -158,6 +159,12 @@ def get_arguments() -> tuple[argparse.ArgumentParser, argparse.Namespace]:
         default=None,
         help="Listen ip address for incoming CoAP packets",
     )
+    parser.add_argument(
+        "--supports_scripts",
+        "-ss",
+        action="store_true",
+        help="Check if device supports scripts",
+    )
 
     arguments = parser.parse_args()
 
@@ -212,6 +219,8 @@ async def main() -> None:
         )
         if args.update_ws:
             await update_outbound_ws(options, args.init, args.update_ws)
+        elif args.supports_scripts:
+            await check_rpc_device_supports_scripts(options, gen)
         else:
             await test_single(options, args.init, gen)
     else: