API documentation

The Nice G.O. API client for Python.

To start, see the NiceGOApi class.

ApiError

Bases: NiceGOError

API error.

Source code in src/nice_go/_exceptions.py

class ApiError(NiceGOError):
    """API error."""

AuthFailedError

Bases: NiceGOError

Authentication failed. Check your credentials.

Source code in src/nice_go/_exceptions.py

class AuthFailedError(NiceGOError):
    """Authentication failed. Check your credentials."""

Barrier dataclass

Class representing a barrier.

Attributes:

Name	Type	Description
id	str	The ID of the barrier.
type	str	The type of the barrier.
controlLevel	str	The control level of the barrier.
attr	list[dict[str, str]]	A list of attributes of the barrier.
state	BarrierState	The state of the barrier.
api	NiceGOApi	The NiceGO API object.

Methods:

Name	Description
open	Open the barrier.
close	Close the barrier.
light_on	Turn on the light of the barrier.
light_off	Turn off the light of the barrier.
get_attr	Get the value of an attribute.

Source code in src/nice_go/_barrier.py

@dataclass
class Barrier:
    """Class representing a barrier.

    Attributes:
        id (str): The ID of the barrier.
        type (str): The type of the barrier.
        controlLevel (str): The control level of the barrier.
        attr (list[dict[str, str]]): A list of attributes of the barrier.
        state (BarrierState): The state of the barrier.
        api (NiceGOApi): The NiceGO API object.

    Methods:
        open: Open the barrier.
        close: Close the barrier.
        light_on: Turn on the light of the barrier.
        light_off: Turn off the light of the barrier.
        get_attr: Get the value of an attribute.
    """

    id: str
    type: str
    controlLevel: str  # noqa: N815
    attr: list[dict[str, str]]
    state: BarrierState
    api: NiceGOApi

    async def open(self) -> bool:
        """Open the barrier.

        Returns:
            A boolean indicating whether the command was successful.
        """
        return await self.api.open_barrier(self.id)

    async def close(self) -> bool:
        """Close the barrier.

        Returns:
            A boolean indicating whether the command was successful.
        """
        return await self.api.close_barrier(self.id)

    async def light_on(self) -> bool:
        """Turn on the light of the barrier.

        Returns:
            A boolean indicating whether the command was successful.
        """
        return await self.api.light_on(self.id)

    async def light_off(self) -> bool:
        """Turn off the light of the barrier.

        Returns:
            A boolean indicating whether the command was successful.
        """
        return await self.api.light_off(self.id)

    async def get_attr(self, key: str) -> str:
        """Get the value of an attribute.

        Args:
            key (str): The key of the attribute.

        Returns:
            The value of the attribute.

        Raises:
            KeyError: If the attribute with the given key is not found.
        """
        attr = next((attr for attr in self.attr if attr["key"] == key), None)
        if attr is None:
            msg = f"Attribute with key {key} not found."
            raise KeyError(msg)
        return attr["value"]

    async def vacation_mode_on(self) -> None:
        """Turn on vacation mode for the barrier."""
        await self.api.vacation_mode_on(self.id)

    async def vacation_mode_off(self) -> None:
        """Turn off vacation mode for the barrier."""
        await self.api.vacation_mode_off(self.id)

close() -> bool async

Close the barrier.

Returns:

Type	Description
bool	A boolean indicating whether the command was successful.

Source code in src/nice_go/_barrier.py

async def close(self) -> bool:
    """Close the barrier.

    Returns:
        A boolean indicating whether the command was successful.
    """
    return await self.api.close_barrier(self.id)

get_attr(key: str) -> str async

Get the value of an attribute.

Parameters:

Name	Type	Description	Default
key	str	The key of the attribute.	required

Returns:

Type	Description
str	The value of the attribute.

Raises:

Type	Description
KeyError	If the attribute with the given key is not found.

Source code in src/nice_go/_barrier.py

async def get_attr(self, key: str) -> str:
    """Get the value of an attribute.

    Args:
        key (str): The key of the attribute.

    Returns:
        The value of the attribute.

    Raises:
        KeyError: If the attribute with the given key is not found.
    """
    attr = next((attr for attr in self.attr if attr["key"] == key), None)
    if attr is None:
        msg = f"Attribute with key {key} not found."
        raise KeyError(msg)
    return attr["value"]

light_off() -> bool async

Turn off the light of the barrier.

Returns:

Type	Description
bool	A boolean indicating whether the command was successful.

Source code in src/nice_go/_barrier.py

async def light_off(self) -> bool:
    """Turn off the light of the barrier.

    Returns:
        A boolean indicating whether the command was successful.
    """
    return await self.api.light_off(self.id)

light_on() -> bool async

Turn on the light of the barrier.

Returns:

Type	Description
bool	A boolean indicating whether the command was successful.

Source code in src/nice_go/_barrier.py

async def light_on(self) -> bool:
    """Turn on the light of the barrier.

    Returns:
        A boolean indicating whether the command was successful.
    """
    return await self.api.light_on(self.id)

open() -> bool async

Open the barrier.

Returns:

Type	Description
bool	A boolean indicating whether the command was successful.

Source code in src/nice_go/_barrier.py

async def open(self) -> bool:
    """Open the barrier.

    Returns:
        A boolean indicating whether the command was successful.
    """
    return await self.api.open_barrier(self.id)

vacation_mode_off() -> None async

Turn off vacation mode for the barrier.

Source code in src/nice_go/_barrier.py

async def vacation_mode_off(self) -> None:
    """Turn off vacation mode for the barrier."""
    await self.api.vacation_mode_off(self.id)

vacation_mode_on() -> None async

Turn on vacation mode for the barrier.

Source code in src/nice_go/_barrier.py

async def vacation_mode_on(self) -> None:
    """Turn on vacation mode for the barrier."""
    await self.api.vacation_mode_on(self.id)

BarrierState dataclass

Class representing the state of a barrier.

Attributes:

Name	Type	Description
deviceId	str	The device ID of the barrier.
reported	dict[str, Any]	The reported state of the barrier.
timestamp	str	The timestamp of the last update.
version	str	The version of the barrier.
connectionState	ConnectionState | None	The connection state of the barrier.

Source code in src/nice_go/_barrier.py

@dataclass
class BarrierState:
    """Class representing the state of a barrier.

    Attributes:
        deviceId (str): The device ID of the barrier.
        reported (dict[str, Any]): The reported state of the barrier.
        timestamp (str): The timestamp of the last update.
        version (str): The version of the barrier.
        connectionState (ConnectionState | None): The connection state of the barrier.
    """

    deviceId: str  # noqa: N815
    reported: dict[str, Any]
    timestamp: str
    version: str
    connectionState: ConnectionState | None  # noqa: N815

ConnectionState dataclass

Class representing the connection state of a barrier.

Attributes:

Name	Type	Description
connected	bool	A boolean indicating whether the barrier is connected.
updatedTimestamp	datetime	The timestamp of the last update.

Source code in src/nice_go/_barrier.py

@dataclass
class ConnectionState:
    """Class representing the connection state of a barrier.

    Attributes:
        connected (bool): A boolean indicating whether the barrier is connected.
        updatedTimestamp (datetime): The timestamp of the last update.
    """

    connected: bool
    updatedTimestamp: datetime  # noqa: N815

NiceGOApi

A class to interact with the Nice G.O. API.

This class provides methods to authenticate, connect to the WebSocket API, and interact with barriers.

Attributes:

Name	Type	Description
id_token	str | None	The ID token.

Methods:

Name	Description
event	Decorator to add an event listener.
authenticate_refresh	Authenticate using a refresh token.
authenticate	Authenticate using username and password.
connect	Connect to the WebSocket API.
subscribe	Subscribe to a receiver.
unsubscribe	Unsubscribe from a receiver.
close	Close the connection.
get_all_barriers	Get all barriers.

Source code in src/nice_go/nice_go_api.py

class NiceGOApi:
    """A class to interact with the Nice G.O. API.

    This class provides methods to authenticate, connect to the WebSocket API,
    and interact with barriers.

    Attributes:
        id_token (str | None): The ID token.

    Functions:
        event: Decorator to add an event listener.
        authenticate_refresh: Authenticate using a refresh token.
        authenticate: Authenticate using username and password.
        connect: Connect to the WebSocket API.
        subscribe: Subscribe to a receiver.
        unsubscribe: Unsubscribe from a receiver.
        close: Close the connection.
        get_all_barriers: Get all barriers.
    """

    def __init__(self) -> None:
        """Initialize the NiceGOApi object."""
        self.id_token: str | None = None
        self._closing_task: asyncio.Task[None] | None = None
        self._device_ws: WebSocketClient | None = None
        self._endpoints: dict[str, Any] | None = None
        self._session: aiohttp.ClientSession | None = None
        self._event_tasks: set[asyncio.Task[None]] = set()
        self._events_ws: WebSocketClient | None = None
        self._device_connected: bool = False
        self._events_connected: bool = False
        self._events: dict[str, list[Callable[..., Coroutine[Any, Any, Any]]]] = {}

        self.event(self.on_device_connected)
        self.event(self.on_events_connected)

    async def on_device_connected(self) -> None:
        """Handle the device connected event."""
        self._device_connected = True
        if self._device_connected and self._events_connected:
            # Only dispatch when both feeds are connected
            self._dispatch("connected")

    async def on_events_connected(self) -> None:
        """Handle the events connected event."""
        self._events_connected = True
        if self._device_connected and self._events_connected:
            # Only dispatch when both feeds are connected
            self._dispatch("connected")

    def event(self, coro: CoroT) -> CoroT:
        """Decorator to add an event listener. Just a wrapper around `listen`.

        Info:
            This can only decorate coroutine functions.

        Args:
            coro (CoroT): The coroutine function to decorate.

        Examples:
            You can use this decorator to add event listeners to the API object.
            Some events include but are not limited to:

            - `connection_lost`: Triggered when the connection to the WebSocket API is
                lost.
            - `connected`: Triggered when the connection to the WebSocket API is
                established.
            - `data`: Triggered when data is received from an active subscription.
                See `subscribe`.

            >>> @api.event
            ... async def on_data(
            ...     data: dict[str, Any] | None = None,
            ... ) -> None:
            ...     if data is not None:
            ...         print(data)
        """
        self.listen(coro.__name__, coro)
        return coro

    def listen(self, event_name: str, coro: CoroT) -> Callable[[], None]:
        """Add an event listener.

        Args:
            event_name (str): The name of the event.
            coro (CoroT): The coroutine to run when the event is dispatched.

        Returns:
            A function to remove the event listener.

        Examples:
            You can use this method to add event listeners to the API object.
            Some events include but are not limited to:

            - `connection_lost`: Triggered when the connection to the WebSocket API is
                lost.
            - `connected`: Triggered when the connection to the WebSocket API is
                established.
            - `data`: Triggered when data is received from an active subscription.
                See `subscribe`.

            >>> def on_data(data: dict[str, Any] | None = None) -> None:
            ...     if data is not None:
            ...         print(data)
            ...
            >>> remove_listener = api.listen("data", on_data)
        """
        if not asyncio.iscoroutinefunction(coro):
            msg = "The decorated function must be a coroutine"
            raise TypeError(msg)

        _LOGGER.debug("Adding event listener %s", coro.__name__)

        self._events.setdefault(event_name, []).append(coro)
        return lambda: self._events[event_name].remove(coro)

    async def _run_event(
        self,
        coro: Callable[..., Coroutine[Any, Any, Any]],
        event_name: str,
        data: dict[str, Any] | None = None,
    ) -> None:
        """Run an event coroutine. For internal use only.

        Args:
            coro (Callable[..., Coroutine[Any, Any, Any]]): The coroutine to run.
            event_name (str): The name of the event.
            data (dict[str, Any] | None): The data to pass to the event coroutine.
        """
        kwargs = {}
        if data is not None:
            kwargs["data"] = data
        try:
            await coro(**kwargs)
        except asyncio.CancelledError:
            pass
        except Exception:
            _LOGGER.exception("Error while handling event %s", event_name)

    def _schedule_event(
        self,
        coro: Callable[..., Coroutine[Any, Any, Any]],
        event_name: str,
        data: dict[str, Any] | None,
    ) -> None:
        """Schedule an event to be dispatched. For internal use only.

        Args:
            coro (Callable[..., Coroutine[Any, Any, Any]]): The coroutine to schedule.
            event_name (str): The name of the event.
            data (dict[str, Any] | None): The data to pass to the event coroutine.
        """
        wrapped = self._run_event(coro, event_name, data)
        task = asyncio.create_task(wrapped, name=f"NiceGO: {event_name}")
        self._event_tasks.add(task)  # See RUF006
        task.add_done_callback(self._event_tasks.discard)

    def _dispatch(self, event: str, data: dict[str, Any] | None = None) -> None:
        """Dispatch an event to listeners. For internal use only.

        Args:
            event (str): The name of the event.
            data (dict[str, Any] | None): The data to pass to the event coroutine.
        """
        method = f"on_{event}"

        coros = self._events.get(method, [])

        if not coros:
            _LOGGER.debug("No listeners for event %s", event)
            return

        _LOGGER.debug("Dispatching event %s", event)
        for coro in coros:
            self._schedule_event(coro, method, data)

    async def authenticate_refresh(
        self,
        refresh_token: str,
        session: aiohttp.ClientSession,
    ) -> None:
        """Authenticate using a previous obtained refresh token.

        Args:
            refresh_token (str): The refresh token.
            session (aiohttp.ClientSession): The client session to use.

        Raises:
            AuthFailedError: If the authentication fails.
            ApiError: If an API error occurs.
        """
        self._session = session
        await self.__authenticate(None, None, refresh_token)

    async def authenticate(
        self,
        user_name: str,
        password: str,
        session: aiohttp.ClientSession,
    ) -> str | None:
        """Authenticate using username and password.

        Args:
            user_name (str): The username.
            password (str): The password.
            session (aiohttp.ClientSession): The client session to use.

        Returns:
            The refresh token.

        Raises:
            AuthFailedError: If the authentication fails.
            ApiError: If an API error occurs.
        """
        self._session = session
        return await self.__authenticate(user_name, password, None)

    async def __authenticate(
        self,
        user_name: str | None,
        password: str | None,
        refresh_token: str | None,
    ) -> str | None:
        """Authenticate using username and password or refresh token.

        Args:
            user_name (str | None): The username.
            password (str | None): The password.
            refresh_token (str | None): The refresh token.

        Returns:
            The refresh token.

        Raises:
            AuthFailedError: If the authentication fails.
            ApiError: If an API error occurs.
        """
        try:
            _LOGGER.debug("Authenticating")

            if self._session is None:
                msg = "ClientSession not provided"
                raise ValueError(msg)

            # Get the endpoints
            data = await self._session.get(ENDPOINTS_URL)
            endpoints = await data.json()
            self._endpoints = endpoints["endpoints"]

            if self._endpoints is None:
                msg = "Endpoints not available"
                raise ApiError(msg)

            authenticator = AwsCognitoAuthenticator(
                self._endpoints["Config"]["Region"],
                self._endpoints["Config"]["ClientId"],
                self._endpoints["Config"]["UserPoolId"],
                self._endpoints["Config"]["IdentityPoolId"],
            )

            if user_name and password:
                token = await asyncio.to_thread(
                    authenticator.get_new_token,
                    user_name,
                    password,
                )
            elif refresh_token:
                token = await asyncio.to_thread(
                    authenticator.refresh_token,
                    refresh_token,
                )

            _LOGGER.debug("Authenticated")

            self.id_token = token.id_token
        except botocore.exceptions.ClientError as e:
            _LOGGER.exception("Exception while authenticating")
            if e.response["Error"]["Code"] == "NotAuthorizedException":
                raise AuthFailedError from e
            raise ApiError from e
        else:
            return token.refresh_token

    @property
    def closed(self) -> bool:
        """Check if the connection is closed."""
        return self._closing_task is not None

    async def _poll_device_ws(self) -> None:
        """Continuously polls the device WebSocket to maintain an active connection.
        This function will repeatedly call the poll method on the WebSocket if it is
        initialized.

        Returns:
            None
        """
        if self._device_ws is None:
            return
        while True:
            await self._device_ws.poll()

    async def _poll_events_ws(self) -> None:
        """Continuously polls the device WebSocket to maintain an active connection.
        This function will repeatedly call the poll method on the WebSocket if it is
        initialized.

        Returns:
            None
        """

        if self._events_ws is None:
            return
        while True:
            await self._events_ws.poll()

    async def _check_response_errors(self, response: dict[str, Any]) -> None:
        """Checks a GraphQL response for errors, namely for expired tokens.

        Args:
            response (dict[str, Any]): The response to check.

        Raises:
            AuthFailedError: If the ID token is expired.
            ApiError: If an API error occurs.

        Returns:
            None
        """

        if errors := response.get("errors"):
            error = errors[0]
            if error["errorType"] == "UnauthorizedException":
                raise AuthFailedError(error)
            raise ApiError(error)

    @retry(
        wait=wait_random_exponential(multiplier=1, min=1, max=10),
        retry=_RetryIfReconnect()
        & retry_if_exception_type(
            (
                OSError,
                WebSocketError,
                aiohttp.ClientError,
                asyncio.TimeoutError,
                ReconnectWebSocketError,
            ),
        ),
        reraise=True,
        before_sleep=before_sleep_log(_LOGGER, logging.DEBUG),
    )
    async def connect(self, *, reconnect: bool = True) -> None:
        """Connect to the WebSocket API.

        Warning:
            No events will be dispatched until you subscribe to a receiver.
            Typically, you should pass the `organization` attribute of a barrier
            object to the `subscribe` method to start receiving data. Don't ask me
            why `organization` specifically, I don't know either.

        Danger:
            This method will block the event loop until the connection is closed.
            If you want to run this method in the background, you should run it in a
            separate thread or process.

        Args:
            reconnect (bool): Whether to reconnect if the connection is lost.

        Raises:
            NoAuthError: If the ID token is not available.
            ApiError: If an API error occurs.
            WebSocketError: If an error occurs while connecting.
        """
        try:
            if self.id_token is None:
                raise NoAuthError

            if self._endpoints is None:
                msg = "Endpoints not available"
                raise ApiError(msg)

            if self._session is None:
                msg = "ClientSession not provided"
                raise ValueError(msg)

            self._reconnect = reconnect

            device_url = self._endpoints["GraphQL"]["device"]["wss"]
            events_url = self._endpoints["GraphQL"]["events"]["wss"]

            _LOGGER.debug("Connecting to WebSocket API %s", device_url)

            self._device_ws = WebSocketClient(client_session=self._session)
            await self._device_ws.connect(
                self.id_token,
                yarl.URL(device_url),
                "device",
                self._dispatch,
                yarl.URL(self._endpoints["GraphQL"]["device"]["https"]).host,
            )
            self._events_ws = WebSocketClient(client_session=self._session)
            await self._events_ws.connect(
                self.id_token,
                yarl.URL(events_url),
                "events",
                self._dispatch,
                yarl.URL(self._endpoints["GraphQL"]["events"]["https"]).host,
            )

            _LOGGER.debug("Connected to WebSocket API")

            device_task = asyncio.create_task(self._poll_device_ws())
            events_task = asyncio.create_task(self._poll_events_ws())

            with contextlib.suppress(asyncio.CancelledError):
                done, pending = await asyncio.wait(
                    [device_task, events_task],
                    return_when=asyncio.FIRST_EXCEPTION,
                )

            with contextlib.suppress(UnboundLocalError):
                if exceptions := [
                    task.exception() for task in done if task.exception()
                ]:
                    for p in pending:
                        p.cancel()
                    # Make sure both WS are closed
                    await self._events_ws.close()
                    await self._device_ws.close()
                    raise exceptions[0]  # type: ignore[misc]
        except (
            OSError,
            WebSocketError,
            aiohttp.ClientError,
            asyncio.TimeoutError,
            ReconnectWebSocketError,
        ) as e:
            self._dispatch("connection_lost", {"exception": e})
            self._device_connected = False
            self._events_connected = False
            if not reconnect:
                _LOGGER.debug("Connection lost, not reconnecting")
                await self.close()
                raise

            if self.closed:
                return

            _LOGGER.debug("Connection lost, retrying...")

            # Raising triggers retry
            raise

    async def subscribe(self, receiver: str) -> list[str]:
        """Subscribe to a receiver.

        Args:
            receiver (str): The receiver to subscribe to.

        Returns:
            The subscription IDs. You can pass this into the `unsubscribe` method to
                unsubscribe from the receiver.

        Raises:
            WebSocketError: If no WebSocket connection is available.
        """
        if self._device_ws is None:
            msg = "No WebSocket connection"
            raise WebSocketError(msg)
        if self._events_ws is None:
            msg = "No WebSocket connection"
            raise WebSocketError(msg)

        _LOGGER.debug("Subscribing to receiver %s", receiver)

        return [
            await self._device_ws.subscribe(receiver),
            await self._events_ws.subscribe(receiver),
        ]

    async def unsubscribe(self, subscription_id: str) -> None:
        """Unsubscribe from a receiver.

        Args:
            subscription_id (str): The subscription ID to unsubscribe from.

        Raises:
            WebSocketError: If no WebSocket connection is available
        """
        if self._device_ws is None:
            msg = "No WebSocket connection"
            raise WebSocketError(msg)
        if self._events_ws is None:
            msg = "No WebSocket connection"
            raise WebSocketError(msg)

        _LOGGER.debug("Unsubscribing from subscription %s", subscription_id)

        await self._device_ws.unsubscribe(subscription_id)
        await self._events_ws.unsubscribe(subscription_id)

    async def close(self) -> None:
        """Close the connection.

        Raises:
            NoAuthError: If the ID token is not available.
        """

        async def _close() -> None:
            if self._device_ws:
                await self._device_ws.close()
            if self._events_ws:
                await self._events_ws.close()

        _LOGGER.debug("Closing connection")

        self._closing_task = asyncio.create_task(_close())
        await self._closing_task

    async def get_all_barriers(self) -> list[Barrier]:
        """Get all barriers.

        Returns:
            A list of all barriers.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Getting all barriers")
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template("get_all_barriers", None),
        )
        data = await response.json()

        _LOGGER.debug("Got all barriers")
        _LOGGER.debug("Data: %s", data)

        await self._check_response_errors(data)

        barriers = []

        for device in data["data"]["devicesListAll"]["devices"]:
            if device["state"]["connectionState"] is not None:
                connection_state = ConnectionState(
                    device["state"]["connectionState"]["connected"],
                    device["state"]["connectionState"]["updatedTimestamp"],
                )
            else:
                connection_state = None
            barrier_state = BarrierState(
                device["state"]["deviceId"],
                json.loads(device["state"]["reported"]),
                device["state"]["timestamp"],
                device["state"]["version"],
                connection_state,
            )
            barrier = Barrier(
                device["id"],
                device["type"],
                device["controlLevel"],
                device["attr"],
                barrier_state,
                self,
            )
            barriers.append(barrier)

        return barriers

    async def open_barrier(self, barrier_id: str) -> bool:
        """Open a barrier.

        Args:
            barrier_id (str): The ID of the barrier to open.

        Returns:
            Whether the barrier was opened successfully.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Opening barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template("open_barrier", {"barrier_id": barrier_id}),
        )
        data = await response.json()

        _LOGGER.debug("Opening barrier response: %s", data)
        await self._check_response_errors(data)

        result: bool = data["data"]["devicesControl"]

        return result

    async def close_barrier(self, barrier_id: str) -> bool:
        """Close a barrier.

        Args:
            barrier_id (str): The ID of the barrier to close.

        Returns:
            Whether the barrier was closed successfully."""
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Closing barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template(
                "close_barrier",
                {"barrier_id": barrier_id},
            ),
        )
        data = await response.json()

        _LOGGER.debug("Closing barrier response: %s", data)
        await self._check_response_errors(data)

        result: bool = data["data"]["devicesControl"]

        return result

    async def light_on(self, barrier_id: str) -> bool:
        """Turn the light on.

        Args:
            barrier_id (str): The ID of the barrier to turn the light on.

        Returns:
            Whether the light was turned on successfully.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Turning light on for barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template("light_on", {"barrier_id": barrier_id}),
        )
        data = await response.json()

        _LOGGER.debug("Turning light on response: %s", data)
        await self._check_response_errors(data)

        result: bool = data["data"]["devicesControl"]

        return result

    async def light_off(self, barrier_id: str) -> bool:
        """Turn the light off.

        Args:
            barrier_id (str): The ID of the barrier to turn the light off.

        Returns:
            Whether the light was turned off successfully.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Turning light off for barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template("light_off", {"barrier_id": barrier_id}),
        )
        data = await response.json()

        _LOGGER.debug("Turning light off response: %s", data)
        await self._check_response_errors(data)

        result: bool = data["data"]["devicesControl"]

        return result

    async def vacation_mode_on(self, barrier_id: str) -> None:
        """Turn vacation mode on.

        Args:
            barrier_id (str): The ID of the barrier to turn vacation mode on.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Turning vacation mode on for barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template(
                "vacation_mode_on",
                {"barrier_id": barrier_id},
            ),
        )
        await response.json()

    async def vacation_mode_off(self, barrier_id: str) -> None:
        """Turn vacation mode off.

        Args:
            barrier_id (str): The ID of the barrier to turn vacation mode off.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Turning vacation mode off for barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template(
                "vacation_mode_off",
                {"barrier_id": barrier_id},
            ),
        )
        await response.json()

