Module pyControl4.websocket

Handles Websocket connections to a Control4 Director, allowing for real-time updates using callbacks.

Classes

class C4Websocket (ip: str,
session_no_verify_ssl: aiohttp.ClientSession | None = None,
connect_callback: Callable[..., Any] | None = None,
disconnect_callback: Callable[..., Any] | None = None)

Expand source code

class C4Websocket:
    def __init__(
        self,
        ip: str,
        session_no_verify_ssl: aiohttp.ClientSession | None = None,
        connect_callback: Callable[..., Any] | None = None,
        disconnect_callback: Callable[..., Any] | None = None,
    ):
        """Creates a Control4 Websocket object.

        Parameters:
            `ip` - The IP address of the Control4 Director/Controller.

            `session_no_verify_ssl` - (Optional) Allows the use of an
                        `aiohttp.ClientSession` object
                        for all network requests. This
                        session will not be closed by the library.
                        If not provided, the library will open and
                        close its own `ClientSession`s as needed.
                        This session is also passed to the underlying
                        socketio/engineio client to avoid blocking
                        `ssl.create_default_context()` calls inside
                        the event loop.

            `connect_callback` - (Optional) A callback to be called when the
                Websocket connection is opened or reconnected after a network
                error.

            `disconnect_callback` - (Optional) A callback to be called when
                the Websocket connection is lost due to a network error.
        """
        self.base_url: str = f"https://{ip}"
        self.wss_url: str = f"wss://{ip}"
        self.session: aiohttp.ClientSession | None = session_no_verify_ssl
        self.connect_callback: Callable[..., Any] | None = connect_callback
        self.disconnect_callback: Callable[..., Any] | None = disconnect_callback

        self._item_callbacks: dict[int, list[Callable[..., Any]]] = dict()
        self._sio: socketio.AsyncClient | None = None

    @property
    def item_callbacks(self) -> MappingProxyType[int, list[Callable[..., Any]]]:
        """Returns a read-only view of registered item ids (key) and their
        callbacks (value). Use add_item_callback() or remove_item_callback()
        to modify callbacks.
        """
        return MappingProxyType(self._item_callbacks)

    def add_item_callback(self, item_id: int, callback: Callable[..., Any]) -> None:
        """Register a callback to receive updates about an item.
        Parameters:
            `item_id` - The Control4 item ID.
            `callback` - The callback to be called when an update is received
                for the provided item id.
        """
        _LOGGER.debug("Subscribing to updates for item id: %s", item_id)

        if item_id not in self._item_callbacks:
            self._item_callbacks[item_id] = []

        # Avoid duplicates
        if callback not in self._item_callbacks[item_id]:
            self._item_callbacks[item_id].append(callback)

    def remove_item_callback(
        self, item_id: int, callback: Callable[..., Any] | None = None
    ) -> None:
        """Unregister callback(s) for an item.
        Parameters:
            `item_id` - The Control4 item ID.
            `callback` - (Optional) Specific callback to remove. If None,
                removes all callbacks for this item_id.
        """
        if item_id not in self._item_callbacks:
            return

        if callback is None:
            # Remove all callbacks for this item_id
            del self._item_callbacks[item_id]
        else:
            # Remove a specific callback
            try:
                self._item_callbacks[item_id].remove(callback)
                # If no more callbacks, remove the entry
                if not self._item_callbacks[item_id]:
                    del self._item_callbacks[item_id]
            except ValueError:
                pass

    async def sio_connect(self, director_bearer_token: str) -> None:
        """Start WebSockets connection and listen, using the provided
        director_bearer_token to authenticate with the Control4 Director. If a
        connection already exists, it will be disconnected and a new connection
        will be created.

        This function should be called using a new token every 86400 seconds (the
        expiry time of the director tokens), otherwise the Control4 Director will
        stop sending WebSocket messages.

        Parameters:
            `director_bearer_token` - The bearer token used to authenticate
                with the Director. See
                `pyControl4.account.C4Account.get_director_bearer_token`
                for how to get this.
        """
        # Disconnect previous sio object
        await self.sio_disconnect()

        if self.session is not None:
            # Create a new session using the caller's connector so engineio
            # can safely close it in _reset() without affecting the caller's
            # session.  Setting ssl_verify=True prevents engineio from
            # creating its own no-verify SSLContext, allowing the
            # connector's SSL configuration to take effect.
            http_session = aiohttp.ClientSession(
                connector=self.session.connector, connector_owner=False
            )
            self._sio = socketio.AsyncClient(ssl_verify=True, http_session=http_session)
        else:
            self._sio = socketio.AsyncClient(ssl_verify=False)
        self._sio.register_namespace(
            _C4DirectorNamespace(
                token=director_bearer_token,
                url=self.base_url,
                callback=self._callback,
                session=self.session,
                connect_callback=self.connect_callback,
                disconnect_callback=self.disconnect_callback,
            )
        )
        await self._sio.connect(
            self.wss_url,
            transports=["websocket"],
            headers={"JWT": director_bearer_token},
        )

    async def sio_disconnect(self) -> None:
        """Disconnects the WebSockets connection, if it has been created."""
        if isinstance(self._sio, socketio.AsyncClient):
            await self._sio.disconnect()

    async def _callback(self, message: Any) -> None:
        if "status" in message:
            _LOGGER.debug(f'Subscription {message["status"]}')
            return
        if isinstance(message, list):
            for m in message:
                await self._process_message(m)
        else:
            await self._process_message(message)

    async def _process_message(self, message: Any) -> None:
        """Process an incoming event message."""
        _LOGGER.debug(message)
        device_id = message.get("iddevice") if isinstance(message, dict) else None
        if device_id is None:
            _LOGGER.debug("Received message without iddevice field")
            return

        callbacks = self._item_callbacks.get(device_id, [])
        if not callbacks:
            _LOGGER.debug(f"No Callback for device id {device_id}")
            return

        for callback in callbacks[:]:
            try:
                if isinstance(message, list):
                    for m in message:
                        await callback(device_id, m)
                else:
                    await callback(device_id, message)
            except Exception as exc:
                _LOGGER.warning(f"Captured exception during callback: {str(exc)}")

