Add OverkizClientSettings for RTS default duration and future extensibility

pyoverkiz/client.py

@@ -21,11 +21,12 @@
 from backoff.types import Details
 
 from pyoverkiz._case import decamelize
-from pyoverkiz.action_queue import ActionQueue, ActionQueueSettings
+from pyoverkiz.action_queue import ActionQueue
 from pyoverkiz.auth import AuthStrategy, Credentials, build_auth_strategy
+from pyoverkiz.client_settings import OverkizClientSettings
 from pyoverkiz.const import SUPPORTED_SERVERS
 from pyoverkiz.converter import converter
-from pyoverkiz.enums import APIType, ExecutionMode, Server
+from pyoverkiz.enums import APIType, ExecutionMode, Protocol, Server
 from pyoverkiz.exceptions import (
     ExecutionQueueFullError,
     InvalidEventListenerIdError,
@@ -38,6 +39,7 @@
 )
 from pyoverkiz.models import (
     Action,
+    Command,
     Device,
     Event,
     Execution,
@@ -172,6 +174,7 @@ class OverkizClient:
     _ssl: ssl.SSLContext | bool = True
     _auth: AuthStrategy
     _action_queue: ActionQueue | None = None
+    _rts_command_duration: int | None = None
 
     def __init__(
         self,
@@ -180,15 +183,15 @@ def __init__(
         credentials: Credentials,
         verify_ssl: bool = True,
         session: ClientSession | None = None,
-        action_queue: bool | ActionQueueSettings = False,
+        settings: OverkizClientSettings | None = None,
     ) -> None:
         """Constructor.
 
         :param server: ServerConfig
         :param credentials: Credentials for authentication
         :param verify_ssl: Enable SSL certificate verification
         :param session: optional ClientSession
-        :param action_queue: enable batching or provide queue settings (default False)
+        :param settings: behavioral settings for the client (default None)
         """
         self.server_config = self._normalize_server(server)
 
@@ -206,23 +209,15 @@ def __init__(
             # Use the prebuilt SSL context with disabled strict validation for local API.
             self._ssl = SSL_CONTEXT_LOCAL_API
 
-        # Initialize action queue if enabled
-        queue_settings: ActionQueueSettings | None
-        if isinstance(action_queue, ActionQueueSettings):
-            queue_settings = action_queue
-        elif isinstance(action_queue, bool):
-            queue_settings = ActionQueueSettings() if action_queue else None
-        else:
-            raise TypeError(
-                "action_queue must be a bool or ActionQueueSettings, "
-                f"got {type(action_queue).__name__}"
-            )
+        # Apply behavioral settings
+        resolved_settings = settings or OverkizClientSettings()
+        self._rts_command_duration = resolved_settings.rts_command_duration
 
-        if queue_settings:
-            queue_settings.validate()
+        if resolved_settings.action_queue:
+            resolved_settings.action_queue.validate()
             self._action_queue = ActionQueue(
                 executor=self._execute_action_group_direct,
-                settings=queue_settings,
+                settings=resolved_settings.action_queue,
             )
 
         self._auth = build_auth_strategy(
@@ -494,6 +489,48 @@ async def get_api_version(self) -> str:
 
         return cast(str, response["protocolVersion"])
 
+    def _apply_rts_duration(self, actions: list[Action]) -> list[Action]:
+        """Append rts_command_duration to RTS commands that accept an extra parameter.
+
+        Returns a new list of actions with modified commands where applicable.
+        The original actions are not mutated.
+        """
+        if self._rts_command_duration is None:
+            return actions
+
+        device_index: dict[str, Device] = {d.device_url: d for d in self.devices}
+
+        result: list[Action] = []
+        for action in actions:
+            device = device_index.get(action.device_url)
+
+            if device is None or device.identifier.protocol != Protocol.RTS:
+                result.append(action)
+                continue
+
+            new_commands: list[Command] = []
+            for cmd in action.commands:
+                cmd_def = device.get_command_definition(str(cmd.name))
+                current_count = len(cmd.parameters) if cmd.parameters else 0
+
+                if cmd_def and current_count < cmd_def.nparams:
+                    new_commands.append(
+                        Command(
+                            name=cmd.name,
+                            parameters=[
+                                *(cmd.parameters or []),
+                                self._rts_command_duration,
+                            ],
+                            type=cmd.type,
+                        )
+                    )
+                else:
+                    new_commands.append(cmd)
+
+            result.append(Action(device_url=action.device_url, commands=new_commands))
+
+        return result
+
     @retry_on_too_many_executions
     @retry_on_auth_error
     async def _execute_action_group_direct(
@@ -544,6 +581,8 @@ async def execute_action_group(
         Returns:
             The ``exec_id`` identifying the execution on the server.
         """
+        actions = self._apply_rts_duration(actions)
+
         if self._action_queue:
             queued = await self._action_queue.add(actions, mode, label)
             return await queued

pyoverkiz/client_settings.py

@@ -0,0 +1,18 @@
+"""Behavioral configuration for OverkizClient."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from pyoverkiz.action_queue import ActionQueueSettings
+
+
+@dataclass(frozen=True, slots=True)
+class OverkizClientSettings:
+    """Behavioral configuration for OverkizClient.
+
+    All fields are optional and default to passive behavior.
+    """
+
+    action_queue: ActionQueueSettings | None = None
+    rts_command_duration: int | None = None

pyoverkiz/models.py

@@ -440,6 +440,14 @@ def select_first_command(self, commands: list[str | OverkizCommand]) -> str | No
             return None
         return self.definition.commands.select(commands)
 
+    def get_command_definition(
+        self, command: str | OverkizCommand
+    ) -> CommandDefinition | None:
+        """Return the CommandDefinition for a command, or None if unavailable."""
+        if self.definition is None:
+            return None
+        return self.definition.commands.get(str(command))
+
     def get_state_value(self, state: str) -> StateType | None:
         """Get value of a single state, or None if not found or None."""
         return self.states.select_value([state])

tests/test_client.py

@@ -10,13 +10,17 @@
 import pytest
 
 from pyoverkiz import exceptions
+from pyoverkiz.action_queue import ActionQueueSettings
+from pyoverkiz.auth import UsernamePasswordCredentials
 from pyoverkiz.client import OverkizClient
+from pyoverkiz.client_settings import OverkizClientSettings
 from pyoverkiz.enums import (
     APIType,
     DataType,
     ExecutionState,
     ExecutionSubType,
     ExecutionType,
+    Server,
 )
 from pyoverkiz.models import (
     Action,
@@ -1203,3 +1207,32 @@ async def test_local_schedule_persisted_action_group_unknown_object(
             await local_client.schedule_persisted_action_group(
                 "00000000-0000-0000-0000-000000000000", 9999999999
             )
+
+
+class TestOverkizClientSettings:
+    """Tests for the OverkizClientSettings integration with OverkizClient."""
+
+    def test_client_with_settings_none(self, client: OverkizClient) -> None:
+        """Client without settings has no action queue and no RTS duration."""
+        assert client._action_queue is None
+        assert client._rts_command_duration is None
+
+    @pytest.mark.asyncio
+    async def test_client_with_rts_duration(self) -> None:
+        """Client stores RTS command duration from settings."""
+        client = OverkizClient(
+            server=Server.SOMFY_EUROPE,
+            credentials=UsernamePasswordCredentials("user", "pass"),
+            settings=OverkizClientSettings(rts_command_duration=0),
+        )
+        assert client._rts_command_duration == 0
+
+    @pytest.mark.asyncio
+    async def test_client_with_action_queue_via_settings(self) -> None:
+        """Client creates action queue from settings."""
+        client = OverkizClient(
+            server=Server.SOMFY_EUROPE,
+            credentials=UsernamePasswordCredentials("user", "pass"),
+            settings=OverkizClientSettings(action_queue=ActionQueueSettings()),
+        )
+        assert client._action_queue is not None

tests/test_client_queue_integration.py

@@ -8,6 +8,7 @@
 from pyoverkiz.action_queue import ActionQueueSettings
 from pyoverkiz.auth import UsernamePasswordCredentials
 from pyoverkiz.client import OverkizClient
+from pyoverkiz.client_settings import OverkizClientSettings
 from pyoverkiz.enums import OverkizCommand, Server
 from pyoverkiz.models import Action, Command
 
@@ -18,7 +19,6 @@ async def test_client_without_queue_executes_immediately():
     client = OverkizClient(
         server=Server.SOMFY_EUROPE,
         credentials=UsernamePasswordCredentials("test@example.com", "test"),
-        action_queue=False,
     )
 
     action = Action(
@@ -48,7 +48,7 @@ async def test_client_with_queue_batches_actions():
     client = OverkizClient(
         server=Server.SOMFY_EUROPE,
         credentials=UsernamePasswordCredentials("test@example.com", "test"),
-        action_queue=ActionQueueSettings(delay=0.1),
+        settings=OverkizClientSettings(action_queue=ActionQueueSettings(delay=0.1)),
     )
 
     actions = [
@@ -98,7 +98,9 @@ async def test_client_manual_flush():
     client = OverkizClient(
         server=Server.SOMFY_EUROPE,
         credentials=UsernamePasswordCredentials("test@example.com", "test"),
-        action_queue=ActionQueueSettings(delay=10.0),  # Long delay
+        settings=OverkizClientSettings(
+            action_queue=ActionQueueSettings(delay=10.0)
+        ),  # Long delay
     )
 
     action = Action(
@@ -138,7 +140,7 @@ async def test_client_close_flushes_queue():
     client = OverkizClient(
         server=Server.SOMFY_EUROPE,
         credentials=UsernamePasswordCredentials("test@example.com", "test"),
-        action_queue=ActionQueueSettings(delay=10.0),
+        settings=OverkizClientSettings(action_queue=ActionQueueSettings(delay=10.0)),
     )
 
     action = Action(
@@ -171,9 +173,11 @@ async def test_client_queue_respects_max_actions():
     client = OverkizClient(
         server=Server.SOMFY_EUROPE,
         credentials=UsernamePasswordCredentials("test@example.com", "test"),
-        action_queue=ActionQueueSettings(
-            delay=10.0,
-            max_actions=2,  # Max 2 actions
+        settings=OverkizClientSettings(
+            action_queue=ActionQueueSettings(
+                delay=10.0,
+                max_actions=2,  # Max 2 actions
+            ),
         ),
     )
 

tests/test_client_settings.py

@@ -0,0 +1,24 @@
+"""Tests for OverkizClientSettings."""
+
+from pyoverkiz.action_queue import ActionQueueSettings
+from pyoverkiz.client_settings import OverkizClientSettings
+
+
+def test_defaults():
+    """Default settings have no queue and no RTS duration."""
+    settings = OverkizClientSettings()
+    assert settings.action_queue is None
+    assert settings.rts_command_duration is None
+
+
+def test_with_rts_duration():
+    """RTS command duration can be set."""
+    settings = OverkizClientSettings(rts_command_duration=0)
+    assert settings.rts_command_duration == 0
+
+
+def test_with_action_queue_settings():
+    """Passing ActionQueueSettings stores it directly."""
+    qs = ActionQueueSettings(delay=1.0, max_actions=10)
+    settings = OverkizClientSettings(action_queue=qs)
+    assert settings.action_queue is qs

tests/test_models.py

@@ -981,3 +981,56 @@ def test_persisted_action_group_id_property_returns_str(self):
         result = group.id
         assert isinstance(result, str)
         assert result == "abc-123"
+
+
+def test_get_command_definition_found():
+    """Device.get_command_definition() returns CommandDefinition when command exists."""
+    from pyoverkiz.models import CommandDefinition
+
+    device = _make_device(
+        {
+            **RAW_DEVICES,
+            "definition": {
+                **RAW_DEVICES["definition"],
+                "commands": [{"commandName": "open", "nparams": 0}],
+            },
+        }
+    )
+    cd = device.get_command_definition("open")
+    assert cd is not None
+    assert isinstance(cd, CommandDefinition)
+    assert cd.nparams == 0
+
+
+def test_get_command_definition_not_found():
+    """Device.get_command_definition() returns None when command doesn't exist."""
+    device = _make_device(
+        {
+            **RAW_DEVICES,
+            "definition": {
+                **RAW_DEVICES["definition"],
+                "commands": [],
+            },
+        }
+    )
+    assert device.get_command_definition("open") is None
+
+
+def test_get_command_definition_no_definition():
+    """Device.get_command_definition() returns None when device has no definition."""
+    from pyoverkiz.enums import ProductType
+    from pyoverkiz.models import States
+
+    device = Device(
+        attributes=States(),
+        available=True,
+        enabled=True,
+        label="Test",
+        device_url="io://1234-5678-9012/1",
+        controllable_name="test",
+        definition=None,
+        type=ProductType.ACTUATOR,
+        widget="SomeWidget",
+        ui_class="RollerShutter",
+    )
+    assert device.get_command_definition("open") is None