ha-ninlight/custom_components/ninlight/coordinator.py

"""Data coordinator for one ninlight lamp."""

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Any

from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, HomeAssistantError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed

from .api import (
    NinlightAuthError,
    NinlightBusyError,
    NinlightClient,
    NinlightError,
    NinlightRequestError,
)
from .const import CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL, DOMAIN, MAX_POLL_FAILURES

_LOGGER = logging.getLogger(__name__)

type NinlightConfigEntry = ConfigEntry[NinlightCoordinator]


@dataclass(frozen=True, slots=True)
class NinlightData:
    """Last known lamp state, system info and settings."""

    state: dict[str, Any]
    info: dict[str, Any]
    # Only the settings the lamp exposes; a lamp on firmware without sACN carries no
    # "sacn" block here, and the entities that need it report themselves unavailable.
    settings: dict[str, Any]

    @property
    def sacn(self) -> dict[str, Any] | None:
        """The lamp's sACN settings, or None on firmware that predates the feature."""
        block = self.settings.get("sacn")
        return block if isinstance(block, dict) else None

    @property
    def sacn_info(self) -> dict[str, Any] | None:
        """The sACN status from system/info, or None on firmware without it."""
        block = self.info.get("sacn")
        return block if isinstance(block, dict) else None


class NinlightCoordinator(DataUpdateCoordinator[NinlightData]):
    """Polls one lamp; every request to the lamp goes through `lock`."""

    config_entry: NinlightConfigEntry

    def __init__(
        self, hass: HomeAssistant, entry: NinlightConfigEntry, client: NinlightClient
    ) -> None:
        super().__init__(
            hass,
            _LOGGER,
            config_entry=entry,
            name=f"{DOMAIN} {client.host}",
            update_interval=timedelta(
                seconds=entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
            ),
        )
        self.client = client
        # The W600 web server has a listen backlog of 4: never talk to one lamp concurrently.
        self.lock = asyncio.Lock()
        # While a firmware update runs, the update entity watches the lamp itself.
        self.ota_in_progress = False
        # Consecutive poll failures; a short outage should not flip the lamp unavailable.
        self._consecutive_poll_failures = 0

    async def _async_update_data(self) -> NinlightData:
        if self.ota_in_progress and self.data is not None:
            return self.data
        try:
            async with self.lock:
                state = await self.client.get_state()
                info = await self.client.get_info()
                # The sACN settings live behind system/settings, not device/state - so the
                # config entities for them need this third request per cycle.
                settings = await self.client.get_settings()
        except NinlightAuthError as err:
            raise ConfigEntryAuthFailed(f"{self.client.host}: token rejected") from err
        except NinlightError as err:
            self._consecutive_poll_failures += 1
            if self.data is not None and self._consecutive_poll_failures < MAX_POLL_FAILURES:
                _LOGGER.debug(
                    "%s: poll failed (%d/%d), keeping last known state: %s",
                    self.client.host,
                    self._consecutive_poll_failures,
                    MAX_POLL_FAILURES,
                    err,
                )
                return self.data
            raise UpdateFailed(f"{self.client.host}: {err}") from err
        self._consecutive_poll_failures = 0
        return NinlightData(state, info, settings)

    async def async_call[T](self, action: Callable[[NinlightClient], Awaitable[T]]) -> T:
        """Run one request under the lock and translate lamp errors for the UI."""
        try:
            async with self.lock:
                return await action(self.client)
        except NinlightAuthError as err:
            self.config_entry.async_start_reauth(self.hass)
            raise HomeAssistantError(
                translation_domain=DOMAIN, translation_key="auth_failed"
            ) from err
        except NinlightBusyError as err:
            raise HomeAssistantError(
                translation_domain=DOMAIN, translation_key="update_busy"
            ) from err
        except NinlightRequestError as err:
            raise HomeAssistantError(
                translation_domain=DOMAIN,
                translation_key="request_failed",
                translation_placeholders={"message": err.message},
            ) from err
        except NinlightError as err:
            raise HomeAssistantError(
                translation_domain=DOMAIN, translation_key="cannot_connect"
            ) from err

    async def async_set_sacn(self, **fields: Any) -> None:
        """Write sACN settings and adopt the lamp's own answer.

        The lamp returns only {"ok": true}, so the settings are read back instead of
        guessed: it clamps and validates on its own (the start_slot limit depends on the
        slot layout, for one), and an entity that shows what was asked for rather than what
        was stored would lie the moment the two differ.
        """

        async def write_and_read(client: NinlightClient) -> dict[str, Any]:
            await client.set_sacn(**fields)
            return await client.get_settings()

        settings = await self.async_call(write_and_read)
        self._consecutive_poll_failures = 0
        self.async_set_updated_data(NinlightData(self.data.state, self.data.info, settings))

    async def async_set_state(self, **fields: Any) -> None:
        """Write and adopt the lamp's complete answer directly - no debounced refresh."""
        state = await self.async_call(lambda client: client.set_state(**fields))
        self._consecutive_poll_failures = 0
        self.async_set_updated_data(NinlightData(state, self.data.info, self.data.settings))