alexei.dolgolyov
579553a850
HACS-compatible custom component split out from the main LedGrab repo. Creates light, switch, sensor, number, and select entities for each configured LedGrab device.
110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
"""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
|
|
|
|
_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 = []
|
|
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)
|
|
)
|
|
|
|
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)
|