closed: bool property

Check if the connection is closed.

__authenticate(user_name: str | None, password: str | None, refresh_token: str | None) -> str | None async

Authenticate using username and password or refresh token.

Parameters:

Name	Type	Description	Default
user_name	str | None	The username.	required
password	str | None	The password.	required
refresh_token	str | None	The refresh token.	required

Returns:

Type	Description
str | None	The refresh token.

Raises:

Type	Description
AuthFailedError	If the authentication fails.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def __authenticate(
    self,
    user_name: str | None,
    password: str | None,
    refresh_token: str | None,
) -> str | None:
    """Authenticate using username and password or refresh token.

    Args:
        user_name (str | None): The username.
        password (str | None): The password.
        refresh_token (str | None): The refresh token.

    Returns:
        The refresh token.

    Raises:
        AuthFailedError: If the authentication fails.
        ApiError: If an API error occurs.
    """
    try:
        _LOGGER.debug("Authenticating")

        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)

        # Get the endpoints
        data = await self._session.get(ENDPOINTS_URL)
        endpoints = await data.json()
        self._endpoints = endpoints["endpoints"]

        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        authenticator = AwsCognitoAuthenticator(
            self._endpoints["Config"]["Region"],
            self._endpoints["Config"]["ClientId"],
            self._endpoints["Config"]["UserPoolId"],
            self._endpoints["Config"]["IdentityPoolId"],
        )

        if user_name and password:
            token = await asyncio.to_thread(
                authenticator.get_new_token,
                user_name,
                password,
            )
        elif refresh_token:
            token = await asyncio.to_thread(
                authenticator.refresh_token,
                refresh_token,
            )

        _LOGGER.debug("Authenticated")

        self.id_token = token.id_token
    except botocore.exceptions.ClientError as e:
        _LOGGER.exception("Exception while authenticating")
        if e.response["Error"]["Code"] == "NotAuthorizedException":
            raise AuthFailedError from e
        raise ApiError from e
    else:
        return token.refresh_token

__init__() -> None

Initialize the NiceGOApi object.

Source code in src/nice_go/nice_go_api.py

def __init__(self) -> None:
    """Initialize the NiceGOApi object."""
    self.id_token: str | None = None
    self._closing_task: asyncio.Task[None] | None = None
    self._device_ws: WebSocketClient | None = None
    self._endpoints: dict[str, Any] | None = None
    self._session: aiohttp.ClientSession | None = None
    self._event_tasks: set[asyncio.Task[None]] = set()
    self._events_ws: WebSocketClient | None = None
    self._device_connected: bool = False
    self._events_connected: bool = False
    self._events: dict[str, list[Callable[..., Coroutine[Any, Any, Any]]]] = {}

    self.event(self.on_device_connected)
    self.event(self.on_events_connected)

_check_response_errors(response: dict[str, Any]) -> None async

Checks a GraphQL response for errors, namely for expired tokens.

Parameters:

Name	Type	Description	Default
response	dict[str, Any]	The response to check.	required

Raises:

Type	Description
AuthFailedError	If the ID token is expired.
ApiError	If an API error occurs.

Returns:

Type	Description
None	None

Source code in src/nice_go/nice_go_api.py

async def _check_response_errors(self, response: dict[str, Any]) -> None:
    """Checks a GraphQL response for errors, namely for expired tokens.

    Args:
        response (dict[str, Any]): The response to check.

    Raises:
        AuthFailedError: If the ID token is expired.
        ApiError: If an API error occurs.

    Returns:
        None
    """

    if errors := response.get("errors"):
        error = errors[0]
        if error["errorType"] == "UnauthorizedException":
            raise AuthFailedError(error)
        raise ApiError(error)

_dispatch(event: str, data: dict[str, Any] | None = None) -> None

Dispatch an event to listeners. For internal use only.

Parameters:

Name	Type	Description	Default
event	str	The name of the event.	required
data	dict[str, Any] | None	The data to pass to the event coroutine.	None

Source code in src/nice_go/nice_go_api.py

def _dispatch(self, event: str, data: dict[str, Any] | None = None) -> None:
    """Dispatch an event to listeners. For internal use only.

    Args:
        event (str): The name of the event.
        data (dict[str, Any] | None): The data to pass to the event coroutine.
    """
    method = f"on_{event}"

    coros = self._events.get(method, [])

    if not coros:
        _LOGGER.debug("No listeners for event %s", event)
        return

    _LOGGER.debug("Dispatching event %s", event)
    for coro in coros:
        self._schedule_event(coro, method, data)

_poll_device_ws() -> None async

Continuously polls the device WebSocket to maintain an active connection. This function will repeatedly call the poll method on the WebSocket if it is initialized.

Returns:

Type	Description
None	None

Source code in src/nice_go/nice_go_api.py

async def _poll_device_ws(self) -> None:
    """Continuously polls the device WebSocket to maintain an active connection.
    This function will repeatedly call the poll method on the WebSocket if it is
    initialized.

    Returns:
        None
    """
    if self._device_ws is None:
        return
    while True:
        await self._device_ws.poll()

_poll_events_ws() -> None async

Continuously polls the device WebSocket to maintain an active connection. This function will repeatedly call the poll method on the WebSocket if it is initialized.

Returns:

Type	Description
None	None

Source code in src/nice_go/nice_go_api.py

async def _poll_events_ws(self) -> None:
    """Continuously polls the device WebSocket to maintain an active connection.
    This function will repeatedly call the poll method on the WebSocket if it is
    initialized.

    Returns:
        None
    """

    if self._events_ws is None:
        return
    while True:
        await self._events_ws.poll()

_run_event(coro: Callable[..., Coroutine[Any, Any, Any]], event_name: str, data: dict[str, Any] | None = None) -> None async

Run an event coroutine. For internal use only.

Parameters:

Name	Type	Description	Default
coro	Callable[..., Coroutine[Any, Any, Any]]	The coroutine to run.	required
event_name	str	The name of the event.	required
data	dict[str, Any] | None	The data to pass to the event coroutine.	None

Source code in src/nice_go/nice_go_api.py

async def _run_event(
    self,
    coro: Callable[..., Coroutine[Any, Any, Any]],
    event_name: str,
    data: dict[str, Any] | None = None,
) -> None:
    """Run an event coroutine. For internal use only.

    Args:
        coro (Callable[..., Coroutine[Any, Any, Any]]): The coroutine to run.
        event_name (str): The name of the event.
        data (dict[str, Any] | None): The data to pass to the event coroutine.
    """
    kwargs = {}
    if data is not None:
        kwargs["data"] = data
    try:
        await coro(**kwargs)
    except asyncio.CancelledError:
        pass
    except Exception:
        _LOGGER.exception("Error while handling event %s", event_name)

_schedule_event(coro: Callable[..., Coroutine[Any, Any, Any]], event_name: str, data: dict[str, Any] | None) -> None

Schedule an event to be dispatched. For internal use only.

Parameters:

Name	Type	Description	Default
coro	Callable[..., Coroutine[Any, Any, Any]]	The coroutine to schedule.	required
event_name	str	The name of the event.	required
data	dict[str, Any] | None	The data to pass to the event coroutine.	required

Source code in src/nice_go/nice_go_api.py

def _schedule_event(
    self,
    coro: Callable[..., Coroutine[Any, Any, Any]],
    event_name: str,
    data: dict[str, Any] | None,
) -> None:
    """Schedule an event to be dispatched. For internal use only.

    Args:
        coro (Callable[..., Coroutine[Any, Any, Any]]): The coroutine to schedule.
        event_name (str): The name of the event.
        data (dict[str, Any] | None): The data to pass to the event coroutine.
    """
    wrapped = self._run_event(coro, event_name, data)
    task = asyncio.create_task(wrapped, name=f"NiceGO: {event_name}")
    self._event_tasks.add(task)  # See RUF006
    task.add_done_callback(self._event_tasks.discard)

authenticate(user_name: str, password: str, session: aiohttp.ClientSession) -> str | None async

Authenticate using username and password.

Parameters:

Name	Type	Description	Default
user_name	str	The username.	required
password	str	The password.	required
session	ClientSession	The client session to use.	required

Returns:

Type	Description
str | None	The refresh token.

Raises:

Type	Description
AuthFailedError	If the authentication fails.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def authenticate(
    self,
    user_name: str,
    password: str,
    session: aiohttp.ClientSession,
) -> str | None:
    """Authenticate using username and password.

    Args:
        user_name (str): The username.
        password (str): The password.
        session (aiohttp.ClientSession): The client session to use.

    Returns:
        The refresh token.

    Raises:
        AuthFailedError: If the authentication fails.
        ApiError: If an API error occurs.
    """
    self._session = session
    return await self.__authenticate(user_name, password, None)

authenticate_refresh(refresh_token: str, session: aiohttp.ClientSession) -> None async

Authenticate using a previous obtained refresh token.

Parameters:

Name	Type	Description	Default
refresh_token	str	The refresh token.	required
session	ClientSession	The client session to use.	required

Raises:

Type	Description
AuthFailedError	If the authentication fails.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def authenticate_refresh(
    self,
    refresh_token: str,
    session: aiohttp.ClientSession,
) -> None:
    """Authenticate using a previous obtained refresh token.

    Args:
        refresh_token (str): The refresh token.
        session (aiohttp.ClientSession): The client session to use.

    Raises:
        AuthFailedError: If the authentication fails.
        ApiError: If an API error occurs.
    """
    self._session = session
    await self.__authenticate(None, None, refresh_token)

close() -> None async

Close the connection.

Raises:

Type	Description
NoAuthError	If the ID token is not available.

Source code in src/nice_go/nice_go_api.py

async def close(self) -> None:
    """Close the connection.

    Raises:
        NoAuthError: If the ID token is not available.
    """

    async def _close() -> None:
        if self._device_ws:
            await self._device_ws.close()
        if self._events_ws:
            await self._events_ws.close()

    _LOGGER.debug("Closing connection")

    self._closing_task = asyncio.create_task(_close())
    await self._closing_task

close_barrier(barrier_id: str) -> bool async

Close a barrier.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to close.	required

Returns:

Type	Description
bool	Whether the barrier was closed successfully.

Source code in src/nice_go/nice_go_api.py

async def close_barrier(self, barrier_id: str) -> bool:
    """Close a barrier.

    Args:
        barrier_id (str): The ID of the barrier to close.

    Returns:
        Whether the barrier was closed successfully."""
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Closing barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template(
            "close_barrier",
            {"barrier_id": barrier_id},
        ),
    )
    data = await response.json()

    _LOGGER.debug("Closing barrier response: %s", data)
    await self._check_response_errors(data)

    result: bool = data["data"]["devicesControl"]

    return result

