hms-mediaengine/hms_app/config.py

"""Persistente Konfiguration: config.json, Defaults, Validierung."""

from __future__ import annotations

import copy
import json
from pathlib import Path

DEFAULTS: dict = {
    "web": {"bind": "0.0.0.0", "port": 8080},
    "preview": {"width": 640, "height": 360, "fps": 15,
                "jpeg_quality": 80},
    "output": {"fullscreen": False},
    "media": {"dir": "media", "thumbs_dir": "thumbs",
              "max_upload_mb": 2048},
    "artnet": {
        "enabled": True,
        "listen_port": 6454,
        "universes": [0],
        "master_channel": 1,
        "layer_start_channel": 2,
        "playback_channel": 6,
        "retrigger_channel": 7,
        "blackout_channel": 8,
        "fx1_channel": None,
        "fx2_channel": None,
        "signal_loss_policy": "hold",
        "signal_loss_timeout_s": 10,
    },
    "engine": {"max_layers": 8, "loop_default": True},
}


def deep_merge(base: dict, override: dict) -> dict:
    out = dict(base)
    for k, v in override.items():
        if isinstance(v, dict) and isinstance(out.get(k), dict):
            out[k] = deep_merge(out[k], v)
        else:
            out[k] = v
    return out


def load(root: Path) -> dict:
    """Lädt config.json und füllt fehlende Schlüssel mit Defaults."""
    file = root / "config.json"
    data: dict = {}
    if file.exists():
        try:
            data = json.loads(file.read_text("utf-8"))
        except (OSError, ValueError) as e:
            print(f"[Config] FEHLER config.json: {e} – nutze Defaults")
    if not isinstance(data, dict):
        data = {}
    return deep_merge(DEFAULTS, data)


def save(root: Path, cfg: dict) -> bool:
    file = root / "config.json"
    try:
        tmp = file.with_suffix(".json.tmp")
        tmp.write_text(
            json.dumps(cfg, ensure_ascii=False, indent=2), "utf-8")
        tmp.replace(file)
        return True
    except OSError as e:
        print(f"[Config] Schreiben fehlgeschlagen: {e}")
        return False


def validate(new: dict) -> tuple[dict | None, str | None]:
    """Prüft ein komplettes Settings-Objekt. Rückgabe: (normalized, error)."""
    merged = deep_merge(copy.deepcopy(DEFAULTS), new if isinstance(new, dict) else {})
    errors: list[str] = []

    def _int(d, key, lo, hi, label):
        try:
            v = int(d[key])
        except (KeyError, TypeError, ValueError):
            errors.append(label)
            return None
        if not lo <= v <= hi:
            errors.append(label)
            return None
        d[key] = v
        return v

    w = merged["web"]
    _int(w, "port", 1, 65535, "web.port")
    if not isinstance(w.get("bind"), str):
        errors.append("web.bind")

    p = merged["preview"]
    _int(p, "width", 64, 3840, "preview.width")
    _int(p, "height", 64, 2160, "preview.height")
    _int(p, "fps", 1, 60, "preview.fps")
    _int(p, "jpeg_quality", 10, 95, "preview.jpeg_quality")

    m = merged["media"]
    _int(m, "max_upload_mb", 1, 8192, "media.max_upload_mb")
    for k in ("dir", "thumbs_dir"):
        if not isinstance(m.get(k), str) or not m[k] or "/" in m[k] \
                or m[k] in (".", ".."):
            errors.append(f"media.{k}")

    a = merged["artnet"]
    _int(a, "listen_port", 1, 65535, "artnet.listen_port")
    uni = a.get("universes")
    if (not isinstance(uni, list) or not uni
            or len(set(uni)) != len(uni)
            or any(not isinstance(u, int) or not 0 <= u <= 32767
                   for u in uni)):
        errors.append("artnet.universes")
    for k in ("master_channel", "layer_start_channel", "playback_channel",
              "retrigger_channel", "blackout_channel", "fx1_channel",
              "fx2_channel"):
        v = a.get(k)
        if v is not None and (not isinstance(v, int)
                              or not 1 <= v <= 512):
            errors.append(f"artnet.{k}")
    _int(a, "signal_loss_timeout_s", 1, 3600,
         "artnet.signal_loss_timeout_s")
    if a.get("signal_loss_policy") not in ("hold", "fade_black"):
        errors.append("artnet.signal_loss_policy")

    e = merged["engine"]
    _int(e, "max_layers", 1, 16, "engine.max_layers")
    if not isinstance(e.get("loop_default"), bool):
        errors.append("engine.loop_default")
    if not isinstance(merged["output"].get("fullscreen"), bool):
        errors.append("output.fullscreen")

    if errors:
        return None, "ungültig: " + ", ".join(errors)
    return merged, None