hms-mediaengine/packages/audio_analysis/hms_audio/mapping.py

"""Audio-Mapping-Engine (PLAN.md §20.4).

Jedes Audiofeature kann über ein Binding auf einen Parameter wirken:

Audiofeature → Gate/Threshold → Normalisierung → Gain → Kurve →
Attack/Release → Min/Max → optional Quantisierung → Zielparameter

Bindings sind speicherbar, aktivierbar und priorisierbar (§20.4).
Ohne-Audio-Modulatoren: LFO, Random, Envelope, Step Sequencer (§20.5).
"""

from __future__ import annotations

import math
import random
from dataclasses import dataclass, field
from enum import StrEnum

from hms_audio import AudioFeatures


class CurveType(StrEnum):
    """Anwendungskurven (§20.4)."""

    LINEAR = "linear"
    QUADRATIC = "quadratic"
    CUBIC = "cubic"
    EXPONENTIAL = "exponential"


def apply_curve(value: float, curve: CurveType) -> float:
    """Wendet eine Kurve auf einen 0..1-Wert an."""
    value = max(0.0, min(1.0, value))
    if curve is CurveType.LINEAR:
        return value
    if curve is CurveType.QUADRATIC:
        return value * value
    if curve is CurveType.CUBIC:
        return value * value * value
    if curve is CurveType.EXPONENTIAL:
        return math.pow(value, 4.0) if value > 0 else 0.0
    return value


@dataclass
class AudioBinding:
    """Ein Audio→Parameter-Binding (§20.4).

    Pipeline: Gate → Normalize → Gain → Curve → Attack/Release → Clamp.
    """

    id: str
    feature: str  # rms, peak, bass, mid, treble, beat, beat_phase
    parameter_path: str
    threshold: float = 0.05  # Gate: Feature muss darüber liegen
    gain: float = 1.0
    curve: CurveType = CurveType.LINEAR
    attack_s: float = 0.01  # Anstiegszeit
    release_s: float = 0.1  # Abfallzeit
    min_value: float = 0.0
    max_value: float = 1.0
    enabled: bool = True
    # Interner Zustand
    _current: float = field(default=0.0, repr=False)
    _last_update_ns: int = field(default=0, repr=False)

    def process(self, features: AudioFeatures, now_ns: int) -> float:
        """Verarbeitet ein Feature-Snapshot; gibt den Parameterwert zurück.

        Attack/Release: exponentielle Glättung mit Zeitschritten.
        """
        if not self.enabled:
            return self._current

        raw = getattr(features, self.feature, 0.0)
        if isinstance(raw, bool):
            raw = 1.0 if raw else 0.0

        # Gate: unter Schwelle → 0
        if raw < self.threshold:
            raw = 0.0
        else:
            raw = (raw - self.threshold) / (1.0 - self.threshold)

        # Gain + Kurve
        shaped = apply_curve(min(raw * self.gain, 1.0), self.curve)

        # Attack/Release mit dt
        if self._last_update_ns > 0:
            dt_s = (now_ns - self._last_update_ns) / 1e9
            if dt_s > 0:
                if shaped > self._current:
                    rate = dt_s / max(self.attack_s, 0.001)
                else:
                    rate = dt_s / max(self.release_s, 0.001)
                self._current += (shaped - self._current) * min(rate, 1.0)
        else:
            self._current = shaped

        self._last_update_ns = now_ns
        # Clamp auf Min/Max
        return self.min_value + self._current * (self.max_value - self.min_value)


