"""Adaptive Quality Controller (PLAN.md §5.2). Verbindliche Regeln: - Hysterese: höchstens eine Stufenänderung je Regelintervall - Abwertung schnell auf anhaltende Last, Aufwertung deutlich langsamer - Mindesthaltezeit je Stufe gegen Oszillation (kein Pumpen) - geschützte Größen bleiben unverändert: physische Auflösung, Refresh, Layer-Reihenfolge, aktive Layer, DMX-Zuordnung, Parametersemantik - Wechsel nur an Framegrenze atomar anwenden; alle Varianten vorab kompiliert """ from __future__ import annotations import enum import time from dataclasses import dataclass, field class QualityLevel(enum.IntEnum): LOW = 0 MEDIUM = 1 HIGH = 2 # Abwertungsschwellen p99 (ms) je aktueller Stufe _DOWNGRADE_MS = {QualityLevel.HIGH: 14.0, QualityLevel.MEDIUM: 15.0} # Aufwertungsschwellen p99 (ms): deutliche Reserve nötig _UPGRADE_MS = {QualityLevel.MEDIUM: 10.0, QualityLevel.LOW: 8.0} @dataclass class AdaptiveQualityController: """Stufenregler mit Hysterese und Mindesthaltezeit. step(p99_frame_ms) führt höchstens eine Stufenänderung je Intervall aus und gibt die aktuelle Stufe zurück; der Aufrufer wendet sie atomar an der Framegrenze an. Upgrade braucht deutlich mehr gute Intervalle als Downgrade schlechte, damit kein sichtbares Pumpen entsteht. """ interval_ms: int = 500 min_hold_ms: int = 2000 downgrade_intervals: int = 2 upgrade_intervals: int = 6 level: QualityLevel = QualityLevel.HIGH _last_change_ns: int = field(default_factory=time.monotonic_ns, repr=False) _bad_intervals: int = field(default=0, repr=False) _good_intervals: int = field(default=0, repr=False) _reason: str = "" def step(self, p99_frame_ms: float) -> QualityLevel: """Ein Regelschritt; gibt die (ggf. geänderte) Stufe zurück.""" now = time.monotonic_ns() held_ms = (now - self._last_change_ns) / 1_000_000 budget = _DOWNGRADE_MS.get(self.level) if budget is not None and p99_frame_ms > budget: self._bad_intervals += 1 self._good_intervals = 0 if ( self._bad_intervals >= self.downgrade_intervals and held_ms >= self.min_hold_ms and self.level is not QualityLevel.LOW ): self.level = QualityLevel(self.level - 1) self._last_change_ns = now self._bad_intervals = 0 self._reason = f"p99 {p99_frame_ms:.2f}ms > budget {budget}ms" else: self._bad_intervals = 0 target = _UPGRADE_MS.get(self.level) if target is not None and p99_frame_ms < target: self._good_intervals += 1 if ( self._good_intervals >= self.upgrade_intervals and held_ms >= self.min_hold_ms and self.level is not QualityLevel.HIGH ): self.level = QualityLevel(self.level + 1) self._last_change_ns = now self._good_intervals = 0 self._reason = f"p99 {p99_frame_ms:.2f}ms < reserve {target}ms" else: self._good_intervals = 0 return self.level @property def last_reason(self) -> str: return self._reason