Add execute_action_group method and remove other command execution methods. (#1862)

- `client.execute_command()` and `client.execute_commands()` are
replaced by `client.execute_action_group()`

- `client.execute_action_group()` now supports multiple execution modes
(high priority, internal, geolocated)
- `client.execute_action_group()` now supports multiple device actions
in the same request

The current execution methods are poorly typed and do not support
concurrent execution across multiple devices, which makes it impossible
to properly work around TooManyExecutionsException and
TooManyConcurrentRequestsException.

The main change is the move from `client.execute_command()` and
`client.execute_commands()` to a single `client.execute_action_group()`.
An action group takes a list of actions, each of which can include
multiple device actions, including multiple commands per action.

```python3
await client.execute_action_group(
    actions=[
        Action(
            device_url="io://1234-5678-1234/12345678",
            commands=[
                Command(name="down"),
                Command(name="refresh")
            ]
        )
    ],
    label="Execution via Home Assistant"
)
```

New (mode) options like high priority will be possible now:

```python3
await client.execute_action_group(
    actions=[
        Action(
            device_url="io://1234-5678-1234/12345678",
            commands=[
                Command(name=OverkizCommand.SET_CLOSURE, parameters=[0])
            ]
        )
    ],
    label="Execution via Home Assistant",
    mode=CommandMode.HIGH_PRIORITY
)
```

This could serve as a foundation for grouping commands that are executed
within a short time window, for example when triggered by a scene or
automation in Home Assistant. Requests issued close together could be
batched and sent as a single action group, reducing the impact of
current Overkiz limitations.

The open question is where this queue should live: inside this
integration itself, or as part of the Home Assistant core
implementation.

---------

Co-authored-by: Copilot <[email protected]>

Author: iMicknl

README.md

@@ -39,7 +39,8 @@ import time
 
 from pyoverkiz.const import SUPPORTED_SERVERS
 from pyoverkiz.client import OverkizClient
-from pyoverkiz.enums import Server
+from pyoverkiz.models import Action
+from pyoverkiz.enums import Server, OverkizCommand
 
 USERNAME = ""
 PASSWORD = ""
@@ -61,6 +62,19 @@ async def main() -> None:
             print(f"{device.label} ({device.id}) - {device.controllable_name}")
             print(f"{device.widget} - {device.ui_class}")
 
+        await client.execute_action_group(
+            actions=[
+                Action(
+                    device_url="io://1234-5678-1234/12345678",
+                    commands=[
+                        Command(name=OverkizCommand.SET_CLOSURE, parameters=[100])
+                    ]
+                )
+            ],
+            label="Execution via Python",
+            # mode=CommandMode.HIGH_PRIORITY
+        )
+
         while True:
             events = await client.fetch_events()
             print(events)

pyoverkiz/client.py

@@ -39,7 +39,7 @@
     SOMFY_CLIENT_SECRET,
     SUPPORTED_SERVERS,
 )
-from pyoverkiz.enums import APIType, Server
+from pyoverkiz.enums import APIType, CommandMode, Server
 from pyoverkiz.exceptions import (
     AccessDeniedToGatewayException,
     BadCredentialsException,
@@ -71,8 +71,8 @@
     UnknownUserException,
 )
 from pyoverkiz.models import (
+    Action,
     ActionGroup,
-    Command,
     Device,
     Event,
     Execution,
@@ -87,6 +87,7 @@
     State,
 )
 from pyoverkiz.obfuscate import obfuscate_sensitive_data
+from pyoverkiz.serializers import prepare_payload
 from pyoverkiz.types import JSON
 
 
@@ -630,40 +631,42 @@ async def get_api_version(self) -> str:
 
     @retry_on_too_many_executions
     @retry_on_auth_error