connect(*, reconnect: bool = True) -> None async

Connect to the WebSocket API.

Warning

No events will be dispatched until you subscribe to a receiver. Typically, you should pass the organization attribute of a barrier object to the subscribe method to start receiving data. Don't ask me why organization specifically, I don't know either.

Danger

This method will block the event loop until the connection is closed. If you want to run this method in the background, you should run it in a separate thread or process.

Parameters:

Name	Type	Description	Default
reconnect	bool	Whether to reconnect if the connection is lost.	True

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ApiError	If an API error occurs.
WebSocketError	If an error occurs while connecting.

Source code in src/nice_go/nice_go_api.py

@retry(
    wait=wait_random_exponential(multiplier=1, min=1, max=10),
    retry=_RetryIfReconnect()
    & retry_if_exception_type(
        (
            OSError,
            WebSocketError,
            aiohttp.ClientError,
            asyncio.TimeoutError,
            ReconnectWebSocketError,
        ),
    ),
    reraise=True,
    before_sleep=before_sleep_log(_LOGGER, logging.DEBUG),
)
async def connect(self, *, reconnect: bool = True) -> None:
    """Connect to the WebSocket API.

    Warning:
        No events will be dispatched until you subscribe to a receiver.
        Typically, you should pass the `organization` attribute of a barrier
        object to the `subscribe` method to start receiving data. Don't ask me
        why `organization` specifically, I don't know either.

    Danger:
        This method will block the event loop until the connection is closed.
        If you want to run this method in the background, you should run it in a
        separate thread or process.

    Args:
        reconnect (bool): Whether to reconnect if the connection is lost.

    Raises:
        NoAuthError: If the ID token is not available.
        ApiError: If an API error occurs.
        WebSocketError: If an error occurs while connecting.
    """
    try:
        if self.id_token is None:
            raise NoAuthError

        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)

        self._reconnect = reconnect

        device_url = self._endpoints["GraphQL"]["device"]["wss"]
        events_url = self._endpoints["GraphQL"]["events"]["wss"]

        _LOGGER.debug("Connecting to WebSocket API %s", device_url)

        self._device_ws = WebSocketClient(client_session=self._session)
        await self._device_ws.connect(
            self.id_token,
            yarl.URL(device_url),
            "device",
            self._dispatch,
            yarl.URL(self._endpoints["GraphQL"]["device"]["https"]).host,
        )
        self._events_ws = WebSocketClient(client_session=self._session)
        await self._events_ws.connect(
            self.id_token,
            yarl.URL(events_url),
            "events",
            self._dispatch,
            yarl.URL(self._endpoints["GraphQL"]["events"]["https"]).host,
        )

        _LOGGER.debug("Connected to WebSocket API")

        device_task = asyncio.create_task(self._poll_device_ws())
        events_task = asyncio.create_task(self._poll_events_ws())

        with contextlib.suppress(asyncio.CancelledError):
            done, pending = await asyncio.wait(
                [device_task, events_task],
                return_when=asyncio.FIRST_EXCEPTION,
            )

        with contextlib.suppress(UnboundLocalError):
            if exceptions := [
                task.exception() for task in done if task.exception()
            ]:
                for p in pending:
                    p.cancel()
                # Make sure both WS are closed
                await self._events_ws.close()
                await self._device_ws.close()
                raise exceptions[0]  # type: ignore[misc]
    except (
        OSError,
        WebSocketError,
        aiohttp.ClientError,
        asyncio.TimeoutError,
        ReconnectWebSocketError,
    ) as e:
        self._dispatch("connection_lost", {"exception": e})
        self._device_connected = False
        self._events_connected = False
        if not reconnect:
            _LOGGER.debug("Connection lost, not reconnecting")
            await self.close()
            raise

        if self.closed:
            return

        _LOGGER.debug("Connection lost, retrying...")

        # Raising triggers retry
        raise

event(coro: CoroT) -> CoroT

Decorator to add an event listener. Just a wrapper around listen.

Info

This can only decorate coroutine functions.

Parameters:

Name	Type	Description	Default
coro	CoroT	The coroutine function to decorate.	required

Examples:

You can use this decorator to add event listeners to the API object. Some events include but are not limited to:

• connection_lost: Triggered when the connection to the WebSocket API is lost.
• connected: Triggered when the connection to the WebSocket API is established.
• data: Triggered when data is received from an active subscription. See subscribe.

>>> @api.event
... async def on_data(
...     data: dict[str, Any] | None = None,
... ) -> None:
...     if data is not None:
...         print(data)

Source code in src/nice_go/nice_go_api.py

def event(self, coro: CoroT) -> CoroT:
    """Decorator to add an event listener. Just a wrapper around `listen`.

    Info:
        This can only decorate coroutine functions.

    Args:
        coro (CoroT): The coroutine function to decorate.

    Examples:
        You can use this decorator to add event listeners to the API object.
        Some events include but are not limited to:

        - `connection_lost`: Triggered when the connection to the WebSocket API is
            lost.
        - `connected`: Triggered when the connection to the WebSocket API is
            established.
        - `data`: Triggered when data is received from an active subscription.
            See `subscribe`.

        >>> @api.event
        ... async def on_data(
        ...     data: dict[str, Any] | None = None,
        ... ) -> None:
        ...     if data is not None:
        ...         print(data)
    """
    self.listen(coro.__name__, coro)
    return coro

get_all_barriers() -> list[Barrier] async

Get all barriers.

Returns:

Type	Description
list[Barrier]	A list of all barriers.

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def get_all_barriers(self) -> list[Barrier]:
    """Get all barriers.

    Returns:
        A list of all barriers.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Getting all barriers")
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template("get_all_barriers", None),
    )
    data = await response.json()

    _LOGGER.debug("Got all barriers")
    _LOGGER.debug("Data: %s", data)

    await self._check_response_errors(data)

    barriers = []

    for device in data["data"]["devicesListAll"]["devices"]:
        if device["state"]["connectionState"] is not None:
            connection_state = ConnectionState(
                device["state"]["connectionState"]["connected"],
                device["state"]["connectionState"]["updatedTimestamp"],
            )
        else:
            connection_state = None
        barrier_state = BarrierState(
            device["state"]["deviceId"],
            json.loads(device["state"]["reported"]),
            device["state"]["timestamp"],
            device["state"]["version"],
            connection_state,
        )
        barrier = Barrier(
            device["id"],
            device["type"],
            device["controlLevel"],
            device["attr"],
            barrier_state,
            self,
        )
        barriers.append(barrier)

    return barriers

light_off(barrier_id: str) -> bool async

Turn the light off.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to turn the light off.	required

Returns:

Type	Description
bool	Whether the light was turned off successfully.

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def light_off(self, barrier_id: str) -> bool:
    """Turn the light off.

    Args:
        barrier_id (str): The ID of the barrier to turn the light off.

    Returns:
        Whether the light was turned off successfully.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Turning light off for barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template("light_off", {"barrier_id": barrier_id}),
    )
    data = await response.json()

    _LOGGER.debug("Turning light off response: %s", data)
    await self._check_response_errors(data)

    result: bool = data["data"]["devicesControl"]

    return result

light_on(barrier_id: str) -> bool async

Turn the light on.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to turn the light on.	required

Returns:

Type	Description
bool	Whether the light was turned on successfully.

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def light_on(self, barrier_id: str) -> bool:
    """Turn the light on.

    Args:
        barrier_id (str): The ID of the barrier to turn the light on.

    Returns:
        Whether the light was turned on successfully.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Turning light on for barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template("light_on", {"barrier_id": barrier_id}),
    )
    data = await response.json()

    _LOGGER.debug("Turning light on response: %s", data)
    await self._check_response_errors(data)

    result: bool = data["data"]["devicesControl"]

    return result

listen(event_name: str, coro: CoroT) -> Callable[[], None]

Add an event listener.

Parameters:

Name	Type	Description	Default
event_name	str	The name of the event.	required
coro	CoroT	The coroutine to run when the event is dispatched.	required

Returns:

Type	Description
Callable[[], None]	A function to remove the event listener.

Examples:

You can use this method to add event listeners to the API object. Some events include but are not limited to:

• connection_lost: Triggered when the connection to the WebSocket API is lost.
• connected: Triggered when the connection to the WebSocket API is established.
• data: Triggered when data is received from an active subscription. See subscribe.

>>> def on_data(data: dict[str, Any] | None = None) -> None:
...     if data is not None:
...         print(data)
...
>>> remove_listener = api.listen("data", on_data)

Source code in src/nice_go/nice_go_api.py

def listen(self, event_name: str, coro: CoroT) -> Callable[[], None]:
    """Add an event listener.

    Args:
        event_name (str): The name of the event.
        coro (CoroT): The coroutine to run when the event is dispatched.

    Returns:
        A function to remove the event listener.

    Examples:
        You can use this method to add event listeners to the API object.
        Some events include but are not limited to:

        - `connection_lost`: Triggered when the connection to the WebSocket API is
            lost.
        - `connected`: Triggered when the connection to the WebSocket API is
            established.
        - `data`: Triggered when data is received from an active subscription.
            See `subscribe`.

        >>> def on_data(data: dict[str, Any] | None = None) -> None:
        ...     if data is not None:
        ...         print(data)
        ...
        >>> remove_listener = api.listen("data", on_data)
    """
    if not asyncio.iscoroutinefunction(coro):
        msg = "The decorated function must be a coroutine"
        raise TypeError(msg)

    _LOGGER.debug("Adding event listener %s", coro.__name__)

    self._events.setdefault(event_name, []).append(coro)
    return lambda: self._events[event_name].remove(coro)

on_device_connected() -> None async

Handle the device connected event.

Source code in src/nice_go/nice_go_api.py

async def on_device_connected(self) -> None:
    """Handle the device connected event."""
    self._device_connected = True
    if self._device_connected and self._events_connected:
        # Only dispatch when both feeds are connected
        self._dispatch("connected")

on_events_connected() -> None async

Handle the events connected event.

Source code in src/nice_go/nice_go_api.py

async def on_events_connected(self) -> None:
    """Handle the events connected event."""
    self._events_connected = True
    if self._device_connected and self._events_connected:
        # Only dispatch when both feeds are connected
        self._dispatch("connected")

open_barrier(barrier_id: str) -> bool async

Open a barrier.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to open.	required

Returns:

Type	Description
bool	Whether the barrier was opened successfully.

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def open_barrier(self, barrier_id: str) -> bool:
    """Open a barrier.

    Args:
        barrier_id (str): The ID of the barrier to open.

    Returns:
        Whether the barrier was opened successfully.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Opening barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template("open_barrier", {"barrier_id": barrier_id}),
    )
    data = await response.json()

    _LOGGER.debug("Opening barrier response: %s", data)
    await self._check_response_errors(data)

    result: bool = data["data"]["devicesControl"]

    return result

subscribe(receiver: str) -> list[str] async

Subscribe to a receiver.

Parameters:

Name	Type	Description	Default
receiver	str	The receiver to subscribe to.	required

Returns:

Type	Description
list[str]	The subscription IDs. You can pass this into the unsubscribe method to unsubscribe from the receiver.

Raises:

Type	Description
WebSocketError	If no WebSocket connection is available.

Source code in src/nice_go/nice_go_api.py

async def subscribe(self, receiver: str) -> list[str]:
    """Subscribe to a receiver.

    Args:
        receiver (str): The receiver to subscribe to.

    Returns:
        The subscription IDs. You can pass this into the `unsubscribe` method to
            unsubscribe from the receiver.

    Raises:
        WebSocketError: If no WebSocket connection is available.
    """
    if self._device_ws is None:
        msg = "No WebSocket connection"
        raise WebSocketError(msg)
    if self._events_ws is None:
        msg = "No WebSocket connection"
        raise WebSocketError(msg)

    _LOGGER.debug("Subscribing to receiver %s", receiver)

    return [
        await self._device_ws.subscribe(receiver),
        await self._events_ws.subscribe(receiver),
    ]

unsubscribe(subscription_id: str) -> None async

Unsubscribe from a receiver.

Parameters:

Name	Type	Description	Default
subscription_id	str	The subscription ID to unsubscribe from.	required

Raises:

Type	Description
WebSocketError	If no WebSocket connection is available

Source code in src/nice_go/nice_go_api.py

async def unsubscribe(self, subscription_id: str) -> None:
    """Unsubscribe from a receiver.

    Args:
        subscription_id (str): The subscription ID to unsubscribe from.

    Raises:
        WebSocketError: If no WebSocket connection is available
    """
    if self._device_ws is None:
        msg = "No WebSocket connection"
        raise WebSocketError(msg)
    if self._events_ws is None:
        msg = "No WebSocket connection"
        raise WebSocketError(msg)

    _LOGGER.debug("Unsubscribing from subscription %s", subscription_id)

    await self._device_ws.unsubscribe(subscription_id)
    await self._events_ws.unsubscribe(subscription_id)

vacation_mode_off(barrier_id: str) -> None async

Turn vacation mode off.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to turn vacation mode off.	required

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def vacation_mode_off(self, barrier_id: str) -> None:
    """Turn vacation mode off.

    Args:
        barrier_id (str): The ID of the barrier to turn vacation mode off.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Turning vacation mode off for barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template(
            "vacation_mode_off",
            {"barrier_id": barrier_id},
        ),
    )
    await response.json()

vacation_mode_on(barrier_id: str) -> None async

Turn vacation mode on.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to turn vacation mode on.	required

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def vacation_mode_on(self, barrier_id: str) -> None:
    """Turn vacation mode on.

    Args:
        barrier_id (str): The ID of the barrier to turn vacation mode on.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Turning vacation mode on for barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template(
            "vacation_mode_on",
            {"barrier_id": barrier_id},
        ),
    )
    await response.json()

NiceGOError

Bases: Exception

Base exception for Nice G.O. API.

Source code in src/nice_go/_exceptions.py

class NiceGOError(Exception):
    """Base exception for Nice G.O. API."""

NoAuthError

Bases: NiceGOError

Not authenticated exception.

Source code in src/nice_go/_exceptions.py

class NoAuthError(NiceGOError):
    """Not authenticated exception."""

WebSocketError

Bases: NiceGOError

WebSocket error.

Source code in src/nice_go/_exceptions.py

class WebSocketError(NiceGOError):
    """WebSocket error."""

_authentication_tokens

Holds the tokens retrieved from authentication.

AuthenticationTokens

A class to handle authentication tokens.

Initializes an AuthenticationTokens object with the provided data dictionary. It extracts the 'IdToken' and 'RefreshToken' if available, storing them as attributes.

Parameters:

Name	Type	Description	Default
data	dict[str, str]	A dictionary containing authentication tokens.	required

Attributes:

Name	Type	Description
id_token	str	The IdToken retrieved from authentication.
refresh_token	str | None	The RefreshToken retrieved from authentication. If no RefreshToken is available, it is set to None.

Source code in src/nice_go/_authentication_tokens.py

class AuthenticationTokens:
    """
    A class to handle authentication tokens.

    Initializes an AuthenticationTokens object with the provided data dictionary. It
    extracts the 'IdToken' and 'RefreshToken' if available, storing them as attributes.

    Args:
        data (dict[str, str]): A dictionary containing authentication tokens.

    Attributes:
        id_token (str): The IdToken retrieved from authentication.
        refresh_token (str | None): The RefreshToken retrieved from authentication. If
            no RefreshToken is available, it is set to None.

    """

    def __init__(self, data: dict[str, str]) -> None:
        """Initialize the AuthenticationTokens object."""
        self.id_token = data["IdToken"]
        self.refresh_token: str | None = None
        try:
            self.refresh_token = data["RefreshToken"]
        except KeyError:
            self.refresh_token = None

__init__(data: dict[str, str]) -> None

Initialize the AuthenticationTokens object.

Source code in src/nice_go/_authentication_tokens.py

def __init__(self, data: dict[str, str]) -> None:
    """Initialize the AuthenticationTokens object."""
    self.id_token = data["IdToken"]
    self.refresh_token: str | None = None
    try:
        self.refresh_token = data["RefreshToken"]
    except KeyError:
        self.refresh_token = None

_aws_cognito_authenticator

AWS Cognito authentication and identity management.

This module provides a class to handle AWS Cognito authentication and identity management.

Info

You do not need to use this module directly. It is used by the nice_go_api module to authenticate with AWS Cognito.

AwsCognitoAuthenticator

Handles AWS Cognito authentication and identity management.

This class provides methods to authenticate with AWS Cognito and retrieve authentication tokens. It can be used to refresh tokens or to get new tokens by providing a username and password.

Parameters:

Name	Type	Description	Default
region_name	str	The AWS region name.	required
client_id	str	The AWS Cognito client ID.	required
pool_id	str	The AWS Cognito pool ID.	required
identity_pool_id	str	The AWS Cognito identity pool ID.	required

Attributes:

Name	Type	Description
region_name	str	The AWS region name.
identity_pool_id	str	The AWS Cognito identity pool ID.
client_id	str	The AWS Cognito client ID.
pool_id	str	The AWS Cognito pool ID.
session	Session	The botocore session object.

Source code in src/nice_go/_aws_cognito_authenticator.py

