HMS MediaEngine Agent
2b1e6af6c0
- AudioFeatures: Peak, RMS, FFT-Spektrum, Bass/Low-Mid/Mid/High-Mid/ Treble, Spectral Flux, Beat, BPM, Beat-Phase, Confidence, Stilleerkennung; timestamped mit monotoner Zeitbasis (§20.2, §20.3) - BeatDetector: adaptiver Schwellwert ueber Flux-Fenster, Mindestabstand, BPM-Median ueber Inter-Beat-Intervalle, Sentinel-Fix fuer ersten Beat - RingBuffer: begrenzter Kreisring statt unkontrollierter Queues (§20.3, §33) - AudioBinding (§20.4): Gate/Threshold, Normalisierung, Gain, Kurve (linear/quadratic/cubic/exponential), Attack/Release, Min/Max - Modulatoren ohne Audio (§20.5): LFO Sine/Triangle/Saw/Square, Random mit Seed (deterministisch), Step Sequencer BPM-synchron - ModulatorEngine: verwaltet Audio-Bindings und Modulatoren; Ergebnisse ueber Parameter-Engine mit AUDIO-Prioritaet 6 (§11.2) - 29 Unit-Tests: RMS/Peak, RingBuffer-Kapazitaet, FFT-Peak-Frequenz, Band-Energie, Flux, BPM-Recovery (120 BPM), Min-Interval, Silent-Tone-Analyse, Kurven, Binding-Pipeline, LFO-Periodizitaet, Random-Seed-Determinismus, Sequencer-Cycling - Gesamtsuite 586 gruen, Ruff gruen
318 lines
10 KiB
Python
318 lines
10 KiB
Python
"""Audio-Analyse-Engine (PLAN.md §20).
|
||
|
||
- Peak und RMS
|
||
- FFT-Spektrum mit konfigurierbaren Frequenzbändern
|
||
- Bass, Low-Mid, Mid, High-Mid, Treble
|
||
- Spectral Flux / Onset
|
||
- Beat-Trigger und BPM-Schätzung
|
||
- Beat-Phase und Confidence
|
||
|
||
Kein LLM, keine Cloudanfrage im Audiothread (§20.3). Ringbuffer statt
|
||
unkontrollierter Queues. Feature-Snapshots timestamped mit der gemeinsamen
|
||
monotonen Zeitbasis (§12.2).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class AudioFeatures:
|
||
"""Feature-Snapshot einer Analyse-Periode (§20.2, timestamped §20.3)."""
|
||
|
||
rms: float = 0.0
|
||
peak: float = 0.0
|
||
bass: float = 0.0
|
||
low_mid: float = 0.0
|
||
mid: float = 0.0
|
||
high_mid: float = 0.0
|
||
treble: float = 0.0
|
||
spectral_flux: float = 0.0
|
||
beat: bool = False
|
||
beat_confidence: float = 0.0
|
||
bpm: float = 0.0
|
||
beat_phase: float = 0.0
|
||
silence: bool = True
|
||
monotonic_ns: int = 0
|
||
|
||
|
||
@dataclass
|
||
class BandConfig:
|
||
"""Frequenzband-Konfiguration in Hz (§20.2: konfigurierbare Bänder)."""
|
||
|
||
bass_max: float = 250.0
|
||
low_mid_max: float = 800.0
|
||
mid_max: float = 2500.0
|
||
high_mid_max: float = 8000.0
|
||
treble_max: float = 20000.0
|
||
|
||
|
||
class RingBuffer:
|
||
"""Kreisring für Audio-Samples (§20.3: Ringbuffer statt Queues)."""
|
||
|
||
def __init__(self, capacity: int) -> None:
|
||
if capacity <= 0:
|
||
raise ValueError("capacity must be positive")
|
||
self._data = [0.0] * capacity
|
||
self._size = 0
|
||
self._head = 0
|
||
self._capacity = capacity
|
||
|
||
def push(self, value: float) -> None:
|
||
self._data[self._head] = value
|
||
self._head = (self._head + 1) % self._capacity
|
||
self._size = min(self._size + 1, self._capacity)
|
||
|
||
def extend(self, values: list[float]) -> None:
|
||
for v in values:
|
||
self.push(v)
|
||
|
||
def latest(self, count: int) -> list[float]:
|
||
"""Die letzten `count` Werte in chronologischer Reihenfolge."""
|
||
count = min(count, self._size)
|
||
result = []
|
||
start = (self._head - count) % self._capacity
|
||
for i in range(count):
|
||
result.append(self._data[(start + i) % self._capacity])
|
||
return result
|
||
|
||
def __len__(self) -> int:
|
||
return self._size
|
||
|
||
@property
|
||
def capacity(self) -> int:
|
||
return self._capacity
|
||
|
||
|
||
def compute_rms(samples: list[float]) -> float:
|
||
"""Root Mean Square (§20.2)."""
|
||
if not samples:
|
||
return 0.0
|
||
return math.sqrt(sum(s * s for s in samples) / len(samples))
|
||
|
||
|
||
def compute_peak(samples: list[float]) -> float:
|
||
"""Absoluter Maximalwert (§20.2)."""
|
||
return max((abs(s) for s in samples), default=0.0)
|
||
|
||
|
||
def compute_fft_magnitude(samples: list[float], sample_rate: float) -> list[float]:
|
||
"""Vereinfachte FFT über DFT (ohne NumPy im Livepfad; für kleine Fenster).
|
||
|
||
Nutzt das Discrete Fourier Transform O(n²). Für Produktionsbetrieb wird
|
||
diese durch GStreamer-FFT oder rustfft ersetzt – hier als plattformneutrale
|
||
Referenzimplementierung mit deterministischen Ergebnissen.
|
||
"""
|
||
n = len(samples)
|
||
if n == 0 or sample_rate <= 0:
|
||
return []
|
||
result: list[float] = []
|
||
for k in range(n // 2):
|
||
real = 0.0
|
||
imag = 0.0
|
||
for t, sample in enumerate(samples):
|
||
angle = 2.0 * math.pi * k * t / n
|
||
real += sample * math.cos(angle)
|
||
imag -= sample * math.sin(angle)
|
||
result.append(math.sqrt(real * real + imag * imag) / n)
|
||
return result
|
||
|
||
|
||
def frequency_of_bin(bin_index: int, fft_size: int, sample_rate: float) -> float:
|
||
"""Frequenz eines FFT-Bins in Hz."""
|
||
if fft_size == 0:
|
||
return 0.0
|
||
return bin_index * sample_rate / fft_size
|
||
|
||
|
||
def compute_band_energy(
|
||
magnitudes: list[float],
|
||
sample_rate: float,
|
||
low_hz: float,
|
||
high_hz: float,
|
||
) -> float:
|
||
"""Energie in einem Frequenzband (normalisiert auf 0..1)."""
|
||
if not magnitudes:
|
||
return 0.0
|
||
fft_size = len(magnitudes) * 2
|
||
total = 0.0
|
||
count = 0
|
||
for i, mag in enumerate(magnitudes):
|
||
freq = frequency_of_bin(i, fft_size, sample_rate)
|
||
if low_hz <= freq < high_hz:
|
||
total += mag
|
||
count += 1
|
||
if count == 0:
|
||
return 0.0
|
||
return min(total / count, 1.0)
|
||
|
||
|
||
def compute_spectral_flux(
|
||
current: list[float],
|
||
previous: list[float],
|
||
) -> float:
|
||
"""Spectral Flux: Summe der positiven Änderungen (§20.2 Onset)."""
|
||
if len(current) != len(previous) or not current:
|
||
return 0.0
|
||
flux = 0.0
|
||
for cur, prev in zip(current, previous, strict=False):
|
||
diff = cur - prev
|
||
if diff > 0:
|
||
flux += diff
|
||
return flux
|
||
|
||
|
||
@dataclass
|
||
class BeatDetector:
|
||
"""Beat-Erkennung über Spectral Flux mit adaptivem Schwellwert (§20.2).
|
||
|
||
- feed(flux): neuer Flux-Wert je Analyse-Periode
|
||
- beat: True bei erkanntem Beat (Schwellwert + Mindestabstand)
|
||
- bpm: Schätzung über Inter-Beat-Intervalle
|
||
- confidence: Verhältnis erkannter Beats zu erwarteten
|
||
"""
|
||
|
||
threshold_factor: float = 1.5 # über Mittelwert des Flux-Fensters
|
||
min_interval_s: float = 0.25 # 240 BPM Maximum
|
||
window_size: int = 43 # ~0.5 s bei 86 Hz Analyse-Rate
|
||
_flux_history: list[float] = field(default_factory=list)
|
||
_last_beat_ns: int = -1 # -1 = noch kein Beat (Sentinel)
|
||
_beat_intervals: list[float] = field(default_factory=list)
|
||
bpm: float = 0.0
|
||
beat_phase: float = 0.0
|
||
confidence: float = 0.0
|
||
beat_active: bool = False
|
||
|
||
def feed(self, flux: float, now_ns: int) -> bool:
|
||
"""Verarbeitet einen Flux-Wert; True bei erkanntem Beat."""
|
||
self._flux_history.append(flux)
|
||
if len(self._flux_history) > self.window_size:
|
||
self._flux_history.pop(0)
|
||
|
||
self.beat_active = False
|
||
if len(self._flux_history) < 4:
|
||
return False
|
||
|
||
mean_flux = sum(self._flux_history) / len(self._flux_history)
|
||
threshold = mean_flux * self.threshold_factor
|
||
|
||
# Mindestabstand prüfen (nicht mehr als 240 BPM)
|
||
# _last_beat_ns == -1 bedeutet: noch kein Beat erkannt → immer zulassen
|
||
if self._last_beat_ns >= 0:
|
||
since_last = (now_ns - self._last_beat_ns) / 1e9
|
||
else:
|
||
since_last = float("inf") # erster Beat ist immer erlaubt
|
||
if flux > threshold and since_last >= self.min_interval_s:
|
||
self.beat_active = True
|
||
interval = since_last if self._last_beat_ns >= 0 else 0.0
|
||
if 0.0 < interval < 3.0: # max 3 s zwischen Beats
|
||
self._beat_intervals.append(interval)
|
||
if len(self._beat_intervals) > 12:
|
||
self._beat_intervals.pop(0)
|
||
# BPM als Median der letzten Intervalle
|
||
sorted_intervals = sorted(self._beat_intervals)
|
||
median = sorted_intervals[len(sorted_intervals) // 2]
|
||
if median > 0:
|
||
self.bpm = 60.0 / median
|
||
self.beat_phase = (now_ns % int(median * 1e9)) / (median * 1e9)
|
||
self._last_beat_ns = now_ns
|
||
self.confidence = min(
|
||
len(self._beat_intervals) / 8.0, 1.0
|
||
)
|
||
return self.beat_active
|
||
|
||
def update_phase(self, now_ns: int) -> None:
|
||
"""Aktualisiert die Beat-Phase kontinuierlich zwischen Beats."""
|
||
if self.bpm > 0:
|
||
period_ns = int((60.0 / self.bpm) * 1e9)
|
||
if period_ns > 0:
|
||
self.beat_phase = (now_ns % period_ns) / period_ns
|
||
|
||
|
||
class AudioAnalyzer:
|
||
"""Vollständige Audio-Analyse pro Periode (§20.2, §20.3).
|
||
|
||
- feed(samples): neue Audiosamples (mono, -1..1)
|
||
- analyze(): berechnet Features und gibt einen AudioFeatures-Snapshot
|
||
- Ringbuffer begrenzt Speicher (§33: kein unbeschränkter Zustand)
|
||
"""
|
||
|
||
SAMPLE_RATE = 44100.0
|
||
WINDOW_SIZE = 512 # FFT-Fenster
|
||
SILENCE_THRESHOLD = 0.001
|
||
|
||
def __init__(self, bands: BandConfig | None = None) -> None:
|
||
self._bands = bands or BandConfig()
|
||
self._samples = RingBuffer(self.WINDOW_SIZE * 2)
|
||
self._prev_magnitudes: list[float] = []
|
||
self._beat_detector = BeatDetector()
|
||
self._last_features = AudioFeatures()
|
||
|
||
@property
|
||
def features(self) -> AudioFeatures:
|
||
return self._last_features
|
||
|
||
def feed(self, samples: list[float]) -> None:
|
||
"""Fügt neue Samples in den Ringbuffer ein."""
|
||
self._samples.extend(samples)
|
||
|
||
def analyze(self, now_ns: int | None = None) -> AudioFeatures:
|
||
"""Berechnet den nächsten Feature-Snapshot.
|
||
|
||
Läuft typischerweise 50-100 mal pro Sekunde (§20.3).
|
||
"""
|
||
now = now_ns if now_ns is not None else time.monotonic_ns()
|
||
window = self._samples.latest(self.WINDOW_SIZE)
|
||
|
||
if len(window) < self.WINDOW_SIZE // 2:
|
||
return self._last_features # nicht genug Daten
|
||
|
||
rms = compute_rms(window)
|
||
peak = compute_peak(window)
|
||
silence = rms < self.SILENCE_THRESHOLD
|
||
|
||
magnitudes = compute_fft_magnitude(window, self.SAMPLE_RATE)
|
||
bass = compute_band_energy(
|
||
magnitudes, self.SAMPLE_RATE, 0, self._bands.bass_max
|
||
)
|
||
low_mid = compute_band_energy(
|
||
magnitudes, self.SAMPLE_RATE, self._bands.bass_max, self._bands.low_mid_max
|
||
)
|
||
mid = compute_band_energy(
|
||
magnitudes, self.SAMPLE_RATE, self._bands.low_mid_max, self._bands.mid_max
|
||
)
|
||
high_mid = compute_band_energy(
|
||
magnitudes, self.SAMPLE_RATE, self._bands.mid_max, self._bands.high_mid_max
|
||
)
|
||
treble = compute_band_energy(
|
||
magnitudes, self.SAMPLE_RATE, self._bands.high_mid_max, self._bands.treble_max
|
||
)
|
||
|
||
flux = compute_spectral_flux(magnitudes, self._prev_magnitudes)
|
||
self._prev_magnitudes = magnitudes
|
||
|
||
beat = self._beat_detector.feed(flux, now)
|
||
self._beat_detector.update_phase(now)
|
||
|
||
features = AudioFeatures(
|
||
rms=rms,
|
||
peak=peak,
|
||
bass=bass,
|
||
low_mid=low_mid,
|
||
mid=mid,
|
||
high_mid=high_mid,
|
||
treble=treble,
|
||
spectral_flux=flux,
|
||
beat=beat,
|
||
beat_confidence=self._beat_detector.confidence,
|
||
bpm=self._beat_detector.bpm,
|
||
beat_phase=self._beat_detector.beat_phase,
|
||
silence=silence,
|
||
monotonic_ns=now,
|
||
)
|
||
self._last_features = features
|
||
return features
|