vf-cuda-grid/controller/cuda_grid_controller/config.py

"""Конфигурация — pydantic models + YAML loader.

Структура YAML:
  broker:
    host: localhost
    port: 1883
    username_env: MQTT_USERNAME
    password_env: MQTT_PASSWORD

  instances:
    - name: livingroom_tv
      zmq_endpoint: tcp://127.0.0.1:5555
      default_layout: quad

  ha_discovery:
    enabled: true
    prefix: homeassistant
    device_name: "CUDA Grid Composer"

  http:
    host: 0.0.0.0
    port: 8080

  log:
    level: INFO
"""

from __future__ import annotations

import os
from pathlib import Path
from typing import Self

import yaml
from pydantic import BaseModel, Field, field_validator


class BrokerCfg(BaseModel):
    host: str = "localhost"
    port: int = 1883
    client_id: str = "cuda-grid-controller"
    username_env: str | None = None
    password_env: str | None = None
    keepalive_sec: int = 30

    @property
    def username(self) -> str | None:
        return os.environ.get(self.username_env) if self.username_env else None

    @property
    def password(self) -> str | None:
        return os.environ.get(self.password_env) if self.password_env else None


class InstanceCfg(BaseModel):
    """Один FFmpeg pipeline = одна cuda_grid filter instance."""

    name: str = Field(description="уникальное имя — становится частью HA entity ID")
    zmq_endpoint: str = Field(
        description="ZMQ endpoint видео-pipeline (cuda_grid + overlay filters)"
    )
    audio_zmq_endpoint: str | None = Field(
        default=None,
        description="ZMQ endpoint отдельного audio sidecar (Phase 5d split-process). "
                    "None = audio_set/intercom вызывают video pipeline (Phase 5a single source)",
    )
    default_layout: str = "quad"
    filter_target: str = Field(
        default="Parsed_cuda_grid_0",
        description="Filter target name в FFmpeg filter graph (для process_command)",
    )
    output_rtsp_url: str | None = Field(
        default=None,
        description="URL куда pipeline push'ит composed stream — controller read'ит для snapshot/preview endpoints",
    )
    audio_sources: list["AudioSourceCfg"] = Field(
        default_factory=list,
        description="Audio sources для astreamselect switching (Phase 5b)",
    )
    audio_filter_target: str = Field(
        default="astreamselect@as",
        description="Target filter для ZMQ команд переключения audio (должен соответствовать pipeline filter_complex)",
    )
    music_volume_target: str = Field(
        default="volume@music",
        description="Target filter для управления громкостью music chain (Phase 5c ducking)",
    )
    intercom_volume_target: str = Field(
        default="volume@intercom",
        description="Target filter для управления громкостью intercom (Phase 5c)",
    )
    music_ducked_volume: float = Field(
        default=0.2, ge=0.0, le=1.0,
        description="Громкость music когда intercom активен (0.2 = -14 dB)",
    )
    layout_filter_target: str = Field(
        default="streamselect@layout",
        description="ZMQ target streamselect filter для runtime layout switching",
    )
    main_cam_filter_target: str = Field(
        default="streamselect@main_cam",
        description="ZMQ target streamselect filter для dynamic main camera (single layout)",
    )
    mpp_main_filter_target: str = Field(
        default="streamselect@mpp_main",
        description="ZMQ target streamselect filter для dynamic main в main_plus_preview layout",
    )
    layout_map: dict[str, int] = Field(
        default_factory=lambda: {"quad": 0, "single": 1, "main_plus_preview": 2},
        description="layout name → streamselect map index (соответствует pipeline filter_complex)",
    )
    snapshot_history: "SnapshotHistoryCfg" = Field(
        default_factory=lambda: SnapshotHistoryCfg(),
        description="Periodic snapshot capture в shared volume",
    )


class SnapshotHistoryCfg(BaseModel):
    enabled: bool = False
    interval_sec: int = Field(default=60, ge=5, le=3600)
    keep_last: int = Field(default=120, ge=1, le=10000,
        description="Сколько последних snapshots держать (FIFO eviction)")
    dir: str = Field(default="/var/lib/cuda-grid/snapshots",
        description="Базовая директория; instance-name становится поддиректорией")


class AudioSourceCfg(BaseModel):
    """Описание audio source в порядке как они добавлены в pipeline -i ..."""
    name: str = Field(description="Уникальное имя для API (e.g. 'europa_plus')")
    index: int = Field(ge=0, description="Index в astreamselect inputs (соответствует порядку -i в pipeline)")
    label: str | None = Field(default=None, description="UI-friendly label (default = name)")

    @field_validator("name")
    @classmethod
    def name_alnum(cls, v: str) -> str:
        if not v.replace("_", "").isalnum():
            raise ValueError(f"instance name '{v}' must be alphanumeric + underscore")
        return v


class HaDiscoveryCfg(BaseModel):
    enabled: bool = True
    prefix: str = "homeassistant"
    device_name: str = "CUDA Grid Composer"
    device_identifier: str = "cuda_grid_controller"


class HttpCfg(BaseModel):
    host: str = "0.0.0.0"
    port: int = 8080


class LogCfg(BaseModel):
    level: str = "INFO"


class Config(BaseModel):
    broker: BrokerCfg = BrokerCfg()
    instances: list[InstanceCfg] = []
    ha_discovery: HaDiscoveryCfg = HaDiscoveryCfg()
    http: HttpCfg = HttpCfg()
    log: LogCfg = LogCfg()
    # Frigate bridge — late import чтобы избежать circular dep
    frigate: dict | None = None  # parsed в FrigateBridgeCfg при runtime
    # Dynamic overlays (charts/chats) — late import тоже
    dynamic_overlays: dict | None = None  # parsed в DynamicRenderer cfg
    icon_dir: str = Field(
        default="/var/lib/cuda-grid/icons",
        description="Shared volume куда controller пишет dynamic PNG; filter (`icon_dir=` option) читает оттуда",
    )

    @classmethod
    def from_yaml(cls, path: Path | str) -> Self:
        with open(path) as f:
            data = yaml.safe_load(f) or {}
        return cls.model_validate(data)