"""Switch platform for LED Screen Controller.""" from __future__ import annotations import logging from typing import Any from homeassistant.components.switch import SwitchEntity from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN, DATA_COORDINATOR from .coordinator import LedGrabCoordinator from .entity import LedGrabPlaylistEntity _LOGGER = logging.getLogger(__name__) async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: """Set up LED Screen Controller switches.""" data = hass.data[DOMAIN][entry.entry_id] coordinator: LedGrabCoordinator = data[DATA_COORDINATOR] entities: list[SwitchEntity] = [] if coordinator.data and "targets" in coordinator.data: for target_id, target_data in coordinator.data["targets"].items(): entities.append( LedGrabSwitch(coordinator, target_id, entry.entry_id) ) if coordinator.data: for clock in coordinator.data.get("sync_clocks", []): entities.append( LedGrabSyncClockSwitch(coordinator, clock["id"], entry.entry_id) ) for playlist in coordinator.data.get("scene_playlists", []): entities.append( LedGrabPlaylistSwitch(coordinator, playlist["id"], entry.entry_id) ) async_add_entities(entities) class LedGrabSwitch(CoordinatorEntity, SwitchEntity): """Representation of a LED Screen Controller target processing switch.""" _attr_has_entity_name = True def __init__( self, coordinator: LedGrabCoordinator, target_id: str, entry_id: str, ) -> None: """Initialize the switch.""" super().__init__(coordinator) self._target_id = target_id self._entry_id = entry_id self._attr_unique_id = f"{target_id}_processing" self._attr_translation_key = "processing" self._attr_icon = "mdi:television-ambient-light" @property def device_info(self) -> dict[str, Any]: """Return device information.""" return {"identifiers": {(DOMAIN, self._target_id)}} @property def is_on(self) -> bool: """Return true if processing is active.""" target_data = self._get_target_data() if not target_data or not target_data.get("state"): return False return target_data["state"].get("processing", False) @property def available(self) -> bool: """Return if entity is available.""" return self._get_target_data() is not None @property def extra_state_attributes(self) -> dict[str, Any]: """Return additional state attributes.""" target_data = self._get_target_data() if not target_data: return {} attrs: dict[str, Any] = {"target_id": self._target_id} state = target_data.get("state") or {} metrics = target_data.get("metrics") or {} if state: attrs["fps_target"] = state.get("fps_target") attrs["fps_actual"] = state.get("fps_actual") if metrics: attrs["frames_processed"] = metrics.get("frames_processed") attrs["errors_count"] = metrics.get("errors_count") attrs["uptime_seconds"] = metrics.get("uptime_seconds") return attrs async def async_turn_on(self, **kwargs: Any) -> None: """Start processing.""" await self.coordinator.start_processing(self._target_id) async def async_turn_off(self, **kwargs: Any) -> None: """Stop processing.""" await self.coordinator.stop_processing(self._target_id) def _get_target_data(self) -> dict[str, Any] | None: """Get target data from coordinator.""" if not self.coordinator.data: return None return self.coordinator.data.get("targets", {}).get(self._target_id) class LedGrabSyncClockSwitch(CoordinatorEntity, SwitchEntity): """Running/paused control for a sync clock. On = clock running (linked animations advance), off = paused (frozen). """ _attr_has_entity_name = True _attr_icon = "mdi:clock-outline" def __init__( self, coordinator: LedGrabCoordinator, clock_id: str, entry_id: str, ) -> None: super().__init__(coordinator) self._clock_id = clock_id self._entry_id = entry_id self._attr_unique_id = f"{clock_id}_running" self._attr_translation_key = "sync_clock_running" @property def device_info(self) -> dict[str, Any]: return {"identifiers": {(DOMAIN, self._clock_id)}} @property def is_on(self) -> bool: clock = self._get_clock() if not clock: return False return bool(clock.get("is_running", False)) @property def available(self) -> bool: return self._get_clock() is not None async def async_turn_on(self, **kwargs: Any) -> None: await self.coordinator.resume_sync_clock(self._clock_id) async def async_turn_off(self, **kwargs: Any) -> None: await self.coordinator.pause_sync_clock(self._clock_id) def _get_clock(self) -> dict[str, Any] | None: if not self.coordinator.data: return None for clock in self.coordinator.data.get("sync_clocks", []): if clock.get("id") == self._clock_id: return clock return None class LedGrabPlaylistSwitch(LedGrabPlaylistEntity, SwitchEntity): """Start/stop control for a scene playlist. On = this playlist is cycling, off = stopped. The server runs at most one playlist at a time, so turning one on stops any other that was running (the other switches flip off on the next refresh). Device/availability/lookup plumbing lives in :class:`LedGrabPlaylistEntity` so it stays in sync with the playlist stats sensors. """ _attr_icon = "mdi:playlist-play" def __init__( self, coordinator: LedGrabCoordinator, playlist_id: str, entry_id: str, ) -> None: super().__init__(coordinator, playlist_id, entry_id) self._attr_unique_id = f"{playlist_id}_active" self._attr_translation_key = "playlist_active" @property def is_on(self) -> bool: playlist = self._get_playlist() if not playlist: return False return bool(playlist.get("is_running", False)) @property def extra_state_attributes(self) -> dict[str, Any]: playlist = self._get_playlist() if not playlist: return {} attrs: dict[str, Any] = { "playlist_id": self._playlist_id, "item_count": len(playlist.get("items") or []), "loop": playlist.get("loop"), "shuffle": playlist.get("shuffle"), "tags": playlist.get("tags") or [], } # Runtime cycling details only apply to whichever playlist is running. state = self._running_state() if state: attrs["current_index"] = state.get("current_index") attrs["current_preset_id"] = state.get("current_preset_id") attrs["step_duration"] = state.get("step_duration") attrs["started_at"] = state.get("started_at") return attrs async def async_turn_on(self, **kwargs: Any) -> None: await self.coordinator.start_playlist(self._playlist_id) async def async_turn_off(self, **kwargs: Any) -> None: # Stop is global; only act if this playlist is the one actually cycling # so toggling an already-idle switch can't stop a different playlist. if self.is_on: await self.coordinator.stop_playlist()