SDK Reference

Interface class for the Overkiz API.

Source code in pyoverkiz/client.py

class OverkizClient:
    """Interface class for the Overkiz API."""

    server_config: ServerConfig
    setup: Setup | None
    devices: list[Device]
    gateways: list[Gateway]
    session: ClientSession
    _ssl: ssl.SSLContext | bool = True
    _auth: AuthStrategy
    _action_queue: ActionQueue | None = None
    _event_listener_id: str | None
    settings: OverkizClientSettings

    @property
    def event_listener_id(self) -> str | None:
        """Return the current event listener ID (read-only)."""
        return self._event_listener_id

    def __init__(
        self,
        *,
        server: ServerConfig | Server | str,
        credentials: Credentials,
        verify_ssl: bool = True,
        session: ClientSession | None = None,
        settings: OverkizClientSettings | None = None,
    ) -> None:
        """Constructor.

        Args:
            server: ServerConfig, Server enum, or server key string.
            credentials: Credentials for authentication.
            verify_ssl: Enable SSL certificate verification.
            session: Optional ClientSession.
            settings: Behavioral settings for the client.
        """
        self.server_config = self._normalize_server(server)

        self.setup: Setup | None = None
        self.devices: list[Device] = []
        self.gateways: list[Gateway] = []
        self._event_listener_id: str | None = None

        self.session = session or ClientSession(
            headers={"User-Agent": USER_AGENT}, timeout=DEFAULT_TIMEOUT
        )
        self._ssl = verify_ssl

        if self.server_config.api_type == APIType.LOCAL and verify_ssl:
            # Use the prebuilt SSL context with disabled strict validation for local API.
            self._ssl = SSL_CONTEXT_LOCAL_API

        self.settings = settings or OverkizClientSettings()

        if self.settings.action_queue:
            self.settings.action_queue.validate()
            self._action_queue = ActionQueue(
                executor=self._execute_action_group_direct,
                settings=self.settings.action_queue,
            )

        self._auth = build_auth_strategy(
            server_config=self.server_config,
            credentials=credentials,
            session=self.session,
            ssl_context=self._ssl,
        )

    async def __aenter__(self) -> Self:
        """Enter async context manager and return the client instance."""
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        """Exit the async context manager and close the client session."""
        await self.close()

    @staticmethod
    def _normalize_server(server: ServerConfig | Server | str) -> ServerConfig:
        """Resolve user-provided server identifiers into a `ServerConfig`."""
        if isinstance(server, ServerConfig):
            return server

        server_key = server.value if isinstance(server, Server) else str(server)

        try:
            return SUPPORTED_SERVERS[server_key]
        except KeyError as error:
            raise OverkizError(
                f"Unknown server '{server_key}'. Provide a supported server key or ServerConfig instance."
            ) from error

    async def close(self) -> None:
        """Close the session."""
        # Flush any pending actions in queue
        if self._action_queue:
            await self._action_queue.shutdown()

        if self.event_listener_id:
            await self.unregister_event_listener()

        await self._auth.close()
        await self.session.close()

    async def login(
        self,
        register_event_listener: bool = True,
    ) -> None:
        """Authenticate and create an API session allowing access to the other operations.

        Caller must provide one of [userId+userPassword, userId+ssoToken, accessToken, jwt].

        Raises:
            BadCredentialsError: When the provided credentials are invalid.
            TooManyAttemptsBannedError: When too many failed login attempts have been made.
            TooManyRequestsError: When the API rate limit has been exceeded.
        """
        await self._auth.login()

        if self.server_config.api_type == APIType.LOCAL:
            if register_event_listener:
                await self.register_event_listener()
            else:
                # Validate local API token by calling a simple endpoint
                await self.get_gateways()

            return

        if register_event_listener:
            await self.register_event_listener()

        return

    @retry_on_auth_error
    async def get_setup(self, refresh: bool = False) -> Setup:
        """Get all data about the connected user setup.

            -> gateways data (serial number, activation state, ...): <gateways/gateway>
            -> setup location: <location>
            -> house places (rooms and floors): <place>
            -> setup devices: <devices>.

        A gateway may be in different modes (mode) regarding to the activated functions (functions).
        A house may be composed of several floors and rooms. The house, floors and rooms are viewed as a place.
        Devices in the house are grouped by type called uiClass. Each device has an associated widget.
        The widget is used to control or to know the device state, whatever the device protocol (controllable): IO, RTS, X10, ... .
        A device can be either an actuator (type=1) or a sensor (type=2).
        Data of one or several devices can be also get by setting the device(s) url as request parameter.

        Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load)
        """
        if self.setup and not refresh:
            return self.setup

        response = await self._get("setup")

        setup = converter.structure(response, Setup)

        # Cache response
        self.setup = setup
        self.gateways = setup.gateways
        self.devices = setup.devices

        return setup

    @retry_on_auth_error
    async def get_diagnostic_data(
        self, mask_sensitive_data: bool = True
    ) -> dict[str, Any]:
        """Get diagnostic data for the connected user setup.

            -> gateways data (serial number, activation state, ...): <gateways/gateway>
            -> setup location: <location>
            -> house places (rooms and floors): <place>
            -> setup devices: <devices>
            -> action groups: <actionGroups>

        By default, this data is masked to not return confidential or PII data.
        Set `mask_sensitive_data` to `False` to return the raw payloads.
        """
        setup, action_groups = await asyncio.gather(
            self._get("setup"),
            self._get("actionGroups"),
        )

        if mask_sensitive_data:
            setup = obfuscate_sensitive_data(setup)
            action_groups = obfuscate_sensitive_data(action_groups)

        return {
            "setup": setup,
            "action_groups": action_groups,
        }

    @retry_on_auth_error
    async def get_devices(self, refresh: bool = False) -> list[Device]:
        """List devices.

        Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load).
        """
        if self.devices and not refresh:
            return self.devices

        response = await self._get("setup/devices")
        devices = converter.structure(response, list[Device])

        # Cache response
        self.devices = devices
        if self.setup:
            self.setup.devices = devices

        return devices

    @retry_on_auth_error
    async def get_gateways(self, refresh: bool = False) -> list[Gateway]:
        """Get every gateways of a connected user setup.

        Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load).
        """
        if self.gateways and not refresh:
            return self.gateways

        response = await self._get("setup/gateways")
        gateways = converter.structure(response, list[Gateway])

        # Cache response
        self.gateways = gateways
        if self.setup:
            self.setup.gateways = gateways

        return gateways

    @retry_on_auth_error
    async def get_execution_history(self) -> list[HistoryExecution]:
        """List past executions and their outcomes."""
        response = await self._get("history/executions")
        return converter.structure(response, list[HistoryExecution])

    @retry_on_auth_error
    async def get_device_definition(self, device_url: str) -> Definition | None:
        """Retrieve a particular setup device definition."""
        response: dict = await self._get(
            f"setup/devices/{urllib.parse.quote_plus(device_url)}"
        )

        raw = response.get("definition")
        if raw is None:
            return None

        return converter.structure(raw, Definition)

    @retry_on_auth_error
    async def get_state(self, device_url: str) -> list[State]:
        """Retrieve states of requested device."""
        response = await self._get(
            f"setup/devices/{urllib.parse.quote_plus(device_url)}/states"
        )
        return converter.structure(response, list[State])

    @retry_on_auth_error
    async def refresh_states(self) -> None:
        """Ask the box to refresh all devices states for protocols supporting that operation."""
        await self._post("setup/devices/states/refresh")

    @retry_on_auth_error
    async def refresh_device_states(self, device_url: str) -> None:
        """Ask the box to refresh all states of the given device for protocols supporting that operation."""
        await self._post(
            f"setup/devices/{urllib.parse.quote_plus(device_url)}/states/refresh"
        )

    @retry_on_concurrent_requests
    async def register_event_listener(self) -> str:
        """Register a new setup event listener on the current session and return a new.

        listener id.
        Only one listener may be registered on a given session.
        Registering an new listener will invalidate the previous one if any.
        Note that registering an event listener drastically reduces the session
        timeout : listening sessions are expected to call the /events/{listenerId}/fetch
        API on a regular basis.
        """
        response = await self._post("events/register")
        listener_id = cast(str, response.get("id"))
        self._event_listener_id = listener_id

        return listener_id

    @retry_on_concurrent_requests
    @retry_on_auth_error
    @retry_on_listener_error
    async def fetch_events(self) -> list[Event]:
        """Fetch new events from a registered event listener. Fetched events are removed.

        from the listener buffer. Return an empty response if no event is available.
        Per-session rate-limit : 1 calls per 1 SECONDS period for this particular
        operation (polling).
        """
        response = await self._post(f"events/{self.event_listener_id}/fetch")
        return converter.structure(response, list[Event])

    async def unregister_event_listener(self) -> None:
        """Unregister an event listener.

        API response status is always 200, even on unknown listener ids.
        """
        await self._post(f"events/{self._event_listener_id}/unregister")
        self._event_listener_id = None

    @retry_on_auth_error
    async def get_current_execution(self, exec_id: str) -> Execution | None:
        """Get a currently running execution by its exec_id.

        Returns None if the execution does not exist.
        """
        response = await self._get(f"exec/current/{exec_id}")
        if not response or not isinstance(response, dict):
            return None

        return converter.structure(response, Execution)

    @retry_on_auth_error
    async def get_current_executions(self) -> list[Execution]:
        """Get all currently running executions."""
        response = await self._get("exec/current")
        return converter.structure(response, list[Execution])

    @retry_on_auth_error
    async def get_api_version(self) -> str:
        """Get the API version (local only)."""
        response = await self._get("apiVersion")

        return cast(str, response["protocolVersion"])

    def _apply_rts_duration(self, actions: list[Action]) -> list[Action]:
        """Apply the default execution duration for RTS commands.

        For RTS commands, the last parameter is always the execution duration
        (default 15s for tilt commands, 30s for movement commands). This
        injects ``default_rts_command_duration`` as the last parameter only
        when the user has not already provided it — i.e., when the command
        has all domain-specific parameters filled but the duration slot is
        still empty (current_count == nparams - 1).

        If the user explicitly passes the duration parameter themselves, it
        is left unchanged.
        """
        duration = self.settings.default_rts_command_duration
        if 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

            updated_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 cmd_def.nparams > 0
                    and current_count == cmd_def.nparams - 1
                ):
                    updated_commands.append(
                        Command(
                            name=cmd.name,
                            parameters=[*(cmd.parameters or []), duration],
                            type=cmd.type,
                        )
                    )
                else:
                    updated_commands.append(cmd)

            result.append(
                Action(device_url=action.device_url, commands=updated_commands)
            )

        return result

    @retry_on_too_many_executions
    @retry_on_auth_error
    async def _execute_action_group_direct(
        self,
        actions: list[Action],
        mode: ExecutionMode | None = None,
        label: str | None = "pyOverkiz",
    ) -> str:
        """Execute a non-persistent action group directly (internal method).

        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)
        """
        payload = {"label": label, "actions": [a.to_payload() for a in actions]}
        url = f"exec/apply/{mode.value}" if mode else "exec/apply"

        response: dict = await self._post(url, prepare_payload(payload))

        return cast(str, response["execId"])

    async def execute_action_group(
        self,
        *,
        actions: list[Action],
        mode: ExecutionMode | None = None,
        label: str | None = "pyOverkiz",
    ) -> str:
        """Execute an ad-hoc action group built from the given actions.

        An action group is a batch of device actions submitted as a single
        execution. Each ``Action`` targets one device and contains one or more
        ``Command`` instances (e.g. ``open``, ``setClosure(50)``). The gateway
        allows at most one action per device per action group.

        When the action queue is enabled, actions are held for a short delay
        and merged with other actions submitted in the same window. Commands
        targeting the same device are combined into a single action. The method
        blocks until the batch executes and returns the resulting exec_id.

        When the action queue is disabled, the action group is sent immediately.

        Args:
            actions: One or more actions to execute. Each action targets a
                single device and holds one or more commands.
            mode: Optional execution mode (``HIGH_PRIORITY``, ``GEOLOCATED``,
                or ``INTERNAL``). Only supported by the Cloud API; the local
                API rejects requests that specify an execution mode (see
                https://github.com/Somfy-Developer/Somfy-TaHoma-Developer-Mode/issues/227).
            label: Human-readable label for the execution.

        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
        return await self._execute_action_group_direct(actions, mode, label)

    async def flush_action_queue(self) -> None:
        """Force flush all pending actions in the queue immediately.

        If action queue is disabled, this method does nothing.
        If there are no pending actions, this method does nothing.
        """
        if self._action_queue:
            await self._action_queue.flush()

    def get_pending_actions_count(self) -> int:
        """Get the approximate number of actions currently waiting in the queue.

        Returns 0 if action queue is disabled. This is a best-effort snapshot
        and may be stale if other coroutines modify the queue concurrently.
        """
        if self._action_queue:
            return self._action_queue.get_pending_count()
        return 0

    @retry_on_auth_error
    async def cancel_execution(self, exec_id: str) -> None:
        """Cancel a running execution by its exec_id."""
        await self._delete(f"exec/current/setup/{exec_id}")

    @retry_on_auth_error
    async def get_action_groups(self) -> list[PersistedActionGroup]:
        """List action groups persisted on the server."""
        response = await self._get("actionGroups")
        return converter.structure(response, list[PersistedActionGroup])

    @retry_on_auth_error
    async def get_places(self) -> Place:
        """Get the hierarchical structure of places (house, rooms, areas, zones).

        The Place model represents a hierarchical organization where the root place is
        typically the house/property, and `sub_places` contains nested child places
        (floors, rooms, areas). This structure can be recursively navigated to build
        a complete map of all locations in the setup. Each place has:
        - `label`: Human-readable name for the place
        - `type`: Numeric identifier for the place type
        - `sub_places`: List of nested places within this location
        """
        response = await self._get("setup/places")
        return converter.structure(response, Place)

    @retry_on_auth_error
    async def execute_persisted_action_group(self, oid: str) -> str:
        """Execute a server-side action group by its OID (see ``get_action_groups``)."""
        response = await self._post(f"exec/{oid}")
        return cast(str, response["execId"])

    @retry_on_auth_error
    async def schedule_persisted_action_group(self, oid: str, timestamp: int) -> str:
        """Schedule a server-side action group for execution at the given timestamp."""
        response = await self._post(f"exec/schedule/{oid}/{timestamp}")
        return cast(str, response["triggerId"])

    @retry_on_auth_error
    async def get_setup_options(self) -> list[Option]:
        """This operation returns all subscribed options of a given setup.

        Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load)
        Access scope : Full enduser API access (enduser/*).
        """
        response = await self._get("setup/options")
        return converter.structure(response, list[Option])

    @retry_on_auth_error
    async def get_setup_option(self, option: str) -> Option | None:
        """This operation returns the selected subscribed option of a given setup.

        For example `developerMode-{gateway_id}` to understand if developer mode is on.
        """
        response = await self._get(f"setup/options/{option}")

        if response:
            return converter.structure(response, Option)

        return None

    @retry_on_auth_error
    async def get_setup_option_parameter(
        self, option: str, parameter: str
    ) -> OptionParameter | None:
        """This operation returns the selected parameters of a given setup and option.

        For example `developerMode-{gateway_id}` and `gatewayId` to understand if developer mode is on.

        If the option is not available, an OverkizError will be thrown.
        If the parameter is not available you will receive None.
        """
        response = await self._get(f"setup/options/{option}/{parameter}")

        if response:
            return converter.structure(response, OptionParameter)

        return None

    @retry_on_auth_error
    async def get_reference_controllable(
        self, controllable_name: str
    ) -> dict[str, Any]:
        """Get a controllable definition."""
        return await self._get(
            f"reference/controllable/{urllib.parse.quote_plus(controllable_name)}"
        )

    @retry_on_auth_error
    async def get_reference_controllable_types(self) -> list[dict[str, Any]]:
        """Get details about all supported controllable types."""
        return await self._get("reference/controllableTypes")

    @retry_on_auth_error
    async def search_reference_devices(
        self, payload: dict[str, Any]
    ) -> DeviceSearchResult:
        """Search reference device models using a POST payload.

        The payload can filter by protocol types, UI classes, widgets, classifiers,
        controllable types, or free-text criteria. Additional flags control whether
        commands, states, attributes and manufacturer references are included.

        Example payload:
            {
                "protocolTypes": ["IO"],
                "withCommands": True,
                "withStates": True,
                "withAttributes": True,
                "withManufacturerReferences": True,
            }
        """
        response = await self._post("reference/devices/search", payload)
        return converter.structure(response, DeviceSearchResult)

    @retry_on_auth_error
    async def get_reference_protocol_types(self) -> list[ProtocolType]:
        """Get details about supported protocol types on that server instance.

        Returns a list of protocol type definitions, each containing:
        - id: Numeric protocol identifier
        - prefix: URL prefix used in device addresses
        - name: Internal protocol name
        - label: Human-readable protocol label
        """
        response = await self._get("reference/protocolTypes")
        return converter.structure(response, list[ProtocolType])

    @retry_on_auth_error
    async def get_reference_timezones(self) -> list[dict[str, Any]]:
        """Get timezones list."""
        return await self._get("reference/timezones")

    @retry_on_auth_error
    async def get_reference_ui_classes(self) -> list[str]:
        """Get a list of all defined UI classes."""
        return await self._get("reference/ui/classes")

    @retry_on_auth_error
    async def get_reference_ui_classifiers(self) -> list[str]:
        """Get a list of all defined UI classifiers."""
        return await self._get("reference/ui/classifiers")

    @retry_on_auth_error
    async def get_reference_ui_profile(self, profile_name: str) -> UIProfileDefinition:
        """Get a description of a given UI profile (or form-factor variant).

        Returns a profile definition containing:
        - name: Profile name
        - commands: Available commands with parameters and descriptions
        - states: Available states with value types and descriptions
        - form_factor: Whether profile is tied to a specific physical device type
        """
        response = await self._get(
            f"reference/ui/profile/{urllib.parse.quote_plus(profile_name)}"
        )
        return converter.structure(response, UIProfileDefinition)

    @retry_on_auth_error
    async def get_reference_ui_profile_names(self) -> list[str]:
        """Get a list of all defined UI profiles (and form-factor variants)."""
        return await self._get("reference/ui/profileNames")

    @retry_on_auth_error
    async def get_reference_ui_widgets(self) -> list[str]:
        """Get a list of all defined UI widgets."""
        return await self._get("reference/ui/widgets")

    @retry_on_auth_error
    async def get_devices_not_up_to_date(self) -> list[Device]:
        """Get all devices whose firmware is not up to date."""
        response = await self._get("setup/devices/notUpToDate")
        return converter.structure(response, list[Device])

    @retry_on_auth_error
    async def get_device_firmware_status(
        self, device_url: str
    ) -> FirmwareStatus | None:
        """Check if a device's firmware is up to date.

        Returns None if the device does not support firmware status checks.
        """
        try:
            response = await self._get(
                f"setup/devices/{urllib.parse.quote_plus(device_url)}/firmwareUpToDate"
            )
        except UnsupportedOperationError:
            return None
        return converter.structure(response, FirmwareStatus)

    @retry_on_auth_error
    async def get_device_firmware_update_capability(self, device_url: str) -> bool:
        """Check if a device supports firmware updates.

        Returns False if the device does not support this query.
        """
        try:
            response = await self._get(
                f"setup/devices/{urllib.parse.quote_plus(device_url)}/firmwareUpdateCapability"
            )
        except UnsupportedOperationError:
            return False
        return cast(bool, response["supportsFirmwareUpdate"])

    @retry_on_auth_error
    async def update_device_firmware(self, device_url: str) -> None:
        """Update a device's firmware to the next available version."""
        await self._put(
            f"setup/devices/{urllib.parse.quote_plus(device_url)}/updateFirmware"
        )

    @retry_on_auth_error
    async def update_all_device_firmwares(self) -> None:
        """Update firmware for all devices that are not up to date."""
        await self._put("setup/devices/updateFirmwares")

    @retry_on_auth_error
    async def get_device_controllables(self) -> dict[str, list[str]]:
        """Get all device URLs grouped by their controllable name.

        Returns a dict mapping controllable names to lists of device URLs
        that use that controllable.
        """
        return await self._get("setup/devices/controllables")

    @retry_on_auth_error
    async def get_device_controllable_definition(
        self, device_url: str
    ) -> Definition | None:
        """Get the controllable definition for a specific device."""
        response = await self._get(
            f"setup/devices/{urllib.parse.quote_plus(device_url)}/controllable"
        )
        if response is None:
            return None
        return converter.structure(response, Definition)

    @retry_on_auth_error
    async def get_device_alternative_controllables(self, device_url: str) -> list[str]:
        """Get all alternative controllable names for a device."""
        return await self._get(
            f"setup/devices/{urllib.parse.quote_plus(device_url)}/alternativeControllables"
        )

    @retry_on_auth_error
    async def get_device_manufacturer_references(
        self, device_url: str
    ) -> list[DeviceManufacturerReference]:
        """Get all manufacturer references for a device."""
        response = await self._get(
            f"setup/devices/{urllib.parse.quote_plus(device_url)}/manufacturerReferences"
        )
        return converter.structure(response, list[DeviceManufacturerReference])

    async def discover_gateways(self) -> list[GatewayCandidate]:
        """Discover selectable gateways.

        Raises:
            UnsupportedOperationError: When the server does not support gateway selection.
        """
        if not isinstance(self._auth, SupportsGatewaySelection):
            raise UnsupportedOperationError(
                f"{self.server_config.name} does not support gateway selection."
            )
        return await self._auth.discover_gateways()

    def select_gateway(self, gateway_id: str) -> None:
        """Select the gateway to scope requests to.

        Raises:
            UnsupportedOperationError: When the server does not support gateway selection.
        """
        if not isinstance(self._auth, SupportsGatewaySelection):
            raise UnsupportedOperationError(
                f"{self.server_config.name} does not support gateway selection."
            )
        self._auth.select_gateway(gateway_id)

    # -----------------------------------------------------------------------
    # Local token management (cloud API)
    # -----------------------------------------------------------------------

    @retry_on_auth_error
    async def generate_local_token(self, gateway_id: str) -> str:
        """Generate a new local API token for a gateway.

        The token must be activated before it can be used.
        """
        response = await self._get(f"config/{gateway_id}/local/tokens/generate")
        return cast(str, response["token"])

    @retry_on_auth_error
    async def activate_local_token(
        self,
        gateway_id: str,
        token: str,
        label: str,
        scope: str = "devmode",
    ) -> str:
        """Activate a local API token with the given label and scope.

        Returns the activation request ID.
        """
        response = await self._post(
            f"config/{gateway_id}/local/tokens",
            payload={"label": label, "token": token, "scope": scope},
        )
        return cast(str, response["requestId"])

    @retry_on_auth_error
    async def get_local_tokens(
        self, gateway_id: str, scope: str = "devmode"
    ) -> list[LocalToken]:
        """Get all active local API tokens for a gateway with a given scope."""
        response = await self._get(f"config/{gateway_id}/local/tokens/{scope}")
        return converter.structure(response, list[LocalToken])

    @retry_on_auth_error
    async def delete_local_token(self, gateway_id: str, uuid: str) -> None:
        """Delete a local API token by its UUID."""
        await self._delete(f"config/{gateway_id}/local/tokens/{uuid}")

    @retry_on_auth_error
    async def open_local_pairing(self, gateway_id: str) -> Any:
        """Put the gateway into local pairing mode (~180 seconds).

        During this window, new tokens can be registered directly on the
        gateway without requiring developer mode.

        Returns the raw response. Observed as an empty dict on success, but the
        shape under other conditions is not yet confirmed, so the response is
        passed through as-is.

        .. warning::
            Experimental (preview). This endpoint is not yet fully validated
            and its behaviour or signature may change in a future release.
        """
        return await self._post(f"config/{gateway_id}/local/openPairing")

    # -----------------------------------------------------------------------
    # Developer mode (cloud API)
    # -----------------------------------------------------------------------

    @retry_on_auth_error
    async def activate_developer_mode(self, gateway_id: str) -> None:
        """Activate developer mode for a gateway.

        Required for Somfy gateways before local API tokens can be used.
        """
        await self._post(f"setup/gateways/{gateway_id}/developerMode")

    @retry_on_auth_error
    async def get_developer_mode(self, gateway_id: str) -> DeveloperMode:
        """Get the developer mode status for a gateway."""
        response = await self._get(f"setup/gateways/{gateway_id}/developerMode")
        return converter.structure(response, DeveloperMode)

    @retry_on_auth_error
    async def deactivate_developer_mode(self, gateway_id: str) -> None:
        """Deactivate developer mode for a gateway.

        .. warning::
            Experimental (preview). This DELETE is not part of the bundled API
            spec and some gateways reject it (``NoSuchResourceError: Request
            method 'DELETE' not supported``). Behaviour may change in a future
            release.
        """
        await self._delete(f"setup/gateways/{gateway_id}/developerMode")

    @retry_on_connection_failure
    async def _get(self, path: str) -> Any:
        """Make a GET request to the OverKiz API."""
        await self._refresh_token_if_expired()

        async with self.session.get(
            f"{self._auth.endpoint}{path}",
            headers=await self._auth.auth_headers(path),
            ssl=self._ssl,
        ) as response:
            return await self._parse_response(response)

    @retry_on_connection_failure
    async def _post(
        self,
        path: str,
        payload: dict[str, Any] | None = None,
        data: dict[str, Any] | None = None,
    ) -> Any:
        """Make a POST request to the OverKiz API."""
        await self._refresh_token_if_expired()

        async with self.session.post(
            f"{self._auth.endpoint}{path}",
            data=data,
            json=payload,
            headers=await self._auth.auth_headers(path),
            ssl=self._ssl,
        ) as response:
            return await self._parse_response(response)

    @retry_on_connection_failure
    async def _put(self, path: str, payload: dict[str, Any] | None = None) -> Any:
        """Make a PUT request to the OverKiz API."""
        await self._refresh_token_if_expired()

        async with self.session.put(
            f"{self._auth.endpoint}{path}",
            json=payload,
            headers=await self._auth.auth_headers(path),
            ssl=self._ssl,
        ) as response:
            return await self._parse_response(response)

    @retry_on_connection_failure
    async def _delete(self, path: str) -> None:
        """Make a DELETE request to the OverKiz API."""
        await self._refresh_token_if_expired()

        async with self.session.delete(
            f"{self._auth.endpoint}{path}",
            headers=await self._auth.auth_headers(path),
            ssl=self._ssl,
        ) as response:
            await check_response(response)

    @staticmethod
    async def _parse_response(response: ClientResponse) -> Any:
        """Check response status and parse JSON body (returns None for 204)."""
        await check_response(response)
        if response.status == HTTPStatus.NO_CONTENT:
            return None
        return await response.json()

    async def _refresh_token_if_expired(self) -> None:
        """Check if token is expired and request a new one."""
        refreshed = await self._auth.refresh_if_needed()

        if refreshed and self.event_listener_id:
            await self.register_event_listener()

event_listener_id: str | None property

Return the current event listener ID (read-only).

__aenter__() -> Self async

Enter async context manager and return the client instance.

Source code in pyoverkiz/client.py

async def __aenter__(self) -> Self:
    """Enter async context manager and return the client instance."""
    return self

__aexit__(exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> None async

Exit the async context manager and close the client session.

Source code in pyoverkiz/client.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> None:
    """Exit the async context manager and close the client session."""
    await self.close()

__init__(*, server: ServerConfig | Server | str, credentials: Credentials, verify_ssl: bool = True, session: ClientSession | None = None, settings: OverkizClientSettings | None = None) -> None

Constructor.

Parameters:

Name	Type	Description	Default
server	ServerConfig | Server | str	ServerConfig, Server enum, or server key string.	required
credentials	Credentials	Credentials for authentication.	required
verify_ssl	bool	Enable SSL certificate verification.	True
session	ClientSession | None	Optional ClientSession.	None
settings	OverkizClientSettings | None	Behavioral settings for the client.	None

Source code in pyoverkiz/client.py

def __init__(
    self,
    *,
    server: ServerConfig | Server | str,
    credentials: Credentials,
    verify_ssl: bool = True,
    session: ClientSession | None = None,
    settings: OverkizClientSettings | None = None,
) -> None:
    """Constructor.

    Args:
        server: ServerConfig, Server enum, or server key string.
        credentials: Credentials for authentication.
        verify_ssl: Enable SSL certificate verification.
        session: Optional ClientSession.
        settings: Behavioral settings for the client.
    """
    self.server_config = self._normalize_server(server)

    self.setup: Setup | None = None
    self.devices: list[Device] = []
    self.gateways: list[Gateway] = []
    self._event_listener_id: str | None = None

    self.session = session or ClientSession(
        headers={"User-Agent": USER_AGENT}, timeout=DEFAULT_TIMEOUT
    )
    self._ssl = verify_ssl

    if self.server_config.api_type == APIType.LOCAL and verify_ssl:
        # Use the prebuilt SSL context with disabled strict validation for local API.
        self._ssl = SSL_CONTEXT_LOCAL_API

    self.settings = settings or OverkizClientSettings()

    if self.settings.action_queue:
        self.settings.action_queue.validate()
        self._action_queue = ActionQueue(
            executor=self._execute_action_group_direct,
            settings=self.settings.action_queue,
        )

    self._auth = build_auth_strategy(
        server_config=self.server_config,
        credentials=credentials,
        session=self.session,
        ssl_context=self._ssl,
    )

activate_developer_mode(gateway_id: str) -> None async

Activate developer mode for a gateway.

Required for Somfy gateways before local API tokens can be used.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def activate_developer_mode(self, gateway_id: str) -> None:
    """Activate developer mode for a gateway.

    Required for Somfy gateways before local API tokens can be used.
    """
    await self._post(f"setup/gateways/{gateway_id}/developerMode")

activate_local_token(gateway_id: str, token: str, label: str, scope: str = 'devmode') -> str async

Activate a local API token with the given label and scope.

Returns the activation request ID.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def activate_local_token(
    self,
    gateway_id: str,
    token: str,
    label: str,
    scope: str = "devmode",
) -> str:
    """Activate a local API token with the given label and scope.

    Returns the activation request ID.
    """
    response = await self._post(
        f"config/{gateway_id}/local/tokens",
        payload={"label": label, "token": token, "scope": scope},
    )
    return cast(str, response["requestId"])

cancel_execution(exec_id: str) -> None async

Cancel a running execution by its exec_id.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def cancel_execution(self, exec_id: str) -> None:
    """Cancel a running execution by its exec_id."""
    await self._delete(f"exec/current/setup/{exec_id}")

close() -> None async

Close the session.

Source code in pyoverkiz/client.py

async def close(self) -> None:
    """Close the session."""
    # Flush any pending actions in queue
    if self._action_queue:
        await self._action_queue.shutdown()

    if self.event_listener_id:
        await self.unregister_event_listener()

    await self._auth.close()
    await self.session.close()

deactivate_developer_mode(gateway_id: str) -> None async

Deactivate developer mode for a gateway.

.. warning:: Experimental (preview). This DELETE is not part of the bundled API spec and some gateways reject it (NoSuchResourceError: Request method 'DELETE' not supported). Behaviour may change in a future release.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def deactivate_developer_mode(self, gateway_id: str) -> None:
    """Deactivate developer mode for a gateway.

    .. warning::
        Experimental (preview). This DELETE is not part of the bundled API
        spec and some gateways reject it (``NoSuchResourceError: Request
        method 'DELETE' not supported``). Behaviour may change in a future
        release.
    """
    await self._delete(f"setup/gateways/{gateway_id}/developerMode")

delete_local_token(gateway_id: str, uuid: str) -> None async

Delete a local API token by its UUID.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def delete_local_token(self, gateway_id: str, uuid: str) -> None:
    """Delete a local API token by its UUID."""
    await self._delete(f"config/{gateway_id}/local/tokens/{uuid}")

discover_gateways() -> list[GatewayCandidate] async

Discover selectable gateways.

Raises:

Type	Description
UnsupportedOperationError	When the server does not support gateway selection.

Source code in pyoverkiz/client.py

async def discover_gateways(self) -> list[GatewayCandidate]:
    """Discover selectable gateways.

    Raises:
        UnsupportedOperationError: When the server does not support gateway selection.
    """
    if not isinstance(self._auth, SupportsGatewaySelection):
        raise UnsupportedOperationError(
            f"{self.server_config.name} does not support gateway selection."
        )
    return await self._auth.discover_gateways()

execute_action_group(*, actions: list[Action], mode: ExecutionMode | None = None, label: str | None = 'pyOverkiz') -> str async

Execute an ad-hoc action group built from the given actions.

An action group is a batch of device actions submitted as a single execution. Each Action targets one device and contains one or more Command instances (e.g. open, setClosure(50)). The gateway allows at most one action per device per action group.

When the action queue is enabled, actions are held for a short delay and merged with other actions submitted in the same window. Commands targeting the same device are combined into a single action. The method blocks until the batch executes and returns the resulting exec_id.

When the action queue is disabled, the action group is sent immediately.

Parameters:

Name	Type	Description	Default
actions	list[Action]	One or more actions to execute. Each action targets a single device and holds one or more commands.	required
mode	ExecutionMode | None	Optional execution mode (HIGH_PRIORITY, GEOLOCATED, or INTERNAL). Only supported by the Cloud API; the local API rejects requests that specify an execution mode (see https://github.com/Somfy-Developer/Somfy-TaHoma-Developer-Mode/issues/227).	None
label	str | None	Human-readable label for the execution.	'pyOverkiz'

Returns:

Type	Description
str	The exec_id identifying the execution on the server.

Source code in pyoverkiz/client.py

async def execute_action_group(
    self,
    *,
    actions: list[Action],
    mode: ExecutionMode | None = None,
    label: str | None = "pyOverkiz",
) -> str:
    """Execute an ad-hoc action group built from the given actions.

    An action group is a batch of device actions submitted as a single
    execution. Each ``Action`` targets one device and contains one or more
    ``Command`` instances (e.g. ``open``, ``setClosure(50)``). The gateway
    allows at most one action per device per action group.

    When the action queue is enabled, actions are held for a short delay
    and merged with other actions submitted in the same window. Commands
    targeting the same device are combined into a single action. The method
    blocks until the batch executes and returns the resulting exec_id.

    When the action queue is disabled, the action group is sent immediately.

    Args:
        actions: One or more actions to execute. Each action targets a
            single device and holds one or more commands.
        mode: Optional execution mode (``HIGH_PRIORITY``, ``GEOLOCATED``,
            or ``INTERNAL``). Only supported by the Cloud API; the local
            API rejects requests that specify an execution mode (see
            https://github.com/Somfy-Developer/Somfy-TaHoma-Developer-Mode/issues/227).
        label: Human-readable label for the execution.

    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
    return await self._execute_action_group_direct(actions, mode, label)

execute_persisted_action_group(oid: str) -> str async

Execute a server-side action group by its OID (see get_action_groups).

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def execute_persisted_action_group(self, oid: str) -> str:
    """Execute a server-side action group by its OID (see ``get_action_groups``)."""
    response = await self._post(f"exec/{oid}")
    return cast(str, response["execId"])

fetch_events() -> list[Event] async

Fetch new events from a registered event listener. Fetched events are removed.

from the listener buffer. Return an empty response if no event is available. Per-session rate-limit : 1 calls per 1 SECONDS period for this particular operation (polling).

Source code in pyoverkiz/client.py

@retry_on_concurrent_requests
@retry_on_auth_error
@retry_on_listener_error
async def fetch_events(self) -> list[Event]:
    """Fetch new events from a registered event listener. Fetched events are removed.

    from the listener buffer. Return an empty response if no event is available.
    Per-session rate-limit : 1 calls per 1 SECONDS period for this particular
    operation (polling).
    """
    response = await self._post(f"events/{self.event_listener_id}/fetch")
    return converter.structure(response, list[Event])

flush_action_queue() -> None async

Force flush all pending actions in the queue immediately.

If action queue is disabled, this method does nothing. If there are no pending actions, this method does nothing.

Source code in pyoverkiz/client.py

async def flush_action_queue(self) -> None:
    """Force flush all pending actions in the queue immediately.

    If action queue is disabled, this method does nothing.
    If there are no pending actions, this method does nothing.
    """
    if self._action_queue:
        await self._action_queue.flush()

generate_local_token(gateway_id: str) -> str async

Generate a new local API token for a gateway.

The token must be activated before it can be used.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def generate_local_token(self, gateway_id: str) -> str:
    """Generate a new local API token for a gateway.

    The token must be activated before it can be used.
    """
    response = await self._get(f"config/{gateway_id}/local/tokens/generate")
    return cast(str, response["token"])

get_action_groups() -> list[PersistedActionGroup] async

List action groups persisted on the server.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_action_groups(self) -> list[PersistedActionGroup]:
    """List action groups persisted on the server."""
    response = await self._get("actionGroups")
    return converter.structure(response, list[PersistedActionGroup])

get_api_version() -> str async

Get the API version (local only).

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_api_version(self) -> str:
    """Get the API version (local only)."""
    response = await self._get("apiVersion")

    return cast(str, response["protocolVersion"])

get_current_execution(exec_id: str) -> Execution | None async

Get a currently running execution by its exec_id.

Returns None if the execution does not exist.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_current_execution(self, exec_id: str) -> Execution | None:
    """Get a currently running execution by its exec_id.

    Returns None if the execution does not exist.
    """
    response = await self._get(f"exec/current/{exec_id}")
    if not response or not isinstance(response, dict):
        return None

    return converter.structure(response, Execution)

get_current_executions() -> list[Execution] async

Get all currently running executions.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_current_executions(self) -> list[Execution]:
    """Get all currently running executions."""
    response = await self._get("exec/current")
    return converter.structure(response, list[Execution])

get_developer_mode(gateway_id: str) -> DeveloperMode async

Get the developer mode status for a gateway.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_developer_mode(self, gateway_id: str) -> DeveloperMode:
    """Get the developer mode status for a gateway."""
    response = await self._get(f"setup/gateways/{gateway_id}/developerMode")
    return converter.structure(response, DeveloperMode)

get_device_alternative_controllables(device_url: str) -> list[str] async

Get all alternative controllable names for a device.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_device_alternative_controllables(self, device_url: str) -> list[str]:
    """Get all alternative controllable names for a device."""
    return await self._get(
        f"setup/devices/{urllib.parse.quote_plus(device_url)}/alternativeControllables"
    )

get_device_controllable_definition(device_url: str) -> Definition | None async

Get the controllable definition for a specific device.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_device_controllable_definition(
    self, device_url: str
) -> Definition | None:
    """Get the controllable definition for a specific device."""
    response = await self._get(
        f"setup/devices/{urllib.parse.quote_plus(device_url)}/controllable"
    )
    if response is None:
        return None
    return converter.structure(response, Definition)

get_device_controllables() -> dict[str, list[str]] async

Get all device URLs grouped by their controllable name.

Returns a dict mapping controllable names to lists of device URLs that use that controllable.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_device_controllables(self) -> dict[str, list[str]]:
    """Get all device URLs grouped by their controllable name.

    Returns a dict mapping controllable names to lists of device URLs
    that use that controllable.
    """
    return await self._get("setup/devices/controllables")

get_device_definition(device_url: str) -> Definition | None async

Retrieve a particular setup device definition.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_device_definition(self, device_url: str) -> Definition | None:
    """Retrieve a particular setup device definition."""
    response: dict = await self._get(
        f"setup/devices/{urllib.parse.quote_plus(device_url)}"
    )

    raw = response.get("definition")
    if raw is None:
        return None

    return converter.structure(raw, Definition)

get_device_firmware_status(device_url: str) -> FirmwareStatus | None async

Check if a device's firmware is up to date.

Returns None if the device does not support firmware status checks.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_device_firmware_status(
    self, device_url: str
) -> FirmwareStatus | None:
    """Check if a device's firmware is up to date.

    Returns None if the device does not support firmware status checks.
    """
    try:
        response = await self._get(
            f"setup/devices/{urllib.parse.quote_plus(device_url)}/firmwareUpToDate"
        )
    except UnsupportedOperationError:
        return None
    return converter.structure(response, FirmwareStatus)

get_device_firmware_update_capability(device_url: str) -> bool async

Check if a device supports firmware updates.

Returns False if the device does not support this query.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_device_firmware_update_capability(self, device_url: str) -> bool:
    """Check if a device supports firmware updates.

    Returns False if the device does not support this query.
    """
    try:
        response = await self._get(
            f"setup/devices/{urllib.parse.quote_plus(device_url)}/firmwareUpdateCapability"
        )
    except UnsupportedOperationError:
        return False
    return cast(bool, response["supportsFirmwareUpdate"])

get_device_manufacturer_references(device_url: str) -> list[DeviceManufacturerReference] async

Get all manufacturer references for a device.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_device_manufacturer_references(
    self, device_url: str
) -> list[DeviceManufacturerReference]:
    """Get all manufacturer references for a device."""
    response = await self._get(
        f"setup/devices/{urllib.parse.quote_plus(device_url)}/manufacturerReferences"
    )
    return converter.structure(response, list[DeviceManufacturerReference])

get_devices(refresh: bool = False) -> list[Device] async

List devices.

Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load).

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_devices(self, refresh: bool = False) -> list[Device]:
    """List devices.

    Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load).
    """
    if self.devices and not refresh:
        return self.devices

    response = await self._get("setup/devices")
    devices = converter.structure(response, list[Device])

    # Cache response
    self.devices = devices
    if self.setup:
        self.setup.devices = devices

    return devices

get_devices_not_up_to_date() -> list[Device] async

Get all devices whose firmware is not up to date.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_devices_not_up_to_date(self) -> list[Device]:
    """Get all devices whose firmware is not up to date."""
    response = await self._get("setup/devices/notUpToDate")
    return converter.structure(response, list[Device])

get_diagnostic_data(mask_sensitive_data: bool = True) -> dict[str, Any] async

Get diagnostic data for the connected user setup.

-> gateways data (serial number, activation state, ...): <gateways/gateway>
-> setup location: <location>
-> house places (rooms and floors): <place>
-> setup devices: <devices>
-> action groups: <actionGroups>

By default, this data is masked to not return confidential or PII data. Set mask_sensitive_data to False to return the raw payloads.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_diagnostic_data(
    self, mask_sensitive_data: bool = True
) -> dict[str, Any]:
    """Get diagnostic data for the connected user setup.

        -> gateways data (serial number, activation state, ...): <gateways/gateway>
        -> setup location: <location>
        -> house places (rooms and floors): <place>
        -> setup devices: <devices>
        -> action groups: <actionGroups>

    By default, this data is masked to not return confidential or PII data.
    Set `mask_sensitive_data` to `False` to return the raw payloads.
    """
    setup, action_groups = await asyncio.gather(
        self._get("setup"),
        self._get("actionGroups"),
    )

    if mask_sensitive_data:
        setup = obfuscate_sensitive_data(setup)
        action_groups = obfuscate_sensitive_data(action_groups)

    return {
        "setup": setup,
        "action_groups": action_groups,
    }

get_execution_history() -> list[HistoryExecution] async

List past executions and their outcomes.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_execution_history(self) -> list[HistoryExecution]:
    """List past executions and their outcomes."""
    response = await self._get("history/executions")
    return converter.structure(response, list[HistoryExecution])

get_gateways(refresh: bool = False) -> list[Gateway] async

Get every gateways of a connected user setup.

Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load).

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_gateways(self, refresh: bool = False) -> list[Gateway]:
    """Get every gateways of a connected user setup.

    Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load).
    """
    if self.gateways and not refresh:
        return self.gateways

    response = await self._get("setup/gateways")
    gateways = converter.structure(response, list[Gateway])

    # Cache response
    self.gateways = gateways
    if self.setup:
        self.setup.gateways = gateways

    return gateways

get_local_tokens(gateway_id: str, scope: str = 'devmode') -> list[LocalToken] async

Get all active local API tokens for a gateway with a given scope.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_local_tokens(
    self, gateway_id: str, scope: str = "devmode"
) -> list[LocalToken]:
    """Get all active local API tokens for a gateway with a given scope."""
    response = await self._get(f"config/{gateway_id}/local/tokens/{scope}")
    return converter.structure(response, list[LocalToken])

get_pending_actions_count() -> int

Get the approximate number of actions currently waiting in the queue.

Returns 0 if action queue is disabled. This is a best-effort snapshot and may be stale if other coroutines modify the queue concurrently.

Source code in pyoverkiz/client.py

def get_pending_actions_count(self) -> int:
    """Get the approximate number of actions currently waiting in the queue.

    Returns 0 if action queue is disabled. This is a best-effort snapshot
    and may be stale if other coroutines modify the queue concurrently.
    """
    if self._action_queue:
        return self._action_queue.get_pending_count()
    return 0

get_places() -> Place async

Get the hierarchical structure of places (house, rooms, areas, zones).

The Place model represents a hierarchical organization where the root place is typically the house/property, and sub_places contains nested child places (floors, rooms, areas). This structure can be recursively navigated to build a complete map of all locations in the setup. Each place has: - label: Human-readable name for the place - type: Numeric identifier for the place type - sub_places: List of nested places within this location

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_places(self) -> Place:
    """Get the hierarchical structure of places (house, rooms, areas, zones).

    The Place model represents a hierarchical organization where the root place is
    typically the house/property, and `sub_places` contains nested child places
    (floors, rooms, areas). This structure can be recursively navigated to build
    a complete map of all locations in the setup. Each place has:
    - `label`: Human-readable name for the place
    - `type`: Numeric identifier for the place type
    - `sub_places`: List of nested places within this location
    """
    response = await self._get("setup/places")
    return converter.structure(response, Place)

get_reference_controllable(controllable_name: str) -> dict[str, Any] async

Get a controllable definition.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_reference_controllable(
    self, controllable_name: str
) -> dict[str, Any]:
    """Get a controllable definition."""
    return await self._get(
        f"reference/controllable/{urllib.parse.quote_plus(controllable_name)}"
    )

get_reference_controllable_types() -> list[dict[str, Any]] async

Get details about all supported controllable types.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_reference_controllable_types(self) -> list[dict[str, Any]]:
    """Get details about all supported controllable types."""
    return await self._get("reference/controllableTypes")

get_reference_protocol_types() -> list[ProtocolType] async

Get details about supported protocol types on that server instance.

Returns a list of protocol type definitions, each containing: - id: Numeric protocol identifier - prefix: URL prefix used in device addresses - name: Internal protocol name - label: Human-readable protocol label

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_reference_protocol_types(self) -> list[ProtocolType]:
    """Get details about supported protocol types on that server instance.

    Returns a list of protocol type definitions, each containing:
    - id: Numeric protocol identifier
    - prefix: URL prefix used in device addresses
    - name: Internal protocol name
    - label: Human-readable protocol label
    """
    response = await self._get("reference/protocolTypes")
    return converter.structure(response, list[ProtocolType])

get_reference_timezones() -> list[dict[str, Any]] async

Get timezones list.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_reference_timezones(self) -> list[dict[str, Any]]:
    """Get timezones list."""
    return await self._get("reference/timezones")

get_reference_ui_classes() -> list[str] async

Get a list of all defined UI classes.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_reference_ui_classes(self) -> list[str]:
    """Get a list of all defined UI classes."""
    return await self._get("reference/ui/classes")

get_reference_ui_classifiers() -> list[str] async

Get a list of all defined UI classifiers.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_reference_ui_classifiers(self) -> list[str]:
    """Get a list of all defined UI classifiers."""
    return await self._get("reference/ui/classifiers")

get_reference_ui_profile(profile_name: str) -> UIProfileDefinition async

Get a description of a given UI profile (or form-factor variant).

Returns a profile definition containing: - name: Profile name - commands: Available commands with parameters and descriptions - states: Available states with value types and descriptions - form_factor: Whether profile is tied to a specific physical device type

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_reference_ui_profile(self, profile_name: str) -> UIProfileDefinition:
    """Get a description of a given UI profile (or form-factor variant).

    Returns a profile definition containing:
    - name: Profile name
    - commands: Available commands with parameters and descriptions
    - states: Available states with value types and descriptions
    - form_factor: Whether profile is tied to a specific physical device type
    """
    response = await self._get(
        f"reference/ui/profile/{urllib.parse.quote_plus(profile_name)}"
    )
    return converter.structure(response, UIProfileDefinition)

get_reference_ui_profile_names() -> list[str] async

Get a list of all defined UI profiles (and form-factor variants).

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_reference_ui_profile_names(self) -> list[str]:
    """Get a list of all defined UI profiles (and form-factor variants)."""
    return await self._get("reference/ui/profileNames")

get_reference_ui_widgets() -> list[str] async

Get a list of all defined UI widgets.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_reference_ui_widgets(self) -> list[str]:
    """Get a list of all defined UI widgets."""
    return await self._get("reference/ui/widgets")

get_setup(refresh: bool = False) -> Setup async

Get all data about the connected user setup.

-> gateways data (serial number, activation state, ...): <gateways/gateway>
-> setup location: <location>
-> house places (rooms and floors): <place>
-> setup devices: <devices>.

A gateway may be in different modes (mode) regarding to the activated functions (functions). A house may be composed of several floors and rooms. The house, floors and rooms are viewed as a place. Devices in the house are grouped by type called uiClass. Each device has an associated widget. The widget is used to control or to know the device state, whatever the device protocol (controllable): IO, RTS, X10, ... . A device can be either an actuator (type=1) or a sensor (type=2). Data of one or several devices can be also get by setting the device(s) url as request parameter.

Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load)

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_setup(self, refresh: bool = False) -> Setup:
    """Get all data about the connected user setup.

        -> gateways data (serial number, activation state, ...): <gateways/gateway>
        -> setup location: <location>
        -> house places (rooms and floors): <place>
        -> setup devices: <devices>.

    A gateway may be in different modes (mode) regarding to the activated functions (functions).
    A house may be composed of several floors and rooms. The house, floors and rooms are viewed as a place.
    Devices in the house are grouped by type called uiClass. Each device has an associated widget.
    The widget is used to control or to know the device state, whatever the device protocol (controllable): IO, RTS, X10, ... .
    A device can be either an actuator (type=1) or a sensor (type=2).
    Data of one or several devices can be also get by setting the device(s) url as request parameter.

    Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load)
    """
    if self.setup and not refresh:
        return self.setup

    response = await self._get("setup")

    setup = converter.structure(response, Setup)

    # Cache response
    self.setup = setup
    self.gateways = setup.gateways
    self.devices = setup.devices

    return setup

get_setup_option(option: str) -> Option | None async

This operation returns the selected subscribed option of a given setup.

For example developerMode-{gateway_id} to understand if developer mode is on.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_setup_option(self, option: str) -> Option | None:
    """This operation returns the selected subscribed option of a given setup.

    For example `developerMode-{gateway_id}` to understand if developer mode is on.
    """
    response = await self._get(f"setup/options/{option}")

    if response:
        return converter.structure(response, Option)

    return None

get_setup_option_parameter(option: str, parameter: str) -> OptionParameter | None async

This operation returns the selected parameters of a given setup and option.

For example developerMode-{gateway_id} and gatewayId to understand if developer mode is on.

If the option is not available, an OverkizError will be thrown. If the parameter is not available you will receive None.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_setup_option_parameter(
    self, option: str, parameter: str
) -> OptionParameter | None:
    """This operation returns the selected parameters of a given setup and option.

    For example `developerMode-{gateway_id}` and `gatewayId` to understand if developer mode is on.

    If the option is not available, an OverkizError will be thrown.
    If the parameter is not available you will receive None.
    """
    response = await self._get(f"setup/options/{option}/{parameter}")

    if response:
        return converter.structure(response, OptionParameter)

    return None

get_setup_options() -> list[Option] async

This operation returns all subscribed options of a given setup.

Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load) Access scope : Full enduser API access (enduser/*).

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_setup_options(self) -> list[Option]:
    """This operation returns all subscribed options of a given setup.

    Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load)
    Access scope : Full enduser API access (enduser/*).
    """
    response = await self._get("setup/options")
    return converter.structure(response, list[Option])

get_state(device_url: str) -> list[State] async

Retrieve states of requested device.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def get_state(self, device_url: str) -> list[State]:
    """Retrieve states of requested device."""
    response = await self._get(
        f"setup/devices/{urllib.parse.quote_plus(device_url)}/states"
    )
    return converter.structure(response, list[State])

login(register_event_listener: bool = True) -> None async

Authenticate and create an API session allowing access to the other operations.

Caller must provide one of [userId+userPassword, userId+ssoToken, accessToken, jwt].

Raises:

Type	Description
BadCredentialsError	When the provided credentials are invalid.
TooManyAttemptsBannedError	When too many failed login attempts have been made.
TooManyRequestsError	When the API rate limit has been exceeded.

Source code in pyoverkiz/client.py

async def login(
    self,
    register_event_listener: bool = True,
) -> None:
    """Authenticate and create an API session allowing access to the other operations.

    Caller must provide one of [userId+userPassword, userId+ssoToken, accessToken, jwt].

    Raises:
        BadCredentialsError: When the provided credentials are invalid.
        TooManyAttemptsBannedError: When too many failed login attempts have been made.
        TooManyRequestsError: When the API rate limit has been exceeded.
    """
    await self._auth.login()

    if self.server_config.api_type == APIType.LOCAL:
        if register_event_listener:
            await self.register_event_listener()
        else:
            # Validate local API token by calling a simple endpoint
            await self.get_gateways()

        return

    if register_event_listener:
        await self.register_event_listener()

    return

open_local_pairing(gateway_id: str) -> Any async

Put the gateway into local pairing mode (~180 seconds).

During this window, new tokens can be registered directly on the gateway without requiring developer mode.

Returns the raw response. Observed as an empty dict on success, but the shape under other conditions is not yet confirmed, so the response is passed through as-is.

.. warning:: Experimental (preview). This endpoint is not yet fully validated and its behaviour or signature may change in a future release.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def open_local_pairing(self, gateway_id: str) -> Any:
    """Put the gateway into local pairing mode (~180 seconds).

    During this window, new tokens can be registered directly on the
    gateway without requiring developer mode.

    Returns the raw response. Observed as an empty dict on success, but the
    shape under other conditions is not yet confirmed, so the response is
    passed through as-is.

    .. warning::
        Experimental (preview). This endpoint is not yet fully validated
        and its behaviour or signature may change in a future release.
    """
    return await self._post(f"config/{gateway_id}/local/openPairing")

refresh_device_states(device_url: str) -> None async

Ask the box to refresh all states of the given device for protocols supporting that operation.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def refresh_device_states(self, device_url: str) -> None:
    """Ask the box to refresh all states of the given device for protocols supporting that operation."""
    await self._post(
        f"setup/devices/{urllib.parse.quote_plus(device_url)}/states/refresh"
    )

refresh_states() -> None async

Ask the box to refresh all devices states for protocols supporting that operation.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def refresh_states(self) -> None:
    """Ask the box to refresh all devices states for protocols supporting that operation."""
    await self._post("setup/devices/states/refresh")

register_event_listener() -> str async

Register a new setup event listener on the current session and return a new.

listener id. Only one listener may be registered on a given session. Registering an new listener will invalidate the previous one if any. Note that registering an event listener drastically reduces the session timeout : listening sessions are expected to call the /events/{listenerId}/fetch API on a regular basis.

Source code in pyoverkiz/client.py

@retry_on_concurrent_requests
async def register_event_listener(self) -> str:
    """Register a new setup event listener on the current session and return a new.

    listener id.
    Only one listener may be registered on a given session.
    Registering an new listener will invalidate the previous one if any.
    Note that registering an event listener drastically reduces the session
    timeout : listening sessions are expected to call the /events/{listenerId}/fetch
    API on a regular basis.
    """
    response = await self._post("events/register")
    listener_id = cast(str, response.get("id"))
    self._event_listener_id = listener_id

    return listener_id

schedule_persisted_action_group(oid: str, timestamp: int) -> str async

Schedule a server-side action group for execution at the given timestamp.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def schedule_persisted_action_group(self, oid: str, timestamp: int) -> str:
    """Schedule a server-side action group for execution at the given timestamp."""
    response = await self._post(f"exec/schedule/{oid}/{timestamp}")
    return cast(str, response["triggerId"])

search_reference_devices(payload: dict[str, Any]) -> DeviceSearchResult async

Search reference device models using a POST payload.

The payload can filter by protocol types, UI classes, widgets, classifiers, controllable types, or free-text criteria. Additional flags control whether commands, states, attributes and manufacturer references are included.

Example payload

{ "protocolTypes": ["IO"], "withCommands": True, "withStates": True, "withAttributes": True, "withManufacturerReferences": True, }

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def search_reference_devices(
    self, payload: dict[str, Any]
) -> DeviceSearchResult:
    """Search reference device models using a POST payload.

    The payload can filter by protocol types, UI classes, widgets, classifiers,
    controllable types, or free-text criteria. Additional flags control whether
    commands, states, attributes and manufacturer references are included.

    Example payload:
        {
            "protocolTypes": ["IO"],
            "withCommands": True,
            "withStates": True,
            "withAttributes": True,
            "withManufacturerReferences": True,
        }
    """
    response = await self._post("reference/devices/search", payload)
    return converter.structure(response, DeviceSearchResult)

select_gateway(gateway_id: str) -> None

Select the gateway to scope requests to.

Raises:

Type	Description
UnsupportedOperationError	When the server does not support gateway selection.

Source code in pyoverkiz/client.py

def select_gateway(self, gateway_id: str) -> None:
    """Select the gateway to scope requests to.

    Raises:
        UnsupportedOperationError: When the server does not support gateway selection.
    """
    if not isinstance(self._auth, SupportsGatewaySelection):
        raise UnsupportedOperationError(
            f"{self.server_config.name} does not support gateway selection."
        )
    self._auth.select_gateway(gateway_id)

unregister_event_listener() -> None async

Unregister an event listener.

API response status is always 200, even on unknown listener ids.

Source code in pyoverkiz/client.py

async def unregister_event_listener(self) -> None:
    """Unregister an event listener.

    API response status is always 200, even on unknown listener ids.
    """
    await self._post(f"events/{self._event_listener_id}/unregister")
    self._event_listener_id = None

update_all_device_firmwares() -> None async

Update firmware for all devices that are not up to date.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def update_all_device_firmwares(self) -> None:
    """Update firmware for all devices that are not up to date."""
    await self._put("setup/devices/updateFirmwares")

update_device_firmware(device_url: str) -> None async

Update a device's firmware to the next available version.

Source code in pyoverkiz/client.py

@retry_on_auth_error
async def update_device_firmware(self, device_url: str) -> None:
    """Update a device's firmware to the next available version."""
    await self._put(
        f"setup/devices/{urllib.parse.quote_plus(device_url)}/updateFirmware"
    )

Models representing Overkiz API payloads and convenient accessors.

Action

An action consists of multiple commands related to a single device, identified by its device URL.

to_payload() -> dict[str, object]

Return a JSON-serializable payload for this action (snake_case).

The final camelCase conversion is handled by the client.

ActionGroup

An action group is composed of one or more actions.

Each action is related to a single setup device (designated by its device URL) and is composed of one or more commands to be executed on that device.

Command

Represents an OverKiz Command.

to_payload() -> 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.

CommandDefinition

Metadata for a single command definition (name and parameter count).

CommandDefinitions

Bases: Mapping[str, CommandDefinition]

Container for command definitions implementing Mapping[str, CommandDefinition].

__contains__(name: object) -> bool

Return True if a command with name exists.

__getitem__(command: str) -> CommandDefinition

Return the command definition or raise KeyError if missing.

__init__(commands: list[CommandDefinition] | None = None) -> None

Build the inner list and index from CommandDefinition objects.

__iter__() -> Iterator[str]

Iterate over command names.

__len__() -> int

Return number of command definitions.

first(commands: list[str | OverkizCommand]) -> str | None

Return the first command name that exists in this definition, or None.

has_any(commands: list[str | OverkizCommand]) -> bool

Return True if any of the given commands exist in this definition.

CommandParameter

Command parameter definition.

CommandPrototype

Command prototype defining parameters.

Connectivity

Connectivity metadata for a gateway update box.

DataProperty

Data property with qualified name and value.

Definition

Definition of device capabilities: command definitions, state definitions and UI hints.

DeveloperMode

Developer mode status for a gateway.

Device

Representation of a device in the setup including parsed fields and states.

__attrs_post_init__() -> None

Resolve computed fields from device URL and definition fallbacks.

first_command(commands: list[str | OverkizCommand]) -> str | None

Return first supported command name from list, or None.

get_command_definition(command: str | OverkizCommand) -> CommandDefinition | None

Return the CommandDefinition for a command, or None if unavailable.

supports_any_command(commands: list[str | OverkizCommand]) -> bool

Check if device supports any of the commands.

supports_command(command: str | OverkizCommand) -> bool

Check if device supports a command.

DeviceAttributeDefinition

Attribute definition from the device catalog.

DeviceAvailableEvent

Bases: DeviceEvent

A device became available.

DeviceCommandDefinition

Full command definition from the device catalog.

DeviceCreatedEvent

Bases: _DeviceMetadataEvent

A device was created.

DeviceDisabledEvent

Bases: DeviceEvent

A device was disabled.

DeviceEvent

Bases: Event

Any event about a device; device_url identifies it (required).

Narrow to this to handle all device events uniformly, or to a leaf subtype (e.g. DeviceStateChangedEvent) for that event's full payload.

DeviceIdentifier

Parsed components from a device URL.

is_sub_device: bool property

Return True if this identifier represents a sub-device (subsystem_id > 1).

__attrs_post_init__() -> None

Compute base_device_url from protocol, gateway_id and device_address.

from_device_url(device_url: str) -> DeviceIdentifier classmethod

Parse a device URL into its structured identifier components.

DeviceManufacturerReference

Manufacturer reference for a specific device (setup endpoint).

DeviceRemovedEvent

Bases: DeviceEvent

A device was removed.

DeviceSearchResult

Result from POST /reference/devices/search.

DeviceStateChangedEvent

Bases: DeviceEvent

One or more states of a device changed (high-level).

DeviceStateDefinition

Full state definition from the device catalog.

DeviceTypeDefinition

Complete device type definition from the reference devices search.

DeviceUnavailableEvent

Bases: DeviceEvent

A device became unavailable.

DeviceUpdatedEvent

Bases: _DeviceMetadataEvent

A device was updated.

Event

Base Overkiz event; structured into a subtype by name (see converter).

EventState

Bases: State

State variant used when parsing event payloads (casts string values).

__attrs_post_init__() -> None

Cast string values based on declared data type.

Execution

Execution occurrence with owner, state and action group metadata.

ExecutionEvent

Bases: Event

Any event about an execution; exec_id identifies it (required).

Narrow to this to handle all execution events uniformly, or to a leaf subtype (e.g. ExecutionStateChangedEvent) for that event's full payload.

ExecutionRegisteredEvent

Bases: ExecutionEvent

A new execution was registered.

ExecutionStateChangedEvent

Bases: ExecutionEvent

An execution state has changed; new_state/old_state are required.

FailureEvent

Bases: Event

Any *FailedEvent: failure reason plus the operation's scope id.

By design every *FailedEvent name structures into this one type rather than a per-name class: consumers branch on "did it fail, and why" (failure_type), not on which of the ~30 operations failed. The specific operation is identified by event.name.

gateway_id / device_url / protocol_type cover the documented failure payloads.

Feature

Feature flags exposed by a setup or gateway.

FirmwareStatus

Firmware status of a device.

Gateway

Representation of a gateway, including connectivity and partner info.

id: str property

Alias for gateway_id.

GatewayActivatedEvent

Bases: GatewayEvent

A gateway was activated.

GatewayActiveProtocolsChangedEvent

Bases: GatewayEvent

A gateway's active protocols changed.

GatewayAliveEvent

Bases: GatewayEvent

A gateway became reachable.

GatewayAssociatedEvent

Bases: GatewayEvent

A gateway was associated with the setup.

GatewayAttachedEvent

Bases: GatewayEvent

A gateway was attached.

GatewayBootEvent

Bases: GatewayEvent

A gateway booted.

GatewayDeactivatedEvent

Bases: GatewayEvent

A gateway was deactivated.

GatewayDetachedEvent

Bases: GatewayEvent

A gateway was detached.

GatewayDissociatedEvent

Bases: GatewayEvent

A gateway was dissociated from the setup.

GatewayDownEvent

Bases: GatewayEvent

A gateway became unreachable.

GatewayDownOptionsChangedEvent

Bases: GatewayEvent

A gateway's down-detection timeout changed.

GatewayEvent

Bases: Event

Any gateway event; gateway_id identifies it (required).

Narrow to this to handle all gateway events uniformly, or to a leaf subtype (e.g. GatewayDownEvent) for that event's full payload.

GatewayFirmwareUpdateCompletedEvent

Bases: GatewayEvent

A gateway firmware update completed.

GatewayFirmwareUpdatedEvent

Bases: GatewayEvent

A gateway's firmware was updated.

GatewayFunctionChangedEvent

Bases: GatewayEvent

A gateway function was enabled or disabled.

GatewayMigratedEvent

Bases: GatewayEvent

A gateway was migrated.

GatewayModeChangedEvent

Bases: GatewayEvent

A gateway's mode changed.

GatewayPlaceUpdatedEvent

Bases: GatewayEvent

A gateway's place was updated.

GatewayProtocolDownEvent

Bases: GatewayEvent

A gateway protocol became unavailable.

GatewayProtocolReadyEvent

Bases: GatewayEvent

A gateway protocol became available.

GatewaySynchronizationEndedEvent

Bases: GatewayEvent

A gateway synchronization ended.

GatewaySynchronizationStartedEvent

Bases: GatewayEvent

A gateway synchronization started.

GatewayTimeReliabilityChangedEvent

Bases: GatewayEvent

A gateway's time reliability changed.

HistoryExecution

A recorded execution entry containing details and its list of commands.

HistoryExecutionCommand

A command within a recorded historical execution, including its status and parameters.

LocalToken

A local API token for direct gateway access.

Location

Geographical and address metadata for a Setup.

ManufacturerReference

Manufacturer reference from the device catalog.

ManufacturerReferenceTag

Tag within a manufacturer reference.

Option

A subscribed option for a setup including parameters.

OptionParameter

Key/value pair representing option parameter.

Partner

Partner details for a gateway or service provider.

PersistedActionGroup

Bases: ActionGroup

A server-persisted action group returned by the /actionGroups endpoint.

id: str property

Alias for oid.

Place

Hierarchical representation of a location (house, room, area) in a setup.

id: str property

Alias for oid.

ProtocolType

Protocol type definition from the reference API.

ServerConfig

Connection target details for an Overkiz-compatible server.

Setup

Representation of a complete setup returned by the Overkiz API.

State

A single device state with typed accessors for its value.

value_as_bool: bool | None property

Return the boolean value or raise on type mismatch.

value_as_dict: dict[str, Any] | None property

Return the dict value or raise if state is not a JSON object.

value_as_float: float | None property

Return the float value, allow int->float conversion; raise on type mismatch.

value_as_int: int | None property

Return the integer value or None if not set; raise on type mismatch.

value_as_list: list[Any] | None property

Return the list value or raise if state is not a JSON array.

value_as_str: str | None property

Return the string value or raise on type mismatch.

StateDefinition

Definition metadata for a state (qualified name, type and possible values).

__attrs_post_init__() -> None

Resolve qualified_name from either name or qualified_name.

StateDefinitions

Bases: Mapping[str, StateDefinition]

Container for state definitions implementing Mapping[str, StateDefinition].

__contains__(name: object) -> bool

Return True if a state definition with name exists.

__getitem__(name: str) -> StateDefinition

Return the state definition or raise KeyError if missing.

__init__(definitions: list[StateDefinition] | None = None) -> None

Build the inner list and index from StateDefinition objects.

__iter__() -> Iterator[str]

Iterate over qualified names.

__len__() -> int

Return number of state definitions.

first(names: list[str | OverkizState]) -> StateDefinition | None

Return the first StateDefinition whose qualified_name matches, or None.

has_any(names: list[str | OverkizState]) -> bool

Return True if any of the given state definitions exist.

StatePrototype

State prototype defining value constraints.

States

Bases: Mapping[str, State]

Container of State objects implementing Mapping[str, State].

__contains__(name: object) -> bool

Return True if a state with the given name exists in the container.

__getitem__(name: str) -> State

Return the State with the given name or raise KeyError if missing.

__init__(states: list[State] | None = None) -> None

Create a States container from a list of State objects or empty.

__iter__() -> Iterator[str]

Return an iterator over state names.

__len__() -> int

Return number of states in the container.

__setitem__(name: str, state: State) -> None

Set or append a State identified by name.

first(names: list[StateName]) -> State | None

Return the first State that exists and has a non-None value, or None.

first_value(names: list[StateName]) -> StateType

Return the value of the first State with a non-None value, or None.

get_value(name: StateName) -> StateType

Return the value of a state by name, or None if missing.

has_any_value(names: list[StateName]) -> bool

Return True if any of the given states exist with a non-None value.

has_value(name: StateName) -> bool

Return True if the state exists and has a non-None value.

For a pure existence check (ignoring the value), use name in states.

UIProfileCommand

UI profile command definition.

UIProfileDefinition

UI profile definition from the reference API.

Describes device capabilities through available commands and states.

UIProfileState

UI profile state definition.

ValuePrototype

Value prototype defining parameter/state value constraints.

Zone

A Zone groups related devices inside a place.

ZoneCreatedEvent

Bases: _ZoneMutationEvent

A zone was created.

ZoneDeletedEvent

Bases: ZoneEvent

A zone was deleted.

ZoneEvent

Bases: Event

Any zone event; zone_oid identifies it (required).

Narrow to this to handle all zone events uniformly, or to a leaf subtype (e.g. ZoneCreatedEvent) for that event's full payload.

ZoneItem

An item inside a Zone (device reference).

ZoneUpdatedEvent

Bases: _ZoneMutationEvent

A zone was updated.

Convenience re-exports for the enums package.

APIType

Bases: StrEnum

API types supported by the client (cloud or local).

DataType

Bases: IntEnum

Data type constants used to interpret State values.

EventName

Bases: UnknownEnumMixin, StrEnum

Enumeration of event names emitted by Overkiz.

ExecutionMode

Bases: StrEnum

Execution mode flags (e.g., high priority or geolocated).

ExecutionState

Bases: UnknownEnumMixin, StrEnum

Execution lifecycle states.

ExecutionSubType

Bases: UnknownEnumMixin, StrEnum

Subtypes for execution reasons or sources.

ExecutionType

Bases: UnknownEnumMixin, StrEnum

High-level execution categories returned by the API.

FailureType

Bases: UnknownEnumMixin, IntEnum

Failure type codes returned by the API.

GatewaySubType

Bases: UnknownEnumMixin, IntEnum

Sub-type enumeration for gateways to identify specific models/variants.

beautify_name: str property

Return a human friendly name for the gateway sub-type.

GatewayType

Bases: UnknownEnumMixin, IntEnum

Enumeration of known gateway types returned by Overkiz.

beautify_name: str property

Return a human friendly name for the gateway type.

MeasuredValueType

Bases: StrEnum

See https://github.com/Somfy-Developer/Somfy-TaHoma-Developer-Mode/issues/68#issuecomment-1281474879.

OverkizAttribute

Bases: StrEnum

Device attributes used by Overkiz.

OverkizCommand

Bases: StrEnum

Device commands used by Overkiz.

OverkizCommandParam

Bases: StrEnum

Parameter used by Overkiz commands and/or states.

OverkizState

Bases: StrEnum

Device states used by Overkiz.

ProductType

Bases: IntEnum

Product type constants used to interpret Device types.

Protocol

Bases: UnknownEnumMixin, StrEnum

Protocol used by Overkiz.

Values have been retrieved from /reference/protocolTypes

Server

Bases: StrEnum

Known named Overkiz server endpoints.

StateDefinitionType

Bases: UnknownEnumMixin, StrEnum

Type of a state definition describing its value semantics.

UIClass

Bases: UnknownEnumMixin, StrEnum

Enumeration of UI classes used to describe device categories and behaviors.

UIClassifier

Bases: UnknownEnumMixin, StrEnum

Enumeration of UI classifiers used to categorize device types.

UIProfile

Bases: UnknownEnumMixin, StrEnum

UI Profiles define device capabilities through commands and states.

Each profile describes what a device can do (commands) and what information it provides (states). Form factor indicates if the profile is tied to a specific physical device type.

UIWidget

Bases: UnknownEnumMixin, StrEnum

Enumeration of UI widgets used by Overkiz for device presentation.

UpdateBoxStatus

Bases: StrEnum

Status of the gateway update box indicating its updateability.

UpdateCriticityLevel

Bases: UnknownEnumMixin, StrEnum

Criticity level of an available gateway update.

Errors defined for Overkiz API and its integrations.

AccessDeniedToGatewayError

Bases: ResourceAccessDeniedError

Raised when access is denied to the gateway.

This often happens when the user is not the owner of the gateway.

ActionGroupSetupNotFoundError

Bases: BaseOverkizError

Raised when an action group setup cannot be determined for a gateway.

ApplicationNotAllowedError

Bases: ResourceAccessDeniedError

Raised when the setup cannot be accessed through the application.

BadCredentialsError

Bases: BaseOverkizError

Raised when invalid credentials are provided.

BaseOverkizError

Bases: Exception

Base error for Overkiz errors.

BrandtBadCredentialsError

Bases: BadCredentialsError

Raised when the Brandt middleware rejects the username/password.

BrandtServiceError

Bases: BaseOverkizError

Raised when the Brandt middleware returns an unexpected response.

CozyTouchBadCredentialsError

Bases: BadCredentialsError

Raised when invalid credentials are provided to CozyTouch authentication API.

CozyTouchServiceError

Bases: BaseOverkizError

Raised when an error occurs while communicating with the CozyTouch API.

DuplicateActionOnDeviceError

Bases: BaseOverkizError

Raised when another action already exists for the same device.

ExecutionQueueFullError

Bases: BaseOverkizError

Raised when the execution queue is full.

InvalidCommandError

Bases: BaseOverkizError

Raised when an invalid command is provided.

InvalidEventListenerIdError

Bases: BaseOverkizError

Raised when an invalid event listener ID is provided.

InvalidTokenError

Bases: BaseOverkizError

Raised when an invalid token is provided.

MaintenanceError

Bases: ServiceUnavailableError

Raised when the service is under maintenance.

MissingAPIKeyError

Bases: BaseOverkizError

Raised when the API key is missing.

MissingAuthorizationTokenError

Bases: ResourceAccessDeniedError

Raised when the authorization token is missing.

NexityBadCredentialsError

Bases: BadCredentialsError

Raised when invalid credentials are provided to Nexity authentication API.

NexityServiceError

Bases: BaseOverkizError

Raised when an error occurs while communicating with the Nexity API.

NoGatewaySelectedError

Bases: BaseOverkizError

No gateway selected for a server that requires gateway selection.

NoOverkizUserError

Bases: ResourceAccessDeniedError

Raised when the account authenticated but has no Overkiz user on this server.

Login succeeds, but the token is not bound to an Overkiz user on this endpoint. Seen for Somfy accounts that need the site-scoped multi-account flow rather than the classic single-endpoint login.

NoRegisteredEventListenerError

Bases: BaseOverkizError

Raised when no event listener is registered.

NoSuchActionGroupError

Bases: BaseOverkizError

Raised when the requested action group does not exist.

NoSuchDeviceError

Bases: BaseOverkizError

Raised when the requested device does not exist.

NoSuchResourceError

Bases: BaseOverkizError

Raised when an invalid API call is made.

NoSuchTokenError

Bases: BaseOverkizError

Raised when no such token exists.

NotAuthenticatedError

Bases: ResourceAccessDeniedError

Raised when the user is not authenticated.

OverkizError

Bases: BaseOverkizError

Raised when an undefined error occurs while communicating with the Overkiz API.

ResourceAccessDeniedError

Bases: BaseOverkizError

Raised when the API returns a RESOURCE_ACCESS_DENIED error.

ServiceUnavailableError

Bases: BaseOverkizError

Raised when the service is unavailable.

SessionAndBearerInSameRequestError

Bases: BaseOverkizError

Raised when both session and bearer are provided in the same request.

SomfyBadCredentialsError

Bases: BadCredentialsError

Raised when invalid credentials are provided to Somfy authentication API.

SomfyServiceError

Bases: BaseOverkizError

Raised when an error occurs while communicating with the Somfy API.

TooManyAttemptsBannedError

Bases: BaseOverkizError

Raised when too many attempts are made and the user is (temporarily) banned.

TooManyConcurrentRequestsError

Bases: BaseOverkizError

Raised when too many concurrent requests are made.

TooManyExecutionsError

Bases: BaseOverkizError

Raised when too many executions are requested.

TooManyRequestsError

Bases: BaseOverkizError

Raised when too many requests are made.

UnknownObjectError

Bases: BaseOverkizError

Raised when an unknown object is provided.

UnknownUserError

Bases: BaseOverkizError

Raised when an unknown user is provided.

UnsupportedOperationError

Bases: BaseOverkizError

Raised when an operation is not supported by the device or gateway.

Credentials for authentication strategies.

Credentials

Bases: ABC

Abstract base class for auth credentials.

LocalTokenCredentials dataclass

Bases: TokenCredentials

Credentials using a local API token.

RexelOAuthCodeCredentials dataclass

Bases: Credentials

Credentials using Rexel OAuth2 authorization code with PKCE.

RexelTokenCredentials dataclass

Bases: Credentials

Rexel credentials backed by an externally-managed access token.

Use this when the OAuth2 lifecycle is owned outside pyoverkiz (for example Home Assistant's application_credentials platform). Supply either an async access_token_callback (called before each request, so the owner can refresh and persist tokens) or a static access_token for simple standalone or test use. gateway_id pre-selects a gateway on reload, skipping discovery.

__post_init__() -> None

Require at least one access-token source.

TokenCredentials dataclass

Bases: Credentials

Credentials using an (API) token.

UsernamePasswordCredentials dataclass

Bases: Credentials

Credentials using username and password.

Base classes for authentication strategies.

GatewayCandidate dataclass

A selectable Overkiz gateway behind a multi-account directory.

SupportsGatewaySelection

Bases: Protocol

Optional capability: discover and select among multiple gateways.

selected_gateway: str | None property

Return the currently selected gateway id, or None.

discover_gateways() -> list[GatewayCandidate] async

Return all selectable gateways for the authenticated account.

select_gateway(gateway_id: str) -> None

Select the gateway to scope subsequent requests to.