Trixy7/trixy_core/audio_morpher/dsp.py

# -*- coding: utf-8 -*-
"""
Basis-DSP-Funktionen fuer Voice-Morphing.

Pitch-Shift, VTLP, Breathiness, Shimmer, Fade, Speed — rein numpy/scipy.
"""

from __future__ import annotations

import numpy as np


def pitch_shift(audio: np.ndarray, sample_rate: int, semitones: float) -> np.ndarray:
    """
    Pitch-Shift via Sinc-Resampling.

    Positive Semitones = hoeher (Signal wird kuerzer).
    Negative Semitones = tiefer (Signal wird laenger).
    Nutzt scipy.signal.resample (bandlimitierte Sinc-Interpolation).
    """
    from scipy.signal import resample

    if abs(semitones) < 0.05:
        return audio.copy()

    audio = audio.astype(np.float64)
    factor = 2.0 ** (semitones / 12.0)

    new_len = int(len(audio) / factor)
    if new_len < 10:
        return audio.copy()

    result = resample(audio, new_len)

    # Amplitude beibehalten
    orig_rms = np.sqrt(np.mean(audio ** 2)) + 1e-10
    result_rms = np.sqrt(np.mean(result ** 2)) + 1e-10
    if result_rms > 1e-10:
        result *= orig_rms / result_rms

    return result


def speed_change(audio: np.ndarray, sample_rate: int, factor: float) -> np.ndarray:
    """
    Aendert die Geschwindigkeit ohne Pitch-Aenderung.

    factor > 1.0 = schneller (kuerzeres Signal).
    factor < 1.0 = langsamer (laengeres Signal).
    Nutzt STFT-basiertes Phase-Vocoder-Prinzip via Resampling.
    """
    from scipy.signal import resample

    if abs(factor - 1.0) < 0.005:
        return audio.copy()

    audio = audio.astype(np.float64)
    new_len = int(len(audio) / factor)
    if new_len < 10:
        return audio.copy()

    result = resample(audio, new_len)

    # Amplitude beibehalten
    orig_rms = np.sqrt(np.mean(audio ** 2)) + 1e-10
    result_rms = np.sqrt(np.mean(result ** 2)) + 1e-10
    if result_rms > 1e-10:
        result *= orig_rms / result_rms

    return result


