from __future__ import annotations 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.homeassistant.service import HomeAssistantService from app.patch.service import PatchService class LiveDeskService: def __init__( self, engine: DmxEngine, home_assistant: HomeAssistantService | None = None, session_factory: async_sessionmaker[AsyncSession] = SessionLocal, ) -> None: self.engine = engine self.patch = PatchService(session_factory) self.home_assistant = home_assistant self._values_by_patch: dict[int, dict[int, int]] = {} self._values_by_home_assistant_mapping: dict[int, dict[int, int]] = {} async def snapshot(self) -> dict[str, object]: patches = await self.patch.list_instances() items: list[dict[str, object]] = [] for patch in patches: relative_values = self._values_by_patch.get(int(patch["id"]), {}) channels: list[dict[str, object]] = [] for channel in patch.get("channels", []): if not isinstance(channel, dict): continue relative_index = int(channel.get("index", 0)) absolute_channel = int(patch["start_address"]) + relative_index - 1 channels.append( { "index": relative_index, "absolute_channel": absolute_channel, "key": str(channel.get("key", f"channel-{relative_index}")), "display_name": str(channel.get("display_name", f"Channel {relative_index}")), "precedence": str(channel.get("precedence", "ltp")), "resolution": int(channel.get("resolution", 8)), "value": int(relative_values.get(relative_index, 0)), } ) items.append( { "id": f"patch:{patch['id']}", "source_type": "patch", "source_id": patch["id"], "patch_id": patch["id"], "name": patch["name"], "manufacturer": patch["manufacturer"], "model": patch["model"], "mode_key": patch["mode_key"], "universe": patch["universe"], "start_address": patch["start_address"], "end_address": patch["end_address"], "channel_count": patch["channel_count"], "entity_id": None, "status": "active" if patch.get("enabled", True) else "disabled", "last_sent_summary": None, "channels": channels, } ) if self.home_assistant is not None: mappings = (await self.home_assistant.list_mappings())["items"] for mapping in mappings: if not isinstance(mapping, dict): continue mapping_id = int(mapping["id"]) relative_values = self._values_by_home_assistant_mapping.get(mapping_id, {}) channels = [] for channel in self._build_home_assistant_channels(mapping): relative_index = int(channel["index"]) absolute_channel = int(mapping["start_address"]) + relative_index - 1 channels.append( { "index": relative_index, "absolute_channel": absolute_channel, "key": str(channel["key"]), "display_name": str(channel["display_name"]), "precedence": str(channel["precedence"]), "resolution": 8, "value": int(relative_values.get(relative_index, 0)), } ) items.append( { "id": f"home_assistant:{mapping_id}", "source_type": "home_assistant", "source_id": mapping_id, "patch_id": None, "name": mapping["name"], "manufacturer": "Home Assistant", "model": mapping["fixture_type"], "mode_key": f"HA {mapping['fixture_type']}", "universe": mapping["universe"], "start_address": mapping["start_address"], "end_address": int(mapping["start_address"]) + int(mapping["channel_span"]) - 1, "channel_count": mapping["channel_span"], "entity_id": mapping["entity_id"], "status": mapping["status"], "last_sent_summary": mapping["last_sent_summary"], "channels": channels, } ) return {"items": items} async def set_patch_values(self, patch_id: int, values: dict[int, int]) -> dict[str, object]: patches = await self.patch.list_instances() patch = next((item for item in patches if int(item["id"]) == patch_id), None) if patch is None: raise LookupError("Patch not found") channel_map = { int(channel["index"]): channel for channel in patch.get("channels", []) if isinstance(channel, dict) and "index" in channel } sanitized_values: dict[int, int] = {} absolute_values: dict[int, int] = {} precedence_map: dict[int, str] = {} for relative_index, raw_value in values.items(): if relative_index not in channel_map: continue value = max(0, min(255, int(raw_value))) sanitized_values[relative_index] = value absolute_channel = int(patch["start_address"]) + relative_index - 1 absolute_values[absolute_channel] = value precedence_map[absolute_channel] = str(channel_map[relative_index].get("precedence", "ltp")).lower() self._values_by_patch[patch_id] = sanitized_values self.engine.set_layer( FrameLayer.from_channel_values( name=f"mixer:patch:{patch_id}", priority=90, values=absolute_values, precedence_map=precedence_map, universe=int(patch["universe"]), ) ) return await self.snapshot() async def clear_patch(self, patch_id: int) -> dict[str, object]: self._values_by_patch.pop(patch_id, None) self.engine.remove_layer(f"mixer:patch:{patch_id}") return await self.snapshot() async def set_home_assistant_mapping_values(self, mapping_id: int, values: dict[int, int]) -> dict[str, object]: if self.home_assistant is None: raise LookupError("Home Assistant live mixer is not available") mappings = (await self.home_assistant.list_mappings())["items"] mapping = next((item for item in mappings if int(item["id"]) == mapping_id), None) if mapping is None: raise LookupError("Home Assistant mapping not found") channel_map = { int(channel["index"]): channel for channel in self._build_home_assistant_channels(mapping) } sanitized_values: dict[int, int] = {} absolute_values: dict[int, int] = {} precedence_map: dict[int, str] = {} for relative_index, raw_value in values.items(): if relative_index not in channel_map: continue value = max(0, min(255, int(raw_value))) sanitized_values[relative_index] = value absolute_channel = int(mapping["start_address"]) + relative_index - 1 absolute_values[absolute_channel] = value precedence_map[absolute_channel] = str(channel_map[relative_index].get("precedence", "ltp")).lower() self._values_by_home_assistant_mapping[mapping_id] = sanitized_values self.engine.set_layer( FrameLayer.from_channel_values( name=f"mixer:home-assistant:{mapping_id}", priority=90, values=absolute_values, precedence_map=precedence_map, universe=int(mapping["universe"]), ) ) return await self.snapshot() async def clear_home_assistant_mapping(self, mapping_id: int) -> dict[str, object]: self._values_by_home_assistant_mapping.pop(mapping_id, None) self.engine.remove_layer(f"mixer:home-assistant:{mapping_id}") return await self.snapshot() def _build_home_assistant_channels(self, mapping: dict[str, object]) -> list[dict[str, object]]: fixture_type = str(mapping["fixture_type"]) has_master = bool(mapping.get("master_dimmer", False)) if fixture_type == "switch": return [{"index": 1, "key": "Switch", "display_name": "State", "precedence": "ltp"}] if fixture_type == "scene": return [{"index": 1, "key": "Scene", "display_name": "Trigger", "precedence": "ltp"}] if fixture_type == "automation": return [{"index": 1, "key": "Automation", "display_name": "Trigger", "precedence": "ltp"}] if fixture_type == "dimmer": return [{"index": 1, "key": "Dimmer", "display_name": "Dimmer", "precedence": "ltp"}] channels: list[dict[str, object]] = [] next_index = 1 if has_master: channels.append({"index": next_index, "key": "Master", "display_name": "Master", "precedence": "ltp"}) next_index += 1 if fixture_type == "rgb": names = ["Red", "Green", "Blue"] elif fixture_type == "rgbw": names = ["Red", "Green", "Blue", "White"] elif fixture_type == "cct": names = ["Warm", "Cold"] else: names = ["Channel"] for name in names: channels.append({"index": next_index, "key": name, "display_name": name, "precedence": "ltp"}) next_index += 1 return channels