class AwsCognitoAuthenticator:
    """Handles AWS Cognito authentication and identity management.

    This class provides methods to authenticate with AWS Cognito and retrieve
    authentication tokens. It can be used to refresh tokens or to get new tokens
    by providing a username and password.

    Args:
        region_name (str): The AWS region name.
        client_id (str): The AWS Cognito client ID.
        pool_id (str): The AWS Cognito pool ID.
        identity_pool_id (str): The AWS Cognito identity pool ID.

    Attributes:
        region_name (str): The AWS region name.
        identity_pool_id (str): The AWS Cognito identity pool ID.
        client_id (str): The AWS Cognito client ID.
        pool_id (str): The AWS Cognito pool ID.
        session (botocore.session.Session): The botocore session object.
    """

    def __init__(
        self,
        region_name: str,
        client_id: str,
        pool_id: str,
        identity_pool_id: str,
    ) -> None:
        """Initialize the AwsCognitoAuthenticator object."""
        self.region_name = region_name
        self.identity_pool_id = identity_pool_id
        self.client_id = client_id
        self.pool_id = pool_id
        self.session = get_session()

    def refresh_token(self, refresh_token: str) -> AuthenticationTokens:
        """Regenerates the token by providing a refresh token.

        Args:
            refresh_token (str): The refresh token.

        Returns:
            The new authentication tokens.
        """
        _LOGGER.debug("Refreshing token")
        cognito_identity_provider = boto3.client("cognito-idp", self.region_name)
        resp = cognito_identity_provider.initiate_auth(
            AuthFlow="REFRESH_TOKEN_AUTH",
            AuthParameters={
                "REFRESH_TOKEN": refresh_token,
            },
            ClientId=self.client_id,
        )
        _LOGGER.debug("Token refreshed")
        return AuthenticationTokens(resp["AuthenticationResult"])

    def get_new_token(self, username: str, password: str) -> AuthenticationTokens:
        """Gets the initial token by providing username and password.

        Args:
            username (str): The username.
            password (str): The password.

        Returns:
            The new authentication tokens.
        """
        _LOGGER.debug("Getting new token")
        cognito_identity_provider = boto3.client("cognito-idp", self.region_name)
        # Start the authentication flow
        aws_srp = AWSSRP(
            username=username,
            password=password,
            pool_id=self.pool_id,
            client_id=self.client_id,
            client=cognito_identity_provider,
        )

        _LOGGER.debug("Initiating auth")

        auth_params = aws_srp.get_auth_params()
        resp = cognito_identity_provider.initiate_auth(
            AuthFlow="USER_SRP_AUTH",
            AuthParameters=auth_params,
            ClientId=self.client_id,
        )

        _LOGGER.debug("Auth initiated, responding to challenge")

        challenge_response = aws_srp.process_challenge(
            resp["ChallengeParameters"],
            auth_params,
        )

        # Respond to PASSWORD_VERIFIER
        resp = cognito_identity_provider.respond_to_auth_challenge(
            ClientId=self.client_id,
            ChallengeName="PASSWORD_VERIFIER",
            ChallengeResponses=challenge_response,
        )

        _LOGGER.debug("Token received")

        return AuthenticationTokens(resp["AuthenticationResult"])

__init__(region_name: str, client_id: str, pool_id: str, identity_pool_id: str) -> None

Initialize the AwsCognitoAuthenticator object.

Source code in src/nice_go/_aws_cognito_authenticator.py

def __init__(
    self,
    region_name: str,
    client_id: str,
    pool_id: str,
    identity_pool_id: str,
) -> None:
    """Initialize the AwsCognitoAuthenticator object."""
    self.region_name = region_name
    self.identity_pool_id = identity_pool_id
    self.client_id = client_id
    self.pool_id = pool_id
    self.session = get_session()

get_new_token(username: str, password: str) -> AuthenticationTokens

Gets the initial token by providing username and password.

Parameters:

Name	Type	Description	Default
username	str	The username.	required
password	str	The password.	required

Returns:

Type	Description
AuthenticationTokens	The new authentication tokens.

Source code in src/nice_go/_aws_cognito_authenticator.py

def get_new_token(self, username: str, password: str) -> AuthenticationTokens:
    """Gets the initial token by providing username and password.

    Args:
        username (str): The username.
        password (str): The password.

    Returns:
        The new authentication tokens.
    """
    _LOGGER.debug("Getting new token")
    cognito_identity_provider = boto3.client("cognito-idp", self.region_name)
    # Start the authentication flow
    aws_srp = AWSSRP(
        username=username,
        password=password,
        pool_id=self.pool_id,
        client_id=self.client_id,
        client=cognito_identity_provider,
    )

    _LOGGER.debug("Initiating auth")

    auth_params = aws_srp.get_auth_params()
    resp = cognito_identity_provider.initiate_auth(
        AuthFlow="USER_SRP_AUTH",
        AuthParameters=auth_params,
        ClientId=self.client_id,
    )

    _LOGGER.debug("Auth initiated, responding to challenge")

    challenge_response = aws_srp.process_challenge(
        resp["ChallengeParameters"],
        auth_params,
    )

    # Respond to PASSWORD_VERIFIER
    resp = cognito_identity_provider.respond_to_auth_challenge(
        ClientId=self.client_id,
        ChallengeName="PASSWORD_VERIFIER",
        ChallengeResponses=challenge_response,
    )

    _LOGGER.debug("Token received")

    return AuthenticationTokens(resp["AuthenticationResult"])

refresh_token(refresh_token: str) -> AuthenticationTokens

Regenerates the token by providing a refresh token.

Parameters:

Name	Type	Description	Default
refresh_token	str	The refresh token.	required

Returns:

Type	Description
AuthenticationTokens	The new authentication tokens.

Source code in src/nice_go/_aws_cognito_authenticator.py

def refresh_token(self, refresh_token: str) -> AuthenticationTokens:
    """Regenerates the token by providing a refresh token.

    Args:
        refresh_token (str): The refresh token.

    Returns:
        The new authentication tokens.
    """
    _LOGGER.debug("Refreshing token")
    cognito_identity_provider = boto3.client("cognito-idp", self.region_name)
    resp = cognito_identity_provider.initiate_auth(
        AuthFlow="REFRESH_TOKEN_AUTH",
        AuthParameters={
            "REFRESH_TOKEN": refresh_token,
        },
        ClientId=self.client_id,
    )
    _LOGGER.debug("Token refreshed")
    return AuthenticationTokens(resp["AuthenticationResult"])

_barrier

Module containing classes for barriers.

This module contains classes for barriers and their states. The Barrier class provides methods to interact with the barrier, such as opening and closing it, and turning the light on and off.

Classes:

Name	Description
ConnectionState	Represents the connection state of a barrier.
BarrierState	Represents the state of a barrier.
Barrier	Represents a barrier.

Barrier dataclass

Class representing a barrier.

Attributes:

Name	Type	Description
id	str	The ID of the barrier.
type	str	The type of the barrier.
controlLevel	str	The control level of the barrier.
attr	list[dict[str, str]]	A list of attributes of the barrier.
state	BarrierState	The state of the barrier.
api	NiceGOApi	The NiceGO API object.

Methods:

Name	Description
open	Open the barrier.
close	Close the barrier.
light_on	Turn on the light of the barrier.
light_off	Turn off the light of the barrier.
get_attr	Get the value of an attribute.

Source code in src/nice_go/_barrier.py

@dataclass
class Barrier:
    """Class representing a barrier.

    Attributes:
        id (str): The ID of the barrier.
        type (str): The type of the barrier.
        controlLevel (str): The control level of the barrier.
        attr (list[dict[str, str]]): A list of attributes of the barrier.
        state (BarrierState): The state of the barrier.
        api (NiceGOApi): The NiceGO API object.

    Methods:
        open: Open the barrier.
        close: Close the barrier.
        light_on: Turn on the light of the barrier.
        light_off: Turn off the light of the barrier.
        get_attr: Get the value of an attribute.
    """

    id: str
    type: str
    controlLevel: str  # noqa: N815
    attr: list[dict[str, str]]
    state: BarrierState
    api: NiceGOApi

    async def open(self) -> bool:
        """Open the barrier.

        Returns:
            A boolean indicating whether the command was successful.
        """
        return await self.api.open_barrier(self.id)

    async def close(self) -> bool:
        """Close the barrier.

        Returns:
            A boolean indicating whether the command was successful.
        """
        return await self.api.close_barrier(self.id)

    async def light_on(self) -> bool:
        """Turn on the light of the barrier.

        Returns:
            A boolean indicating whether the command was successful.
        """
        return await self.api.light_on(self.id)

    async def light_off(self) -> bool:
        """Turn off the light of the barrier.

        Returns:
            A boolean indicating whether the command was successful.
        """
        return await self.api.light_off(self.id)

    async def get_attr(self, key: str) -> str:
        """Get the value of an attribute.

        Args:
            key (str): The key of the attribute.

        Returns:
            The value of the attribute.

        Raises:
            KeyError: If the attribute with the given key is not found.
        """
        attr = next((attr for attr in self.attr if attr["key"] == key), None)
        if attr is None:
            msg = f"Attribute with key {key} not found."
            raise KeyError(msg)
        return attr["value"]

    async def vacation_mode_on(self) -> None:
        """Turn on vacation mode for the barrier."""
        await self.api.vacation_mode_on(self.id)

    async def vacation_mode_off(self) -> None:
        """Turn off vacation mode for the barrier."""
        await self.api.vacation_mode_off(self.id)

close() -> bool async

Close the barrier.

Returns:

Type	Description
bool	A boolean indicating whether the command was successful.

Source code in src/nice_go/_barrier.py

async def close(self) -> bool:
    """Close the barrier.

    Returns:
        A boolean indicating whether the command was successful.
    """
    return await self.api.close_barrier(self.id)

get_attr(key: str) -> str async

Get the value of an attribute.

Parameters:

Name	Type	Description	Default
key	str	The key of the attribute.	required

Returns:

Type	Description
str	The value of the attribute.

Raises:

Type	Description
KeyError	If the attribute with the given key is not found.

Source code in src/nice_go/_barrier.py

async def get_attr(self, key: str) -> str:
    """Get the value of an attribute.

    Args:
        key (str): The key of the attribute.

    Returns:
        The value of the attribute.

    Raises:
        KeyError: If the attribute with the given key is not found.
    """
    attr = next((attr for attr in self.attr if attr["key"] == key), None)
    if attr is None:
        msg = f"Attribute with key {key} not found."
        raise KeyError(msg)
    return attr["value"]

light_off() -> bool async

Turn off the light of the barrier.

Returns:

Type	Description
bool	A boolean indicating whether the command was successful.

Source code in src/nice_go/_barrier.py

async def light_off(self) -> bool:
    """Turn off the light of the barrier.

    Returns:
        A boolean indicating whether the command was successful.
    """
    return await self.api.light_off(self.id)

light_on() -> bool async

Turn on the light of the barrier.

Returns:

Type	Description
bool	A boolean indicating whether the command was successful.

Source code in src/nice_go/_barrier.py

async def light_on(self) -> bool:
    """Turn on the light of the barrier.

    Returns:
        A boolean indicating whether the command was successful.
    """
    return await self.api.light_on(self.id)

open() -> bool async

Open the barrier.

Returns:

Type	Description
bool	A boolean indicating whether the command was successful.

Source code in src/nice_go/_barrier.py

async def open(self) -> bool:
    """Open the barrier.

    Returns:
        A boolean indicating whether the command was successful.
    """
    return await self.api.open_barrier(self.id)

vacation_mode_off() -> None async

Turn off vacation mode for the barrier.

Source code in src/nice_go/_barrier.py

async def vacation_mode_off(self) -> None:
    """Turn off vacation mode for the barrier."""
    await self.api.vacation_mode_off(self.id)

vacation_mode_on() -> None async

Turn on vacation mode for the barrier.

Source code in src/nice_go/_barrier.py

async def vacation_mode_on(self) -> None:
    """Turn on vacation mode for the barrier."""
    await self.api.vacation_mode_on(self.id)

BarrierState dataclass

Class representing the state of a barrier.

Attributes:

Name	Type	Description
deviceId	str	The device ID of the barrier.
reported	dict[str, Any]	The reported state of the barrier.
timestamp	str	The timestamp of the last update.
version	str	The version of the barrier.
connectionState	ConnectionState | None	The connection state of the barrier.

Source code in src/nice_go/_barrier.py

@dataclass
class BarrierState:
    """Class representing the state of a barrier.

    Attributes:
        deviceId (str): The device ID of the barrier.
        reported (dict[str, Any]): The reported state of the barrier.
        timestamp (str): The timestamp of the last update.
        version (str): The version of the barrier.
        connectionState (ConnectionState | None): The connection state of the barrier.
    """

    deviceId: str  # noqa: N815
    reported: dict[str, Any]
    timestamp: str
    version: str
    connectionState: ConnectionState | None  # noqa: N815

ConnectionState dataclass

Class representing the connection state of a barrier.

Attributes:

Name	Type	Description
connected	bool	A boolean indicating whether the barrier is connected.
updatedTimestamp	datetime	The timestamp of the last update.

Source code in src/nice_go/_barrier.py

@dataclass
class ConnectionState:
    """Class representing the connection state of a barrier.

    Attributes:
        connected (bool): A boolean indicating whether the barrier is connected.
        updatedTimestamp (datetime): The timestamp of the last update.
    """

    connected: bool
    updatedTimestamp: datetime  # noqa: N815

_const

Constants for Nice G.O. API.

_exceptions

Exceptions for Nice G.O. API.

ApiError

Bases: NiceGOError

API error.

Source code in src/nice_go/_exceptions.py

class ApiError(NiceGOError):
    """API error."""

AuthFailedError

Bases: NiceGOError

Authentication failed. Check your credentials.

Source code in src/nice_go/_exceptions.py

class AuthFailedError(NiceGOError):
    """Authentication failed. Check your credentials."""

NiceGOError

Bases: Exception

Base exception for Nice G.O. API.

Source code in src/nice_go/_exceptions.py

class NiceGOError(Exception):
    """Base exception for Nice G.O. API."""

NoAuthError

Bases: NiceGOError

Not authenticated exception.

Source code in src/nice_go/_exceptions.py

class NoAuthError(NiceGOError):
    """Not authenticated exception."""

ReconnectWebSocketError

Bases: WebSocketError

Reconnect WebSocket error.

Source code in src/nice_go/_exceptions.py

class ReconnectWebSocketError(WebSocketError):
    """Reconnect WebSocket error."""

WebSocketError

Bases: NiceGOError

WebSocket error.

Source code in src/nice_go/_exceptions.py

class WebSocketError(NiceGOError):
    """WebSocket error."""

_util

Utilities for the nice_go package.

get_request_template(request_name: str, arguments: dict[str, str] | None) -> Any async

Get a request template with optional arguments.

Parameters:

Name	Type	Description	Default
request_name	str	The name of the request template.	required
arguments	dict[str, str] | None	Optional arguments to replace in the template.	required

Returns:

Type	Description
Any	The request template with optional arguments.

Source code in src/nice_go/_util.py

async def get_request_template(
    request_name: str,
    arguments: dict[str, str] | None,
) -> Any:
    """Get a request template with optional arguments.

    Args:
        request_name: The name of the request template.
        arguments: Optional arguments to replace in the template.

    Returns:
        The request template with optional arguments.
    """
    template = json.dumps(REQUEST_TEMPLATES[request_name])
    if arguments:
        for key, value in arguments.items():
            template = template.replace(f"${key}", value)
    return json.loads(template)

_ws_client

This module contains the WebSocketClient class, which is used to interact with the WebSocket server.

Classes:

Name	Description
WebSocketClient	A class that represents a WebSocket client.

EventListener

Bases: NamedTuple

A class representing an event listener.

Source code in src/nice_go/_ws_client.py

class EventListener(NamedTuple):
    """A class representing an event listener."""

    predicate: Callable[[dict[str, Any]], bool] | None
    event: str
    result: Callable[[dict[str, Any]], Any] | None
    future: asyncio.Future[Any]

WebSocketClient

A class that represents a WebSocket client.

Attributes:

Name	Type	Description
ws	ClientWebSocketResponse | None	The WebSocket connection.
_dispatch_listeners	list[EventListener]	A list of event listeners.
_subscriptions	list[str]	A list of subscription IDs.

Source code in src/nice_go/_ws_client.py