-    async def execute_command(
+    async def execute_action_group(
         self,
-        device_url: str,
-        command: Command | str,
+        actions: list[Action],
+        mode: CommandMode | None = None,
         label: str | None = "python-overkiz-api",
     ) -> str:
-        """Send a command."""
-        if isinstance(command, str):
-            command = Command(command)
+        """Execute a non-persistent action group.
 
-        response: str = await self.execute_commands(device_url, [command], label)
+        The executed action group does not have to be persisted on the server before use.
+        Per-session rate-limit : 1 calls per 28min 48s period for all operations of the same category (exec)
+        """
+        # Build a logical (snake_case) payload using model helpers and convert it
+        # to the exact JSON schema expected by the API (camelCase + small fixes).
+        payload = {"label": label, "actions": [a.to_payload() for a in actions]}
+
+        # Prepare final payload with camelCase keys and special abbreviation handling
+        final_payload = prepare_payload(payload)
+
+        if mode == CommandMode.GEOLOCATED:
+            url = "exec/apply/geolocated"
+        elif mode == CommandMode.INTERNAL:
+            url = "exec/apply/internal"
+        elif mode == CommandMode.HIGH_PRIORITY:
+            url = "exec/apply/highPriority"
+        else:
+            url = "exec/apply"
 
-        return response
+        response: dict = await self.__post(url, final_payload)
+
+        return cast(str, response["execId"])
 
     @retry_on_auth_error
     async def cancel_command(self, exec_id: str) -> None:
         """Cancel a running setup-level execution."""
         await self.__delete(f"/exec/current/setup/{exec_id}")
 
-    @retry_on_auth_error
-    async def execute_commands(
-        self,
-        device_url: str,
-        commands: list[Command],
-        label: str | None = "python-overkiz-api",
-    ) -> str:
-        """Send several commands in one call."""
-        payload = {
-            "label": label,
-            "actions": [{"deviceURL": device_url, "commands": commands}],
-        }
-        response: dict = await self.__post("exec/apply", payload)
-        return cast(str, response["execId"])
-
     @retry_on_auth_error
     async def get_action_groups(self) -> list[ActionGroup]:
         """List the action groups (scenarios)."""

pyoverkiz/models.py

@@ -22,6 +22,7 @@
     UIWidget,
     UpdateBoxStatus,
 )
+from pyoverkiz.enums.command import OverkizCommand, OverkizCommandParam
 from pyoverkiz.enums.protocol import Protocol
 from pyoverkiz.obfuscate import obfuscate_email, obfuscate_id, obfuscate_string
 from pyoverkiz.types import DATA_TYPE_TO_PYTHON, StateType
@@ -465,19 +466,44 @@ def __len__(self) -> int:
     get = __getitem__
 
 
-class Command(dict):
+@define(init=False, kw_only=True)
+class Command:
     """Represents an OverKiz Command."""
 
-    name: str
-    parameters: list[str | int | float] | None
+    type: int | None = None
+    name: OverkizCommand
+    parameters: list[str | int | float | OverkizCommandParam] | None
 
     def __init__(
-        self, name: str, parameters: list[str | int | float] | None = None, **_: Any
+        self,
+        name: OverkizCommand,
+        parameters: list[str | int | float | OverkizCommandParam] | None = None,
+        type: int | None = None,
+        **_: Any,
     ):
         """Initialize a command instance and mirror fields into dict base class."""
         self.name = name
         self.parameters = parameters
-        dict.__init__(self, name=name, parameters=parameters)
+        self.type = type
+
+    def to_payload(self) -> dict[str, object]:
+        """Return a JSON-serializable payload for this command.
+
+        The payload uses snake_case keys; the client will convert to camelCase
+        and apply small key fixes (like `deviceURL`) before sending.
+        """
+        payload: dict[str, object] = {"name": str(self.name)}
+
+        if self.type is not None:
+            payload["type"] = self.type
+
+        if self.parameters is not None:
+            payload["parameters"] = [
+                p if isinstance(p, (str, int, float, bool)) else str(p)
+                for p in self.parameters  # type: ignore[arg-type]
+            ]
+
+        return payload
 
 
 @define(init=False, kw_only=True)
@@ -600,10 +626,22 @@ class Action:
     device_url: str
     commands: list[Command]
 
-    def __init__(self, device_url: str, commands: list[dict[str, Any]]):
+    def __init__(self, device_url: str, commands: list[dict[str, Any] | Command]):
         """Initialize Action from API data and convert nested commands."""
         self.device_url = device_url
-        self.commands = [Command(**c) for c in commands] if commands else []
+        self.commands = [
+            c if isinstance(c, Command) else Command(**c) for c in commands
+        ]
+
+    def to_payload(self) -> dict[str, object]:
+        """Return a JSON-serializable payload for this action (snake_case).
+
+        The final camelCase conversion is handled by the client.
+        """
+        return {
+            "device_url": self.device_url,
+            "commands": [c.to_payload() for c in self.commands],
+        }
 
 
 @define(init=False, kw_only=True)

pyoverkiz/serializers.py

