from __future__ import annotations import asyncio import logging from dataclasses import dataclass from datetime import UTC, datetime from time import monotonic from typing import Any, Protocol import httpx from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.core.database import SessionLocal from app.dmx.engine import DmxEngine from app.models.entities import Setting logger = logging.getLogger(__name__) HA_CONFIG_KEY = "home_assistant_config" HA_MAPPINGS_KEY = "home_assistant_mappings" SWITCH_ON_THRESHOLD = 140 SWITCH_OFF_THRESHOLD = 115 EDGE_TRIGGER_THRESHOLD = 128 ON_OFF_ENTITY_DOMAINS = {"light", "switch", "input_boolean"} TRIGGER_ENTITY_DOMAINS = {"scene", "automation"} class HomeAssistantAdapter(Protocol): async def get_api_root(self, base_url: str, token: str) -> dict[str, object]: ... async def get_config(self, base_url: str, token: str) -> dict[str, object]: ... async def list_entities(self, base_url: str, token: str) -> list[dict[str, object]]: ... async def call_service( self, base_url: str, token: str, domain: str, service: str, data: dict[str, object], ) -> dict[str, object]: ... class HttpxHomeAssistantAdapter: def __init__(self, timeout_s: float = 2.0) -> None: self.timeout_s = timeout_s async def get_api_root(self, base_url: str, token: str) -> dict[str, object]: return await self._get_json(base_url, token, "/api/") async def get_config(self, base_url: str, token: str) -> dict[str, object]: return await self._get_json(base_url, token, "/api/config") async def list_entities(self, base_url: str, token: str) -> list[dict[str, object]]: payload = await self._get_json(base_url, token, "/api/states") if not isinstance(payload, list): return [] return [item for item in payload if isinstance(item, dict)] async def call_service( self, base_url: str, token: str, domain: str, service: str, data: dict[str, object], ) -> dict[str, object]: async with httpx.AsyncClient(timeout=self.timeout_s) as client: response = await client.post( f"{base_url.rstrip('/')}/api/services/{domain}/{service}", headers=self._headers(token), json=data, ) response.raise_for_status() payload = response.json() return payload if isinstance(payload, dict) else {"result": payload} async def _get_json(self, base_url: str, token: str, path: str) -> dict[str, object] | list[object]: async with httpx.AsyncClient(timeout=self.timeout_s) as client: response = await client.get( f"{base_url.rstrip('/')}{path}", headers=self._headers(token), ) response.raise_for_status() payload = response.json() if isinstance(payload, (dict, list)): return payload return {} def _headers(self, token: str) -> dict[str, str]: return { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } @dataclass(slots=True) class MappingRuntime: last_dmx_values: list[int] last_sent_raw_values: list[int] | None = None last_sent_summary: str | None = None last_success_at: datetime | None = None last_error: str | None = None last_error_at: datetime | None = None last_service: str | None = None in_flight: bool = False pending_raw_values: list[int] | None = None resync_required: bool = False switch_state: bool | None = None class HomeAssistantService: def __init__( self, engine: DmxEngine, session_factory: async_sessionmaker[AsyncSession] = SessionLocal, adapter: HomeAssistantAdapter | None = None, ) -> None: self.engine = engine self._session_factory = session_factory self._adapter = adapter or HttpxHomeAssistantAdapter() self._config = self._default_config() self._token = "" self._mappings: list[dict[str, object]] = [] self._entities_by_id: dict[str, dict[str, object]] = {} self._mapping_runtime: dict[int, MappingRuntime] = {} self._mapping_tasks: dict[int, asyncio.Task[None]] = {} self._running = False self._task: asyncio.Task[None] | None = None self._dispatch_count = 0 self._error_count = 0 self._last_error: str | None = None self._last_successful_call_at: datetime | None = None self._last_connection_success_at: datetime | None = None self._last_connection_error: str | None = None self._ha_version: str | None = None self._auth_ok = False self._reachable = False async def startup(self) -> None: await self._load() self._running = True self._task = asyncio.create_task(self._loop(), name="tuxdmx-home-assistant-bridge") async def shutdown(self) -> None: self._running = False for task in list(self._mapping_tasks.values()): task.cancel() for task in list(self._mapping_tasks.values()): try: await task except asyncio.CancelledError: pass self._mapping_tasks.clear() if self._task is not None: self._task.cancel() try: await self._task except asyncio.CancelledError: pass self._task = None async def get_config(self) -> dict[str, object]: return { **self._config, "has_token": bool(self._token), "token_mask": "********" if self._token else "", "mapping_count": len(self._mappings), "dispatch_count": self._dispatch_count, "error_count": self._error_count, "last_error": self._last_error, "last_successful_call_at": self._iso(self._last_successful_call_at), "last_connection_success_at": self._iso(self._last_connection_success_at), "last_connection_error": self._last_connection_error, "ha_version": self._ha_version, "auth_ok": self._auth_ok, "reachable": self._reachable, } async def save_config(self, payload: dict[str, object]) -> dict[str, object]: normalized = self._normalize_config(payload) incoming_token = str(payload.get("token", "") or "").strip() if incoming_token: self._token = incoming_token self._config = normalized await self._persist_setting( HA_CONFIG_KEY, { **normalized, "token": self._token, }, ) return await self.get_config() async def test_connection(self) -> dict[str, object]: base_url = str(self._config["base_url"]) if not base_url or not self._token: self._reachable = False self._auth_ok = False self._last_connection_error = "Base URL eller token mangler." return { "reachable": False, "auth_ok": False, "ha_version": self._ha_version, "last_error": self._last_connection_error, "last_successful_call_at": self._iso(self._last_successful_call_at), "last_connection_success_at": self._iso(self._last_connection_success_at), } try: await self._adapter.get_api_root(base_url, self._token) config = await self._adapter.get_config(base_url, self._token) except httpx.HTTPStatusError as exc: self._reachable = True self._auth_ok = exc.response.status_code != 401 self._last_connection_error = self._summarize_exception(exc) return { "reachable": self._reachable, "auth_ok": False, "ha_version": self._ha_version, "last_error": self._last_connection_error, "last_successful_call_at": self._iso(self._last_successful_call_at), "last_connection_success_at": self._iso(self._last_connection_success_at), } except Exception as exc: self._reachable = False self._auth_ok = False self._last_connection_error = self._summarize_exception(exc) return { "reachable": False, "auth_ok": False, "ha_version": self._ha_version, "last_error": self._last_connection_error, "last_successful_call_at": self._iso(self._last_successful_call_at), "last_connection_success_at": self._iso(self._last_connection_success_at), } self._reachable = True self._auth_ok = True self._last_connection_error = None self._last_connection_success_at = datetime.now(UTC) self._ha_version = str(config.get("version", "") or self._ha_version or "") return { "reachable": True, "auth_ok": True, "ha_version": self._ha_version, "last_error": None, "last_successful_call_at": self._iso(self._last_successful_call_at), "last_connection_success_at": self._iso(self._last_connection_success_at), } async def list_entities(self) -> dict[str, object]: if not self._config["base_url"] or not self._token: return {"items": []} entities = await self._adapter.list_entities(str(self._config["base_url"]), self._token) items: list[dict[str, object]] = [] self._entities_by_id.clear() for entity in entities: normalized = self._normalize_entity(entity) if normalized is None: continue self._entities_by_id[str(normalized["entity_id"])] = normalized items.append(normalized) items.sort(key=lambda item: (str(item["domain"]), str(item["friendly_name"]), str(item["entity_id"]))) self._last_connection_success_at = datetime.now(UTC) self._reachable = True self._auth_ok = True return {"items": items} async def list_mappings(self) -> dict[str, object]: return {"items": [self._serialize_mapping(mapping) for mapping in self._mappings]} async def create_mapping(self, payload: dict[str, object]) -> dict[str, object]: mapping = self._normalize_mapping(payload) mapping["id"] = self._next_mapping_id() self._mappings.append(mapping) self._ensure_runtime(int(mapping["id"]), self._channel_span(mapping)) await self._persist_mappings() return self._serialize_mapping(mapping) async def update_mapping(self, mapping_id: int, payload: dict[str, object]) -> dict[str, object]: mapping = self._find_mapping(mapping_id) updated = self._normalize_mapping({**mapping, **payload, "id": mapping_id}) updated["id"] = mapping_id index = next(index for index, item in enumerate(self._mappings) if int(item["id"]) == mapping_id) self._mappings[index] = updated runtime = self._ensure_runtime(mapping_id, self._channel_span(updated)) runtime.resync_required = True await self._persist_mappings() return self._serialize_mapping(updated) async def delete_mapping(self, mapping_id: int) -> dict[str, object]: mapping = self._find_mapping(mapping_id) task = self._mapping_tasks.pop(mapping_id, None) if task is not None: task.cancel() self._mappings = [item for item in self._mappings if int(item["id"]) != mapping_id] self._mapping_runtime.pop(mapping_id, None) await self._persist_mappings() return {"deleted": mapping_id, "entity_id": mapping["entity_id"]} async def test_mapping(self, mapping_id: int) -> dict[str, object]: mapping = self._find_mapping(mapping_id) raw_values = self._build_test_values(mapping) runtime = self._ensure_runtime(mapping_id, self._channel_span(mapping)) result = await self._dispatch_mapping(mapping, raw_values, runtime, force=True) runtime.resync_required = True return { "status": "sent" if result else "skipped", "mapping_id": mapping_id, "entity_id": mapping["entity_id"], } async def dispatch_once(self) -> None: await self._dispatch_enabled_mappings() await self._drain_mapping_tasks() async def _loop(self) -> None: try: while self._running: await self._dispatch_enabled_mappings() await asyncio.sleep(0.05) except asyncio.CancelledError: raise async def _dispatch_enabled_mappings(self) -> None: if not bool(self._config["enabled"]): return if not self._config["base_url"] or not self._token: return self._prune_orphaned_runtimes() for mapping in self._mappings: mapping_id = int(mapping["id"]) runtime = self._ensure_runtime(mapping_id, self._channel_span(mapping)) frame = self.engine.get_frame(int(mapping["universe"])) raw_values = self._read_raw_values(mapping, frame.values) previous_observed = runtime.last_dmx_values.copy() runtime.last_dmx_values = raw_values.copy() if not bool(mapping["enabled"]): continue if runtime.in_flight: runtime.pending_raw_values = raw_values.copy() continue if runtime.resync_required: runtime.resync_required = False self._schedule_dispatch(mapping, runtime, raw_values, force=True) continue if not self._should_dispatch(mapping, runtime, previous_observed, raw_values): continue self._schedule_dispatch(mapping, runtime, raw_values) def _schedule_dispatch( self, mapping: dict[str, object], runtime: MappingRuntime, raw_values: list[int], *, force: bool = False, ) -> None: mapping_id = int(mapping["id"]) runtime.in_flight = True runtime.pending_raw_values = None task = asyncio.create_task( self._dispatch_task(mapping_id, raw_values.copy(), force=force), name=f"tuxdmx-ha-mapping-{mapping_id}", ) self._mapping_tasks[mapping_id] = task async def _dispatch_task(self, mapping_id: int, raw_values: list[int], *, force: bool = False) -> None: try: mapping = self._find_mapping(mapping_id) except LookupError: return runtime = self._ensure_runtime(mapping_id, self._channel_span(mapping)) try: await self._dispatch_mapping(mapping, raw_values, runtime, force=force) finally: runtime.in_flight = False self._mapping_tasks.pop(mapping_id, None) pending = runtime.pending_raw_values.copy() if runtime.pending_raw_values is not None else None runtime.pending_raw_values = None if pending is not None and bool(mapping.get("enabled", True)): if runtime.resync_required: runtime.resync_required = False self._schedule_dispatch(mapping, runtime, pending, force=True) elif runtime.last_sent_raw_values != pending: self._schedule_dispatch(mapping, runtime, pending) async def _dispatch_mapping( self, mapping: dict[str, object], raw_values: list[int], runtime: MappingRuntime, *, force: bool = False, ) -> bool: call = self._build_service_call(mapping, raw_values, runtime, force=force) if call is None: return False domain, service, data, summary = call try: await self._adapter.call_service(str(self._config["base_url"]), self._token, domain, service, data) except Exception as exc: message = self._summarize_exception(exc) runtime.last_error = message runtime.last_error_at = datetime.now(UTC) self._error_count += 1 self._last_error = message self._last_connection_error = message self._reachable = not isinstance(exc, (httpx.ConnectError, httpx.ReadTimeout, httpx.ConnectTimeout)) if isinstance(exc, httpx.HTTPStatusError): self._auth_ok = exc.response.status_code != 401 logger.warning("Home Assistant dispatch failed: %s", message) return False runtime.last_error = None runtime.last_service = f"{domain}.{service}" runtime.last_sent_summary = summary runtime.last_success_at = datetime.now(UTC) runtime.last_sent_raw_values = raw_values.copy() if mapping["fixture_type"] == "switch": normalized = self._normalized_value(raw_values[0], mapping) if normalized >= SWITCH_ON_THRESHOLD: runtime.switch_state = True elif normalized <= SWITCH_OFF_THRESHOLD: runtime.switch_state = False self._dispatch_count += 1 self._last_error = None self._last_successful_call_at = runtime.last_success_at self._last_connection_success_at = runtime.last_success_at self._last_connection_error = None self._reachable = True self._auth_ok = True return True def _should_dispatch( self, mapping: dict[str, object], runtime: MappingRuntime, previous_observed: list[int], raw_values: list[int], ) -> bool: fixture_type = str(mapping["fixture_type"]) if fixture_type in {"scene", "automation"}: previous = previous_observed[0] if previous_observed else 0 current = raw_values[0] if raw_values else 0 return previous < EDGE_TRIGGER_THRESHOLD <= current if fixture_type == "switch": current = self._normalized_value(raw_values[0], mapping) if runtime.switch_state is None: if current >= SWITCH_ON_THRESHOLD: return True if current <= SWITCH_OFF_THRESHOLD: return True return False if not runtime.switch_state and current >= SWITCH_ON_THRESHOLD: return True if runtime.switch_state and current <= SWITCH_OFF_THRESHOLD: return True return False previous_sent = runtime.last_sent_raw_values if previous_sent is None: return True deadband = max(0, int(mapping["deadband"])) if max(abs(current - old) for current, old in zip(raw_values, previous_sent, strict=False)) < deadband: return False rate_limit_hz = max(0.1, float(mapping["rate_limit_hz"])) interval = 1.0 / rate_limit_hz if runtime.last_success_at is None: return True return (monotonic() - self._datetime_to_monotonic_reference(runtime.last_success_at)) >= interval def _datetime_to_monotonic_reference(self, value: datetime) -> float: delta = datetime.now(UTC) - value return monotonic() - max(0.0, delta.total_seconds()) def _build_service_call( self, mapping: dict[str, object], raw_values: list[int], runtime: MappingRuntime, *, force: bool = False, ) -> tuple[str, str, dict[str, object], str] | None: fixture_type = str(mapping["fixture_type"]) entity_id = str(mapping["entity_id"]) transition = max(0.0, int(mapping["fade_ms"]) / 1000) entity_domain = self._entity_domain(entity_id) if fixture_type == "switch": normalized = self._normalized_value(raw_values[0], mapping) if entity_domain not in ON_OFF_ENTITY_DOMAINS: runtime.last_error = f"Entity {entity_id} understoetter ikke on/off-routing for fixturetype switch." runtime.last_error_at = datetime.now(UTC) return None if normalized >= SWITCH_ON_THRESHOLD or force: return (entity_domain, "turn_on", {"entity_id": entity_id}, f"{entity_domain}.turn_on {entity_id}") if normalized <= SWITCH_OFF_THRESHOLD: return (entity_domain, "turn_off", {"entity_id": entity_id}, f"{entity_domain}.turn_off {entity_id}") return None if fixture_type == "scene": if not force and self._normalized_value(raw_values[0], mapping) < EDGE_TRIGGER_THRESHOLD: return None if entity_domain != "scene": runtime.last_error = f"Entity {entity_id} er ikke en scene." runtime.last_error_at = datetime.now(UTC) return None return ("scene", "turn_on", {"entity_id": entity_id}, f"scene.turn_on {entity_id}") if fixture_type == "automation": if not force and self._normalized_value(raw_values[0], mapping) < EDGE_TRIGGER_THRESHOLD: return None if entity_domain != "automation": runtime.last_error = f"Entity {entity_id} er ikke en automation." runtime.last_error_at = datetime.now(UTC) return None return ("automation", "trigger", {"entity_id": entity_id}, f"automation.trigger {entity_id}") capabilities = self._entities_by_id.get(entity_id) if capabilities is None: runtime.last_error = "Entity-capabilities mangler. Hent HA-enheder eller test forbindelsen først." runtime.last_error_at = datetime.now(UTC) return None if entity_domain != "light": runtime.last_error = f"Entity {entity_id} understoetter ikke fixturetype {fixture_type}." runtime.last_error_at = datetime.now(UTC) return None if fixture_type == "dimmer": brightness = self._normalized_value(raw_values[0], mapping) if brightness <= 0 and not force: return ("light", "turn_off", {"entity_id": entity_id, "transition": transition}, f"light.turn_off {entity_id}") payload: dict[str, object] = {"entity_id": entity_id, "transition": transition} if bool(capabilities.get("supports_brightness", False)): payload["brightness"] = max(1, brightness) return ("light", "turn_on", payload, f"light.turn_on {entity_id} brightness={payload.get('brightness', 'on')}") if fixture_type == "rgb": return self._build_rgb_call(mapping, raw_values, capabilities, transition, include_white=False, force=force) if fixture_type == "rgbw": return self._build_rgb_call(mapping, raw_values, capabilities, transition, include_white=True, force=force) if fixture_type == "cct": return self._build_cct_call(mapping, raw_values, capabilities, transition, force=force) return None def _build_rgb_call( self, mapping: dict[str, object], raw_values: list[int], capabilities: dict[str, object], transition: float, *, include_white: bool, force: bool, ) -> tuple[str, str, dict[str, object], str] | None: entity_id = str(mapping["entity_id"]) has_master = bool(mapping["master_dimmer"]) offset = 1 if has_master else 0 master = self._normalized_value(raw_values[0], mapping) if has_master else 255 colors = [self._normalized_value(value, mapping) for value in raw_values[offset : offset + 3]] white = self._normalized_value(raw_values[offset + 3], mapping) if include_white else 0 scaled_colors = [int(round(color * (master / 255))) for color in colors] scaled_white = int(round(white * (master / 255))) if include_white else 0 brightness = master if has_master else max(scaled_colors + ([scaled_white] if include_white else [0])) if brightness <= 0 and not force: return ("light", "turn_off", {"entity_id": entity_id, "transition": transition}, f"light.turn_off {entity_id}") payload: dict[str, object] = {"entity_id": entity_id, "transition": transition} if bool(capabilities.get("supports_brightness", False)): payload["brightness"] = max(1, brightness) if include_white: if bool(capabilities.get("supports_rgbw", False)): payload["rgbw_color"] = scaled_colors + [scaled_white] return ( "light", "turn_on", payload, f"light.turn_on {entity_id} rgbw={payload['rgbw_color']} brightness={payload.get('brightness', 'on')}", ) if bool(capabilities.get("supports_rgbww", False)): payload["rgbww_color"] = scaled_colors + [scaled_white, 0] return ( "light", "turn_on", payload, f"light.turn_on {entity_id} rgbww={payload['rgbww_color']} brightness={payload.get('brightness', 'on')}", ) return None if bool(capabilities.get("supports_rgb", False)): payload["rgb_color"] = scaled_colors return ( "light", "turn_on", payload, f"light.turn_on {entity_id} rgb={payload['rgb_color']} brightness={payload.get('brightness', 'on')}", ) if bool(capabilities.get("supports_rgbw", False)): payload["rgbw_color"] = scaled_colors + [0] return ( "light", "turn_on", payload, f"light.turn_on {entity_id} rgbw={payload['rgbw_color']} brightness={payload.get('brightness', 'on')}", ) if bool(capabilities.get("supports_rgbww", False)): payload["rgbww_color"] = scaled_colors + [0, 0] return ( "light", "turn_on", payload, f"light.turn_on {entity_id} rgbww={payload['rgbww_color']} brightness={payload.get('brightness', 'on')}", ) return None def _build_cct_call( self, mapping: dict[str, object], raw_values: list[int], capabilities: dict[str, object], transition: float, *, force: bool, ) -> tuple[str, str, dict[str, object], str] | None: if not bool(capabilities.get("supports_color_temp_kelvin", False)): return None entity_id = str(mapping["entity_id"]) has_master = bool(mapping["master_dimmer"]) offset = 1 if has_master else 0 master = self._normalized_value(raw_values[0], mapping) if has_master else 255 warm = self._normalized_value(raw_values[offset], mapping) cold = self._normalized_value(raw_values[offset + 1], mapping) brightness = master if has_master else max(warm, cold) if brightness <= 0 and not force: return ("light", "turn_off", {"entity_id": entity_id, "transition": transition}, f"light.turn_off {entity_id}") total = max(1, warm + cold) cool_ratio = cold / total kelvin = int(round(2200 + (cool_ratio * 4300))) payload: dict[str, object] = { "entity_id": entity_id, "color_temp_kelvin": kelvin, "transition": transition, } if bool(capabilities.get("supports_brightness", False)): payload["brightness"] = max(1, brightness) return ( "light", "turn_on", payload, f"light.turn_on {entity_id} kelvin={kelvin} brightness={payload.get('brightness', 'on')}", ) def _build_test_values(self, mapping: dict[str, object]) -> list[int]: fixture_type = str(mapping["fixture_type"]) if fixture_type in {"scene", "automation", "switch"}: return [255] if fixture_type == "dimmer": return [255] if fixture_type == "rgb": return [255, 255, 80, 80] if bool(mapping["master_dimmer"]) else [255, 80, 80] if fixture_type == "rgbw": return [255, 255, 80, 80, 40] if bool(mapping["master_dimmer"]) else [255, 80, 80, 40] if fixture_type == "cct": return [255, 255, 120] if bool(mapping["master_dimmer"]) else [255, 120] return [255] def _read_raw_values(self, mapping: dict[str, object], values: list[int]) -> list[int]: start_address = int(mapping["start_address"]) span = self._channel_span(mapping) if start_address < 1 or start_address + span - 1 > 512: return [0] * span return values[start_address - 1 : start_address - 1 + span] def _channel_span(self, mapping: dict[str, object]) -> int: fixture_type = str(mapping["fixture_type"]) has_master = bool(mapping["master_dimmer"]) if fixture_type in {"dimmer", "switch", "scene", "automation"}: return 1 if fixture_type == "rgb": return 4 if has_master else 3 if fixture_type == "rgbw": return 5 if has_master else 4 if fixture_type == "cct": return 3 if has_master else 2 return 1 def _normalized_value(self, raw_value: int, mapping: dict[str, object]) -> int: clamped = max(0, min(255, int(raw_value))) if bool(mapping["invert_channel"]): clamped = 255 - clamped minimum = max(0, min(255, int(mapping["min_value"]))) maximum = max(minimum, min(255, int(mapping["max_value"]))) return int(round(minimum + ((clamped / 255) * (maximum - minimum)))) async def _load(self) -> None: async with self._session_factory() as session: config_setting = await session.get(Setting, HA_CONFIG_KEY) mappings_setting = await session.get(Setting, HA_MAPPINGS_KEY) if config_setting is not None and isinstance(config_setting.value, dict): raw_config = dict(config_setting.value) self._token = str(raw_config.get("token", "") or "").strip() self._config = self._normalize_config(raw_config) if mappings_setting is not None and isinstance(mappings_setting.value, dict): items = mappings_setting.value.get("items", []) if isinstance(items, list): self._mappings = [self._normalize_mapping(item) for item in items if isinstance(item, dict)] for mapping in self._mappings: self._ensure_runtime(int(mapping["id"]), self._channel_span(mapping)) async def _persist_setting(self, key: str, value: dict[str, object]) -> None: async with self._session_factory() as session: setting = await session.get(Setting, key) if setting is None: setting = Setting(key=key, value=value, updated_at=datetime.now(UTC)) session.add(setting) else: setting.value = value setting.updated_at = datetime.now(UTC) await session.commit() async def _persist_mappings(self) -> None: await self._persist_setting(HA_MAPPINGS_KEY, {"items": self._mappings}) def _default_config(self) -> dict[str, object]: return { "enabled": False, "base_url": "", "default_universe": 10, } def _normalize_config(self, payload: dict[str, object]) -> dict[str, object]: return { "enabled": bool(payload.get("enabled", False)), "base_url": str(payload.get("base_url", "") or "").strip(), "default_universe": max(1, min(63999, int(payload.get("default_universe", 10) or 10))), } def _normalize_mapping(self, payload: dict[str, object]) -> dict[str, object]: fixture_type = str(payload.get("fixture_type", "dimmer")).lower() if fixture_type not in {"dimmer", "rgb", "rgbw", "cct", "switch", "scene", "automation"}: fixture_type = "dimmer" mapping = { "id": int(payload.get("id", 0) or 0), "name": str(payload.get("name", "") or "").strip() or str(payload.get("entity_id", "") or "").strip(), "universe": max(1, min(63999, int(payload.get("universe", self._config["default_universe"]) or self._config["default_universe"]))), "start_address": max(1, min(512, int(payload.get("start_address", 1) or 1))), "fixture_type": fixture_type, "entity_id": str(payload.get("entity_id", "") or "").strip(), "rate_limit_hz": max(0.1, min(30.0, float(payload.get("rate_limit_hz", 5) or 5))), "deadband": max(0, min(255, int(payload.get("deadband", 2) or 2))), "fade_ms": max(0, min(10000, int(payload.get("fade_ms", 0) or 0))), "invert_channel": bool(payload.get("invert_channel", False)), "min_value": max(0, min(255, int(payload.get("min_value", 0) or 0))), "max_value": max(0, min(255, int(payload.get("max_value", 255) or 255))), "enabled": bool(payload.get("enabled", True)), "master_dimmer": bool(payload.get("master_dimmer", fixture_type in {"rgb", "rgbw", "cct"})), } if mapping["max_value"] < mapping["min_value"]: mapping["max_value"] = mapping["min_value"] if not mapping["name"]: mapping["name"] = mapping["entity_id"] or f"{fixture_type}-{mapping['universe']}-{mapping['start_address']}" return mapping def _normalize_entity(self, payload: dict[str, object]) -> dict[str, object] | None: entity_id = str(payload.get("entity_id", "") or "").strip() if "." not in entity_id: return None domain = entity_id.split(".", 1)[0] if domain not in {"light", "switch", "input_boolean", "scene", "automation"}: return None attributes = payload.get("attributes", {}) attr_map = attributes if isinstance(attributes, dict) else {} supported_modes = attr_map.get("supported_color_modes", []) modes = {str(mode).lower() for mode in supported_modes if isinstance(mode, str)} color_mode = str(attr_map.get("color_mode", "") or "").lower() if color_mode: modes.add(color_mode) friendly_name = str(attr_map.get("friendly_name", "") or "") supports_brightness = domain == "light" and any(mode not in {"onoff"} for mode in modes) supports_rgb = domain == "light" and any(mode in {"rgb", "hs", "xy"} for mode in modes) supports_rgbw = domain == "light" and "rgbw" in modes supports_rgbww = domain == "light" and "rgbww" in modes supports_color_temp_kelvin = domain == "light" and "color_temp" in modes return { "entity_id": entity_id, "domain": domain, "friendly_name": friendly_name, "state": str(payload.get("state", "")), "supports_brightness": supports_brightness, "supports_rgb": supports_rgb, "supports_rgbw": supports_rgbw, "supports_rgbww": supports_rgbww, "supports_color_temp_kelvin": supports_color_temp_kelvin, } def _entity_domain(self, entity_id: str) -> str: if "." not in entity_id: return "" return entity_id.split(".", 1)[0].strip().lower() def _serialize_mapping(self, mapping: dict[str, object]) -> dict[str, object]: runtime = self._ensure_runtime(int(mapping["id"]), self._channel_span(mapping)) status = "disabled" if bool(mapping["enabled"]): status = "sending" if runtime.in_flight else "active" if runtime.last_error: status = "error" return { **mapping, "channel_span": self._channel_span(mapping), "status": status, "last_dmx_values": runtime.last_dmx_values, "last_sent_summary": runtime.last_sent_summary, "last_service": runtime.last_service, "last_success_at": self._iso(runtime.last_success_at), "last_error": runtime.last_error, "last_error_at": self._iso(runtime.last_error_at), "in_flight": runtime.in_flight, } def _next_mapping_id(self) -> int: return max((int(mapping["id"]) for mapping in self._mappings), default=0) + 1 def _find_mapping(self, mapping_id: int) -> dict[str, object]: for mapping in self._mappings: if int(mapping["id"]) == mapping_id: return mapping raise LookupError("Home Assistant mapping not found") def _ensure_runtime(self, mapping_id: int, span: int) -> MappingRuntime: runtime = self._mapping_runtime.get(mapping_id) if runtime is None: runtime = MappingRuntime(last_dmx_values=[0] * span) self._mapping_runtime[mapping_id] = runtime return runtime if len(runtime.last_dmx_values) != span: runtime.last_dmx_values = [0] * span runtime.last_sent_raw_values = None runtime.pending_raw_values = None return runtime def _prune_orphaned_runtimes(self) -> None: active_ids = {int(mapping["id"]) for mapping in self._mappings} for mapping_id in list(self._mapping_runtime): if mapping_id not in active_ids: self._mapping_runtime.pop(mapping_id, None) async def _drain_mapping_tasks(self) -> None: while self._mapping_tasks: tasks = list(self._mapping_tasks.values()) await asyncio.gather(*tasks, return_exceptions=True) def _iso(self, value: datetime | None) -> str | None: return value.isoformat() if value is not None else None def _summarize_exception(self, exc: Exception) -> str: if isinstance(exc, httpx.HTTPStatusError): return f"HTTP {exc.response.status_code} fra Home Assistant" if isinstance(exc, (httpx.ConnectTimeout, httpx.ReadTimeout)): return "Timeout ved kald til Home Assistant" if isinstance(exc, httpx.ConnectError): return "Home Assistant kunne ikke kontaktes" return str(exc) or exc.__class__.__name__