class WebSocketClient:
    """A class that represents a WebSocket client.

    Attributes:
        ws (aiohttp.ClientWebSocketResponse | None): The WebSocket connection.
        _dispatch_listeners (list[EventListener]): A list of event listeners.
        _subscriptions (list[str]): A list of subscription IDs.
    """

    def __init__(self, client_session: aiohttp.ClientSession) -> None:
        """Initialize the WebSocketClient."""
        self.ws: aiohttp.ClientWebSocketResponse | None = None
        self._dispatch_listeners: list[EventListener] = []
        self._subscriptions: list[str] = []
        self.client_session = client_session
        self.reconnecting = False
        self._timeout_task: asyncio.Task[None] | None = None

    def _redact_message(self, message: str | dict[str, Any]) -> Any:
        """Redact sensitive information from a message.

        Args:
            message: The message to redact.

        Returns:
            The redacted message.
        """
        if isinstance(message, dict):
            return json.loads(json.dumps(message).replace(self.id_token, "<REDACTED>"))
        return message.replace(self.id_token, "<REDACTED>")

    async def _watch_keepalive(self) -> None:
        """A task that handles the timeout for the WebSocket connection.

        Raises:
            WebSocketError: If the WebSocket connection is closed.
        """
        if self.ws is None or self.ws.closed:
            msg = "WebSocket connection is closed"
            raise WebSocketError(msg)
        await asyncio.sleep(self._timeout / 1000)
        _LOGGER.debug("WebSocket keepalive timeout reached, reconnecting")
        await self._reconnect()

    async def _reconnect(self) -> None:
        """Reconnect to the WebSocket server.

        Raises:
            WebSocketError: If the WebSocket connection is closed or an error occurs.
        """
        if self.ws is None or self.ws.closed:
            msg = "WebSocket connection is closed"
            raise WebSocketError(msg)
        self.reconnecting = True
        _LOGGER.debug("Reconnecting to WebSocket server")
        await self.close()
        raise ReconnectWebSocketError

    async def connect(
        self,
        id_token: str,
        endpoint: yarl.URL,
        api_type: str,
        dispatch: Callable[[str, dict[str, Any] | None], None],
        host: str | None = None,
    ) -> None:
        """Connect to the WebSocket server.

        Args:
            client_session: The aiohttp ClientSession.
            id_token: The IdToken retrieved from authentication.
            endpoint: The endpoint URL.
            dispatch: The dispatch function.
            host: The host URL.

        Raises:
            ValueError: If host is not provided.
            WebSocketError: If the WebSocket connection is closed or an error occurs.
        """
        if host is None:
            msg = "host must be provided"
            raise ValueError(msg)

        self._dispatch = dispatch
        self.id_token = id_token
        self.host = host
        self.api_type = api_type  # Should be "device" or "events"
        self._endpoint = endpoint

        raw_header = {
            "Authorization": id_token,
            "host": host,
        }
        # Base64 encode the header
        header = base64.b64encode(json.dumps(raw_header).encode()).decode()
        # Construct the URL
        url = endpoint.with_query({"header": header, "payload": "e30="})

        # URL contains sensitive information, so we don't want to log it
        _LOGGER.debug("Connecting to WebSocket server at %s", endpoint)

        headers = {"sec-websocket-protocol": "graphql-ws"}
        self.ws = await self.client_session.ws_connect(url, headers=headers)

        await self.init()

    async def init(self) -> None:
        """Initialize the WebSocket connection.

        Raises:
            WebSocketError: If the WebSocket connection is closed or an error occurs.
        """
        if self.ws is None or self.ws.closed:
            msg = "WebSocket connection is closed"
            raise WebSocketError(msg)
        _LOGGER.debug("Initializing WebSocket connection")
        await self.send({"type": "connection_init"})
        try:
            _LOGGER.debug("Waiting for connection_ack")
            message = await self.ws.receive(timeout=10)
            data = json.loads(message.data)
            _LOGGER.debug("Received message: %s", data)
            if data["type"] != "connection_ack":
                msg = f'Expected connection_ack, but received {data["type"]}'
                raise WebSocketError(
                    msg,
                )
        except asyncio.TimeoutError as e:
            msg = "Connection to the websocket server timed out"
            raise WebSocketError(msg) from e
        _LOGGER.debug("Received connection_ack, WebSocket connection established")

        self._timeout = data.get("payload", {}).get("timeout", 300000)
        self._timeout_task = asyncio.create_task(self._watch_keepalive())
        self._dispatch(f"{self.api_type}_connected", None)

    async def send(self, message: str | dict[str, Any]) -> None:
        """Send a message to the WebSocket server.

        Args:
            message: The message to send.

        Raises:
            WebSocketError: If the WebSocket connection is closed
        """
        if self.ws is None or self.ws.closed:
            msg = "WebSocket connection is closed"
            raise WebSocketError(msg)
        redacted_message = self._redact_message(message)
        _LOGGER.debug("Sending message: %s", redacted_message)
        if isinstance(message, dict):
            await self.ws.send_json(message)
        else:
            await self.ws.send_str(message)

    async def close(self) -> None:
        """Close the WebSocket connection.

        Raises:
            WebSocketError: If the WebSocket connection is closed
        """
        if self.ws is None or self.ws.closed:
            return
        _LOGGER.debug("Closing WebSocket client")
        # Unsubscribe from all subscriptions
        for subscription_id in self._subscriptions:
            _LOGGER.debug("Unsubscribing from subscription %s", subscription_id)
            await self.unsubscribe(subscription_id)
        if self._timeout_task is not None and not self._timeout_task.done():
            _LOGGER.debug("Cancelling keepalive task")
            # Cancel the keepalive task
            self._timeout_task.cancel()
            try:
                await self._timeout_task
            except asyncio.CancelledError:
                _LOGGER.debug("Keepalive task was cancelled")
            except Exception:
                _LOGGER.exception("Exception occurred while cancelling keepalive task")
        _LOGGER.debug("Closing WebSocket connection")
        await self.ws.close()
        _LOGGER.debug("WebSocket connection closed")

    async def poll(self) -> None:
        """Poll the WebSocket connection for messages.

        Raises:
            WebSocketError: If the WebSocket connection is closed or an error occurs.
        """
        if self.ws is None or self.ws.closed:
            error_msg = "WebSocket connection is closed"
            raise WebSocketError(error_msg)
        msg = await self.ws.receive(timeout=300.0)
        if msg.type == aiohttp.WSMsgType.TEXT:
            await self.received_message(msg.data)
        elif msg.type == aiohttp.WSMsgType.ERROR:
            error_msg = f"WebSocket connection closed with error {msg}"
            raise WebSocketError(error_msg)
        elif msg.type in (
            aiohttp.WSMsgType.CLOSE,
            aiohttp.WSMsgType.CLOSING,
            aiohttp.WSMsgType.CLOSED,
        ):
            error_msg = "WebSocket connection closed"
            if self._timeout_task is not None and not self._timeout_task.done():
                # Cancel the keepalive task
                self._timeout_task.cancel()
            if self.reconnecting:
                # Don't raise an error, just return
                return
            raise WebSocketError(error_msg)

    def load_message(self, message: str) -> Any:
        """Load a message from a string.

        Args:
            message: The message to load.

        Returns:
            The parsed message.

        Raises:
            WebSocketError: If the message is not valid JSON.
        """
        try:
            parsed_message = json.loads(message)
        except json.JSONDecodeError as e:
            msg = f"Received invalid JSON message: {message}"
            raise WebSocketError(msg) from e

        return parsed_message

    def dispatch_message(self, message: dict[str, Any]) -> None:
        """Dispatch a message to the appropriate handler.

        Args:
            message: The message to dispatch.

        Raises:
            WebSocketError: If the message type is not valid.
        """
        if message["type"] == "data":
            if self.api_type == "events" and message["payload"]["data"]["eventsFeed"][
                "item"
            ]["eventId"] == ("event-error-barrier-obstructed"):
                self._dispatch(
                    "barrier_obstructed",
                    message["payload"]["data"]["eventsFeed"]["item"],
                )
            if self.api_type == "events":
                return

            self._dispatch(message["type"], message["payload"])
        elif message["type"] == "error":
            msg = f"Received error message: {message}"
            raise WebSocketError(msg)
        elif message["type"] == "ka":
            _LOGGER.debug("Received keepalive message")
            if self._timeout_task is not None and not self._timeout_task.done():
                # Restart the keepalive task
                self._timeout_task.cancel()
                self._timeout_task = asyncio.create_task(self._watch_keepalive())
        else:
            _LOGGER.debug("Received message of type %s: %s", message["type"], message)

    async def received_message(self, message: str) -> None:
        """Handle a received message.

        Args:
            message: The message to handle.

        Raises:
            WebSocketError: If the message does not contain 'type
        """
        _LOGGER.debug("Received message: %s", message)
        parsed_message = self.load_message(message)
        if "type" not in message:
            msg = f"Received message does not contain 'type', got {message}"
            raise WebSocketError(msg)
        _LOGGER.debug("Dispatching message")
        self.dispatch_message(parsed_message)

        removed = []
        for index, entry in enumerate(self._dispatch_listeners):
            if entry.event != parsed_message["type"]:
                continue

            future = entry.future
            if future.cancelled():
                removed.append(index)
                continue

            if entry.predicate is not None:
                try:
                    valid = entry.predicate(parsed_message)
                except Exception as e:  # noqa: BLE001
                    future.set_exception(e)
                    removed.append(index)
                    continue
            else:
                valid = True

            if valid:
                ret = (
                    parsed_message
                    if entry.result is None
                    else entry.result(parsed_message)
                )
                future.set_result(ret)
                removed.append(index)

                _LOGGER.debug("Event %s occurred, no longer waiting", entry.event)

        for index in reversed(removed):
            del self._dispatch_listeners[index]

    def wait_for(
        self,
        event: str,
        predicate: Callable[[dict[str, Any]], bool] | None = None,
        result: Callable[[dict[str, Any]], Any] | None = None,
    ) -> asyncio.Future[Any]:
        """Wait for an event to occur.

        Args:
            event: The event to wait for.
            predicate: A predicate function.
            result: A result function.

        Returns:
            A future that resolves when the event occurs.

        Raises:
            WebSocketError: If the event is not valid.
        """
        _LOGGER.debug("Waiting for event %s", event)
        future: asyncio.Future[dict[str, Any]] = asyncio.Future()
        self._dispatch_listeners.append(EventListener(predicate, event, result, future))
        return future

    async def subscribe(self, receiver: str) -> str:
        """Subscribe to the WebSocket server.

        Args:
            receiver: The receiver ID. Typically, it's the organization ID, which can be
                found in the attributes of any barrier. (Don't ask me why.)

        Returns:
            The subscription ID.

        Raises:
            WebSocketError: If the subscription times out.
        """
        subscription_id = str(uuid.uuid4())
        payload = await get_request_template(
            "subscribe" if self.api_type == "device" else "event_subscribe",
            {
                "receiver_id": receiver,
                "uuid": subscription_id,
                "id_token": self.id_token,
                "host": self.host,
            },
        )
        _LOGGER.debug(
            "Subscribing to receiver %s with subscription ID %s",
            receiver,
            subscription_id,
        )
        await self.send(payload)

        def _predicate(message: dict[str, Any]) -> bool:
            valid: bool = (
                message["type"] == "start_ack" and message["id"] == subscription_id
            )
            _LOGGER.debug("Checking if start_ack is valid: %s", valid)
            return valid

        try:
            await asyncio.wait_for(self.wait_for("start_ack", _predicate), timeout=10)
        except asyncio.TimeoutError as e:
            msg = "Subscription to the websocket server timed out"
            raise WebSocketError(msg) from e

        _LOGGER.debug("Subscription successful")

        self._subscriptions.append(subscription_id)

        _LOGGER.debug("Subscription added")

        return subscription_id

    async def unsubscribe(self, subscription_id: str) -> None:
        """Unsubscribe from the WebSocket server.

        Args:
            subscription_id: The subscription ID.

        Raises:
            WebSocketError: If the WebSocket connection is closed
        """
        try:
            self._subscriptions.remove(subscription_id)
        except ValueError:
            _LOGGER.debug("Subscription %s not found", subscription_id)
            return
        finally:
            _LOGGER.debug("Removing subscription %s", subscription_id)
            payload = await get_request_template("unsubscribe", {"id": subscription_id})
            _LOGGER.debug("Unsubscribing from subscription %s", subscription_id)
            await self.send(payload)
            _LOGGER.debug("Unsubscribed from subscription %s", subscription_id)

    @property
    def closed(self) -> bool:
        """Check if the WebSocket connection is closed.

        Returns:
            True if the WebSocket connection is closed, False otherwise.
        """
        return True if self.ws is None else self.ws.closed

closed: bool property

Check if the WebSocket connection is closed.

Returns:

Type	Description
bool	True if the WebSocket connection is closed, False otherwise.

__init__(client_session: aiohttp.ClientSession) -> None

Initialize the WebSocketClient.

Source code in src/nice_go/_ws_client.py

def __init__(self, client_session: aiohttp.ClientSession) -> None:
    """Initialize the WebSocketClient."""
    self.ws: aiohttp.ClientWebSocketResponse | None = None
    self._dispatch_listeners: list[EventListener] = []
    self._subscriptions: list[str] = []
    self.client_session = client_session
    self.reconnecting = False
    self._timeout_task: asyncio.Task[None] | None = None

_reconnect() -> None async

Reconnect to the WebSocket server.

Raises:

Type	Description
WebSocketError	If the WebSocket connection is closed or an error occurs.

Source code in src/nice_go/_ws_client.py

async def _reconnect(self) -> None:
    """Reconnect to the WebSocket server.

    Raises:
        WebSocketError: If the WebSocket connection is closed or an error occurs.
    """
    if self.ws is None or self.ws.closed:
        msg = "WebSocket connection is closed"
        raise WebSocketError(msg)
    self.reconnecting = True
    _LOGGER.debug("Reconnecting to WebSocket server")
    await self.close()
    raise ReconnectWebSocketError

_redact_message(message: str | dict[str, Any]) -> Any

Redact sensitive information from a message.

Parameters:

Name	Type	Description	Default
message	str | dict[str, Any]	The message to redact.	required

Returns:

Type	Description
Any	The redacted message.

Source code in src/nice_go/_ws_client.py

def _redact_message(self, message: str | dict[str, Any]) -> Any:
    """Redact sensitive information from a message.

    Args:
        message: The message to redact.

    Returns:
        The redacted message.
    """
    if isinstance(message, dict):
        return json.loads(json.dumps(message).replace(self.id_token, "<REDACTED>"))
    return message.replace(self.id_token, "<REDACTED>")

_watch_keepalive() -> None async

A task that handles the timeout for the WebSocket connection.

Raises:

Type	Description
WebSocketError	If the WebSocket connection is closed.

Source code in src/nice_go/_ws_client.py

async def _watch_keepalive(self) -> None:
    """A task that handles the timeout for the WebSocket connection.

    Raises:
        WebSocketError: If the WebSocket connection is closed.
    """
    if self.ws is None or self.ws.closed:
        msg = "WebSocket connection is closed"
        raise WebSocketError(msg)
    await asyncio.sleep(self._timeout / 1000)
    _LOGGER.debug("WebSocket keepalive timeout reached, reconnecting")
    await self._reconnect()

close() -> None async

Close the WebSocket connection.

Raises:

Type	Description
WebSocketError	If the WebSocket connection is closed

Source code in src/nice_go/_ws_client.py

async def close(self) -> None:
    """Close the WebSocket connection.

    Raises:
        WebSocketError: If the WebSocket connection is closed
    """
    if self.ws is None or self.ws.closed:
        return
    _LOGGER.debug("Closing WebSocket client")
    # Unsubscribe from all subscriptions
    for subscription_id in self._subscriptions:
        _LOGGER.debug("Unsubscribing from subscription %s", subscription_id)
        await self.unsubscribe(subscription_id)
    if self._timeout_task is not None and not self._timeout_task.done():
        _LOGGER.debug("Cancelling keepalive task")
        # Cancel the keepalive task
        self._timeout_task.cancel()
        try:
            await self._timeout_task
        except asyncio.CancelledError:
            _LOGGER.debug("Keepalive task was cancelled")
        except Exception:
            _LOGGER.exception("Exception occurred while cancelling keepalive task")
    _LOGGER.debug("Closing WebSocket connection")
    await self.ws.close()
    _LOGGER.debug("WebSocket connection closed")

connect(id_token: str, endpoint: yarl.URL, api_type: str, dispatch: Callable[[str, dict[str, Any] | None], None], host: str | None = None) -> None async

Connect to the WebSocket server.

Parameters:

Name	Type	Description	Default
client_session		The aiohttp ClientSession.	required
id_token	str	The IdToken retrieved from authentication.	required
endpoint	URL	The endpoint URL.	required
dispatch	Callable[[str, dict[str, Any] | None], None]	The dispatch function.	required
host	str | None	The host URL.	None

Raises:

Type	Description
ValueError	If host is not provided.
WebSocketError	If the WebSocket connection is closed or an error occurs.

Source code in src/nice_go/_ws_client.py

async def connect(
    self,
    id_token: str,
    endpoint: yarl.URL,
    api_type: str,
    dispatch: Callable[[str, dict[str, Any] | None], None],
    host: str | None = None,
) -> None:
    """Connect to the WebSocket server.

    Args:
        client_session: The aiohttp ClientSession.
        id_token: The IdToken retrieved from authentication.
        endpoint: The endpoint URL.
        dispatch: The dispatch function.
        host: The host URL.

    Raises:
        ValueError: If host is not provided.
        WebSocketError: If the WebSocket connection is closed or an error occurs.
    """
    if host is None:
        msg = "host must be provided"
        raise ValueError(msg)

    self._dispatch = dispatch
    self.id_token = id_token
    self.host = host
    self.api_type = api_type  # Should be "device" or "events"
    self._endpoint = endpoint

    raw_header = {
        "Authorization": id_token,
        "host": host,
    }
    # Base64 encode the header
    header = base64.b64encode(json.dumps(raw_header).encode()).decode()
    # Construct the URL
    url = endpoint.with_query({"header": header, "payload": "e30="})

    # URL contains sensitive information, so we don't want to log it
    _LOGGER.debug("Connecting to WebSocket server at %s", endpoint)

    headers = {"sec-websocket-protocol": "graphql-ws"}
    self.ws = await self.client_session.ws_connect(url, headers=headers)

    await self.init()

dispatch_message(message: dict[str, Any]) -> None

Dispatch a message to the appropriate handler.

Parameters:

Name	Type	Description	Default
message	dict[str, Any]	The message to dispatch.	required

Raises:

Type	Description
WebSocketError	If the message type is not valid.

Source code in src/nice_go/_ws_client.py

def dispatch_message(self, message: dict[str, Any]) -> None:
    """Dispatch a message to the appropriate handler.

    Args:
        message: The message to dispatch.

    Raises:
        WebSocketError: If the message type is not valid.
    """
    if message["type"] == "data":
        if self.api_type == "events" and message["payload"]["data"]["eventsFeed"][
            "item"
        ]["eventId"] == ("event-error-barrier-obstructed"):
            self._dispatch(
                "barrier_obstructed",
                message["payload"]["data"]["eventsFeed"]["item"],
            )
        if self.api_type == "events":
            return

        self._dispatch(message["type"], message["payload"])
    elif message["type"] == "error":
        msg = f"Received error message: {message}"
        raise WebSocketError(msg)
    elif message["type"] == "ka":
        _LOGGER.debug("Received keepalive message")
        if self._timeout_task is not None and not self._timeout_task.done():
            # Restart the keepalive task
            self._timeout_task.cancel()
            self._timeout_task = asyncio.create_task(self._watch_keepalive())
    else:
        _LOGGER.debug("Received message of type %s: %s", message["type"], message)

init() -> None async

Initialize the WebSocket connection.

Raises:

Type	Description
WebSocketError	If the WebSocket connection is closed or an error occurs.

Source code in src/nice_go/_ws_client.py

async def init(self) -> None:
    """Initialize the WebSocket connection.

    Raises:
        WebSocketError: If the WebSocket connection is closed or an error occurs.
    """
    if self.ws is None or self.ws.closed:
        msg = "WebSocket connection is closed"
        raise WebSocketError(msg)
    _LOGGER.debug("Initializing WebSocket connection")
    await self.send({"type": "connection_init"})
    try:
        _LOGGER.debug("Waiting for connection_ack")
        message = await self.ws.receive(timeout=10)
        data = json.loads(message.data)
        _LOGGER.debug("Received message: %s", data)
        if data["type"] != "connection_ack":
            msg = f'Expected connection_ack, but received {data["type"]}'
            raise WebSocketError(
                msg,
            )
    except asyncio.TimeoutError as e:
        msg = "Connection to the websocket server timed out"
        raise WebSocketError(msg) from e
    _LOGGER.debug("Received connection_ack, WebSocket connection established")

    self._timeout = data.get("payload", {}).get("timeout", 300000)
    self._timeout_task = asyncio.create_task(self._watch_keepalive())
    self._dispatch(f"{self.api_type}_connected", None)