@@ -0,0 +1,39 @@
+"""Helpers for preparing API payloads.
+
+This module centralizes JSON key formatting and any small transport-specific
+fixes (like mapping "deviceUrl" -> "deviceURL"). Models should produce
+logical snake_case payloads and the client should call `prepare_payload`
+before sending the payload to Overkiz.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import humps
+
+# Small mapping for keys that need special casing beyond simple camelCase.
+_ABBREV_MAP: dict[str, str] = {"deviceUrl": "deviceURL"}
+
+
+def _fix_abbreviations(obj: Any) -> Any:
+    if isinstance(obj, dict):
+        out = {}
+        for k, v in obj.items():
+            k = _ABBREV_MAP.get(k, k)
+            out[k] = _fix_abbreviations(v)
+        return out
+    if isinstance(obj, list):
+        return [_fix_abbreviations(i) for i in obj]
+    return obj
+
+
+def prepare_payload(payload: Any) -> Any:
+    """Convert snake_case payload to API-ready camelCase and apply fixes.
+
+    Example:
+        payload = {"device_url": "x", "commands": [{"name": "close"}]}
+        => {"deviceURL": "x", "commands": [{"name": "close"}]}
+    """
+    camel = humps.camelize(payload)
+    return _fix_abbreviations(camel)

tests/test_client.py

@@ -379,6 +379,41 @@ async def test_get_setup_options(
             for option in options:
                 assert isinstance(option, Option)
 
+    @pytest.mark.asyncio
+    async def test_execute_action_group_omits_none_fields(self, client: OverkizClient):
+        """Ensure `type` and `parameters` that are None are omitted from the request payload."""
+        from pyoverkiz.enums.command import OverkizCommand
+        from pyoverkiz.models import Action, Command
+
+        action = Action(
+            "rts://2025-8464-6867/16756006",
+            [Command(name=OverkizCommand.CLOSE, parameters=None, type=None)],
+        )
+
+        resp = MockResponse('{"execId": "exec-123"}')
+
+        with patch.object(aiohttp.ClientSession, "post") as mock_post:
+            mock_post.return_value = resp
+
+            exec_id = await client.execute_action_group([action])
+
+            assert exec_id == "exec-123"
+
+            assert mock_post.called
+            _, kwargs = mock_post.call_args
+            sent_json = kwargs.get("json")
+            assert sent_json is not None
+
+            # The client should have converted payload to camelCase and applied
+            # abbreviation fixes (deviceURL) before sending.
+            action_sent = sent_json["actions"][0]
+            assert action_sent.get("deviceURL") == action.device_url
+
+            cmd = action_sent["commands"][0]
+            assert "type" not in cmd
+            assert "parameters" not in cmd
+            assert cmd["name"] == "close"
+
     @pytest.mark.parametrize(
         "fixture_name, option_name, instance",
         [

tests/test_models.py

@@ -300,3 +300,33 @@ def test_bad_list_value(self):
         state = State(name="state", type=DataType.BOOLEAN, value=False)
         with pytest.raises(TypeError):
             assert state.value_as_list
+
+
+def test_command_to_payload_omits_none():
+    """Command.to_payload omits None fields from the resulting payload."""
+    from pyoverkiz.enums.command import OverkizCommand
+    from pyoverkiz.models import Command
+
+    cmd = Command(name=OverkizCommand.CLOSE, parameters=None, type=None)
+    payload = cmd.to_payload()
+
+    assert payload == {"name": "close"}
+
+
+def test_action_to_payload_and_parameters_conversion():
+    """Action.to_payload converts nested Command enums to primitives."""
+    from pyoverkiz.enums.command import OverkizCommand, OverkizCommandParam
+    from pyoverkiz.models import Action, Command
+
+    cmd = Command(
+        name=OverkizCommand.SET_LEVEL, parameters=[10, OverkizCommandParam.A], type=1
+    )
+    action = Action("rts://2025-8464-6867/16756006", [cmd])
+
+    payload = action.to_payload()
+
+    assert payload["device_url"] == "rts://2025-8464-6867/16756006"
+    assert payload["commands"][0]["name"] == "setLevel"
+    assert payload["commands"][0]["type"] == 1
+    # parameters should be converted to primitives (enum -> str)
+    assert payload["commands"][0]["parameters"] == [10, "A"]

tests/test_serializers.py

@@ -0,0 +1,37 @@
+"""Tests for pyoverkiz.serializers."""
+
+from __future__ import annotations
+
+from pyoverkiz.serializers import prepare_payload
+
+
+def test_prepare_payload_camelizes_and_fixes_device_url():
+    """Test that prepare_payload converts snake_case to camelCase and fixes abbreviations."""
+    payload = {
+        "label": "test",
+        "actions": [{"device_url": "rts://1/2", "commands": [{"name": "close"}]}],
+    }
+
+    final = prepare_payload(payload)
+
+    assert final["label"] == "test"
+    assert "deviceURL" in final["actions"][0]
+    assert final["actions"][0]["deviceURL"] == "rts://1/2"
+
+
+def test_prepare_payload_nested_lists_and_dicts():
+    """Test that prepare_payload handles nested lists and dicts correctly."""
+    payload = {
+        "actions": [
+            {
+                "device_url": "rts://1/2",
+                "commands": [{"name": "setLevel", "parameters": [10, "A"]}],
+            }
+        ]
+    }
+
+    final = prepare_payload(payload)
+
+    cmd = final["actions"][0]["commands"][0]
+    assert cmd["name"] == "setLevel"
+    assert cmd["parameters"] == [10, "A"]