TuxDMX-WebUI/backend/app/dmx/output_service.py

from __future__ import annotations

from datetime import UTC, datetime
from typing import Literal

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.core.config import get_settings
from app.core.database import SessionLocal
from app.dmx.backends import (
    ArtNetDmxBackend,
    ArtNetNode,
    OlaDmxBackend,
    SimulatorDmxBackend,
    discover_artnet_nodes,
)
from app.dmx.engine import DmxEngine
from app.models.entities import Setting


DMX_OUTPUT_SETTING_KEY = "dmx_output"
BackendKind = Literal["simulator", "ola", "artnet"]


class DmxOutputService:
    def __init__(
        self,
        engine: DmxEngine,
        session_factory: async_sessionmaker[AsyncSession] = SessionLocal,
    ) -> None:
        self.engine = engine
        self._session_factory = session_factory
        self._settings = get_settings()
        self._config = self._default_config()
        self._artnet_nodes: list[ArtNetNode] = []

    async def startup(self) -> None:
        await self._load_config()
        self.engine.configure_backend(self._build_backend(self._config))

    async def get_config(self) -> dict[str, object]:
        return {
            **self._config,
            "artnet_nodes": [self._serialize_node(node) for node in self._artnet_nodes],
        }

    async def save_config(self, payload: dict[str, object]) -> dict[str, object]:
        normalized = self._normalize_config(payload)
        self._config = normalized
        await self._persist_config(normalized)
        backend = self._build_backend(normalized)
        if self.engine.is_running:
            await self.engine.replace_backend(backend)
        else:
            self.engine.configure_backend(backend)
        return await self.get_config()

    async def discover_artnet(self, timeout_s: float = 1.0) -> dict[str, object]:
        nodes = await self._discover(timeout_s)
        self._artnet_nodes = nodes
        return {
            "items": [self._serialize_node(node) for node in nodes],
            "count": len(nodes),
        }

    async def _discover(self, timeout_s: float) -> list[ArtNetNode]:
        return await self._to_thread_discovery(timeout_s)

    async def _to_thread_discovery(self, timeout_s: float) -> list[ArtNetNode]:
        import asyncio

        return await asyncio.to_thread(discover_artnet_nodes, timeout_s)

    async def _load_config(self) -> None:
        async with self._session_factory() as session:
            setting = await session.get(Setting, DMX_OUTPUT_SETTING_KEY)
        if setting is None or not isinstance(setting.value, dict):
            return
        self._config = self._normalize_config(setting.value)

    async def _persist_config(self, config: dict[str, object]) -> None:
        async with self._session_factory() as session:
            setting = await session.get(Setting, DMX_OUTPUT_SETTING_KEY)
            if setting is None:
                setting = Setting(
                    key=DMX_OUTPUT_SETTING_KEY,
                    value=config,
                    updated_at=datetime.now(UTC),
                )
                session.add(setting)
            else:
                setting.value = config
                setting.updated_at = datetime.now(UTC)
            await session.commit()

    def _default_config(self) -> dict[str, object]:
        backend: BackendKind = "simulator" if self._settings.simulator_enabled else "ola"
        return {
            "backend": backend,
            "universe": self._settings.ola_universe,
            "output_port": self._settings.ola_output_port or "",
            "target_host": "",
        }

    def _normalize_config(self, payload: dict[str, object]) -> dict[str, object]:
        backend = str(payload.get("backend", self._default_config()["backend"])).lower()
        if backend not in {"simulator", "ola", "artnet"}:
            backend = "simulator"
        universe = int(payload.get("universe", self._settings.ola_universe) or self._settings.ola_universe)
        output_port = str(payload.get("output_port", "") or "").strip()
        target_host = str(payload.get("target_host", "") or "").strip()
        if backend == "artnet" and not target_host:
            target_host = "255.255.255.255"
        if backend == "simulator":
            output_port = "simulator"
        return {
            "backend": backend,
            "universe": max(1, min(63999, universe)),
            "output_port": output_port,
            "target_host": target_host,
        }

    def _build_backend(self, config: dict[str, object]):
        backend = str(config["backend"])
        universe = int(config["universe"])
        output_port = str(config.get("output_port", "") or "") or None
        if backend == "ola":
            return OlaDmxBackend(
                universe=universe,
                output_port=output_port,
                send_timeout_s=self._settings.ola_send_timeout_ms / 1000,
            )
        if backend == "artnet":
            target_host = str(config.get("target_host", "") or "255.255.255.255")
            return ArtNetDmxBackend(
                universe=universe,
                target_host=target_host,
                output_port=output_port or target_host,
            )
        return SimulatorDmxBackend(universe=universe)

    def _serialize_node(self, node: ArtNetNode) -> dict[str, object]:
        return {
            "ip": node.ip,
            "short_name": node.short_name,
            "long_name": node.long_name,
            "label": node.label,
            "net": node.net,
            "sub_switch": node.sub_switch,
            "port_count": node.port_count,
            "raw_port_address": node.raw_port_address,
        }