thomas
1f110866f5
CI / backend (pull_request) Canceled after 0s
CI / shell (pull_request) Canceled after 0s
CI / frontend (pull_request) Canceled after 0s
CI / arm64-smoke (pull_request) Canceled after 0s
CI / backend (push) Canceled after 0s
CI / shell (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / arm64-smoke (push) Canceled after 0s
870 lines
38 KiB
Python
870 lines
38 KiB
Python
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__
|