load_message(message: str) -> Any

Load a message from a string.

Parameters:

Name	Type	Description	Default
message	str	The message to load.	required

Returns:

Type	Description
Any	The parsed message.

Raises:

Type	Description
WebSocketError	If the message is not valid JSON.

Source code in src/nice_go/_ws_client.py

def load_message(self, message: str) -> Any:
    """Load a message from a string.

    Args:
        message: The message to load.

    Returns:
        The parsed message.

    Raises:
        WebSocketError: If the message is not valid JSON.
    """
    try:
        parsed_message = json.loads(message)
    except json.JSONDecodeError as e:
        msg = f"Received invalid JSON message: {message}"
        raise WebSocketError(msg) from e

    return parsed_message

poll() -> None async

Poll the WebSocket connection for messages.

Raises:

Type	Description
WebSocketError	If the WebSocket connection is closed or an error occurs.

Source code in src/nice_go/_ws_client.py

async def poll(self) -> None:
    """Poll the WebSocket connection for messages.

    Raises:
        WebSocketError: If the WebSocket connection is closed or an error occurs.
    """
    if self.ws is None or self.ws.closed:
        error_msg = "WebSocket connection is closed"
        raise WebSocketError(error_msg)
    msg = await self.ws.receive(timeout=300.0)
    if msg.type == aiohttp.WSMsgType.TEXT:
        await self.received_message(msg.data)
    elif msg.type == aiohttp.WSMsgType.ERROR:
        error_msg = f"WebSocket connection closed with error {msg}"
        raise WebSocketError(error_msg)
    elif msg.type in (
        aiohttp.WSMsgType.CLOSE,
        aiohttp.WSMsgType.CLOSING,
        aiohttp.WSMsgType.CLOSED,
    ):
        error_msg = "WebSocket connection closed"
        if self._timeout_task is not None and not self._timeout_task.done():
            # Cancel the keepalive task
            self._timeout_task.cancel()
        if self.reconnecting:
            # Don't raise an error, just return
            return
        raise WebSocketError(error_msg)

received_message(message: str) -> None async

Handle a received message.

Parameters:

Name	Type	Description	Default
message	str	The message to handle.	required

Raises:

Type	Description
WebSocketError	If the message does not contain 'type

Source code in src/nice_go/_ws_client.py

async def received_message(self, message: str) -> None:
    """Handle a received message.

    Args:
        message: The message to handle.

    Raises:
        WebSocketError: If the message does not contain 'type
    """
    _LOGGER.debug("Received message: %s", message)
    parsed_message = self.load_message(message)
    if "type" not in message:
        msg = f"Received message does not contain 'type', got {message}"
        raise WebSocketError(msg)
    _LOGGER.debug("Dispatching message")
    self.dispatch_message(parsed_message)

    removed = []
    for index, entry in enumerate(self._dispatch_listeners):
        if entry.event != parsed_message["type"]:
            continue

        future = entry.future
        if future.cancelled():
            removed.append(index)
            continue

        if entry.predicate is not None:
            try:
                valid = entry.predicate(parsed_message)
            except Exception as e:  # noqa: BLE001
                future.set_exception(e)
                removed.append(index)
                continue
        else:
            valid = True

        if valid:
            ret = (
                parsed_message
                if entry.result is None
                else entry.result(parsed_message)
            )
            future.set_result(ret)
            removed.append(index)

            _LOGGER.debug("Event %s occurred, no longer waiting", entry.event)

    for index in reversed(removed):
        del self._dispatch_listeners[index]

send(message: str | dict[str, Any]) -> None async

Send a message to the WebSocket server.

Parameters:

Name	Type	Description	Default
message	str | dict[str, Any]	The message to send.	required

Raises:

Type	Description
WebSocketError	If the WebSocket connection is closed

Source code in src/nice_go/_ws_client.py

async def send(self, message: str | dict[str, Any]) -> None:
    """Send a message to the WebSocket server.

    Args:
        message: The message to send.

    Raises:
        WebSocketError: If the WebSocket connection is closed
    """
    if self.ws is None or self.ws.closed:
        msg = "WebSocket connection is closed"
        raise WebSocketError(msg)
    redacted_message = self._redact_message(message)
    _LOGGER.debug("Sending message: %s", redacted_message)
    if isinstance(message, dict):
        await self.ws.send_json(message)
    else:
        await self.ws.send_str(message)

subscribe(receiver: str) -> str async

Subscribe to the WebSocket server.

Parameters:

Name	Type	Description	Default
receiver	str	The receiver ID. Typically, it's the organization ID, which can be found in the attributes of any barrier. (Don't ask me why.)	required

Returns:

Type	Description
str	The subscription ID.

Raises:

Type	Description
WebSocketError	If the subscription times out.

Source code in src/nice_go/_ws_client.py

async def subscribe(self, receiver: str) -> str:
    """Subscribe to the WebSocket server.

    Args:
        receiver: The receiver ID. Typically, it's the organization ID, which can be
            found in the attributes of any barrier. (Don't ask me why.)

    Returns:
        The subscription ID.

    Raises:
        WebSocketError: If the subscription times out.
    """
    subscription_id = str(uuid.uuid4())
    payload = await get_request_template(
        "subscribe" if self.api_type == "device" else "event_subscribe",
        {
            "receiver_id": receiver,
            "uuid": subscription_id,
            "id_token": self.id_token,
            "host": self.host,
        },
    )
    _LOGGER.debug(
        "Subscribing to receiver %s with subscription ID %s",
        receiver,
        subscription_id,
    )
    await self.send(payload)

    def _predicate(message: dict[str, Any]) -> bool:
        valid: bool = (
            message["type"] == "start_ack" and message["id"] == subscription_id
        )
        _LOGGER.debug("Checking if start_ack is valid: %s", valid)
        return valid

    try:
        await asyncio.wait_for(self.wait_for("start_ack", _predicate), timeout=10)
    except asyncio.TimeoutError as e:
        msg = "Subscription to the websocket server timed out"
        raise WebSocketError(msg) from e

    _LOGGER.debug("Subscription successful")

    self._subscriptions.append(subscription_id)

    _LOGGER.debug("Subscription added")

    return subscription_id

unsubscribe(subscription_id: str) -> None async

Unsubscribe from the WebSocket server.

Parameters:

Name	Type	Description	Default
subscription_id	str	The subscription ID.	required

Raises:

Type	Description
WebSocketError	If the WebSocket connection is closed

Source code in src/nice_go/_ws_client.py

async def unsubscribe(self, subscription_id: str) -> None:
    """Unsubscribe from the WebSocket server.

    Args:
        subscription_id: The subscription ID.

    Raises:
        WebSocketError: If the WebSocket connection is closed
    """
    try:
        self._subscriptions.remove(subscription_id)
    except ValueError:
        _LOGGER.debug("Subscription %s not found", subscription_id)
        return
    finally:
        _LOGGER.debug("Removing subscription %s", subscription_id)
        payload = await get_request_template("unsubscribe", {"id": subscription_id})
        _LOGGER.debug("Unsubscribing from subscription %s", subscription_id)
        await self.send(payload)
        _LOGGER.debug("Unsubscribed from subscription %s", subscription_id)

wait_for(event: str, predicate: Callable[[dict[str, Any]], bool] | None = None, result: Callable[[dict[str, Any]], Any] | None = None) -> asyncio.Future[Any]

Wait for an event to occur.

Parameters:

Name	Type	Description	Default
event	str	The event to wait for.	required
predicate	Callable[[dict[str, Any]], bool] | None	A predicate function.	None
result	Callable[[dict[str, Any]], Any] | None	A result function.	None

Returns:

Type	Description
Future[Any]	A future that resolves when the event occurs.

Raises:

Type	Description
WebSocketError	If the event is not valid.

Source code in src/nice_go/_ws_client.py

def wait_for(
    self,
    event: str,
    predicate: Callable[[dict[str, Any]], bool] | None = None,
    result: Callable[[dict[str, Any]], Any] | None = None,
) -> asyncio.Future[Any]:
    """Wait for an event to occur.

    Args:
        event: The event to wait for.
        predicate: A predicate function.
        result: A result function.

    Returns:
        A future that resolves when the event occurs.

    Raises:
        WebSocketError: If the event is not valid.
    """
    _LOGGER.debug("Waiting for event %s", event)
    future: asyncio.Future[dict[str, Any]] = asyncio.Future()
    self._dispatch_listeners.append(EventListener(predicate, event, result, future))
    return future

nice_go_api

Parses data from the Nice G.O. API.

This module provides a class to interact with the Nice G.O. API. It allows the user to authenticate, connect to the WebSocket API, and interact with barriers.

Classes:

Name	Description
NiceGOApi	A class to interact with the Nice G.O. API.

NiceGOApi

A class to interact with the Nice G.O. API.

This class provides methods to authenticate, connect to the WebSocket API, and interact with barriers.

Attributes:

Name	Type	Description
id_token	str | None	The ID token.

Methods:

Name	Description
event	Decorator to add an event listener.
authenticate_refresh	Authenticate using a refresh token.
authenticate	Authenticate using username and password.
connect	Connect to the WebSocket API.
subscribe	Subscribe to a receiver.
unsubscribe	Unsubscribe from a receiver.
close	Close the connection.
get_all_barriers	Get all barriers.

Source code in src/nice_go/nice_go_api.py

