ha-ninstrip/custom_components/ninstrip/coordinator.py

"""Datenkoordinator fuer genau ein ninstrip-Geraet."""

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 (
    NinstripAuthError,
    NinstripBusyError,
    NinstripClient,
    NinstripError,
    NinstripRequestError,
)
from .const import CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL, DOMAIN, MAX_POLL_FAILURES

_LOGGER = logging.getLogger(__name__)

type NinstripConfigEntry = ConfigEntry[NinstripCoordinator]


@dataclass(frozen=True, slots=True)
class NinstripData:
    """Letzter bekannter Stand eines Geraets."""

    state: dict[str, Any]
    info: dict[str, Any]
    presets: dict[str, Any]
    # Farbe und Farbtemperatur stehen NICHT in device/state, sondern je
    # Segment. Ohne diese Liste koennte die Lichtentitaet keine Farbe zeigen
    # und keine setzen.
    segments: dict[str, Any]
    # Nur, was das Geraet unter system/settings zeigt. Aeltere Firmware traegt
    # dort keinen sacn-Block - die Entitaeten melden sich dann nicht
    # verfuegbar, statt zu verschwinden.
    settings: dict[str, Any]

    @property
    def sacn(self) -> dict[str, Any] | None:
        """sACN-Einstellungen, oder None bei Firmware ohne sACN."""
        block = self.settings.get("sacn")
        return block if isinstance(block, dict) else None

    @property
    def sacn_info(self) -> dict[str, Any] | None:
        """sACN-Zustand aus system/info, oder None bei Firmware ohne sACN."""
        block = self.info.get("sacn")
        return block if isinstance(block, dict) else None


