from __future__ import annotations from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.core.database import SessionLocal from app.dmx.engine import DmxEngine from app.dmx.frame import FrameLayer from app.models.entities import Scene from app.models.schemas import ScenePayload from app.patch.service import PatchService class SceneService: def __init__( self, engine: DmxEngine, session_factory: async_sessionmaker[AsyncSession] = SessionLocal, ) -> None: self.engine = engine self._session_factory = session_factory self.patch = PatchService(session_factory) self.active_slug: str | None = None async def save(self, payload: ScenePayload, scene_id: int | None = None) -> dict[str, object]: async with self._session_factory() as session: scene = await self._load_for_write(session, payload.slug, scene_id) if scene is None: scene = Scene( name=payload.name, slug=payload.slug, color=payload.color, icon=payload.icon, priority=payload.priority, fade_in_ms=payload.fade_in_ms, fade_out_ms=payload.fade_out_ms, hold_ms=payload.hold_ms, master_limit=payload.master_limit, values=[], targets=[], tags=payload.tags, ) session.add(scene) scene.name = payload.name scene.slug = payload.slug scene.color = payload.color scene.icon = payload.icon scene.priority = payload.priority scene.fade_in_ms = payload.fade_in_ms scene.fade_out_ms = payload.fade_out_ms scene.hold_ms = payload.hold_ms scene.master_limit = payload.master_limit scene.values = [item.model_dump() for item in payload.values] scene.targets = [item.model_dump() for item in payload.targets] scene.tags = payload.tags await session.commit() await session.refresh(scene) return self._serialize_scene(scene) async def list(self) -> list[dict[str, object]]: async with self._session_factory() as session: result = await session.execute(select(Scene).order_by(Scene.priority, Scene.created_at, Scene.id)) return [self._serialize_scene(scene) for scene in result.scalars().all()] async def get(self, scene_id: int) -> dict[str, object]: async with self._session_factory() as session: scene = await session.get(Scene, scene_id) if scene is None: raise LookupError("Scene not found") return self._serialize_scene(scene) async def activate(self, slug: str) -> dict[str, object]: async with self._session_factory() as session: result = await session.execute(select(Scene).where(Scene.slug == slug).limit(1)) scene = result.scalar_one_or_none() if scene is None: raise LookupError("Scene not found") values_by_universe, precedence_map_by_universe = await self._resolve_layer_data(scene) self.engine.set_layer( FrameLayer( name=f"scene:{scene.slug}", priority=scene.priority, values_by_universe=values_by_universe, precedence_map_by_universe=precedence_map_by_universe, ) ) self.active_slug = slug return self._serialize_scene(scene) async def release(self, slug: str) -> None: self.engine.remove_layer(f"scene:{slug}") if self.active_slug == slug: self.active_slug = None async def delete(self, scene_id: int) -> dict[str, object]: async with self._session_factory() as session: scene = await session.get(Scene, scene_id) if scene is None: raise LookupError("Scene not found") slug = scene.slug await self.release(slug) await session.delete(scene) await session.commit() return {"deleted": slug} async def _load_for_write( self, session: AsyncSession, slug: str, scene_id: int | None, ) -> Scene | None: if scene_id is not None: scene = await session.get(Scene, scene_id) if scene is None: raise LookupError("Scene not found") result = await session.execute(select(Scene).where(Scene.slug == slug, Scene.id != scene_id).limit(1)) if result.scalar_one_or_none() is not None: raise ValueError("Scene slug findes allerede") return scene result = await session.execute(select(Scene).where(Scene.slug == slug).limit(1)) return result.scalar_one_or_none() async def _resolve_layer_data(self, scene: Scene) -> tuple[dict[int, dict[int, int]], dict[int, dict[int, str]]]: values_by_universe: dict[int, dict[int, int]] = {1: {}} precedence_map_by_universe: dict[int, dict[int, str]] = {1: {}} for entry in scene.values: if not isinstance(entry, dict): continue if "channel" not in entry: continue channel = int(entry["channel"]) if not 1 <= channel <= 512: continue values_by_universe.setdefault(1, {})[channel] = max(0, min(255, int(entry.get("value", 0)))) precedence_map_by_universe.setdefault(1, {})[channel] = str(entry.get("precedence", "ltp")).lower() if not scene.targets: return values_by_universe, precedence_map_by_universe patches = await self.patch.list_instances() for target in scene.targets: if not isinstance(target, dict): continue matches = self._resolve_target_matches(target, patches) attribute_values = target.get("values", []) if not isinstance(attribute_values, list): continue for patch in matches: for attribute in attribute_values: if not isinstance(attribute, dict): continue channel = self._match_patch_channel(patch, str(attribute.get("attribute", ""))) if channel is None: continue absolute_channel = int(patch["start_address"]) + int(channel["index"]) - 1 universe = int(patch.get("universe", 1)) values_by_universe.setdefault(universe, {})[absolute_channel] = max( 0, min(255, int(attribute.get("value", 0))) ) precedence_map_by_universe.setdefault(universe, {})[absolute_channel] = str( attribute.get("precedence") or channel.get("precedence", "ltp") ).lower() return values_by_universe, precedence_map_by_universe def _resolve_target_matches( self, target: dict[str, object], patches: list[dict[str, object]], ) -> list[dict[str, object]]: target_type = str(target.get("target_type", "fixture")) if target_type == "group": group_name = str(target.get("group_name", "")).strip() if not group_name: return [] group_key = group_name.casefold() return [ patch for patch in patches if any(str(name).casefold() == group_key for name in patch.get("group_names", [])) ] patch_id = target.get("patch_id") if patch_id is None: return [] return [patch for patch in patches if int(patch["id"]) == int(patch_id)] def _match_patch_channel( self, patch: dict[str, object], attribute_name: str, ) -> dict[str, object] | None: normalized = self._normalize_attribute(attribute_name) for channel in patch.get("channels", []): if not isinstance(channel, dict): continue candidates = { self._normalize_attribute(str(channel.get("key", ""))), self._normalize_attribute(str(channel.get("display_name", ""))), } if normalized in candidates: return channel return None def _normalize_attribute(self, value: str) -> str: return "".join(character for character in value.casefold() if character.isalnum()) def _serialize_scene(self, scene: Scene) -> dict[str, object]: return { "id": scene.id, "name": scene.name, "slug": scene.slug, "color": scene.color, "icon": scene.icon, "priority": scene.priority, "fade_in_ms": scene.fade_in_ms, "fade_out_ms": scene.fade_out_ms, "hold_ms": scene.hold_ms, "master_limit": scene.master_limit, "values": scene.values, "targets": scene.targets, "tags": scene.tags, "is_active": scene.slug == self.active_slug, }