alexei.dolgolyov
a666d9eb9c
- Server device exposing CPU/RAM/GPU/temperature/battery sensors via /api/v1/system/performance, plus last-restart timestamp (cached with jitter threshold so the recorder doesn't see poll wobble) and version. - Update entity backed by /api/v1/system/update — installs via /apply, hides the install button when the server reports can_auto_update=false. - Sync-clock entities: reset button, speed number, running switch, and the event listener now refreshes on entity_changed events too. - Bump manifest to 0.4.0.
363 lines
12 KiB
Python
363 lines
12 KiB
Python
"""Number platform for LED Screen Controller (device brightness & HA light settings)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
from homeassistant.components.number import NumberEntity, NumberMode
|
||
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, TARGET_TYPE_HA_LIGHT
|
||
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 number entities."""
|
||
data = hass.data[DOMAIN][entry.entry_id]
|
||
coordinator: LedGrabCoordinator = data[DATA_COORDINATOR]
|
||
|
||
entities: list[NumberEntity] = []
|
||
if coordinator.data:
|
||
for source in coordinator.data.get("css_sources") or []:
|
||
if source.get("source_type") == "api_input":
|
||
entities.append(ApiInputTimeout(coordinator, source, entry.entry_id))
|
||
|
||
for clock in coordinator.data.get("sync_clocks") or []:
|
||
entities.append(SyncClockSpeed(coordinator, clock["id"], entry.entry_id))
|
||
|
||
if coordinator.data and "targets" in coordinator.data:
|
||
devices = coordinator.data.get("devices") or {}
|
||
|
||
for target_id, target_data in coordinator.data["targets"].items():
|
||
info = target_data["info"]
|
||
target_type = info.get("target_type", "led")
|
||
|
||
if target_type == TARGET_TYPE_HA_LIGHT:
|
||
# HA Light target — expose tunable settings
|
||
entities.append(HALightUpdateRate(coordinator, target_id, entry.entry_id))
|
||
entities.append(HALightTransition(coordinator, target_id, entry.entry_id))
|
||
entities.append(HALightMinBrightness(coordinator, target_id, entry.entry_id))
|
||
entities.append(HALightColorTolerance(coordinator, target_id, entry.entry_id))
|
||
continue
|
||
|
||
# LED target — brightness lives on the device
|
||
device_id = info.get("device_id", "")
|
||
if not device_id:
|
||
continue
|
||
|
||
device_data = devices.get(device_id)
|
||
if not device_data:
|
||
continue
|
||
|
||
capabilities = device_data.get("info", {}).get("capabilities") or []
|
||
if "brightness_control" not in capabilities or "static_color" in capabilities:
|
||
continue
|
||
|
||
entities.append(
|
||
LedGrabBrightness(
|
||
coordinator,
|
||
target_id,
|
||
device_id,
|
||
entry.entry_id,
|
||
)
|
||
)
|
||
|
||
async_add_entities(entities)
|
||
|
||
|
||
class LedGrabBrightness(CoordinatorEntity, NumberEntity):
|
||
"""Brightness control for an LED device associated with a target."""
|
||
|
||
_attr_has_entity_name = True
|
||
_attr_native_min_value = 0
|
||
_attr_native_max_value = 255
|
||
_attr_native_step = 1
|
||
_attr_mode = NumberMode.SLIDER
|
||
_attr_icon = "mdi:brightness-6"
|
||
|
||
def __init__(
|
||
self,
|
||
coordinator: LedGrabCoordinator,
|
||
target_id: str,
|
||
device_id: str,
|
||
entry_id: str,
|
||
) -> None:
|
||
"""Initialize the brightness number."""
|
||
super().__init__(coordinator)
|
||
self._target_id = target_id
|
||
self._device_id = device_id
|
||
self._entry_id = entry_id
|
||
self._attr_unique_id = f"{target_id}_brightness"
|
||
self._attr_translation_key = "brightness"
|
||
|
||
@property
|
||
def device_info(self) -> dict[str, Any]:
|
||
"""Return device information."""
|
||
return {"identifiers": {(DOMAIN, self._target_id)}}
|
||
|
||
@property
|
||
def native_value(self) -> float | None:
|
||
"""Return the current brightness value."""
|
||
if not self.coordinator.data:
|
||
return None
|
||
device_data = self.coordinator.data.get("devices", {}).get(self._device_id)
|
||
if not device_data:
|
||
return None
|
||
return device_data.get("brightness")
|
||
|
||
@property
|
||
def available(self) -> bool:
|
||
"""Return if entity is available."""
|
||
if not self.coordinator.data:
|
||
return False
|
||
targets = self.coordinator.data.get("targets", {})
|
||
devices = self.coordinator.data.get("devices", {})
|
||
return self._target_id in targets and self._device_id in devices
|
||
|
||
async def async_set_native_value(self, value: float) -> None:
|
||
"""Set brightness value."""
|
||
await self.coordinator.set_brightness(self._device_id, int(value))
|
||
|
||
|
||
# --- HA Light target number entities ---
|
||
|
||
|
||
class _HALightNumberBase(CoordinatorEntity, NumberEntity):
|
||
"""Base class for HA Light target number entities."""
|
||
|
||
_attr_has_entity_name = True
|
||
_attr_mode = NumberMode.SLIDER
|
||
|
||
def __init__(
|
||
self,
|
||
coordinator: LedGrabCoordinator,
|
||
target_id: str,
|
||
entry_id: str,
|
||
*,
|
||
field_name: str,
|
||
) -> None:
|
||
super().__init__(coordinator)
|
||
self._target_id = target_id
|
||
self._entry_id = entry_id
|
||
self._field_name = field_name
|
||
|
||
@property
|
||
def device_info(self) -> dict[str, Any]:
|
||
return {"identifiers": {(DOMAIN, self._target_id)}}
|
||
|
||
@property
|
||
def native_value(self) -> float | None:
|
||
target_data = self._get_target_data()
|
||
if not target_data:
|
||
return None
|
||
return target_data.get("info", {}).get(self._field_name)
|
||
|
||
@property
|
||
def available(self) -> bool:
|
||
return self._get_target_data() is not None
|
||
|
||
async def async_set_native_value(self, value: float) -> None:
|
||
await self.coordinator.update_target(self._target_id, **{self._field_name: round(value, 2)})
|
||
|
||
def _get_target_data(self) -> dict[str, Any] | None:
|
||
if not self.coordinator.data:
|
||
return None
|
||
return self.coordinator.data.get("targets", {}).get(self._target_id)
|
||
|
||
|
||
class HALightUpdateRate(_HALightNumberBase):
|
||
"""Update rate (Hz) for an HA Light target."""
|
||
|
||
_attr_native_min_value = 0.5
|
||
_attr_native_max_value = 5.0
|
||
_attr_native_step = 0.5
|
||
_attr_native_unit_of_measurement = "Hz"
|
||
_attr_icon = "mdi:update"
|
||
|
||
def __init__(
|
||
self, coordinator: LedGrabCoordinator, target_id: str, entry_id: str
|
||
) -> None:
|
||
super().__init__(coordinator, target_id, entry_id, field_name="update_rate")
|
||
self._attr_unique_id = f"{target_id}_update_rate"
|
||
self._attr_translation_key = "ha_light_update_rate"
|
||
|
||
|
||
class HALightTransition(_HALightNumberBase):
|
||
"""Transition time (seconds) for an HA Light target."""
|
||
|
||
_attr_native_min_value = 0.0
|
||
_attr_native_max_value = 10.0
|
||
_attr_native_step = 0.1
|
||
_attr_native_unit_of_measurement = "s"
|
||
_attr_icon = "mdi:transition-masked"
|
||
|
||
def __init__(
|
||
self, coordinator: LedGrabCoordinator, target_id: str, entry_id: str
|
||
) -> None:
|
||
super().__init__(coordinator, target_id, entry_id, field_name="transition")
|
||
self._attr_unique_id = f"{target_id}_transition"
|
||
self._attr_translation_key = "ha_light_transition"
|
||
|
||
|
||
class HALightMinBrightness(_HALightNumberBase):
|
||
"""Minimum brightness threshold for an HA Light target."""
|
||
|
||
_attr_native_min_value = 0
|
||
_attr_native_max_value = 255
|
||
_attr_native_step = 1
|
||
_attr_icon = "mdi:brightness-4"
|
||
|
||
def __init__(
|
||
self, coordinator: LedGrabCoordinator, target_id: str, entry_id: str
|
||
) -> None:
|
||
super().__init__(coordinator, target_id, entry_id, field_name="min_brightness_threshold")
|
||
self._attr_unique_id = f"{target_id}_min_brightness"
|
||
self._attr_translation_key = "ha_light_min_brightness"
|
||
|
||
|
||
class HALightColorTolerance(_HALightNumberBase):
|
||
"""Color tolerance (RGB delta skip threshold) for an HA Light target."""
|
||
|
||
_attr_native_min_value = 0
|
||
_attr_native_max_value = 50
|
||
_attr_native_step = 1
|
||
_attr_icon = "mdi:palette-outline"
|
||
|
||
def __init__(
|
||
self, coordinator: LedGrabCoordinator, target_id: str, entry_id: str
|
||
) -> None:
|
||
super().__init__(coordinator, target_id, entry_id, field_name="color_tolerance")
|
||
self._attr_unique_id = f"{target_id}_color_tolerance"
|
||
self._attr_translation_key = "ha_light_color_tolerance"
|
||
|
||
|
||
# --- api_input CSS source number entities ---
|
||
|
||
|
||
class ApiInputTimeout(CoordinatorEntity, NumberEntity):
|
||
"""Fallback timeout for an api_input CSS source.
|
||
|
||
The server reverts the strip to ``fallback_color`` when no LED data has
|
||
arrived for this many seconds. Stored server-side as a ``BindableFloat``
|
||
(plain number when unbound, ``{"value", "source_id"}`` when bound).
|
||
"""
|
||
|
||
_attr_has_entity_name = True
|
||
_attr_native_min_value = 0.0
|
||
_attr_native_max_value = 300.0
|
||
_attr_native_step = 0.5
|
||
_attr_native_unit_of_measurement = "s"
|
||
_attr_mode = NumberMode.BOX
|
||
_attr_icon = "mdi:timer-sand"
|
||
|
||
def __init__(
|
||
self,
|
||
coordinator: LedGrabCoordinator,
|
||
source: dict[str, Any],
|
||
entry_id: str,
|
||
) -> None:
|
||
super().__init__(coordinator)
|
||
self._source_id: str = source["id"]
|
||
self._entry_id = entry_id
|
||
self._attr_unique_id = f"{self._source_id}_timeout"
|
||
self._attr_translation_key = "api_input_timeout"
|
||
|
||
@property
|
||
def device_info(self) -> dict[str, Any]:
|
||
return {"identifiers": {(DOMAIN, self._source_id)}}
|
||
|
||
@property
|
||
def native_value(self) -> float | None:
|
||
source = self._get_source()
|
||
if not source:
|
||
return None
|
||
raw = source.get("timeout")
|
||
if isinstance(raw, dict):
|
||
return raw.get("value")
|
||
if isinstance(raw, (int, float)):
|
||
return float(raw)
|
||
return None
|
||
|
||
@property
|
||
def available(self) -> bool:
|
||
return self._get_source() is not None
|
||
|
||
async def async_set_native_value(self, value: float) -> None:
|
||
await self.coordinator.update_source(self._source_id, timeout=round(value, 2))
|
||
await self.coordinator.async_request_refresh()
|
||
|
||
def _get_source(self) -> dict[str, Any] | None:
|
||
if not self.coordinator.data:
|
||
return None
|
||
for source in self.coordinator.data.get("css_sources") or []:
|
||
if source.get("id") == self._source_id:
|
||
return source
|
||
return None
|
||
|
||
|
||
# --- Sync clock number entities ---
|
||
|
||
|
||
class SyncClockSpeed(CoordinatorEntity, NumberEntity):
|
||
"""Speed multiplier for a sync clock (server range 0.1–10.0).
|
||
|
||
Hot-applied — running animations linked to the clock change pace
|
||
immediately without resetting their position.
|
||
"""
|
||
|
||
_attr_has_entity_name = True
|
||
_attr_native_min_value = 0.1
|
||
_attr_native_max_value = 10.0
|
||
_attr_native_step = 0.1
|
||
_attr_mode = NumberMode.SLIDER
|
||
_attr_icon = "mdi:speedometer"
|
||
|
||
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}_speed"
|
||
self._attr_translation_key = "sync_clock_speed"
|
||
|
||
@property
|
||
def device_info(self) -> dict[str, Any]:
|
||
return {"identifiers": {(DOMAIN, self._clock_id)}}
|
||
|
||
@property
|
||
def native_value(self) -> float | None:
|
||
clock = self._get_clock()
|
||
if not clock:
|
||
return None
|
||
speed = clock.get("speed")
|
||
return float(speed) if speed is not None else None
|
||
|
||
@property
|
||
def available(self) -> bool:
|
||
return self._get_clock() is not None
|
||
|
||
async def async_set_native_value(self, value: float) -> None:
|
||
await self.coordinator.update_sync_clock(self._clock_id, speed=round(value, 2))
|
||
|
||
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
|