Trixy7/trixy_core/audio_morpher/dynamics.py

# -*- coding: utf-8 -*-
"""
Dynamik-Effekte fuer Voice-Morphing.

Lautstaerke-abhaengige Pitch-/Tempo-Aenderungen und Pausen-Skalierung.
"""

from __future__ import annotations

import numpy as np


def loud_pitch_boost(
    audio: np.ndarray,
    sample_rate: int,
    boost_semitones: float,
    threshold_db: float = -20.0,
) -> np.ndarray:
    """
    Hebt den Pitch bei lauten Stellen an.

    Berechnet die Lautstaerke in Fenstern und shiftet laute Abschnitte
    um bis zu boost_semitones nach oben (proportional zur Lautstaerke).
    Laengen-neutral: arbeitet frameweise mit gleicher Ausgabelaenge.
    """
    if abs(boost_semitones) < 0.05:
        return audio.copy()

    audio = audio.astype(np.float64)
    frame_len = int(sample_rate * 0.03)  # 30ms Fenster
    if frame_len < 4 or len(audio) < frame_len:
        return audio.copy()

    threshold_linear = 10.0 ** (threshold_db / 20.0)
    output = audio.copy()

    n_frames = len(audio) // frame_len
    for i in range(n_frames):
        start = i * frame_len
        end = start + frame_len
        frame = audio[start:end]

        rms = np.sqrt(np.mean(frame ** 2))
        if rms < threshold_linear:
            continue

        # Proportionaler Boost basierend auf RMS ueber Threshold
        loudness_ratio = min(1.0, (rms - threshold_linear) / (1.0 - threshold_linear + 1e-10))
        semitones = boost_semitones * loudness_ratio

        # Pitch-Shift im Frame (Resampling + Laenge beibehalten)
        factor = 2.0 ** (semitones / 12.0)
        new_len = int(frame_len / factor)
        if new_len < 4:
            continue

        from scipy.signal import resample
        shifted = resample(frame, new_len)

        # Auf Originallaenge bringen (Zuschneiden oder Auffuellen)
        if len(shifted) >= frame_len:
            output[start:end] = shifted[:frame_len]
        else:
            output[start:start + len(shifted)] = shifted
            output[start + len(shifted):end] = 0.0

    return output


def scale_pauses(
    audio: np.ndarray,
    sample_rate: int,
    pause_scale: float,
    silence_threshold_db: float = -40.0,
    min_pause_ms: float = 50.0,
) -> np.ndarray:
    """
    Skaliert die Laenge von Pausen (Stille-Abschnitten).

    pause_scale > 1.0 = laengere Pausen, < 1.0 = kuerzere Pausen.
    Aendert die Gesamtlaenge des Signals.
    """
    if abs(pause_scale - 1.0) < 0.005:
        return audio.copy()

    audio = audio.astype(np.float64)
    threshold_linear = 10.0 ** (silence_threshold_db / 20.0)
    min_pause_samples = int(sample_rate * min_pause_ms / 1000)

    # Stille-Erkennung frameweise (10ms)
    frame_len = int(sample_rate * 0.01)
    if frame_len < 1:
        return audio.copy()

    is_silent = np.zeros(len(audio), dtype=bool)
    n_frames = len(audio) // frame_len
    for i in range(n_frames):
        start = i * frame_len
        end = start + frame_len
        rms = np.sqrt(np.mean(audio[start:end] ** 2))
        if rms < threshold_linear:
            is_silent[start:end] = True

    # Rest am Ende
    if n_frames * frame_len < len(audio):
        tail = audio[n_frames * frame_len:]
        if np.sqrt(np.mean(tail ** 2)) < threshold_linear:
            is_silent[n_frames * frame_len:] = True

    # Segmente sammeln (Sprache / Stille)
    segments: list[tuple[bool, np.ndarray]] = []
    seg_start = 0
    current_silent = is_silent[0]

    for i in range(1, len(audio)):
        if is_silent[i] != current_silent:
            segments.append((current_silent, audio[seg_start:i]))
            seg_start = i
            current_silent = is_silent[i]
    segments.append((current_silent, audio[seg_start:]))

    # Pausen skalieren
    from scipy.signal import resample

    output_parts: list[np.ndarray] = []
    for silent, segment in segments:
        if silent and len(segment) >= min_pause_samples:
            new_len = max(1, int(len(segment) * pause_scale))
            scaled = resample(segment, new_len)
            output_parts.append(scaled)
        else:
            output_parts.append(segment)

    return np.concatenate(output_parts) if output_parts else audio.copy()


def loud_tempo_scale(
    audio: np.ndarray,
    sample_rate: int,
    scale: float,
    threshold_db: float = -20.0,
) -> np.ndarray:
    """
    Skaliert das Tempo bei lauten Stellen.

    scale < 1.0 = laute Stellen schneller, > 1.0 = langsamer.
    Aendert die Gesamtlaenge des Signals.
    """
    if abs(scale - 1.0) < 0.005:
        return audio.copy()

    audio = audio.astype(np.float64)
    threshold_linear = 10.0 ** (threshold_db / 20.0)
    frame_len = int(sample_rate * 0.05)  # 50ms Fenster
    if frame_len < 4 or len(audio) < frame_len:
        return audio.copy()

    from scipy.signal import resample

    output_parts: list[np.ndarray] = []
    n_frames = len(audio) // frame_len

    for i in range(n_frames):
        start = i * frame_len
        end = start + frame_len
        frame = audio[start:end]

        rms = np.sqrt(np.mean(frame ** 2))
        if rms >= threshold_linear:
            # Laut — Tempo skalieren
            loudness_ratio = min(1.0, (rms - threshold_linear) / (1.0 - threshold_linear + 1e-10))
            frame_scale = 1.0 + (scale - 1.0) * loudness_ratio
            new_len = max(4, int(frame_len * frame_scale))
            output_parts.append(resample(frame, new_len))
        else:
            output_parts.append(frame)

    # Rest am Ende
    if n_frames * frame_len < len(audio):
        output_parts.append(audio[n_frames * frame_len:])

    return np.concatenate(output_parts) if output_parts else audio.copy()