class NinstripCoordinator(DataUpdateCoordinator[NinstripData]):
    """Fragt ein Geraet ab; jede Anfrage dorthin laeuft durch `lock`."""

    config_entry: NinstripConfigEntry

    def __init__(
        self, hass: HomeAssistant, entry: NinstripConfigEntry, client: NinstripClient
    ) -> 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
        # Der ESP32-C3 hat einen Kern, den sich WLAN-Stack und LED-Ausgabe
        # teilen. Mehrere gleichzeitige Anfragen bringen nichts ausser Last.
        self.lock = asyncio.Lock()
        self.ota_in_progress = False
        # Die Effektliste steht fest in der Firmware; sie wird einmal geholt
        # und nach einem Update erneut. Jede Abfrage damit zu belasten waere
        # eine sechste Anfrage je Zyklus fuer etwas, das sich nie aendert.
        self.effects: list[dict[str, Any]] = []
        self._effects_fuer: str | None = None
        self._consecutive_poll_failures = 0

    async def _async_update_data(self) -> NinstripData:
        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()
                presets = await self.client.get_presets()
                segments = await self.client.get_segments()
                # Die sACN-Einstellungen liegen hinter system/settings, nicht in
                # device/state - eine vierte Anfrage je Zyklus. Wie ha-ninlight.
                settings = await self.client.get_settings()
                fw = str(info.get("fw_version"))
                if self._effects_fuer != fw:
                    antwort = await self.client.get_effects()
                    roh = antwort.get("effects")
                    self.effects = (
                        [e for e in roh if isinstance(e, dict)] if isinstance(roh, list) else []
                    )
                    self._effects_fuer = fw
        except NinstripAuthError as err:
            raise ConfigEntryAuthFailed(f"{self.client.host}: Token abgelehnt") from err
        except NinstripError as err:
            self._consecutive_poll_failures += 1
            if self.data is not None and self._consecutive_poll_failures < MAX_POLL_FAILURES:
                _LOGGER.debug(
                    "%s: Abfrage fehlgeschlagen (%d/%d), behalte den letzten Stand: %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 NinstripData(state, info, presets, segments, settings)

    async def async_call[T](self, action: Callable[[NinstripClient], Awaitable[T]]) -> T:
        """Fuehrt eine Anfrage unter der Sperre aus und uebersetzt Fehler fuer die Oberflaeche."""
        try:
            async with self.lock:
                return await action(self.client)
        except NinstripAuthError as err:
            self.config_entry.async_start_reauth(self.hass)
            raise HomeAssistantError(
                translation_domain=DOMAIN, translation_key="auth_failed"
            ) from err
        except NinstripBusyError as err:
            raise HomeAssistantError(
                translation_domain=DOMAIN, translation_key="device_busy"
            ) from err
        except NinstripRequestError as err:
            raise HomeAssistantError(
                translation_domain=DOMAIN,
                translation_key="request_failed",
                translation_placeholders={"message": err.message},
            ) from err
        except NinstripError as err:
            raise HomeAssistantError(
                translation_domain=DOMAIN, translation_key="cannot_connect"
            ) from err

    async def async_set_state(self, **fields: Any) -> None:
        """Schreibt und uebernimmt die vollstaendige Antwort des Geraets direkt."""
        state = await self.async_call(lambda client: client.set_state(**fields))
        self._consecutive_poll_failures = 0
        self.async_set_updated_data(
            NinstripData(
                state, self.data.info, self.data.presets, self.data.segments, self.data.settings
            )
        )

    async def async_set_sacn(self, **fields: Any) -> None:
        """sACN-Einstellungen schreiben und die Antwort des Geraets uebernehmen.

        Das Geraet antwortet nur mit {"status": "ok"}, also wird zurueckgelesen
        statt angenommen: es prueft selbst, und eine Entitaet, die das
        Gewuenschte statt des Gespeicherten zeigt, luegt genau dann, wenn
        beides auseinanderlaeuft. Wie ha-ninlight.
        """

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

        settings = await self.async_call(schreiben_und_lesen)
        self._consecutive_poll_failures = 0
        self.async_set_updated_data(
            NinstripData(
                self.data.state,
                self.data.info,
                self.data.presets,
                self.data.segments,
                settings,
            )
        )

    async def async_set_strip(self, **fields: Any) -> None:
        """Strip-Einstellungen schreiben und die Antwort des Geraets uebernehmen.

        Gleiches Muster wie async_set_sacn: das Geraet antwortet nur mit
        {"status": "ok"}, also wird zurueckgelesen statt angenommen.
        """

        async def schreiben_und_lesen(client: NinstripClient) -> dict[str, Any]:
            await client.set_strip(**fields)
            return await client.get_settings()

        settings = await self.async_call(schreiben_und_lesen)
        self._consecutive_poll_failures = 0
        self.async_set_updated_data(
            NinstripData(
                self.data.state,
                self.data.info,
                self.data.presets,
                self.data.segments,
                settings,
            )
        )

    async def async_set_all_segments(self, **fields: Any) -> None:
        """Schreibt dieselben Felder in JEDES Segment und liest zurueck.

        Die Grenzen und Namen der Segmente bleiben dabei stehen - gesetzt wird
        nur, was uebergeben wird. Eine Farbe ueber die Lichtentitaet soll die
        muehsam von Hand gemachte Einteilung des Strips nicht einreissen.

        Geschrieben wird die ganze Liste auf einmal: das Geraet nimmt sie als
        Ganzes an oder gar nicht, und es ist eine Anfrage statt einer je
        Segment. Danach wird zurueckgelesen statt angenommen, wie bei sACN.
        """

        async def schreiben_und_lesen(client: NinstripClient) -> dict[str, Any]:
            alt = self.data.segments.get("segments") or []
            neu = [{**seg, **fields} for seg in alt]
            await client.put_segments(neu)
            return await client.get_segments()

        segments = await self.async_call(schreiben_und_lesen)
        self._consecutive_poll_failures = 0
        self.async_set_updated_data(
            NinstripData(
                self.data.state,
                self.data.info,
                self.data.presets,
                segments,
                self.data.settings,
            )
        )

    async def async_after_change(self) -> None:
        """Nach einem Eingriff, der den ganzen Stand aendern kann (Szene, Segmente)."""
        await self.async_request_refresh()