"""Polling coordinator for the Riedel RRCS integration.""" from __future__ import annotations import logging from dataclasses import dataclass, field from datetime import timedelta from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from .const import DOMAIN from .rrcs import GpioAddress, LogicSource, RRCSClient, RRCSError _LOGGER = logging.getLogger(__name__) @dataclass class RRCSData: """Everything the entities read from.""" connected: bool = False gateway_state: str | None = None version: str | None = None active_xps: int | None = None logic_sources: dict[int, LogicSource] = field(default_factory=dict) gpio_inputs: dict[str, bool] = field(default_factory=dict) gpio_outputs: dict[str, bool] = field(default_factory=dict) class RRCSCoordinator(DataUpdateCoordinator[RRCSData]): """Polls the gateway and folds in pushed notifications between polls.""" def __init__( self, hass: HomeAssistant, entry: ConfigEntry, client: RRCSClient, scan_interval: int, poll_gpio: bool, ) -> None: """Initialise the coordinator.""" super().__init__( hass, _LOGGER, name=f"{DOMAIN} {client.host}", update_interval=timedelta(seconds=scan_interval), config_entry=entry, ) self.client = client self.poll_gpio = poll_gpio self.gpio_inputs: list[GpioAddress] = [] self.gpio_outputs: list[GpioAddress] = [] self.gpio_names: dict[str, str] = {} self._registration: tuple[int, str] | None = None def set_gpios( self, inputs: list[GpioAddress], outputs: list[GpioAddress], names: dict[str, str], ) -> None: """Record the GPIOs this entry exposes as entities.""" self.gpio_inputs = inputs self.gpio_outputs = outputs self.gpio_names = names def set_registration(self, tcp_port: int, url_path: str) -> None: """Remember the notification registration so it can be re-asserted.""" self._registration = (tcp_port, url_path) async def _async_update_data(self) -> RRCSData: """Fetch the current gateway and Artist state.""" data = RRCSData() try: data.connected = await self.client.is_connected_to_artist() data.gateway_state = await self.client.get_state() # The version never changes at runtime, so only ask once. previous = self.data data.version = previous.version if previous else None if data.version is None: data.version = await self.client.get_version() if data.connected: data.logic_sources = await self.client.get_logic_sources() data.active_xps = await self.client.get_active_xp_count() if self.poll_gpio: for address in self.gpio_inputs: data.gpio_inputs[address.key] = await self.client.get_gp_input_state( address ) for address in self.gpio_outputs: data.gpio_outputs[address.key] = ( await self.client.get_gp_output_state(address) ) elif previous is not None: data.gpio_inputs = dict(previous.gpio_inputs) data.gpio_outputs = dict(previous.gpio_outputs) elif previous is not None: # Keep the last known picture rather than blanking every entity # while the gateway is disconnected from the ring. data.logic_sources = dict(previous.logic_sources) data.gpio_inputs = dict(previous.gpio_inputs) data.gpio_outputs = dict(previous.gpio_outputs) except RRCSError as err: raise UpdateFailed(str(err)) from err await self._async_check_registration() return data async def _async_check_registration(self) -> None: """Re-register for notifications if RRCS has forgotten us. RRCS drops a notification channel when it restarts, when the Artist connection is re-established, or when a GetAlive goes unanswered, and it does not tell us that it has. """ if self._registration is None: return tcp_port, url_path = self._registration try: if await self.client.is_registered_for_all_events(tcp_port, url_path): return _LOGGER.info("RRCS notification registration lost; re-registering") await self.client.register_for_all_events(tcp_port, url_path) except RRCSError as err: _LOGGER.warning("Could not refresh RRCS notification registration: %s", err) # --- Push updates -------------------------------------------------------------- def apply_logic_source_change(self, object_id: int, state: bool) -> None: """Fold a LogicSourceChange notification into the current data.""" if self.data is None: return source = self.data.logic_sources.get(object_id) if source is None: # An object we have not enumerated yet; the next poll will pick it up. return self.data.logic_sources[object_id] = LogicSource( object_id=source.object_id, long_name=source.long_name, label=source.label, state=state, ) self.async_set_updated_data(self.data) def apply_gpio_change(self, address: GpioAddress, state: bool) -> None: """Fold a GpInputChange / GpOutputChange notification into the data.""" if self.data is None: return target = self.data.gpio_inputs if address.is_input else self.data.gpio_outputs target[address.key] = state self.async_set_updated_data(self.data) def apply_connection_change(self, connected: bool) -> None: """Fold a ConnectArtistFailure / ConnectArtistRestored notification in.""" if self.data is None: return self.data.connected = connected self.async_set_updated_data(self.data)