class NiceGOApi:
    """A class to interact with the Nice G.O. API.

    This class provides methods to authenticate, connect to the WebSocket API,
    and interact with barriers.

    Attributes:
        id_token (str | None): The ID token.

    Functions:
        event: Decorator to add an event listener.
        authenticate_refresh: Authenticate using a refresh token.
        authenticate: Authenticate using username and password.
        connect: Connect to the WebSocket API.
        subscribe: Subscribe to a receiver.
        unsubscribe: Unsubscribe from a receiver.
        close: Close the connection.
        get_all_barriers: Get all barriers.
    """

    def __init__(self) -> None:
        """Initialize the NiceGOApi object."""
        self.id_token: str | None = None
        self._closing_task: asyncio.Task[None] | None = None
        self._device_ws: WebSocketClient | None = None
        self._endpoints: dict[str, Any] | None = None
        self._session: aiohttp.ClientSession | None = None
        self._event_tasks: set[asyncio.Task[None]] = set()
        self._events_ws: WebSocketClient | None = None
        self._device_connected: bool = False
        self._events_connected: bool = False
        self._events: dict[str, list[Callable[..., Coroutine[Any, Any, Any]]]] = {}

        self.event(self.on_device_connected)
        self.event(self.on_events_connected)

    async def on_device_connected(self) -> None:
        """Handle the device connected event."""
        self._device_connected = True
        if self._device_connected and self._events_connected:
            # Only dispatch when both feeds are connected
            self._dispatch("connected")

    async def on_events_connected(self) -> None:
        """Handle the events connected event."""
        self._events_connected = True
        if self._device_connected and self._events_connected:
            # Only dispatch when both feeds are connected
            self._dispatch("connected")

    def event(self, coro: CoroT) -> CoroT:
        """Decorator to add an event listener. Just a wrapper around `listen`.

        Info:
            This can only decorate coroutine functions.

        Args:
            coro (CoroT): The coroutine function to decorate.

        Examples:
            You can use this decorator to add event listeners to the API object.
            Some events include but are not limited to:

            - `connection_lost`: Triggered when the connection to the WebSocket API is
                lost.
            - `connected`: Triggered when the connection to the WebSocket API is
                established.
            - `data`: Triggered when data is received from an active subscription.
                See `subscribe`.

            >>> @api.event
            ... async def on_data(
            ...     data: dict[str, Any] | None = None,
            ... ) -> None:
            ...     if data is not None:
            ...         print(data)
        """
        self.listen(coro.__name__, coro)
        return coro

    def listen(self, event_name: str, coro: CoroT) -> Callable[[], None]:
        """Add an event listener.

        Args:
            event_name (str): The name of the event.
            coro (CoroT): The coroutine to run when the event is dispatched.

        Returns:
            A function to remove the event listener.

        Examples:
            You can use this method to add event listeners to the API object.
            Some events include but are not limited to:

            - `connection_lost`: Triggered when the connection to the WebSocket API is
                lost.
            - `connected`: Triggered when the connection to the WebSocket API is
                established.
            - `data`: Triggered when data is received from an active subscription.
                See `subscribe`.

            >>> def on_data(data: dict[str, Any] | None = None) -> None:
            ...     if data is not None:
            ...         print(data)
            ...
            >>> remove_listener = api.listen("data", on_data)
        """
        if not asyncio.iscoroutinefunction(coro):
            msg = "The decorated function must be a coroutine"
            raise TypeError(msg)

        _LOGGER.debug("Adding event listener %s", coro.__name__)

        self._events.setdefault(event_name, []).append(coro)
        return lambda: self._events[event_name].remove(coro)

    async def _run_event(
        self,
        coro: Callable[..., Coroutine[Any, Any, Any]],
        event_name: str,
        data: dict[str, Any] | None = None,
    ) -> None:
        """Run an event coroutine. For internal use only.

        Args:
            coro (Callable[..., Coroutine[Any, Any, Any]]): The coroutine to run.
            event_name (str): The name of the event.
            data (dict[str, Any] | None): The data to pass to the event coroutine.
        """
        kwargs = {}
        if data is not None:
            kwargs["data"] = data
        try:
            await coro(**kwargs)
        except asyncio.CancelledError:
            pass
        except Exception:
            _LOGGER.exception("Error while handling event %s", event_name)

    def _schedule_event(
        self,
        coro: Callable[..., Coroutine[Any, Any, Any]],
        event_name: str,
        data: dict[str, Any] | None,
    ) -> None:
        """Schedule an event to be dispatched. For internal use only.

        Args:
            coro (Callable[..., Coroutine[Any, Any, Any]]): The coroutine to schedule.
            event_name (str): The name of the event.
            data (dict[str, Any] | None): The data to pass to the event coroutine.
        """
        wrapped = self._run_event(coro, event_name, data)
        task = asyncio.create_task(wrapped, name=f"NiceGO: {event_name}")
        self._event_tasks.add(task)  # See RUF006
        task.add_done_callback(self._event_tasks.discard)

    def _dispatch(self, event: str, data: dict[str, Any] | None = None) -> None:
        """Dispatch an event to listeners. For internal use only.

        Args:
            event (str): The name of the event.
            data (dict[str, Any] | None): The data to pass to the event coroutine.
        """
        method = f"on_{event}"

        coros = self._events.get(method, [])

        if not coros:
            _LOGGER.debug("No listeners for event %s", event)
            return

        _LOGGER.debug("Dispatching event %s", event)
        for coro in coros:
            self._schedule_event(coro, method, data)

    async def authenticate_refresh(
        self,
        refresh_token: str,
        session: aiohttp.ClientSession,
    ) -> None:
        """Authenticate using a previous obtained refresh token.

        Args:
            refresh_token (str): The refresh token.
            session (aiohttp.ClientSession): The client session to use.

        Raises:
            AuthFailedError: If the authentication fails.
            ApiError: If an API error occurs.
        """
        self._session = session
        await self.__authenticate(None, None, refresh_token)

    async def authenticate(
        self,
        user_name: str,
        password: str,
        session: aiohttp.ClientSession,
    ) -> str | None:
        """Authenticate using username and password.

        Args:
            user_name (str): The username.
            password (str): The password.
            session (aiohttp.ClientSession): The client session to use.

        Returns:
            The refresh token.

        Raises:
            AuthFailedError: If the authentication fails.
            ApiError: If an API error occurs.
        """
        self._session = session
        return await self.__authenticate(user_name, password, None)

    async def __authenticate(
        self,
        user_name: str | None,
        password: str | None,
        refresh_token: str | None,
    ) -> str | None:
        """Authenticate using username and password or refresh token.

        Args:
            user_name (str | None): The username.
            password (str | None): The password.
            refresh_token (str | None): The refresh token.

        Returns:
            The refresh token.

        Raises:
            AuthFailedError: If the authentication fails.
            ApiError: If an API error occurs.
        """
        try:
            _LOGGER.debug("Authenticating")

            if self._session is None:
                msg = "ClientSession not provided"
                raise ValueError(msg)

            # Get the endpoints
            data = await self._session.get(ENDPOINTS_URL)
            endpoints = await data.json()
            self._endpoints = endpoints["endpoints"]

            if self._endpoints is None:
                msg = "Endpoints not available"
                raise ApiError(msg)

            authenticator = AwsCognitoAuthenticator(
                self._endpoints["Config"]["Region"],
                self._endpoints["Config"]["ClientId"],
                self._endpoints["Config"]["UserPoolId"],
                self._endpoints["Config"]["IdentityPoolId"],
            )

            if user_name and password:
                token = await asyncio.to_thread(
                    authenticator.get_new_token,
                    user_name,
                    password,
                )
            elif refresh_token:
                token = await asyncio.to_thread(
                    authenticator.refresh_token,
                    refresh_token,
                )

            _LOGGER.debug("Authenticated")

            self.id_token = token.id_token
        except botocore.exceptions.ClientError as e:
            _LOGGER.exception("Exception while authenticating")
            if e.response["Error"]["Code"] == "NotAuthorizedException":
                raise AuthFailedError from e
            raise ApiError from e
        else:
            return token.refresh_token

    @property
    def closed(self) -> bool:
        """Check if the connection is closed."""
        return self._closing_task is not None

    async def _poll_device_ws(self) -> None:
        """Continuously polls the device WebSocket to maintain an active connection.
        This function will repeatedly call the poll method on the WebSocket if it is
        initialized.

        Returns:
            None
        """
        if self._device_ws is None:
            return
        while True:
            await self._device_ws.poll()

    async def _poll_events_ws(self) -> None:
        """Continuously polls the device WebSocket to maintain an active connection.
        This function will repeatedly call the poll method on the WebSocket if it is
        initialized.

        Returns:
            None
        """

        if self._events_ws is None:
            return
        while True:
            await self._events_ws.poll()

    async def _check_response_errors(self, response: dict[str, Any]) -> None:
        """Checks a GraphQL response for errors, namely for expired tokens.

        Args:
            response (dict[str, Any]): The response to check.

        Raises:
            AuthFailedError: If the ID token is expired.
            ApiError: If an API error occurs.

        Returns:
            None
        """

        if errors := response.get("errors"):
            error = errors[0]
            if error["errorType"] == "UnauthorizedException":
                raise AuthFailedError(error)
            raise ApiError(error)

    @retry(
        wait=wait_random_exponential(multiplier=1, min=1, max=10),
        retry=_RetryIfReconnect()
        & retry_if_exception_type(
            (
                OSError,
                WebSocketError,
                aiohttp.ClientError,
                asyncio.TimeoutError,
                ReconnectWebSocketError,
            ),
        ),
        reraise=True,
        before_sleep=before_sleep_log(_LOGGER, logging.DEBUG),
    )
    async def connect(self, *, reconnect: bool = True) -> None:
        """Connect to the WebSocket API.

        Warning:
            No events will be dispatched until you subscribe to a receiver.
            Typically, you should pass the `organization` attribute of a barrier
            object to the `subscribe` method to start receiving data. Don't ask me
            why `organization` specifically, I don't know either.

        Danger:
            This method will block the event loop until the connection is closed.
            If you want to run this method in the background, you should run it in a
            separate thread or process.

        Args:
            reconnect (bool): Whether to reconnect if the connection is lost.

        Raises:
            NoAuthError: If the ID token is not available.
            ApiError: If an API error occurs.
            WebSocketError: If an error occurs while connecting.
        """
        try:
            if self.id_token is None:
                raise NoAuthError

            if self._endpoints is None:
                msg = "Endpoints not available"
                raise ApiError(msg)

            if self._session is None:
                msg = "ClientSession not provided"
                raise ValueError(msg)

            self._reconnect = reconnect

            device_url = self._endpoints["GraphQL"]["device"]["wss"]
            events_url = self._endpoints["GraphQL"]["events"]["wss"]

            _LOGGER.debug("Connecting to WebSocket API %s", device_url)

            self._device_ws = WebSocketClient(client_session=self._session)
            await self._device_ws.connect(
                self.id_token,
                yarl.URL(device_url),
                "device",
                self._dispatch,
                yarl.URL(self._endpoints["GraphQL"]["device"]["https"]).host,
            )
            self._events_ws = WebSocketClient(client_session=self._session)
            await self._events_ws.connect(
                self.id_token,
                yarl.URL(events_url),
                "events",
                self._dispatch,
                yarl.URL(self._endpoints["GraphQL"]["events"]["https"]).host,
            )

            _LOGGER.debug("Connected to WebSocket API")

            device_task = asyncio.create_task(self._poll_device_ws())
            events_task = asyncio.create_task(self._poll_events_ws())

            with contextlib.suppress(asyncio.CancelledError):
                done, pending = await asyncio.wait(
                    [device_task, events_task],
                    return_when=asyncio.FIRST_EXCEPTION,
                )

            with contextlib.suppress(UnboundLocalError):
                if exceptions := [
                    task.exception() for task in done if task.exception()
                ]:
                    for p in pending:
                        p.cancel()
                    # Make sure both WS are closed
                    await self._events_ws.close()
                    await self._device_ws.close()
                    raise exceptions[0]  # type: ignore[misc]
        except (
            OSError,
            WebSocketError,
            aiohttp.ClientError,
            asyncio.TimeoutError,
            ReconnectWebSocketError,
        ) as e:
            self._dispatch("connection_lost", {"exception": e})
            self._device_connected = False
            self._events_connected = False
            if not reconnect:
                _LOGGER.debug("Connection lost, not reconnecting")
                await self.close()
                raise

            if self.closed:
                return

            _LOGGER.debug("Connection lost, retrying...")

            # Raising triggers retry
            raise

    async def subscribe(self, receiver: str) -> list[str]:
        """Subscribe to a receiver.

        Args:
            receiver (str): The receiver to subscribe to.

        Returns:
            The subscription IDs. You can pass this into the `unsubscribe` method to
                unsubscribe from the receiver.

        Raises:
            WebSocketError: If no WebSocket connection is available.
        """
        if self._device_ws is None:
            msg = "No WebSocket connection"
            raise WebSocketError(msg)
        if self._events_ws is None:
            msg = "No WebSocket connection"
            raise WebSocketError(msg)

        _LOGGER.debug("Subscribing to receiver %s", receiver)

        return [
            await self._device_ws.subscribe(receiver),
            await self._events_ws.subscribe(receiver),
        ]

    async def unsubscribe(self, subscription_id: str) -> None:
        """Unsubscribe from a receiver.

        Args:
            subscription_id (str): The subscription ID to unsubscribe from.

        Raises:
            WebSocketError: If no WebSocket connection is available
        """
        if self._device_ws is None:
            msg = "No WebSocket connection"
            raise WebSocketError(msg)
        if self._events_ws is None:
            msg = "No WebSocket connection"
            raise WebSocketError(msg)

        _LOGGER.debug("Unsubscribing from subscription %s", subscription_id)

        await self._device_ws.unsubscribe(subscription_id)
        await self._events_ws.unsubscribe(subscription_id)

    async def close(self) -> None:
        """Close the connection.

        Raises:
            NoAuthError: If the ID token is not available.
        """

        async def _close() -> None:
            if self._device_ws:
                await self._device_ws.close()
            if self._events_ws:
                await self._events_ws.close()

        _LOGGER.debug("Closing connection")

        self._closing_task = asyncio.create_task(_close())
        await self._closing_task

    async def get_all_barriers(self) -> list[Barrier]:
        """Get all barriers.

        Returns:
            A list of all barriers.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Getting all barriers")
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template("get_all_barriers", None),
        )
        data = await response.json()

        _LOGGER.debug("Got all barriers")
        _LOGGER.debug("Data: %s", data)

        await self._check_response_errors(data)

        barriers = []

        for device in data["data"]["devicesListAll"]["devices"]:
            if device["state"]["connectionState"] is not None:
                connection_state = ConnectionState(
                    device["state"]["connectionState"]["connected"],
                    device["state"]["connectionState"]["updatedTimestamp"],
                )
            else:
                connection_state = None
            barrier_state = BarrierState(
                device["state"]["deviceId"],
                json.loads(device["state"]["reported"]),
                device["state"]["timestamp"],
                device["state"]["version"],
                connection_state,
            )
            barrier = Barrier(
                device["id"],
                device["type"],
                device["controlLevel"],
                device["attr"],
                barrier_state,
                self,
            )
            barriers.append(barrier)

        return barriers

    async def open_barrier(self, barrier_id: str) -> bool:
        """Open a barrier.

        Args:
            barrier_id (str): The ID of the barrier to open.

        Returns:
            Whether the barrier was opened successfully.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Opening barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template("open_barrier", {"barrier_id": barrier_id}),
        )
        data = await response.json()

        _LOGGER.debug("Opening barrier response: %s", data)
        await self._check_response_errors(data)

        result: bool = data["data"]["devicesControl"]

        return result

    async def close_barrier(self, barrier_id: str) -> bool:
        """Close a barrier.

        Args:
            barrier_id (str): The ID of the barrier to close.

        Returns:
            Whether the barrier was closed successfully."""
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Closing barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template(
                "close_barrier",
                {"barrier_id": barrier_id},
            ),
        )
        data = await response.json()

        _LOGGER.debug("Closing barrier response: %s", data)
        await self._check_response_errors(data)

        result: bool = data["data"]["devicesControl"]

        return result

    async def light_on(self, barrier_id: str) -> bool:
        """Turn the light on.

        Args:
            barrier_id (str): The ID of the barrier to turn the light on.

        Returns:
            Whether the light was turned on successfully.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Turning light on for barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template("light_on", {"barrier_id": barrier_id}),
        )
        data = await response.json()

        _LOGGER.debug("Turning light on response: %s", data)
        await self._check_response_errors(data)

        result: bool = data["data"]["devicesControl"]

        return result

    async def light_off(self, barrier_id: str) -> bool:
        """Turn the light off.

        Args:
            barrier_id (str): The ID of the barrier to turn the light off.

        Returns:
            Whether the light was turned off successfully.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Turning light off for barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template("light_off", {"barrier_id": barrier_id}),
        )
        data = await response.json()

        _LOGGER.debug("Turning light off response: %s", data)
        await self._check_response_errors(data)

        result: bool = data["data"]["devicesControl"]

        return result

    async def vacation_mode_on(self, barrier_id: str) -> None:
        """Turn vacation mode on.

        Args:
            barrier_id (str): The ID of the barrier to turn vacation mode on.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Turning vacation mode on for barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template(
                "vacation_mode_on",
                {"barrier_id": barrier_id},
            ),
        )
        await response.json()

    async def vacation_mode_off(self, barrier_id: str) -> None:
        """Turn vacation mode off.

        Args:
            barrier_id (str): The ID of the barrier to turn vacation mode off.

        Raises:
            NoAuthError: If the ID token is not available.
            ValueError: If the ClientSession or Endpoints are not available.
            ApiError: If an API error occurs.
        """
        if self.id_token is None:
            raise NoAuthError
        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)
        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        api_url = self._endpoints["GraphQL"]["device"]["https"]

        _LOGGER.debug("Turning vacation mode off for barrier %s", barrier_id)
        _LOGGER.debug("API URL: %s", api_url)

        headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

        response = await self._session.post(
            api_url,
            headers=headers,
            json=await get_request_template(
                "vacation_mode_off",
                {"barrier_id": barrier_id},
            ),
        )
        await response.json()

closed: bool property

Check if the connection is closed.

__authenticate(user_name: str | None, password: str | None, refresh_token: str | None) -> str | None async

Authenticate using username and password or refresh token.

Parameters:

Name	Type	Description	Default
user_name	str | None	The username.	required
password	str | None	The password.	required
refresh_token	str | None	The refresh token.	required

Returns:

Type	Description
str | None	The refresh token.

Raises:

Type	Description
AuthFailedError	If the authentication fails.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def __authenticate(
    self,
    user_name: str | None,
    password: str | None,
    refresh_token: str | None,
) -> str | None:
    """Authenticate using username and password or refresh token.

    Args:
        user_name (str | None): The username.
        password (str | None): The password.
        refresh_token (str | None): The refresh token.

    Returns:
        The refresh token.

    Raises:
        AuthFailedError: If the authentication fails.
        ApiError: If an API error occurs.
    """
    try:
        _LOGGER.debug("Authenticating")

        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)

        # Get the endpoints
        data = await self._session.get(ENDPOINTS_URL)
        endpoints = await data.json()
        self._endpoints = endpoints["endpoints"]

        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        authenticator = AwsCognitoAuthenticator(
            self._endpoints["Config"]["Region"],
            self._endpoints["Config"]["ClientId"],
            self._endpoints["Config"]["UserPoolId"],
            self._endpoints["Config"]["IdentityPoolId"],
        )

        if user_name and password:
            token = await asyncio.to_thread(
                authenticator.get_new_token,
                user_name,
                password,
            )
        elif refresh_token:
            token = await asyncio.to_thread(
                authenticator.refresh_token,
                refresh_token,
            )

        _LOGGER.debug("Authenticated")

        self.id_token = token.id_token
    except botocore.exceptions.ClientError as e:
        _LOGGER.exception("Exception while authenticating")
        if e.response["Error"]["Code"] == "NotAuthorizedException":
            raise AuthFailedError from e
        raise ApiError from e
    else:
        return token.refresh_token

__init__() -> None

Initialize the NiceGOApi object.

Source code in src/nice_go/nice_go_api.py

def __init__(self) -> None:
    """Initialize the NiceGOApi object."""
    self.id_token: str | None = None
    self._closing_task: asyncio.Task[None] | None = None
    self._device_ws: WebSocketClient | None = None
    self._endpoints: dict[str, Any] | None = None
    self._session: aiohttp.ClientSession | None = None
    self._event_tasks: set[asyncio.Task[None]] = set()
    self._events_ws: WebSocketClient | None = None
    self._device_connected: bool = False
    self._events_connected: bool = False
    self._events: dict[str, list[Callable[..., Coroutine[Any, Any, Any]]]] = {}

    self.event(self.on_device_connected)
    self.event(self.on_events_connected)

_check_response_errors(response: dict[str, Any]) -> None async

Checks a GraphQL response for errors, namely for expired tokens.

Parameters:

Name	Type	Description	Default
response	dict[str, Any]	The response to check.	required

Raises:

Type	Description
AuthFailedError	If the ID token is expired.
ApiError	If an API error occurs.

Returns:

Type	Description
None	None

Source code in src/nice_go/nice_go_api.py

async def _check_response_errors(self, response: dict[str, Any]) -> None:
    """Checks a GraphQL response for errors, namely for expired tokens.

    Args:
        response (dict[str, Any]): The response to check.

    Raises:
        AuthFailedError: If the ID token is expired.
        ApiError: If an API error occurs.

    Returns:
        None
    """

    if errors := response.get("errors"):
        error = errors[0]
        if error["errorType"] == "UnauthorizedException":
            raise AuthFailedError(error)
        raise ApiError(error)

_dispatch(event: str, data: dict[str, Any] | None = None) -> None

Dispatch an event to listeners. For internal use only.

Parameters:

Name	Type	Description	Default
event	str	The name of the event.	required
data	dict[str, Any] | None	The data to pass to the event coroutine.	None

Source code in src/nice_go/nice_go_api.py

def _dispatch(self, event: str, data: dict[str, Any] | None = None) -> None:
    """Dispatch an event to listeners. For internal use only.

    Args:
        event (str): The name of the event.
        data (dict[str, Any] | None): The data to pass to the event coroutine.
    """
    method = f"on_{event}"

    coros = self._events.get(method, [])

    if not coros:
        _LOGGER.debug("No listeners for event %s", event)
        return

    _LOGGER.debug("Dispatching event %s", event)
    for coro in coros:
        self._schedule_event(coro, method, data)

_poll_device_ws() -> None async

Continuously polls the device WebSocket to maintain an active connection. This function will repeatedly call the poll method on the WebSocket if it is initialized.

Returns:

Type	Description
None	None

Source code in src/nice_go/nice_go_api.py

async def _poll_device_ws(self) -> None:
    """Continuously polls the device WebSocket to maintain an active connection.
    This function will repeatedly call the poll method on the WebSocket if it is
    initialized.

    Returns:
        None
    """
    if self._device_ws is None:
        return
    while True:
        await self._device_ws.poll()

_poll_events_ws() -> None async

Continuously polls the device WebSocket to maintain an active connection. This function will repeatedly call the poll method on the WebSocket if it is initialized.

Returns:

Type	Description
None	None

Source code in src/nice_go/nice_go_api.py

async def _poll_events_ws(self) -> None:
    """Continuously polls the device WebSocket to maintain an active connection.
    This function will repeatedly call the poll method on the WebSocket if it is
    initialized.

    Returns:
        None
    """

    if self._events_ws is None:
        return
    while True:
        await self._events_ws.poll()

_run_event(coro: Callable[..., Coroutine[Any, Any, Any]], event_name: str, data: dict[str, Any] | None = None) -> None async

Run an event coroutine. For internal use only.

Parameters:

Name	Type	Description	Default
coro	Callable[..., Coroutine[Any, Any, Any]]	The coroutine to run.	required
event_name	str	The name of the event.	required
data	dict[str, Any] | None	The data to pass to the event coroutine.	None

Source code in src/nice_go/nice_go_api.py

async def _run_event(
    self,
    coro: Callable[..., Coroutine[Any, Any, Any]],
    event_name: str,
    data: dict[str, Any] | None = None,
) -> None:
    """Run an event coroutine. For internal use only.

    Args:
        coro (Callable[..., Coroutine[Any, Any, Any]]): The coroutine to run.
        event_name (str): The name of the event.
        data (dict[str, Any] | None): The data to pass to the event coroutine.
    """
    kwargs = {}
    if data is not None:
        kwargs["data"] = data
    try:
        await coro(**kwargs)
    except asyncio.CancelledError:
        pass
    except Exception:
        _LOGGER.exception("Error while handling event %s", event_name)

_schedule_event(coro: Callable[..., Coroutine[Any, Any, Any]], event_name: str, data: dict[str, Any] | None) -> None

Schedule an event to be dispatched. For internal use only.

Parameters:

Name	Type	Description	Default
coro	Callable[..., Coroutine[Any, Any, Any]]	The coroutine to schedule.	required
event_name	str	The name of the event.	required
data	dict[str, Any] | None	The data to pass to the event coroutine.	required

Source code in src/nice_go/nice_go_api.py

def _schedule_event(
    self,
    coro: Callable[..., Coroutine[Any, Any, Any]],
    event_name: str,
    data: dict[str, Any] | None,
) -> None:
    """Schedule an event to be dispatched. For internal use only.

    Args:
        coro (Callable[..., Coroutine[Any, Any, Any]]): The coroutine to schedule.
        event_name (str): The name of the event.
        data (dict[str, Any] | None): The data to pass to the event coroutine.
    """
    wrapped = self._run_event(coro, event_name, data)
    task = asyncio.create_task(wrapped, name=f"NiceGO: {event_name}")
    self._event_tasks.add(task)  # See RUF006
    task.add_done_callback(self._event_tasks.discard)

authenticate(user_name: str, password: str, session: aiohttp.ClientSession) -> str | None async

Authenticate using username and password.

Parameters:

Name	Type	Description	Default
user_name	str	The username.	required
password	str	The password.	required
session	ClientSession	The client session to use.	required

Returns:

Type	Description
str | None	The refresh token.

Raises:

Type	Description
AuthFailedError	If the authentication fails.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def authenticate(
    self,
    user_name: str,
    password: str,
    session: aiohttp.ClientSession,
) -> str | None:
    """Authenticate using username and password.

    Args:
        user_name (str): The username.
        password (str): The password.
        session (aiohttp.ClientSession): The client session to use.

    Returns:
        The refresh token.

    Raises:
        AuthFailedError: If the authentication fails.
        ApiError: If an API error occurs.
    """
    self._session = session
    return await self.__authenticate(user_name, password, None)

authenticate_refresh(refresh_token: str, session: aiohttp.ClientSession) -> None async

Authenticate using a previous obtained refresh token.

Parameters:

Name	Type	Description	Default
refresh_token	str	The refresh token.	required
session	ClientSession	The client session to use.	required

Raises:

Type	Description
AuthFailedError	If the authentication fails.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def authenticate_refresh(
    self,
    refresh_token: str,
    session: aiohttp.ClientSession,
) -> None:
    """Authenticate using a previous obtained refresh token.

    Args:
        refresh_token (str): The refresh token.
        session (aiohttp.ClientSession): The client session to use.

    Raises:
        AuthFailedError: If the authentication fails.
        ApiError: If an API error occurs.
    """
    self._session = session
    await self.__authenticate(None, None, refresh_token)

close() -> None async

Close the connection.

Raises:

Type	Description
NoAuthError	If the ID token is not available.

Source code in src/nice_go/nice_go_api.py

async def close(self) -> None:
    """Close the connection.

    Raises:
        NoAuthError: If the ID token is not available.
    """

    async def _close() -> None:
        if self._device_ws:
            await self._device_ws.close()
        if self._events_ws:
            await self._events_ws.close()

    _LOGGER.debug("Closing connection")

    self._closing_task = asyncio.create_task(_close())
    await self._closing_task

close_barrier(barrier_id: str) -> bool async

Close a barrier.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to close.	required

Returns:

Type	Description
bool	Whether the barrier was closed successfully.

Source code in src/nice_go/nice_go_api.py

async def close_barrier(self, barrier_id: str) -> bool:
    """Close a barrier.

    Args:
        barrier_id (str): The ID of the barrier to close.

    Returns:
        Whether the barrier was closed successfully."""
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Closing barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template(
            "close_barrier",
            {"barrier_id": barrier_id},
        ),
    )
    data = await response.json()

    _LOGGER.debug("Closing barrier response: %s", data)
    await self._check_response_errors(data)

    result: bool = data["data"]["devicesControl"]

    return result

connect(*, reconnect: bool = True) -> None async

Connect to the WebSocket API.

Warning

No events will be dispatched until you subscribe to a receiver. Typically, you should pass the organization attribute of a barrier object to the subscribe method to start receiving data. Don't ask me why organization specifically, I don't know either.

Danger

This method will block the event loop until the connection is closed. If you want to run this method in the background, you should run it in a separate thread or process.

Parameters:

Name	Type	Description	Default
reconnect	bool	Whether to reconnect if the connection is lost.	True

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ApiError	If an API error occurs.
WebSocketError	If an error occurs while connecting.

Source code in src/nice_go/nice_go_api.py

@retry(
    wait=wait_random_exponential(multiplier=1, min=1, max=10),
    retry=_RetryIfReconnect()
    & retry_if_exception_type(
        (
            OSError,
            WebSocketError,
            aiohttp.ClientError,
            asyncio.TimeoutError,
            ReconnectWebSocketError,
        ),
    ),
    reraise=True,
    before_sleep=before_sleep_log(_LOGGER, logging.DEBUG),
)
async def connect(self, *, reconnect: bool = True) -> None:
    """Connect to the WebSocket API.

    Warning:
        No events will be dispatched until you subscribe to a receiver.
        Typically, you should pass the `organization` attribute of a barrier
        object to the `subscribe` method to start receiving data. Don't ask me
        why `organization` specifically, I don't know either.

    Danger:
        This method will block the event loop until the connection is closed.
        If you want to run this method in the background, you should run it in a
        separate thread or process.

    Args:
        reconnect (bool): Whether to reconnect if the connection is lost.

    Raises:
        NoAuthError: If the ID token is not available.
        ApiError: If an API error occurs.
        WebSocketError: If an error occurs while connecting.
    """
    try:
        if self.id_token is None:
            raise NoAuthError

        if self._endpoints is None:
            msg = "Endpoints not available"
            raise ApiError(msg)

        if self._session is None:
            msg = "ClientSession not provided"
            raise ValueError(msg)

        self._reconnect = reconnect

        device_url = self._endpoints["GraphQL"]["device"]["wss"]
        events_url = self._endpoints["GraphQL"]["events"]["wss"]

        _LOGGER.debug("Connecting to WebSocket API %s", device_url)

        self._device_ws = WebSocketClient(client_session=self._session)
        await self._device_ws.connect(
            self.id_token,
            yarl.URL(device_url),
            "device",
            self._dispatch,
            yarl.URL(self._endpoints["GraphQL"]["device"]["https"]).host,
        )
        self._events_ws = WebSocketClient(client_session=self._session)
        await self._events_ws.connect(
            self.id_token,
            yarl.URL(events_url),
            "events",
            self._dispatch,
            yarl.URL(self._endpoints["GraphQL"]["events"]["https"]).host,
        )

        _LOGGER.debug("Connected to WebSocket API")

        device_task = asyncio.create_task(self._poll_device_ws())
        events_task = asyncio.create_task(self._poll_events_ws())

        with contextlib.suppress(asyncio.CancelledError):
            done, pending = await asyncio.wait(
                [device_task, events_task],
                return_when=asyncio.FIRST_EXCEPTION,
            )

        with contextlib.suppress(UnboundLocalError):
            if exceptions := [
                task.exception() for task in done if task.exception()
            ]:
                for p in pending:
                    p.cancel()
                # Make sure both WS are closed
                await self._events_ws.close()
                await self._device_ws.close()
                raise exceptions[0]  # type: ignore[misc]
    except (
        OSError,
        WebSocketError,
        aiohttp.ClientError,
        asyncio.TimeoutError,
        ReconnectWebSocketError,
    ) as e:
        self._dispatch("connection_lost", {"exception": e})
        self._device_connected = False
        self._events_connected = False
        if not reconnect:
            _LOGGER.debug("Connection lost, not reconnecting")
            await self.close()
            raise

        if self.closed:
            return

        _LOGGER.debug("Connection lost, retrying...")

        # Raising triggers retry
        raise

event(coro: CoroT) -> CoroT

Decorator to add an event listener. Just a wrapper around listen.

Info

This can only decorate coroutine functions.

Parameters:

Name	Type	Description	Default
coro	CoroT	The coroutine function to decorate.	required

Examples:

You can use this decorator to add event listeners to the API object. Some events include but are not limited to:

• connection_lost: Triggered when the connection to the WebSocket API is lost.
• connected: Triggered when the connection to the WebSocket API is established.
• data: Triggered when data is received from an active subscription. See subscribe.

>>> @api.event
... async def on_data(
...     data: dict[str, Any] | None = None,
... ) -> None:
...     if data is not None:
...         print(data)

Source code in src/nice_go/nice_go_api.py

def event(self, coro: CoroT) -> CoroT:
    """Decorator to add an event listener. Just a wrapper around `listen`.

    Info:
        This can only decorate coroutine functions.

    Args:
        coro (CoroT): The coroutine function to decorate.

    Examples:
        You can use this decorator to add event listeners to the API object.
        Some events include but are not limited to:

        - `connection_lost`: Triggered when the connection to the WebSocket API is
            lost.
        - `connected`: Triggered when the connection to the WebSocket API is
            established.
        - `data`: Triggered when data is received from an active subscription.
            See `subscribe`.

        >>> @api.event
        ... async def on_data(
        ...     data: dict[str, Any] | None = None,
        ... ) -> None:
        ...     if data is not None:
        ...         print(data)
    """
    self.listen(coro.__name__, coro)
    return coro

get_all_barriers() -> list[Barrier] async

Get all barriers.

Returns:

Type	Description
list[Barrier]	A list of all barriers.

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def get_all_barriers(self) -> list[Barrier]:
    """Get all barriers.

    Returns:
        A list of all barriers.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Getting all barriers")
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template("get_all_barriers", None),
    )
    data = await response.json()

    _LOGGER.debug("Got all barriers")
    _LOGGER.debug("Data: %s", data)

    await self._check_response_errors(data)

    barriers = []

    for device in data["data"]["devicesListAll"]["devices"]:
        if device["state"]["connectionState"] is not None:
            connection_state = ConnectionState(
                device["state"]["connectionState"]["connected"],
                device["state"]["connectionState"]["updatedTimestamp"],
            )
        else:
            connection_state = None
        barrier_state = BarrierState(
            device["state"]["deviceId"],
            json.loads(device["state"]["reported"]),
            device["state"]["timestamp"],
            device["state"]["version"],
            connection_state,
        )
        barrier = Barrier(
            device["id"],
            device["type"],
            device["controlLevel"],
            device["attr"],
            barrier_state,
            self,
        )
        barriers.append(barrier)

    return barriers

