hms-mediaengine/_entwicklung/packages/parameter_engine/hms_parameter/engine.py

"""Parameter-Engine: Prioritäten, Übernahme, Frame-Snapshot (PLAN.md §11).

Alle Steuerquellen (Browser, Art-Net, später Timeline/Audio/KI) laufen über
diese Engine; direkte Renderer-Zugriffe sind verboten (§11, §33).
"""

from __future__ import annotations

import enum
import time
from dataclasses import dataclass, field

from hms_parameter.paths import validate_parameter_path


class ControlSource(enum.IntEnum):
    """Steuerquellen in Prioritätsordnung (§11.2)."""

    SAFETY = 1    # Not-Aus/Blackout, überstimmt alles
    OPERATOR = 2  # expliziter manueller Override
    CONSOLE = 3   # freigegebenes Lichtpult (Art-Net)
    WEB = 4       # Browser-Livebedienung
    TIMELINE = 5  # reserviert
    AUDIO = 6     # reserviert (Modulatoren)
    AI = 7        # reserviert, niedrigste Priorität


class MergeMode(enum.Enum):
    """Übernahmeverfahren (§11.3)."""

    LTP = "ltp"  # letzte Änderung gewinnt (Standard)
    HTP = "htp"  # höchster Wert gewinnt (optional für Intensität)


@dataclass
class _Binding:
    value: float
    last_change_ns: int


class ParameterFrame:
    """Unveränderlicher Snapshot aller Parameter für genau einen Frame (§11.4)."""

    __slots__ = ("_values", "revision", "created_ns")

    def __init__(self, values: dict[str, float], revision: int) -> None:
        object.__setattr__(self, "_values", dict(values))
        object.__setattr__(self, "revision", revision)
        object.__setattr__(self, "created_ns", time.monotonic_ns())

    def get(self, path: str, default: float = 0.0) -> float:
        return self._values.get(path, default)

    def as_dict(self) -> dict[str, float]:
        return dict(self._values)

    def __contains__(self, path: str) -> bool:
        return path in self._values


class RevisionConflict(Exception):
    """Erwartete Revision stimmt nicht (optimistische Sperre, §23.2)."""

    def __init__(self, current: int, expected: int) -> None:
        self.current = current
        self.expected = expected
        super().__init__(f"revision conflict: current={current}, expected={expected}")


@dataclass
class ParameterEngine:
    """Autoritative Parameter-Instanz des Control Core.

    - set_value: Override einer Quelle mit Prioritätsprüfung
    - release: Rückgabe an nächstniedrigere Quelle (§11.3)
    - snapshot: atomarer Frame-Snapshot (§11.4)
    """

    revision: int = 0
    default: float = 0.0
    merge_mode: MergeMode = MergeMode.LTP
    _bindings: dict[str, dict[ControlSource, _Binding]] = field(
        default_factory=dict, repr=False
    )
    _defaults: dict[str, float] = field(default_factory=dict, repr=False)

    def set_value(
        self,
        path: str,
        value: float,
        source: ControlSource,
        expected_revision: int | None = None,
    ) -> int:
        """Setzt einen Override; gibt die neue Revision zurück."""
        if not validate_parameter_path(path):
            raise ValueError(f"invalid parameter path: {path!r}")
        value = float(value)
        if value != value or value in (float("inf"), float("-inf")):
            raise ValueError(f"value must be finite, got {value}")
        if expected_revision is not None and expected_revision != self.revision:
            raise RevisionConflict(self.revision, expected_revision)

        per_source = self._bindings.setdefault(path, {})
        # Priorität: eine niedrigere Quelle kann eine höhere Quelle nicht
        # verdrängen, aber ihre eigene Bindung jederzeit aktualisieren.
        existing = per_source.get(source)
        now = time.monotonic_ns()
        if existing is None:
            per_source[source] = _Binding(value, now)
            self.revision += 1
        elif self.merge_mode is MergeMode.LTP:
            # LTP: jede Übernahme aktualisiert Bindung und Revision (§11.3)
            per_source[source] = _Binding(value, now)
            self.revision += 1
        elif value > existing.value:
            # HTP: nur ein höherer Wert übernimmt; Maximum bleibt (§11.3)
            per_source[source] = _Binding(value, now)
            self.revision += 1
        return self.revision

    def effective_value(self, path: str) -> float:
        """Wirksamer Wert: höchste Priorität gewinnt; sonst Default (§11.1)."""
        per_source = self._bindings.get(path)
        if not per_source:
            return self._defaults.get(path, self.default)
        source = min(per_source)  # kleinster IntEnum-Wert = höchste Priorität
        return per_source[source].value

    def current_source(self, path: str) -> ControlSource | None:
        per_source = self._bindings.get(path)
        if not per_source:
            return None
        return min(per_source)

    def release(self, path: str, source: ControlSource) -> int:
        """Gibt den Override zurück; nächstniedrigere Quelle übernimmt (§11.3)."""
        per_source = self._bindings.get(path)
        if per_source and source in per_source:
            del per_source[source]
            if not per_source:
                self._bindings.pop(path, None)
            self.revision += 1
        return self.revision

    def snapshot(self) -> ParameterFrame:
        """Atomarer Snapshot aller wirksamen Werte für einen Frame (§11.4)."""
        values = {p: self._defaults[p] for p in self._defaults}
        for path, per_source in self._bindings.items():
            if per_source:
                values[path] = per_source[min(per_source)].value
        return ParameterFrame(values, self.revision)

    def set_default(self, path: str, value: float) -> None:
        if not validate_parameter_path(path):
            raise ValueError(f"invalid parameter path: {path!r}")
        self._defaults[path] = float(value)