From 579553a8509fcf0150319e3dd47dd5096e16bf32 Mon Sep 17 00:00:00 2001 From: "alexei.dolgolyov" Date: Sun, 12 Apr 2026 22:32:46 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20initial=20release=20=E2=80=94=20LedGrab?= =?UTF-8?q?=20Home=20Assistant=20integration?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HACS-compatible custom component split out from the main LedGrab repo. Creates light, switch, sensor, number, and select entities for each configured LedGrab device. --- LICENSE | 21 + README.md | 64 +++ custom_components/ledgrab/__init__.py | 182 ++++++++ custom_components/ledgrab/button.py | 74 +++ custom_components/ledgrab/config_flow.py | 127 ++++++ custom_components/ledgrab/const.py | 22 + custom_components/ledgrab/coordinator.py | 426 ++++++++++++++++++ custom_components/ledgrab/event_listener.py | 95 ++++ custom_components/ledgrab/light.py | 151 +++++++ custom_components/ledgrab/manifest.json | 12 + custom_components/ledgrab/number.py | 233 ++++++++++ custom_components/ledgrab/select.py | 178 ++++++++ custom_components/ledgrab/sensor.py | 225 +++++++++ custom_components/ledgrab/services.yaml | 19 + custom_components/ledgrab/strings.json | 103 +++++ custom_components/ledgrab/switch.py | 109 +++++ .../ledgrab/translations/en.json | 87 ++++ .../ledgrab/translations/ru.json | 87 ++++ hacs.json | 6 + 19 files changed, 2221 insertions(+) create mode 100644 LICENSE create mode 100644 README.md create mode 100644 custom_components/ledgrab/__init__.py create mode 100644 custom_components/ledgrab/button.py create mode 100644 custom_components/ledgrab/config_flow.py create mode 100644 custom_components/ledgrab/const.py create mode 100644 custom_components/ledgrab/coordinator.py create mode 100644 custom_components/ledgrab/event_listener.py create mode 100644 custom_components/ledgrab/light.py create mode 100644 custom_components/ledgrab/manifest.json create mode 100644 custom_components/ledgrab/number.py create mode 100644 custom_components/ledgrab/select.py create mode 100644 custom_components/ledgrab/sensor.py create mode 100644 custom_components/ledgrab/services.yaml create mode 100644 custom_components/ledgrab/strings.json create mode 100644 custom_components/ledgrab/switch.py create mode 100644 custom_components/ledgrab/translations/en.json create mode 100644 custom_components/ledgrab/translations/ru.json create mode 100644 hacs.json diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..2ceb045 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Alexei Dolgolyov + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..c7d4142 --- /dev/null +++ b/README.md @@ -0,0 +1,64 @@ +# LedGrab Home Assistant Integration + +Custom Home Assistant integration for [LedGrab](https://git.dolgolyov-family.by/alexei.dolgolyov/ledgrab) — ambient lighting system that captures screen content and drives LED strips in real time. + +Creates light, switch, sensor, number, and select entities for each configured device, allowing full control and automation from Home Assistant. + +## Installation + +### Option 1: HACS (recommended) + +1. Install [HACS](https://hacs.xyz/docs/setup/download) if you have not already. +2. Open HACS in Home Assistant. +3. Click the three-dot menu, then **Custom repositories**. +4. Add URL: `https://git.dolgolyov-family.by/alexei.dolgolyov/ledgrab-haos-integration` +5. Set category to **Integration** and click **Add**. +6. Search for "LedGrab" in HACS and click **Download**. +7. Restart Home Assistant. +8. Go to **Settings > Devices & Services > Add Integration** and search for "LedGrab". +9. Enter your server URL (e.g., `http://192.168.1.100:8080`) and API key. + +### Option 2: Manual + +Copy the `custom_components/ledgrab/` folder from this repository into your Home Assistant `config/custom_components/` directory, then restart Home Assistant and add the integration as above. + +## Automation Example + +```yaml +automation: + - alias: "Start ambient lighting when TV turns on" + trigger: + - platform: state + entity_id: media_player.living_room_tv + to: "on" + action: + - service: switch.turn_on + target: + entity_id: switch.living_room_tv_processing + + - alias: "Stop ambient lighting when TV turns off" + trigger: + - platform: state + entity_id: media_player.living_room_tv + to: "off" + action: + - service: switch.turn_off + target: + entity_id: switch.living_room_tv_processing +``` + +## Requirements + +- Home Assistant 2023.1.0 or later +- A running [LedGrab](https://git.dolgolyov-family.by/alexei.dolgolyov/ledgrab) server accessible on your network + +## Troubleshooting + +1. Verify the component exists: check that `config/custom_components/ledgrab/` is present. +2. Clear the browser cache and hard-refresh the HA UI. +3. Restart Home Assistant. +4. Check logs at **Settings > System > Logs** and search for `ledgrab`. + +## License + +MIT diff --git a/custom_components/ledgrab/__init__.py b/custom_components/ledgrab/__init__.py new file mode 100644 index 0000000..23f2df8 --- /dev/null +++ b/custom_components/ledgrab/__init__.py @@ -0,0 +1,182 @@ +"""The LED Screen Controller integration.""" + +from __future__ import annotations + +import logging +from datetime import timedelta + +import voluptuous as vol + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant +from homeassistant.helpers import device_registry as dr +from homeassistant.helpers.aiohttp_client import async_get_clientsession + +from .const import ( + DOMAIN, + CONF_SERVER_NAME, + CONF_SERVER_URL, + CONF_API_KEY, + DEFAULT_SCAN_INTERVAL, + TARGET_TYPE_HA_LIGHT, + DATA_COORDINATOR, + DATA_EVENT_LISTENER, +) +from .coordinator import LedGrabCoordinator +from .event_listener import EventStreamListener + +_LOGGER = logging.getLogger(__name__) + +PLATFORMS: list[Platform] = [ + Platform.BUTTON, + Platform.LIGHT, + Platform.SWITCH, + Platform.SENSOR, + Platform.NUMBER, + Platform.SELECT, +] + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up LED Screen Controller from a config entry.""" + server_name = entry.data.get(CONF_SERVER_NAME, "LED Screen Controller") + server_url = entry.data[CONF_SERVER_URL] + api_key = entry.data[CONF_API_KEY] + + session = async_get_clientsession(hass) + coordinator = LedGrabCoordinator( + hass, + session, + server_url, + api_key, + update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL), + ) + + await coordinator.async_config_entry_first_refresh() + + event_listener = EventStreamListener(hass, server_url, api_key, coordinator) + await event_listener.start() + + # Create device entries for each target and remove stale ones + device_registry = dr.async_get(hass) + current_identifiers: set[tuple[str, str]] = set() + if coordinator.data and "targets" in coordinator.data: + 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: + model = "HA Light Target" + else: + model = "LED Target" + device_registry.async_get_or_create( + config_entry_id=entry.entry_id, + identifiers={(DOMAIN, target_id)}, + name=info.get("name", target_id), + manufacturer=server_name, + model=model, + configuration_url=server_url, + ) + current_identifiers.add((DOMAIN, target_id)) + + # Create a single "Scenes" device for scene preset buttons + scenes_identifier = (DOMAIN, f"{entry.entry_id}_scenes") + scene_presets = coordinator.data.get("scene_presets", []) if coordinator.data else [] + if scene_presets: + device_registry.async_get_or_create( + config_entry_id=entry.entry_id, + identifiers={scenes_identifier}, + name="Scenes", + manufacturer=server_name, + model="Scene Presets", + configuration_url=server_url, + ) + current_identifiers.add(scenes_identifier) + + # Remove devices for targets that no longer exist + for device_entry in dr.async_entries_for_config_entry(device_registry, entry.entry_id): + if not device_entry.identifiers & current_identifiers: + _LOGGER.info("Removing stale device: %s", device_entry.name) + device_registry.async_remove_device(device_entry.id) + + # Store data + hass.data.setdefault(DOMAIN, {}) + hass.data[DOMAIN][entry.entry_id] = { + DATA_COORDINATOR: coordinator, + DATA_EVENT_LISTENER: event_listener, + } + + # Track target and scene IDs to detect changes + known_target_ids = set(coordinator.data.get("targets", {}).keys() if coordinator.data else []) + known_scene_ids = set( + p["id"] for p in (coordinator.data.get("scene_presets", []) if coordinator.data else []) + ) + + def _on_coordinator_update() -> None: + """Detect target/scene list changes and trigger reload.""" + nonlocal known_target_ids, known_scene_ids + + if not coordinator.data: + return + + targets = coordinator.data.get("targets", {}) + + # Reload if target or scene list changed + current_ids = set(targets.keys()) + current_scene_ids = set(p["id"] for p in coordinator.data.get("scene_presets", [])) + if current_ids != known_target_ids or current_scene_ids != known_scene_ids: + known_target_ids = current_ids + known_scene_ids = current_scene_ids + _LOGGER.info("Target or scene list changed, reloading integration") + hass.async_create_task(hass.config_entries.async_reload(entry.entry_id)) + + coordinator.async_add_listener(_on_coordinator_update) + + # Register set_leds service (once across all entries) + async def handle_set_leds(call) -> None: + """Handle the set_leds service call.""" + source_id = call.data["source_id"] + segments = call.data["segments"] + # Route to the coordinator that owns this source + for entry_data in hass.data[DOMAIN].values(): + coord = entry_data.get(DATA_COORDINATOR) + if not coord or not coord.data: + continue + source_ids = {s["id"] for s in coord.data.get("css_sources", [])} + if source_id in source_ids: + await coord.push_segments(source_id, segments) + return + _LOGGER.error("No server found with source_id %s", source_id) + + if not hass.services.has_service(DOMAIN, "set_leds"): + hass.services.async_register( + DOMAIN, + "set_leds", + handle_set_leds, + schema=vol.Schema( + { + vol.Required("source_id"): str, + vol.Required("segments"): list, + } + ), + ) + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Unload a config entry.""" + entry_data = hass.data[DOMAIN][entry.entry_id] + await entry_data[DATA_EVENT_LISTENER].shutdown() + + unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + + if unload_ok: + hass.data[DOMAIN].pop(entry.entry_id) + # Unregister service if no entries remain + if not hass.data[DOMAIN]: + hass.services.async_remove(DOMAIN, "set_leds") + + return unload_ok diff --git a/custom_components/ledgrab/button.py b/custom_components/ledgrab/button.py new file mode 100644 index 0000000..2c7f684 --- /dev/null +++ b/custom_components/ledgrab/button.py @@ -0,0 +1,74 @@ +"""Button platform for LED Screen Controller — scene preset activation.""" +from __future__ import annotations + +import logging +from typing import Any + +from homeassistant.components.button import ButtonEntity +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 scene preset buttons.""" + data = hass.data[DOMAIN][entry.entry_id] + coordinator: LedGrabCoordinator = data[DATA_COORDINATOR] + + entities = [] + if coordinator.data: + for preset in coordinator.data.get("scene_presets", []): + entities.append( + SceneActivateButton(coordinator, preset, entry.entry_id) + ) + + async_add_entities(entities) + + +class SceneActivateButton(CoordinatorEntity, ButtonEntity): + """Button that activates a scene preset.""" + + _attr_has_entity_name = True + + def __init__( + self, + coordinator: LedGrabCoordinator, + preset: dict[str, Any], + entry_id: str, + ) -> None: + """Initialize the button.""" + super().__init__(coordinator) + self._preset_id = preset["id"] + self._entry_id = entry_id + self._attr_unique_id = f"{entry_id}_scene_{preset['id']}" + self._attr_translation_key = "activate_scene" + self._attr_translation_placeholders = {"scene_name": preset["name"]} + self._attr_icon = "mdi:palette" + + @property + def device_info(self) -> dict[str, Any]: + """Return device information — all scene buttons belong to the Scenes device.""" + return {"identifiers": {(DOMAIN, f"{self._entry_id}_scenes")}} + + @property + def available(self) -> bool: + """Return if entity is available.""" + if not self.coordinator.data: + return False + return self._preset_id in { + p["id"] for p in self.coordinator.data.get("scene_presets", []) + } + + async def async_press(self) -> None: + """Activate the scene preset.""" + await self.coordinator.activate_scene(self._preset_id) diff --git a/custom_components/ledgrab/config_flow.py b/custom_components/ledgrab/config_flow.py new file mode 100644 index 0000000..eac58ed --- /dev/null +++ b/custom_components/ledgrab/config_flow.py @@ -0,0 +1,127 @@ +"""Config flow for LED Screen Controller integration.""" +from __future__ import annotations + +import logging +from typing import Any +from urllib.parse import urlparse, urlunparse + +import aiohttp +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResult +from homeassistant.helpers.aiohttp_client import async_get_clientsession + +from .const import DOMAIN, CONF_SERVER_NAME, CONF_SERVER_URL, CONF_API_KEY, DEFAULT_TIMEOUT + +_LOGGER = logging.getLogger(__name__) + +STEP_USER_DATA_SCHEMA = vol.Schema( + { + vol.Optional(CONF_SERVER_NAME, default="LED Screen Controller"): str, + vol.Required(CONF_SERVER_URL, default="http://localhost:8080"): str, + vol.Optional(CONF_API_KEY, default=""): str, + } +) + + +def normalize_url(url: str) -> str: + """Normalize URL to ensure port is an integer.""" + parsed = urlparse(url) + + if parsed.port is not None: + netloc = parsed.hostname or "localhost" + port = int(parsed.port) + if port != (443 if parsed.scheme == "https" else 80): + netloc = f"{netloc}:{port}" + parsed = parsed._replace(netloc=netloc) + + return urlunparse(parsed) + + +async def validate_server( + hass: HomeAssistant, server_url: str, api_key: str +) -> dict[str, Any]: + """Validate server connectivity and API key.""" + session = async_get_clientsession(hass) + timeout = aiohttp.ClientTimeout(total=DEFAULT_TIMEOUT) + + # Step 1: Check connectivity via health endpoint (no auth needed) + try: + async with session.get(f"{server_url}/health", timeout=timeout) as resp: + if resp.status != 200: + raise ConnectionError(f"Server returned status {resp.status}") + data = await resp.json() + version = data.get("version", "unknown") + except aiohttp.ClientError as err: + raise ConnectionError(f"Cannot connect to server: {err}") from err + + # Step 2: Validate API key via authenticated endpoint (skip if no key and auth not required) + auth_required = data.get("auth_required", True) + if api_key: + headers = {"Authorization": f"Bearer {api_key}"} + try: + async with session.get( + f"{server_url}/api/v1/output-targets", + headers=headers, + timeout=timeout, + ) as resp: + if resp.status == 401: + raise PermissionError("Invalid API key") + resp.raise_for_status() + except PermissionError: + raise + except aiohttp.ClientError as err: + raise ConnectionError(f"API request failed: {err}") from err + elif auth_required: + raise PermissionError("Server requires an API key") + + return {"version": version} + + +class LedGrabConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for LED Screen Controller.""" + + VERSION = 2 + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Handle the initial step.""" + errors: dict[str, str] = {} + + if user_input is not None: + server_name = user_input.get(CONF_SERVER_NAME, "LED Screen Controller") + server_url = normalize_url(user_input[CONF_SERVER_URL].rstrip("/")) + api_key = user_input[CONF_API_KEY] + + try: + await validate_server(self.hass, server_url, api_key) + + await self.async_set_unique_id(server_url) + self._abort_if_unique_id_configured() + + return self.async_create_entry( + title=server_name, + data={ + CONF_SERVER_NAME: server_name, + CONF_SERVER_URL: server_url, + CONF_API_KEY: api_key, + }, + ) + + except ConnectionError as err: + _LOGGER.error("Connection error: %s", err) + errors["base"] = "cannot_connect" + except PermissionError: + errors["base"] = "invalid_api_key" + except Exception as err: + _LOGGER.exception("Unexpected exception: %s", err) + errors["base"] = "unknown" + + return self.async_show_form( + step_id="user", + data_schema=STEP_USER_DATA_SCHEMA, + errors=errors, + ) diff --git a/custom_components/ledgrab/const.py b/custom_components/ledgrab/const.py new file mode 100644 index 0000000..1f43125 --- /dev/null +++ b/custom_components/ledgrab/const.py @@ -0,0 +1,22 @@ +"""Constants for the LedGrab Home Assistant integration.""" + +DOMAIN = "ledgrab" + +# Configuration +CONF_SERVER_NAME = "server_name" +CONF_SERVER_URL = "server_url" +CONF_API_KEY = "api_key" + +# Default values +DEFAULT_SCAN_INTERVAL = 3 # seconds +DEFAULT_TIMEOUT = 10 # seconds +WS_RECONNECT_DELAY = 5 # seconds +WS_MAX_RECONNECT_DELAY = 60 # seconds + +# Target types +TARGET_TYPE_LED = "led" +TARGET_TYPE_HA_LIGHT = "ha_light" + +# Data keys stored in hass.data[DOMAIN][entry_id] +DATA_COORDINATOR = "coordinator" +DATA_EVENT_LISTENER = "event_listener" diff --git a/custom_components/ledgrab/coordinator.py b/custom_components/ledgrab/coordinator.py new file mode 100644 index 0000000..1a00a0d --- /dev/null +++ b/custom_components/ledgrab/coordinator.py @@ -0,0 +1,426 @@ +"""Data update coordinator for LED Screen Controller.""" + +from __future__ import annotations + +import asyncio +from datetime import timedelta +import logging +from typing import Any + +import aiohttp + +from homeassistant.core import HomeAssistant +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed + +from .const import ( + DOMAIN, + DEFAULT_TIMEOUT, +) + +_LOGGER = logging.getLogger(__name__) + + +class LedGrabCoordinator(DataUpdateCoordinator): + """Class to manage fetching LED Screen Controller data.""" + + def __init__( + self, + hass: HomeAssistant, + session: aiohttp.ClientSession, + server_url: str, + api_key: str, + update_interval: timedelta, + ) -> None: + """Initialize the coordinator.""" + self.server_url = server_url + self.session = session + self.api_key = api_key + self.server_version = "unknown" + self._auth_headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} + self._timeout = aiohttp.ClientTimeout(total=DEFAULT_TIMEOUT) + + super().__init__( + hass, + _LOGGER, + name=DOMAIN, + update_interval=update_interval, + ) + + async def _async_update_data(self) -> dict[str, Any]: + """Fetch data from API.""" + try: + async with asyncio.timeout(DEFAULT_TIMEOUT * 3): + if self.server_version == "unknown": + await self._fetch_server_version() + + targets_list = await self._fetch_targets() + + # Fetch state and metrics for all targets in parallel + targets_data: dict[str, dict[str, Any]] = {} + + async def fetch_target_data(target: dict) -> tuple[str, dict]: + target_id = target["id"] + try: + state, metrics = await asyncio.gather( + self._fetch_target_state(target_id), + self._fetch_target_metrics(target_id), + ) + except Exception as err: + _LOGGER.warning( + "Failed to fetch data for target %s: %s", + target_id, + err, + ) + state = None + metrics = None + + return target_id, { + "info": target, + "state": state, + "metrics": metrics, + } + + results = await asyncio.gather( + *(fetch_target_data(t) for t in targets_list), + return_exceptions=True, + ) + + for r in results: + if isinstance(r, Exception): + _LOGGER.warning("Target fetch failed: %s", r) + continue + target_id, data = r + targets_data[target_id] = data + + # Fetch devices, CSS sources, value sources, and scene presets in parallel + devices_data, css_sources, value_sources, scene_presets = await asyncio.gather( + self._fetch_devices(), + self._fetch_css_sources(), + self._fetch_value_sources(), + self._fetch_scene_presets(), + ) + + return { + "targets": targets_data, + "devices": devices_data, + "css_sources": css_sources, + "value_sources": value_sources, + "scene_presets": scene_presets, + "server_version": self.server_version, + } + + except asyncio.TimeoutError as err: + raise UpdateFailed(f"Timeout fetching data: {err}") from err + except aiohttp.ClientError as err: + raise UpdateFailed(f"Error communicating with API: {err}") from err + + async def _fetch_server_version(self) -> None: + """Fetch server version from health endpoint.""" + try: + async with self.session.get( + f"{self.server_url}/health", + timeout=self._timeout, + ) as resp: + resp.raise_for_status() + data = await resp.json() + self.server_version = data.get("version", "unknown") + except Exception as err: + _LOGGER.warning("Failed to fetch server version: %s", err) + self.server_version = "unknown" + + async def _fetch_targets(self) -> list[dict[str, Any]]: + """Fetch all output targets.""" + async with self.session.get( + f"{self.server_url}/api/v1/output-targets", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + resp.raise_for_status() + data = await resp.json() + return data.get("targets", []) + + async def _fetch_target_state(self, target_id: str) -> dict[str, Any]: + """Fetch target processing state.""" + async with self.session.get( + f"{self.server_url}/api/v1/output-targets/{target_id}/state", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + resp.raise_for_status() + return await resp.json() + + async def _fetch_target_metrics(self, target_id: str) -> dict[str, Any]: + """Fetch target metrics.""" + async with self.session.get( + f"{self.server_url}/api/v1/output-targets/{target_id}/metrics", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + resp.raise_for_status() + return await resp.json() + + async def _fetch_devices(self) -> dict[str, dict[str, Any]]: + """Fetch all devices with capabilities and brightness.""" + try: + async with self.session.get( + f"{self.server_url}/api/v1/devices", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + resp.raise_for_status() + data = await resp.json() + devices = data.get("devices", []) + except Exception as err: + _LOGGER.warning("Failed to fetch devices: %s", err) + return {} + + # Fetch brightness for all capable devices in parallel + async def fetch_device_entry(device: dict) -> tuple[str, dict[str, Any]]: + device_id = device["id"] + entry: dict[str, Any] = {"info": device, "brightness": None} + if "brightness_control" in (device.get("capabilities") or []): + try: + async with self.session.get( + f"{self.server_url}/api/v1/devices/{device_id}/brightness", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + if resp.status == 200: + bri_data = await resp.json() + entry["brightness"] = bri_data.get("brightness") + except Exception as err: + _LOGGER.warning( + "Failed to fetch brightness for device %s: %s", + device_id, + err, + ) + return device_id, entry + + results = await asyncio.gather( + *(fetch_device_entry(d) for d in devices), + return_exceptions=True, + ) + + devices_data: dict[str, dict[str, Any]] = {} + for r in results: + if isinstance(r, Exception): + _LOGGER.warning("Device fetch failed: %s", r) + continue + device_id, entry = r + devices_data[device_id] = entry + + return devices_data + + async def set_brightness(self, device_id: str, brightness: int) -> None: + """Set brightness for a device.""" + async with self.session.put( + f"{self.server_url}/api/v1/devices/{device_id}/brightness", + headers={**self._auth_headers, "Content-Type": "application/json"}, + json={"brightness": brightness}, + timeout=self._timeout, + ) as resp: + if resp.status != 200: + body = await resp.text() + _LOGGER.error( + "Failed to set brightness for device %s: %s %s", + device_id, + resp.status, + body, + ) + resp.raise_for_status() + await self.async_request_refresh() + + async def set_color(self, device_id: str, color: list[int] | None) -> None: + """Set or clear the static color for a device.""" + async with self.session.put( + f"{self.server_url}/api/v1/devices/{device_id}/color", + headers={**self._auth_headers, "Content-Type": "application/json"}, + json={"color": color}, + timeout=self._timeout, + ) as resp: + if resp.status != 200: + body = await resp.text() + _LOGGER.error( + "Failed to set color for device %s: %s %s", + device_id, + resp.status, + body, + ) + resp.raise_for_status() + await self.async_request_refresh() + + async def _fetch_css_sources(self) -> list[dict[str, Any]]: + """Fetch all color strip sources.""" + try: + async with self.session.get( + f"{self.server_url}/api/v1/color-strip-sources", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + resp.raise_for_status() + data = await resp.json() + return data.get("sources", []) + except Exception as err: + _LOGGER.warning("Failed to fetch CSS sources: %s", err) + return [] + + async def _fetch_value_sources(self) -> list[dict[str, Any]]: + """Fetch all value sources.""" + try: + async with self.session.get( + f"{self.server_url}/api/v1/value-sources", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + resp.raise_for_status() + data = await resp.json() + return data.get("sources", []) + except Exception as err: + _LOGGER.warning("Failed to fetch value sources: %s", err) + return [] + + async def _fetch_scene_presets(self) -> list[dict[str, Any]]: + """Fetch all scene presets.""" + try: + async with self.session.get( + f"{self.server_url}/api/v1/scene-presets", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + resp.raise_for_status() + data = await resp.json() + return data.get("presets", []) + except Exception as err: + _LOGGER.warning("Failed to fetch scene presets: %s", err) + return [] + + async def push_colors(self, source_id: str, colors: list[list[int]]) -> None: + """Push flat color array to an api_input CSS source.""" + async with self.session.post( + f"{self.server_url}/api/v1/color-strip-sources/{source_id}/colors", + headers={**self._auth_headers, "Content-Type": "application/json"}, + json={"colors": colors}, + timeout=self._timeout, + ) as resp: + if resp.status not in (200, 204): + body = await resp.text() + _LOGGER.error( + "Failed to push colors to source %s: %s %s", + source_id, + resp.status, + body, + ) + resp.raise_for_status() + + async def push_segments(self, source_id: str, segments: list[dict]) -> None: + """Push segment data to an api_input CSS source.""" + async with self.session.post( + f"{self.server_url}/api/v1/color-strip-sources/{source_id}/colors", + headers={**self._auth_headers, "Content-Type": "application/json"}, + json={"segments": segments}, + timeout=self._timeout, + ) as resp: + if resp.status not in (200, 204): + body = await resp.text() + _LOGGER.error( + "Failed to push segments to source %s: %s %s", + source_id, + resp.status, + body, + ) + resp.raise_for_status() + + async def activate_scene(self, preset_id: str) -> None: + """Activate a scene preset.""" + async with self.session.post( + f"{self.server_url}/api/v1/scene-presets/{preset_id}/activate", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + if resp.status != 200: + body = await resp.text() + _LOGGER.error( + "Failed to activate scene %s: %s %s", + preset_id, + resp.status, + body, + ) + resp.raise_for_status() + await self.async_request_refresh() + + async def update_source(self, source_id: str, **kwargs: Any) -> None: + """Update a color strip source's fields.""" + async with self.session.put( + f"{self.server_url}/api/v1/color-strip-sources/{source_id}", + headers={**self._auth_headers, "Content-Type": "application/json"}, + json=kwargs, + timeout=self._timeout, + ) as resp: + if resp.status != 200: + body = await resp.text() + _LOGGER.error( + "Failed to update source %s: %s %s", + source_id, + resp.status, + body, + ) + resp.raise_for_status() + + async def update_target(self, target_id: str, **kwargs: Any) -> None: + """Update an output target's fields.""" + async with self.session.put( + f"{self.server_url}/api/v1/output-targets/{target_id}", + headers={**self._auth_headers, "Content-Type": "application/json"}, + json=kwargs, + timeout=self._timeout, + ) as resp: + if resp.status != 200: + body = await resp.text() + _LOGGER.error( + "Failed to update target %s: %s %s", + target_id, + resp.status, + body, + ) + resp.raise_for_status() + await self.async_request_refresh() + + async def start_processing(self, target_id: str) -> None: + """Start processing for a target.""" + async with self.session.post( + f"{self.server_url}/api/v1/output-targets/{target_id}/start", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + if resp.status == 409: + _LOGGER.debug("Target %s already processing", target_id) + elif resp.status != 200: + body = await resp.text() + _LOGGER.error( + "Failed to start target %s: %s %s", + target_id, + resp.status, + body, + ) + resp.raise_for_status() + await self.async_request_refresh() + + async def stop_processing(self, target_id: str) -> None: + """Stop processing for a target.""" + async with self.session.post( + f"{self.server_url}/api/v1/output-targets/{target_id}/stop", + headers=self._auth_headers, + timeout=self._timeout, + ) as resp: + if resp.status == 409: + _LOGGER.debug("Target %s already stopped", target_id) + elif resp.status != 200: + body = await resp.text() + _LOGGER.error( + "Failed to stop target %s: %s %s", + target_id, + resp.status, + body, + ) + resp.raise_for_status() + await self.async_request_refresh() diff --git a/custom_components/ledgrab/event_listener.py b/custom_components/ledgrab/event_listener.py new file mode 100644 index 0000000..9fbefc1 --- /dev/null +++ b/custom_components/ledgrab/event_listener.py @@ -0,0 +1,95 @@ +"""WebSocket event listener for server state change notifications.""" +from __future__ import annotations + +import asyncio +import contextlib +import json +import logging + +import aiohttp + +from homeassistant.core import HomeAssistant +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator + +from .const import WS_RECONNECT_DELAY, WS_MAX_RECONNECT_DELAY + +_LOGGER = logging.getLogger(__name__) + + +class EventStreamListener: + """Listens to server WS endpoint for state change events. + + Triggers a coordinator refresh whenever a target starts or stops processing, + so HAOS entities react near-instantly to external state changes. + """ + + def __init__( + self, + hass: HomeAssistant, + server_url: str, + api_key: str, + coordinator: DataUpdateCoordinator, + ) -> None: + self._hass = hass + self._server_url = server_url + self._api_key = api_key + self._coordinator = coordinator + self._task: asyncio.Task | None = None + self._shutting_down = False + + async def start(self) -> None: + """Start listening to the event stream.""" + self._task = self._hass.async_create_background_task( + self._ws_loop(), + "ledgrab_events", + ) + + async def _ws_loop(self) -> None: + """WebSocket connection loop with reconnection.""" + delay = WS_RECONNECT_DELAY + session = async_get_clientsession(self._hass) + ws_base = self._server_url.replace("http://", "ws://").replace( + "https://", "wss://" + ) + url = f"{ws_base}/api/v1/events/ws?token={self._api_key}" + + while not self._shutting_down: + try: + async with session.ws_connect(url) as ws: + delay = WS_RECONNECT_DELAY # reset on successful connect + _LOGGER.debug("Event stream connected") + async for msg in ws: + if msg.type == aiohttp.WSMsgType.TEXT: + try: + data = json.loads(msg.data) + except json.JSONDecodeError: + continue + if data.get("type") == "state_change": + await self._coordinator.async_request_refresh() + elif msg.type in ( + aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.ERROR, + ): + break + except asyncio.CancelledError: + raise + except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as err: + _LOGGER.debug("Event stream connection error: %s", err) + except Exception as err: + _LOGGER.error("Unexpected event stream error: %s", err) + + if self._shutting_down: + break + + await asyncio.sleep(delay) + delay = min(delay * 2, WS_MAX_RECONNECT_DELAY) + + async def shutdown(self) -> None: + """Stop listening.""" + self._shutting_down = True + if self._task: + self._task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._task + self._task = None diff --git a/custom_components/ledgrab/light.py b/custom_components/ledgrab/light.py new file mode 100644 index 0000000..7f6dc6b --- /dev/null +++ b/custom_components/ledgrab/light.py @@ -0,0 +1,151 @@ +"""Light platform for LED Screen Controller (api_input CSS sources).""" +from __future__ import annotations + +import logging +from typing import Any + +from homeassistant.components.light import ( + ATTR_BRIGHTNESS, + ATTR_RGB_COLOR, + ColorMode, + LightEntity, +) +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 api_input lights.""" + data = hass.data[DOMAIN][entry.entry_id] + coordinator: LedGrabCoordinator = data[DATA_COORDINATOR] + + entities = [] + if coordinator.data: + for source in coordinator.data.get("css_sources", []): + if source.get("source_type") == "api_input": + entities.append( + ApiInputLight(coordinator, source, entry.entry_id) + ) + + async_add_entities(entities) + + +class ApiInputLight(CoordinatorEntity, LightEntity): + """Representation of an api_input CSS source as a light entity.""" + + _attr_has_entity_name = True + _attr_color_mode = ColorMode.RGB + _attr_supported_color_modes = {ColorMode.RGB} + _attr_translation_key = "api_input_light" + _attr_icon = "mdi:led-strip-variant" + + def __init__( + self, + coordinator: LedGrabCoordinator, + source: dict[str, Any], + entry_id: str, + ) -> None: + """Initialize the light.""" + super().__init__(coordinator) + self._source_id: str = source["id"] + self._source_name: str = source.get("name", self._source_id) + self._entry_id = entry_id + self._attr_unique_id = f"{self._source_id}_light" + + # Restore state from fallback_color + fallback = self._get_fallback_color() + is_off = fallback == [0, 0, 0] + self._is_on: bool = not is_off + self._rgb_color: tuple[int, int, int] = ( + (255, 255, 255) if is_off else tuple(fallback) # type: ignore[arg-type] + ) + self._brightness: int = 255 + + @property + def device_info(self) -> dict[str, Any]: + """Return device information — one virtual device per api_input source.""" + return { + "identifiers": {(DOMAIN, self._source_id)}, + "name": self._source_name, + "manufacturer": "LedGrab", + "model": "API Input CSS Source", + } + + @property + def name(self) -> str: + """Return the entity name.""" + return self._source_name + + @property + def is_on(self) -> bool: + """Return true if the light is on.""" + return self._is_on + + @property + def rgb_color(self) -> tuple[int, int, int]: + """Return the current RGB color.""" + return self._rgb_color + + @property + def brightness(self) -> int: + """Return the current brightness (0-255).""" + return self._brightness + + async def async_turn_on(self, **kwargs: Any) -> None: + """Turn on the light, optionally setting color and brightness.""" + if ATTR_RGB_COLOR in kwargs: + self._rgb_color = kwargs[ATTR_RGB_COLOR] + if ATTR_BRIGHTNESS in kwargs: + self._brightness = kwargs[ATTR_BRIGHTNESS] + + # Scale RGB by brightness + scale = self._brightness / 255 + r, g, b = self._rgb_color + scaled = [round(r * scale), round(g * scale), round(b * scale)] + + await self.coordinator.push_segments( + self._source_id, + [{"start": 0, "length": 9999, "mode": "solid", "color": scaled}], + ) + # Update fallback_color so the color persists beyond the timeout + await self.coordinator.update_source( + self._source_id, fallback_color=scaled, + ) + self._is_on = True + self.async_write_ha_state() + + async def async_turn_off(self, **kwargs: Any) -> None: + """Turn off the light by pushing black and setting fallback to black.""" + off_color = [0, 0, 0] + await self.coordinator.push_segments( + self._source_id, + [{"start": 0, "length": 9999, "mode": "solid", "color": off_color}], + ) + await self.coordinator.update_source( + self._source_id, fallback_color=off_color, + ) + self._is_on = False + self.async_write_ha_state() + + def _get_fallback_color(self) -> list[int]: + """Read fallback_color from the source config in coordinator data.""" + if not self.coordinator.data: + return [0, 0, 0] + for source in self.coordinator.data.get("css_sources", []): + if source.get("id") == self._source_id: + fallback = source.get("fallback_color") + if fallback and len(fallback) >= 3: + return list(fallback[:3]) + break + return [0, 0, 0] diff --git a/custom_components/ledgrab/manifest.json b/custom_components/ledgrab/manifest.json new file mode 100644 index 0000000..47b3ee5 --- /dev/null +++ b/custom_components/ledgrab/manifest.json @@ -0,0 +1,12 @@ +{ + "domain": "ledgrab", + "name": "LedGrab", + "codeowners": ["@alexeidolgolyov"], + "config_flow": true, + "dependencies": [], + "documentation": "https://git.dolgolyov-family.by/alexei.dolgolyov/ledgrab-haos-integration", + "iot_class": "local_push", + "issue_tracker": "https://git.dolgolyov-family.by/alexei.dolgolyov/ledgrab-haos-integration/issues", + "requirements": ["aiohttp>=3.9.0"], + "version": "0.2.0" +} diff --git a/custom_components/ledgrab/number.py b/custom_components/ledgrab/number.py new file mode 100644 index 0000000..525c7d1 --- /dev/null +++ b/custom_components/ledgrab/number.py @@ -0,0 +1,233 @@ +"""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 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" diff --git a/custom_components/ledgrab/select.py b/custom_components/ledgrab/select.py new file mode 100644 index 0000000..dec5a05 --- /dev/null +++ b/custom_components/ledgrab/select.py @@ -0,0 +1,178 @@ +"""Select platform for LED Screen Controller (CSS source & brightness source).""" + +from __future__ import annotations + +import logging +from typing import Any + +from homeassistant.components.select import SelectEntity +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__) + +NONE_OPTION = "\u2014 None \u2014" + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up LED Screen Controller select entities.""" + data = hass.data[DOMAIN][entry.entry_id] + coordinator: LedGrabCoordinator = data[DATA_COORDINATOR] + + entities: list[SelectEntity] = [] + if coordinator.data and "targets" in coordinator.data: + for target_id, target_data in coordinator.data["targets"].items(): + info = target_data["info"] + target_type = info.get("target_type", "led") + + # Both LED and HA Light targets have a CSS source + entities.append(CSSSourceSelect(coordinator, target_id, entry.entry_id)) + + # Only LED targets have a brightness value source + if target_type != TARGET_TYPE_HA_LIGHT: + entities.append(BrightnessSourceSelect(coordinator, target_id, entry.entry_id)) + + async_add_entities(entities) + + +class CSSSourceSelect(CoordinatorEntity, SelectEntity): + """Select entity for choosing a color strip source for a target.""" + + _attr_has_entity_name = True + _attr_icon = "mdi:palette" + + def __init__( + self, + coordinator: LedGrabCoordinator, + target_id: str, + entry_id: str, + ) -> None: + super().__init__(coordinator) + self._target_id = target_id + self._entry_id = entry_id + self._attr_unique_id = f"{target_id}_css_source" + self._attr_translation_key = "color_strip_source" + + @property + def device_info(self) -> dict[str, Any]: + return {"identifiers": {(DOMAIN, self._target_id)}} + + @property + def options(self) -> list[str]: + if not self.coordinator.data: + return [] + sources = self.coordinator.data.get("css_sources") or [] + return [s["name"] for s in sources] + + @property + def current_option(self) -> str | None: + if not self.coordinator.data: + return None + target_data = self.coordinator.data.get("targets", {}).get(self._target_id) + if not target_data: + return None + current_id = target_data["info"].get("color_strip_source_id", "") + sources = self.coordinator.data.get("css_sources") or [] + for s in sources: + if s["id"] == current_id: + return s["name"] + return None + + @property + def available(self) -> bool: + if not self.coordinator.data: + return False + return self._target_id in self.coordinator.data.get("targets", {}) + + async def async_select_option(self, option: str) -> None: + source_id = self._name_to_id_map().get(option) + if source_id is None: + _LOGGER.error("CSS source not found: %s", option) + return + await self.coordinator.update_target(self._target_id, color_strip_source_id=source_id) + + def _name_to_id_map(self) -> dict[str, str]: + sources = (self.coordinator.data or {}).get("css_sources") or [] + return {s["name"]: s["id"] for s in sources} + + +class BrightnessSourceSelect(CoordinatorEntity, SelectEntity): + """Select entity for choosing a brightness value source for an LED target.""" + + _attr_has_entity_name = True + _attr_icon = "mdi:brightness-auto" + + def __init__( + self, + coordinator: LedGrabCoordinator, + target_id: str, + entry_id: str, + ) -> None: + super().__init__(coordinator) + self._target_id = target_id + self._entry_id = entry_id + self._attr_unique_id = f"{target_id}_brightness_source" + self._attr_translation_key = "brightness_source" + + @property + def device_info(self) -> dict[str, Any]: + return {"identifiers": {(DOMAIN, self._target_id)}} + + @property + def options(self) -> list[str]: + if not self.coordinator.data: + return [NONE_OPTION] + sources = self.coordinator.data.get("value_sources") or [] + return [NONE_OPTION] + [s["name"] for s in sources] + + @property + def current_option(self) -> str | None: + if not self.coordinator.data: + return None + target_data = self.coordinator.data.get("targets", {}).get(self._target_id) + if not target_data: + return None + # BindableFloat: brightness is either a plain float or {"value": float, "source_id": str} + brightness = target_data["info"].get("brightness", "") + if isinstance(brightness, dict): + current_id = brightness.get("source_id", "") + else: + current_id = target_data["info"].get("brightness_value_source_id", "") + if not current_id: + return NONE_OPTION + sources = self.coordinator.data.get("value_sources") or [] + for s in sources: + if s["id"] == current_id: + return s["name"] + return NONE_OPTION + + @property + def available(self) -> bool: + if not self.coordinator.data: + return False + return self._target_id in self.coordinator.data.get("targets", {}) + + async def async_select_option(self, option: str) -> None: + if option == NONE_OPTION: + source_id = "" + else: + name_map = { + s["name"]: s["id"] for s in (self.coordinator.data or {}).get("value_sources") or [] + } + source_id = name_map.get(option) + if source_id is None: + _LOGGER.error("Value source not found: %s", option) + return + await self.coordinator.update_target( + self._target_id, + brightness={"value": 1.0, "source_id": source_id} if source_id else 1.0, + ) diff --git a/custom_components/ledgrab/sensor.py b/custom_components/ledgrab/sensor.py new file mode 100644 index 0000000..62dca1d --- /dev/null +++ b/custom_components/ledgrab/sensor.py @@ -0,0 +1,225 @@ +"""Sensor platform for LED Screen Controller.""" + +from __future__ import annotations + +import logging +from typing import Any + +from homeassistant.components.sensor import ( + SensorDeviceClass, + SensorEntity, + SensorStateClass, +) +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, + TARGET_TYPE_HA_LIGHT, + 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 sensors.""" + data = hass.data[DOMAIN][entry.entry_id] + coordinator: LedGrabCoordinator = data[DATA_COORDINATOR] + + entities: list[SensorEntity] = [] + if coordinator.data and "targets" in coordinator.data: + for target_id, target_data in coordinator.data["targets"].items(): + entities.append(LedGrabFPSSensor(coordinator, target_id, entry.entry_id)) + entities.append( + LedGrabStatusSensor(coordinator, target_id, entry.entry_id) + ) + + # Add mapped lights sensor for HA Light targets + info = target_data["info"] + if info.get("target_type") == TARGET_TYPE_HA_LIGHT: + entities.append(HALightMappedLightsSensor(coordinator, target_id, entry.entry_id)) + + async_add_entities(entities) + + +class LedGrabFPSSensor(CoordinatorEntity, SensorEntity): + """FPS sensor for a LED Screen Controller target.""" + + _attr_has_entity_name = True + _attr_state_class = SensorStateClass.MEASUREMENT + _attr_native_unit_of_measurement = "FPS" + _attr_icon = "mdi:speedometer" + _attr_suggested_display_precision = 1 + + def __init__( + self, + coordinator: LedGrabCoordinator, + target_id: str, + entry_id: str, + ) -> None: + """Initialize the sensor.""" + super().__init__(coordinator) + self._target_id = target_id + self._entry_id = entry_id + self._attr_unique_id = f"{target_id}_fps" + self._attr_translation_key = "fps" + + @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 FPS value.""" + target_data = self._get_target_data() + if not target_data or not target_data.get("state"): + return None + state = target_data["state"] + if not state.get("processing"): + return None + return state.get("fps_actual") + + @property + def extra_state_attributes(self) -> dict[str, Any]: + """Return additional attributes.""" + target_data = self._get_target_data() + if not target_data or not target_data.get("state"): + return {} + return {"fps_target": target_data["state"].get("fps_target")} + + @property + def available(self) -> bool: + """Return if entity is available.""" + return self._get_target_data() is not None + + 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 LedGrabStatusSensor(CoordinatorEntity, SensorEntity): + """Status sensor for a LED Screen Controller target.""" + + _attr_has_entity_name = True + _attr_icon = "mdi:information-outline" + _attr_device_class = SensorDeviceClass.ENUM + _attr_options = ["processing", "idle", "error", "unavailable"] + + def __init__( + self, + coordinator: LedGrabCoordinator, + target_id: str, + entry_id: str, + ) -> None: + """Initialize the sensor.""" + super().__init__(coordinator) + self._target_id = target_id + self._entry_id = entry_id + self._attr_unique_id = f"{target_id}_status" + self._attr_translation_key = "status" + + @property + def device_info(self) -> dict[str, Any]: + """Return device information.""" + return {"identifiers": {(DOMAIN, self._target_id)}} + + @property + def native_value(self) -> str: + """Return the status.""" + target_data = self._get_target_data() + if not target_data: + return "unavailable" + state = target_data.get("state") + if not state: + return "unavailable" + if state.get("processing"): + errors = state.get("errors", []) + if errors: + return "error" + return "processing" + return "idle" + + @property + def available(self) -> bool: + """Return if entity is available.""" + return self._get_target_data() is not None + + 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 HALightMappedLightsSensor(CoordinatorEntity, SensorEntity): + """Sensor showing the number of mapped HA lights for an HA Light target.""" + + _attr_has_entity_name = True + _attr_icon = "mdi:lightbulb-group" + + def __init__( + self, + coordinator: LedGrabCoordinator, + target_id: str, + entry_id: str, + ) -> None: + """Initialize the sensor.""" + super().__init__(coordinator) + self._target_id = target_id + self._entry_id = entry_id + self._attr_unique_id = f"{target_id}_mapped_lights" + self._attr_translation_key = "mapped_lights" + + @property + def device_info(self) -> dict[str, Any]: + """Return device information.""" + return {"identifiers": {(DOMAIN, self._target_id)}} + + @property + def native_value(self) -> int | None: + """Return the number of mapped lights.""" + target_data = self._get_target_data() + if not target_data: + return None + mappings = target_data.get("info", {}).get("light_mappings", []) + return len(mappings) + + @property + def extra_state_attributes(self) -> dict[str, Any]: + """Return light mapping details as attributes.""" + target_data = self._get_target_data() + if not target_data: + return {} + mappings = target_data.get("info", {}).get("light_mappings", []) + entity_ids = [m.get("entity_id", "") for m in mappings] + return { + "entity_ids": entity_ids, + "mappings": [ + { + "entity_id": m.get("entity_id", ""), + "led_start": m.get("led_start", 0), + "led_end": m.get("led_end", -1), + "brightness_scale": m.get("brightness_scale", 1.0), + } + for m in mappings + ], + } + + @property + def available(self) -> bool: + """Return if entity is available.""" + return self._get_target_data() is not None + + 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) diff --git a/custom_components/ledgrab/services.yaml b/custom_components/ledgrab/services.yaml new file mode 100644 index 0000000..8499fdd --- /dev/null +++ b/custom_components/ledgrab/services.yaml @@ -0,0 +1,19 @@ +set_leds: + name: Set LEDs + description: Push segment data to an api_input color strip source + fields: + source_id: + name: Source ID + description: The api_input CSS source ID (e.g., css_abc12345) + required: true + selector: + text: + segments: + name: Segments + description: > + List of segment objects. Each segment has: start (int), length (int), + mode ("solid"/"per_pixel"/"gradient"), color ([R,G,B] for solid), + colors ([[R,G,B],...] for per_pixel/gradient) + required: true + selector: + object: diff --git a/custom_components/ledgrab/strings.json b/custom_components/ledgrab/strings.json new file mode 100644 index 0000000..8060675 --- /dev/null +++ b/custom_components/ledgrab/strings.json @@ -0,0 +1,103 @@ +{ + "config": { + "step": { + "user": { + "title": "Set up LED Screen Controller", + "description": "Enter the URL and API key for your LED Screen Controller server.", + "data": { + "server_name": "Server Name", + "server_url": "Server URL", + "api_key": "API Key" + }, + "data_description": { + "server_name": "Display name for this server in Home Assistant", + "server_url": "URL of your LED Screen Controller server (e.g., http://192.168.1.100:8080)", + "api_key": "API key from your server's configuration file" + } + } + }, + "error": { + "cannot_connect": "Failed to connect to server.", + "invalid_api_key": "Invalid API key.", + "unknown": "Unexpected error occurred." + }, + "abort": { + "already_configured": "This server is already configured." + } + }, + "entity": { + "button": { + "activate_scene": { + "name": "{scene_name}" + } + }, + "light": { + "api_input_light": { + "name": "Light" + } + }, + "switch": { + "processing": { + "name": "Processing" + } + }, + "sensor": { + "fps": { + "name": "FPS" + }, + "status": { + "name": "Status", + "state": { + "processing": "Processing", + "idle": "Idle", + "error": "Error", + "unavailable": "Unavailable" + } + }, + "mapped_lights": { + "name": "Mapped Lights" + } + }, + "number": { + "brightness": { + "name": "Brightness" + }, + "ha_light_update_rate": { + "name": "Update Rate" + }, + "ha_light_transition": { + "name": "Transition" + }, + "ha_light_min_brightness": { + "name": "Min Brightness" + }, + "ha_light_color_tolerance": { + "name": "Color Tolerance" + } + }, + "select": { + "color_strip_source": { + "name": "Color Strip Source" + }, + "brightness_source": { + "name": "Brightness Source" + } + } + }, + "services": { + "set_leds": { + "name": "Set LEDs", + "description": "Push segment data to an api_input color strip source.", + "fields": { + "source_id": { + "name": "Source ID", + "description": "The api_input CSS source ID (e.g., css_abc12345)." + }, + "segments": { + "name": "Segments", + "description": "List of segment objects with start, length, mode, and color/colors fields." + } + } + } + } +} diff --git a/custom_components/ledgrab/switch.py b/custom_components/ledgrab/switch.py new file mode 100644 index 0000000..4a46402 --- /dev/null +++ b/custom_components/ledgrab/switch.py @@ -0,0 +1,109 @@ +"""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) diff --git a/custom_components/ledgrab/translations/en.json b/custom_components/ledgrab/translations/en.json new file mode 100644 index 0000000..2e7cb7a --- /dev/null +++ b/custom_components/ledgrab/translations/en.json @@ -0,0 +1,87 @@ +{ + "config": { + "step": { + "user": { + "title": "Set up LED Screen Controller", + "description": "Enter the URL and API key for your LED Screen Controller server.", + "data": { + "server_name": "Server Name", + "server_url": "Server URL", + "api_key": "API Key" + }, + "data_description": { + "server_name": "Display name for this server in Home Assistant", + "server_url": "URL of your LED Screen Controller server (e.g., http://192.168.1.100:8080)", + "api_key": "API key from your server's configuration file" + } + } + }, + "error": { + "cannot_connect": "Failed to connect to server.", + "invalid_api_key": "Invalid API key.", + "unknown": "Unexpected error occurred." + }, + "abort": { + "already_configured": "This server is already configured." + } + }, + "entity": { + "button": { + "activate_scene": { + "name": "{scene_name}" + } + }, + "light": { + "api_input_light": { + "name": "Light" + } + }, + "switch": { + "processing": { + "name": "Processing" + } + }, + "sensor": { + "fps": { + "name": "FPS" + }, + "status": { + "name": "Status", + "state": { + "processing": "Processing", + "idle": "Idle", + "error": "Error", + "unavailable": "Unavailable" + } + }, + "mapped_lights": { + "name": "Mapped Lights" + } + }, + "number": { + "brightness": { + "name": "Brightness" + }, + "ha_light_update_rate": { + "name": "Update Rate" + }, + "ha_light_transition": { + "name": "Transition" + }, + "ha_light_min_brightness": { + "name": "Min Brightness" + }, + "ha_light_color_tolerance": { + "name": "Color Tolerance" + } + }, + "select": { + "color_strip_source": { + "name": "Color Strip Source" + }, + "brightness_source": { + "name": "Brightness Source" + } + } + } +} diff --git a/custom_components/ledgrab/translations/ru.json b/custom_components/ledgrab/translations/ru.json new file mode 100644 index 0000000..b7e4e1a --- /dev/null +++ b/custom_components/ledgrab/translations/ru.json @@ -0,0 +1,87 @@ +{ + "config": { + "step": { + "user": { + "title": "Настройка LED Screen Controller", + "description": "Введите URL и API-ключ вашего сервера LED Screen Controller.", + "data": { + "server_name": "Имя сервера", + "server_url": "URL сервера", + "api_key": "API-ключ" + }, + "data_description": { + "server_name": "Отображаемое имя сервера в Home Assistant", + "server_url": "URL сервера LED Screen Controller (например, http://192.168.1.100:8080)", + "api_key": "API-ключ из конфигурационного файла сервера" + } + } + }, + "error": { + "cannot_connect": "Не удалось подключиться к серверу.", + "invalid_api_key": "Неверный API-ключ.", + "unknown": "Произошла непредвиденная ошибка." + }, + "abort": { + "already_configured": "Этот сервер уже настроен." + } + }, + "entity": { + "button": { + "activate_scene": { + "name": "{scene_name}" + } + }, + "light": { + "api_input_light": { + "name": "Подсветка" + } + }, + "switch": { + "processing": { + "name": "Обработка" + } + }, + "sensor": { + "fps": { + "name": "FPS" + }, + "status": { + "name": "Статус", + "state": { + "processing": "Обработка", + "idle": "Ожидание", + "error": "Ошибка", + "unavailable": "Недоступен" + } + }, + "mapped_lights": { + "name": "Привязанные светильники" + } + }, + "number": { + "brightness": { + "name": "Яркость" + }, + "ha_light_update_rate": { + "name": "Частота обновления" + }, + "ha_light_transition": { + "name": "Переход" + }, + "ha_light_min_brightness": { + "name": "Мин. яркость" + }, + "ha_light_color_tolerance": { + "name": "Допуск цвета" + } + }, + "select": { + "color_strip_source": { + "name": "Источник цветовой полосы" + }, + "brightness_source": { + "name": "Источник яркости" + } + } + } +} diff --git a/hacs.json b/hacs.json new file mode 100644 index 0000000..6e023f4 --- /dev/null +++ b/hacs.json @@ -0,0 +1,6 @@ +{ + "name": "LedGrab", + "render_readme": true, + "country": ["US"], + "homeassistant": "2023.1.0" +}