light_off(barrier_id: str) -> bool async

Turn the light off.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to turn the light off.	required

Returns:

Type	Description
bool	Whether the light was turned off successfully.

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def light_off(self, barrier_id: str) -> bool:
    """Turn the light off.

    Args:
        barrier_id (str): The ID of the barrier to turn the light off.

    Returns:
        Whether the light was turned off successfully.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Turning light off for barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template("light_off", {"barrier_id": barrier_id}),
    )
    data = await response.json()

    _LOGGER.debug("Turning light off response: %s", data)
    await self._check_response_errors(data)

    result: bool = data["data"]["devicesControl"]

    return result

light_on(barrier_id: str) -> bool async

Turn the light on.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to turn the light on.	required

Returns:

Type	Description
bool	Whether the light was turned on successfully.

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def light_on(self, barrier_id: str) -> bool:
    """Turn the light on.

    Args:
        barrier_id (str): The ID of the barrier to turn the light on.

    Returns:
        Whether the light was turned on successfully.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Turning light on for barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template("light_on", {"barrier_id": barrier_id}),
    )
    data = await response.json()

    _LOGGER.debug("Turning light on response: %s", data)
    await self._check_response_errors(data)

    result: bool = data["data"]["devicesControl"]

    return result

listen(event_name: str, coro: CoroT) -> Callable[[], None]

Add an event listener.

Parameters:

Name	Type	Description	Default
event_name	str	The name of the event.	required
coro	CoroT	The coroutine to run when the event is dispatched.	required

Returns:

Type	Description
Callable[[], None]	A function to remove the event listener.

Examples:

You can use this method to add event listeners to the API object. Some events include but are not limited to:

• connection_lost: Triggered when the connection to the WebSocket API is lost.
• connected: Triggered when the connection to the WebSocket API is established.
• data: Triggered when data is received from an active subscription. See subscribe.

>>> def on_data(data: dict[str, Any] | None = None) -> None:
...     if data is not None:
...         print(data)
...
>>> remove_listener = api.listen("data", on_data)

Source code in src/nice_go/nice_go_api.py

def listen(self, event_name: str, coro: CoroT) -> Callable[[], None]:
    """Add an event listener.

    Args:
        event_name (str): The name of the event.
        coro (CoroT): The coroutine to run when the event is dispatched.

    Returns:
        A function to remove the event listener.

    Examples:
        You can use this method to add event listeners to the API object.
        Some events include but are not limited to:

        - `connection_lost`: Triggered when the connection to the WebSocket API is
            lost.
        - `connected`: Triggered when the connection to the WebSocket API is
            established.
        - `data`: Triggered when data is received from an active subscription.
            See `subscribe`.

        >>> def on_data(data: dict[str, Any] | None = None) -> None:
        ...     if data is not None:
        ...         print(data)
        ...
        >>> remove_listener = api.listen("data", on_data)
    """
    if not asyncio.iscoroutinefunction(coro):
        msg = "The decorated function must be a coroutine"
        raise TypeError(msg)

    _LOGGER.debug("Adding event listener %s", coro.__name__)

    self._events.setdefault(event_name, []).append(coro)
    return lambda: self._events[event_name].remove(coro)

on_device_connected() -> None async

Handle the device connected event.

Source code in src/nice_go/nice_go_api.py

async def on_device_connected(self) -> None:
    """Handle the device connected event."""
    self._device_connected = True
    if self._device_connected and self._events_connected:
        # Only dispatch when both feeds are connected
        self._dispatch("connected")

on_events_connected() -> None async

Handle the events connected event.

Source code in src/nice_go/nice_go_api.py

async def on_events_connected(self) -> None:
    """Handle the events connected event."""
    self._events_connected = True
    if self._device_connected and self._events_connected:
        # Only dispatch when both feeds are connected
        self._dispatch("connected")

open_barrier(barrier_id: str) -> bool async

Open a barrier.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to open.	required

Returns:

Type	Description
bool	Whether the barrier was opened successfully.

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def open_barrier(self, barrier_id: str) -> bool:
    """Open a barrier.

    Args:
        barrier_id (str): The ID of the barrier to open.

    Returns:
        Whether the barrier was opened successfully.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Opening barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template("open_barrier", {"barrier_id": barrier_id}),
    )
    data = await response.json()

    _LOGGER.debug("Opening barrier response: %s", data)
    await self._check_response_errors(data)

    result: bool = data["data"]["devicesControl"]

    return result

subscribe(receiver: str) -> list[str] async

Subscribe to a receiver.

Parameters:

Name	Type	Description	Default
receiver	str	The receiver to subscribe to.	required

Returns:

Type	Description
list[str]	The subscription IDs. You can pass this into the unsubscribe method to unsubscribe from the receiver.

Raises:

Type	Description
WebSocketError	If no WebSocket connection is available.

Source code in src/nice_go/nice_go_api.py

async def subscribe(self, receiver: str) -> list[str]:
    """Subscribe to a receiver.

    Args:
        receiver (str): The receiver to subscribe to.

    Returns:
        The subscription IDs. You can pass this into the `unsubscribe` method to
            unsubscribe from the receiver.

    Raises:
        WebSocketError: If no WebSocket connection is available.
    """
    if self._device_ws is None:
        msg = "No WebSocket connection"
        raise WebSocketError(msg)
    if self._events_ws is None:
        msg = "No WebSocket connection"
        raise WebSocketError(msg)

    _LOGGER.debug("Subscribing to receiver %s", receiver)

    return [
        await self._device_ws.subscribe(receiver),
        await self._events_ws.subscribe(receiver),
    ]

unsubscribe(subscription_id: str) -> None async

Unsubscribe from a receiver.

Parameters:

Name	Type	Description	Default
subscription_id	str	The subscription ID to unsubscribe from.	required

Raises:

Type	Description
WebSocketError	If no WebSocket connection is available

Source code in src/nice_go/nice_go_api.py

async def unsubscribe(self, subscription_id: str) -> None:
    """Unsubscribe from a receiver.

    Args:
        subscription_id (str): The subscription ID to unsubscribe from.

    Raises:
        WebSocketError: If no WebSocket connection is available
    """
    if self._device_ws is None:
        msg = "No WebSocket connection"
        raise WebSocketError(msg)
    if self._events_ws is None:
        msg = "No WebSocket connection"
        raise WebSocketError(msg)

    _LOGGER.debug("Unsubscribing from subscription %s", subscription_id)

    await self._device_ws.unsubscribe(subscription_id)
    await self._events_ws.unsubscribe(subscription_id)

vacation_mode_off(barrier_id: str) -> None async

Turn vacation mode off.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to turn vacation mode off.	required

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def vacation_mode_off(self, barrier_id: str) -> None:
    """Turn vacation mode off.

    Args:
        barrier_id (str): The ID of the barrier to turn vacation mode off.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Turning vacation mode off for barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template(
            "vacation_mode_off",
            {"barrier_id": barrier_id},
        ),
    )
    await response.json()

vacation_mode_on(barrier_id: str) -> None async

Turn vacation mode on.

Parameters:

Name	Type	Description	Default
barrier_id	str	The ID of the barrier to turn vacation mode on.	required

Raises:

Type	Description
NoAuthError	If the ID token is not available.
ValueError	If the ClientSession or Endpoints are not available.
ApiError	If an API error occurs.

Source code in src/nice_go/nice_go_api.py

async def vacation_mode_on(self, barrier_id: str) -> None:
    """Turn vacation mode on.

    Args:
        barrier_id (str): The ID of the barrier to turn vacation mode on.

    Raises:
        NoAuthError: If the ID token is not available.
        ValueError: If the ClientSession or Endpoints are not available.
        ApiError: If an API error occurs.
    """
    if self.id_token is None:
        raise NoAuthError
    if self._session is None:
        msg = "ClientSession not provided"
        raise ValueError(msg)
    if self._endpoints is None:
        msg = "Endpoints not available"
        raise ApiError(msg)

    api_url = self._endpoints["GraphQL"]["device"]["https"]

    _LOGGER.debug("Turning vacation mode on for barrier %s", barrier_id)
    _LOGGER.debug("API URL: %s", api_url)

    headers = {"Authorization": self.id_token, "Content-Type": "application/json"}

    response = await self._session.post(
        api_url,
        headers=headers,
        json=await get_request_template(
            "vacation_mode_on",
            {"barrier_id": barrier_id},
        ),
    )
    await response.json()

_RetryIfReconnect

Bases: retry_base

Retries only if reconnect is set to True.

Source code in src/nice_go/nice_go_api.py

class _RetryIfReconnect(retry_base):
    """Retries only if reconnect is set to True."""

    def __call__(self, retry_state: RetryCallState) -> Any:
        """Check if the retry should be retried."""
        return retry_state.kwargs.get("reconnect", True)

__call__(retry_state: RetryCallState) -> Any

Check if the retry should be retried.

Source code in src/nice_go/nice_go_api.py

def __call__(self, retry_state: RetryCallState) -> Any:
    """Check if the retry should be retried."""
    return retry_state.kwargs.get("reconnect", True)