Creates a Control4 Websocket object.

Parameters

ip - The IP address of the Control4 Director/Controller.

session_no_verify_ssl - (Optional) Allows the use of an aiohttp.ClientSession object for all network requests. This session will not be closed by the library. If not provided, the library will open and close its own ClientSessions as needed. This session is also passed to the underlying socketio/engineio client to avoid blocking ssl.create_default_context() calls inside the event loop.

connect_callback - (Optional) A callback to be called when the Websocket connection is opened or reconnected after a network error.

disconnect_callback - (Optional) A callback to be called when the Websocket connection is lost due to a network error.

Instance variables

prop item_callbacks : MappingProxyType[int, list[Callable[..., Any]]]

Expand source code

@property
def item_callbacks(self) -> MappingProxyType[int, list[Callable[..., Any]]]:
    """Returns a read-only view of registered item ids (key) and their
    callbacks (value). Use add_item_callback() or remove_item_callback()
    to modify callbacks.
    """
    return MappingProxyType(self._item_callbacks)

Returns a read-only view of registered item ids (key) and their callbacks (value). Use add_item_callback() or remove_item_callback() to modify callbacks.

Methods

def add_item_callback(self, item_id: int, callback: Callable[..., Any]) ‑> None

Expand source code

def add_item_callback(self, item_id: int, callback: Callable[..., Any]) -> None:
    """Register a callback to receive updates about an item.
    Parameters:
        `item_id` - The Control4 item ID.
        `callback` - The callback to be called when an update is received
            for the provided item id.
    """
    _LOGGER.debug("Subscribing to updates for item id: %s", item_id)

    if item_id not in self._item_callbacks:
        self._item_callbacks[item_id] = []

    # Avoid duplicates
    if callback not in self._item_callbacks[item_id]:
        self._item_callbacks[item_id].append(callback)

Register a callback to receive updates about an item.

Parameters

item_id - The Control4 item ID. callback - The callback to be called when an update is received for the provided item id.

def remove_item_callback(self, item_id: int, callback: Callable[..., Any] | None = None) ‑> None

Expand source code

def remove_item_callback(
    self, item_id: int, callback: Callable[..., Any] | None = None
) -> None:
    """Unregister callback(s) for an item.
    Parameters:
        `item_id` - The Control4 item ID.
        `callback` - (Optional) Specific callback to remove. If None,
            removes all callbacks for this item_id.
    """
    if item_id not in self._item_callbacks:
        return

    if callback is None:
        # Remove all callbacks for this item_id
        del self._item_callbacks[item_id]
    else:
        # Remove a specific callback
        try:
            self._item_callbacks[item_id].remove(callback)
            # If no more callbacks, remove the entry
            if not self._item_callbacks[item_id]:
                del self._item_callbacks[item_id]
        except ValueError:
            pass

Unregister callback(s) for an item.

Parameters

item_id - The Control4 item ID. callback - (Optional) Specific callback to remove. If None, removes all callbacks for this item_id.

async def sio_connect(self, director_bearer_token: str) ‑> None

Expand source code

async def sio_connect(self, director_bearer_token: str) -> None:
    """Start WebSockets connection and listen, using the provided
    director_bearer_token to authenticate with the Control4 Director. If a
    connection already exists, it will be disconnected and a new connection
    will be created.

    This function should be called using a new token every 86400 seconds (the
    expiry time of the director tokens), otherwise the Control4 Director will
    stop sending WebSocket messages.

    Parameters:
        `director_bearer_token` - The bearer token used to authenticate
            with the Director. See
            `pyControl4.account.C4Account.get_director_bearer_token`
            for how to get this.
    """
    # Disconnect previous sio object
    await self.sio_disconnect()

    if self.session is not None:
        # Create a new session using the caller's connector so engineio
        # can safely close it in _reset() without affecting the caller's
        # session.  Setting ssl_verify=True prevents engineio from
        # creating its own no-verify SSLContext, allowing the
        # connector's SSL configuration to take effect.
        http_session = aiohttp.ClientSession(
            connector=self.session.connector, connector_owner=False
        )
        self._sio = socketio.AsyncClient(ssl_verify=True, http_session=http_session)
    else:
        self._sio = socketio.AsyncClient(ssl_verify=False)
    self._sio.register_namespace(
        _C4DirectorNamespace(
            token=director_bearer_token,
            url=self.base_url,
            callback=self._callback,
            session=self.session,
            connect_callback=self.connect_callback,
            disconnect_callback=self.disconnect_callback,
        )
    )
    await self._sio.connect(
        self.wss_url,
        transports=["websocket"],
        headers={"JWT": director_bearer_token},
    )

Start WebSockets connection and listen, using the provided director_bearer_token to authenticate with the Control4 Director. If a connection already exists, it will be disconnected and a new connection will be created.

This function should be called using a new token every 86400 seconds (the expiry time of the director tokens), otherwise the Control4 Director will stop sending WebSocket messages.

Parameters

director_bearer_token - The bearer token used to authenticate with the Director. See C4Account.get_director_bearer_token() for how to get this.

async def sio_disconnect(self) ‑> None

Expand source code

async def sio_disconnect(self) -> None:
    """Disconnects the WebSockets connection, if it has been created."""
    if isinstance(self._sio, socketio.AsyncClient):
        await self._sio.disconnect()

Disconnects the WebSockets connection, if it has been created.