@dataclass
class LFO:
    """LFO-Modulator ohne Audio (§20.5): Sine/Triangle/Saw/Square."""

    id: str
    waveform: str = "sine"  # sine | triangle | saw | square
    rate_hz: float = 1.0
    min_value: float = 0.0
    max_value: float = 1.0
    phase: float = 0.0

    def process(self, now_ns: int) -> float:
        t = now_ns / 1e9
        phase = (self.phase + t * self.rate_hz) % 1.0
        if self.waveform == "sine":
            raw = 0.5 + 0.5 * math.sin(2.0 * math.pi * phase)
        elif self.waveform == "triangle":
            raw = abs(2.0 * phase - 1.0)
        elif self.waveform == "saw":
            raw = phase
        else:  # square
            raw = 1.0 if phase < 0.5 else 0.0
        return self.min_value + raw * (self.max_value - self.min_value)


@dataclass
class RandomModulator:
    """Random-Modulator mit Seed (§20.5)."""

    id: str
    rate_hz: float = 2.0
    min_value: float = 0.0
    max_value: float = 1.0
    seed: int = 0
    _rng: random.Random = field(default_factory=lambda: random.Random(), repr=False)
    _last_step: int = 0
    _current: float = 0.0

    def __post_init__(self) -> None:
        self._rng = random.Random(self.seed)

    def process(self, now_ns: int) -> float:
        step = int((now_ns / 1e9) * self.rate_hz)
        if step != self._last_step:
            self._last_step = step
            self._current = self._rng.random()
        return self.min_value + self._current * (self.max_value - self.min_value)


@dataclass
class StepSequencer:
    """Step-Sequencer (§20.5): BPM-synchron, 8-16 Steps."""

    id: str
    steps: list[float] = field(default_factory=lambda: [0.0] * 16)
    bpm: float = 120.0
    min_value: float = 0.0
    max_value: float = 1.0

    def process(self, now_ns: int) -> float:
        if not self.steps:
            return self.min_value
        period_s = 60.0 / max(self.bpm, 1.0)
        t = now_ns / 1e9
        step_index = int(t / period_s) % len(self.steps)
        raw = self.steps[step_index]
        return self.min_value + raw * (self.max_value - self.min_value)


class ModulatorEngine:
    """Verwaltet alle Modulatoren und Audio-Bindings (§20.4, §20.5).

    - process_audio(features, now): verarbeitet alle aktiven Audio-Bindings
    - process_modulators(now): verarbeitet LFO/Random/Sequencer
    - Ergebnisse werden über die Parameter-Engine angewendet (§11:
      AUDIO-Priorität 6)
    """

    def __init__(self) -> None:
        self.audio_bindings: dict[str, AudioBinding] = {}
        self.lfos: dict[str, LFO] = {}
        self.randoms: dict[str, RandomModulator] = {}
        self.sequencers: dict[str, StepSequencer] = {}

    def add_audio_binding(self, binding: AudioBinding) -> None:
        self.audio_bindings[binding.id] = binding

    def add_lfo(self, lfo: LFO) -> None:
        self.lfos[lfo.id] = lfo

    def add_random(self, mod: RandomModulator) -> None:
        self.randoms[mod.id] = mod

    def add_sequencer(self, seq: StepSequencer) -> None:
        self.sequencers[seq.id] = seq

    def process_audio(
        self, features: AudioFeatures, now_ns: int
    ) -> dict[str, float]:
        """Verarbeitet alle aktiven Audio-Bindings; Pfad→Wert."""
        results: dict[str, float] = {}
        for binding in self.audio_bindings.values():
            if binding.enabled:
                results[binding.parameter_path] = binding.process(features, now_ns)
        return results

    def process_modulators(self, now_ns: int) -> dict[str, dict[str, float]]:
        """Verarbeitet alle Nicht-Audio-Modulatoren; Typ→(id→Wert)."""
        results: dict[str, dict[str, float]] = {
            "lfo": {},
            "random": {},
            "sequencer": {},
        }
        for lfo_id, lfo in self.lfos.items():
            results["lfo"][lfo_id] = lfo.process(now_ns)
        for mod_id, mod in self.randoms.items():
            results["random"][mod_id] = mod.process(now_ns)
        for seq_id, seq in self.sequencers.items():
            results["sequencer"][seq_id] = seq.process(now_ns)
        return results