def vtlp(audio: np.ndarray, sample_rate: int, alpha: float,
         n_fft: int = 1024, hop_length: int = 256) -> np.ndarray:
    """
    Vocal Tract Length Perturbation — Formant-Verschiebung.

    Warpt NUR die Magnitude-Spektren, behaelt originale Phasen.
    alpha > 1.0 = hoehere Formanten, alpha < 1.0 = tiefere.
    """
    if abs(alpha - 1.0) < 0.005:
        return audio.copy()

    audio = audio.astype(np.float64)
    n_fft_half = n_fft // 2 + 1
    f_nyq = sample_rate / 2.0
    f_hi = min(4800.0, f_nyq * 0.85)

    hop = hop_length
    window = np.hanning(n_fft)
    n_frames = max(1, (len(audio) - n_fft) // hop + 1)

    output = np.zeros(len(audio) + n_fft, dtype=np.float64)
    norm = np.zeros(len(output), dtype=np.float64)

    # Warp-Mapping vorberechnen
    freqs = np.fft.rfftfreq(n_fft, d=1.0 / sample_rate)
    warp_indices = np.zeros(n_fft_half, dtype=np.float64)
    for i in range(n_fft_half):
        warped_freq = _vtlp_warp_freq(freqs[i], alpha, f_hi, f_nyq)
        warp_indices[i] = warped_freq / (sample_rate / n_fft)

    for fi in range(n_frames):
        start = fi * hop
        end = start + n_fft
        if end > len(audio):
            break

        frame = audio[start:end] * window
        spectrum = np.fft.rfft(frame)
        magnitude = np.abs(spectrum)
        phase = np.angle(spectrum)

        # Nur Magnitude warpen
        warped_mag = np.zeros_like(magnitude)
        for i in range(n_fft_half):
            src = warp_indices[i]
            lo = int(np.floor(src))
            hi_idx = lo + 1
            frac = src - lo
            if 0 <= lo < n_fft_half:
                warped_mag[i] += (1.0 - frac) * magnitude[lo]
            if 0 <= hi_idx < n_fft_half:
                warped_mag[i] += frac * magnitude[hi_idx]

        warped_spectrum = warped_mag * np.exp(1j * phase)
        frame_out = np.fft.irfft(warped_spectrum, n=n_fft)

        output[start:end] += frame_out * window
        norm[start:end] += window ** 2

    mask = norm > 1e-8
    output[mask] /= norm[mask]

    return output[:len(audio)]


def _vtlp_warp_freq(freq: float, alpha: float, f_hi: float, f_nyq: float) -> float:
    """Piecewise-lineare Frequenz-Warping (Jaitly & Hinton 2013)."""
    if freq <= f_hi * min(alpha, 1.0):
        return freq * alpha
    return f_nyq - (f_nyq - f_hi * alpha) / (f_nyq - f_hi * min(alpha, 1.0)) * (f_nyq - freq)


def add_breathiness(audio: np.ndarray, sample_rate: int, amount: float) -> np.ndarray:
    """
    Fuegt sanfte Hauchigkeit (Aspiration) hinzu.

    Nutzt Lowpass-gefiltertes Rauschen das der Amplituden-Huellkurve folgt.
    """
    if amount < 0.005:
        return audio

    audio = audio.astype(np.float64)
    noise = np.random.randn(len(audio))

    from scipy.signal import butter, sosfilt
    nyq = sample_rate / 2.0
    cutoff = min(3500.0 / nyq, 0.95)
    sos = butter(2, cutoff, btype="low", output="sos")
    filtered_noise = sosfilt(sos, noise)

    # Amplituden-Huellkurve (50ms Glaettung)
    envelope = np.abs(audio)
    kernel_size = int(sample_rate * 0.05)
    if kernel_size > 1:
        kernel = np.ones(kernel_size) / kernel_size
        envelope = np.convolve(envelope, kernel, mode="same")

    filtered_noise *= envelope
    audio_rms = np.sqrt(np.mean(audio ** 2)) + 1e-10
    noise_rms = np.sqrt(np.mean(filtered_noise ** 2)) + 1e-10
    filtered_noise *= audio_rms / noise_rms

    mixed = audio + amount * filtered_noise

    max_val = np.max(np.abs(mixed))
    if max_val > 1.0:
        mixed /= max_val

    return mixed


def add_shimmer(audio: np.ndarray, sample_rate: int, amount: float) -> np.ndarray:
    """Sanfter Shimmer via niederfrequente Amplitudenmodulation."""
    if amount < 0.005:
        return audio

    audio = audio.astype(np.float64)
    mod_noise = np.random.randn(len(audio))

    from scipy.signal import butter, sosfilt
    nyq = sample_rate / 2.0
    cutoff = min(200.0 / nyq, 0.95)
    sos = butter(2, cutoff, btype="low", output="sos")
    modulation = sosfilt(sos, mod_noise)

    max_mod = np.max(np.abs(modulation))
    if max_mod > 0:
        modulation /= max_mod

    return audio * (1.0 + amount * modulation)


def apply_fade(audio: np.ndarray, sample_rate: int, fade_ms: float = 15.0) -> np.ndarray:
    """Wendet Fade-In/Out an um Klick-Artefakte zu vermeiden."""
    fade_samples = int(sample_rate * fade_ms / 1000)
    if fade_samples < 2 or len(audio) < 2 * fade_samples:
        return audio

    audio = audio.copy()
    audio[:fade_samples] *= np.linspace(0.0, 1.0, fade_samples)
    audio[-fade_samples:] *= np.linspace(1.0, 0.0, fade_samples)

    return audio