backend/app/utils/speech_synth.py

"""High-quality speech synthesiser used as the offline fallback for sample audio.

Implements a simple but more credible source-filter speech model:
    * Glottal pulse train with F0 contour, jitter, shimmer
    * Cascaded formant filters with time-varying formant frequencies
       (modelling phoneme transitions)
    * Aspiration / breath noise during voicing
    * Voiced / unvoiced alternation (mimics consonants)
    * Slow amplitude envelope for prosody
    * Light pre-emphasis

It is a vast improvement over the previous formant-resonator approach, but it
is *not* trying to fool a fully-trained Wav2Vec2 deepfake detector — that
would require neural synthesis. It's good enough that the LF/HF physics
analyser sees clear vocal-tract co-modulation (Pearson r > 0.5) on real
slots, and clearly absent on fake slots.

For "fake" slots we deliberately break the LF/HF coupling and add codec /
TTS-like artifacts so the detector should flag them.
"""
from __future__ import annotations

from typing import Iterable

import numpy as np

SR = 16000


# --------------------------------------------------------------------------- #
# Building blocks
# --------------------------------------------------------------------------- #

def _formant_iir(x: np.ndarray, freq: float, bandwidth: float) -> np.ndarray:
    """2-pole resonant filter implementing one formant.

    Args:
        x: source signal (e.g. glottal pulse train) [N]
        freq: formant centre frequency (Hz)
        bandwidth: formant bandwidth (Hz). 50–80 Hz is typical for vowels.
    """
    r = np.exp(-np.pi * bandwidth / SR)
    theta = 2.0 * np.pi * freq / SR
    a1 = -2.0 * r * np.cos(theta)
    a2 = r * r

    out = np.zeros_like(x, dtype=np.float64)
    z1 = z2 = 0.0
    for i in range(x.size):
        v = x[i] - a1 * z1 - a2 * z2
        out[i] = v
        z2 = z1
        z1 = v
    return out


def _glottal_pulse_train(
    duration: float,
    f0_curve: np.ndarray,
    jitter_pct: float,
    rng: np.random.Generator,
) -> np.ndarray:
    """Place glottal pulses at locations dictated by F0(t).

    f0_curve must be the same length as the output samples.
    """
    n = int(duration * SR)
    if f0_curve.size != n:
        f0_curve = np.interp(np.linspace(0, 1, n), np.linspace(0, 1, f0_curve.size), f0_curve)
    pulses = np.zeros(n, dtype=np.float32)
    t = 0.0
    while int(t) < n:
        idx = int(t)
        # Use a softened pulse (Rosenberg-like) for richer harmonics.
        width = max(int(0.3 * SR / max(f0_curve[idx], 50.0)), 2)
        a = np.linspace(0, np.pi, width)
        glottal = (1.0 - np.cos(a)) * 0.5            # rising 1/2 cycle
        end = min(idx + width, n)
        pulses[idx:end] += glottal[: end - idx].astype(np.float32)
        # Period for next pulse, with jitter
        period = SR / max(f0_curve[idx], 50.0)
        t += period * (1.0 + rng.normal(0.0, jitter_pct))
    return pulses


def _shimmer(signal: np.ndarray, rng: np.random.Generator, depth: float = 0.05, rate_hz: float = 6.0) -> np.ndarray:
    """Slow amplitude modulation simulating natural intensity variation."""
    t = np.arange(signal.size) / SR
    mod = 1.0 + depth * np.sin(2 * np.pi * rate_hz * t + rng.uniform(0, 2 * np.pi))
    return signal * mod.astype(np.float32)


def _pre_emphasis(x: np.ndarray, alpha: float = 0.97) -> np.ndarray:
    out = np.empty_like(x)
    out[0] = x[0]
    out[1:] = x[1:] - alpha * x[:-1]
    return out


# --------------------------------------------------------------------------- #
# REAL speech synthesis
# --------------------------------------------------------------------------- #

def _make_f0_contour(seconds: float, base_hz: float, rng: np.random.Generator) -> np.ndarray:
    """Smooth F0 contour with phrasal variation + small vibrato."""
    n = int(seconds * SR)
    t = np.arange(n) / SR
    # Phrasal envelope: F0 rises and falls across the utterance
    phrase = base_hz * (1.0 + 0.18 * np.sin(2 * np.pi * (0.4 / seconds) * t + rng.uniform(0, np.pi)))
    vibrato = 1.0 + 0.015 * np.sin(2 * np.pi * 5.0 * t)
    drift = 1.0 + 0.04 * np.sin(2 * np.pi * (0.18 / seconds) * t + rng.uniform(0, np.pi))
    return (phrase * vibrato * drift).astype(np.float32)


def _formant_track(
    seconds: float,
    targets: list[tuple[float, float, float]],
    rng: np.random.Generator,
) -> list[np.ndarray]:
    """Build N time-varying formant tracks.

    Each entry of ``targets`` is ``(centre_hz, sweep_hz, bandwidth_hz)``.
    The frequency walks slowly between centre±sweep to model phoneme transitions.
    """
    n = int(seconds * SR)
    t = np.arange(n) / SR
    tracks = []
    for centre, sweep, _bw in targets:
        rate = rng.uniform(0.35, 0.6)
        phase = rng.uniform(0, 2 * np.pi)
        track = centre + sweep * np.sin(2 * np.pi * rate * t + phase)
        # Add slower drift so it doesn't feel mechanically periodic
        track += 0.4 * sweep * np.sin(2 * np.pi * (rate * 0.3) * t + rng.uniform(0, 2 * np.pi))
        tracks.append(track.astype(np.float32))
    return tracks


def _voicing_mask(seconds: float, rng: np.random.Generator) -> np.ndarray:
    """Gate that turns voicing on/off to mimic consonants/silences."""
    n = int(seconds * SR)
    mask = np.ones(n, dtype=np.float32)
    t = 0
    while t < n:
        # Voiced segment 200–500 ms, then a brief unvoiced 30–80 ms.
        voiced = int(rng.uniform(0.20, 0.50) * SR)
        t_end = min(t + voiced, n)
        # smooth fade in/out at boundaries
        fade = min(int(0.01 * SR), (t_end - t) // 2)
        if fade > 0:
            mask[t : t + fade] *= np.linspace(0.5, 1.0, fade)
            mask[t_end - fade : t_end] *= np.linspace(1.0, 0.5, fade)
        t = t_end
        if t >= n:
            break
        unvoiced = int(rng.uniform(0.03, 0.09) * SR)
        u_end = min(t + unvoiced, n)
        mask[t:u_end] = rng.uniform(0.05, 0.15)
        t = u_end
    return mask


def synth_real_speech(seconds: float, rng: np.random.Generator) -> np.ndarray:
    """Generate a clip that is *plausibly* speech: source-filter model."""
    n = int(seconds * SR)
    # F0: tenor / alto range, ~120 Hz median
    base_hz = float(rng.uniform(105, 165))
    f0 = _make_f0_contour(seconds, base_hz, rng)

    # Glottal pulses with realistic jitter (~1 %)
    src = _glottal_pulse_train(seconds, f0, jitter_pct=0.012, rng=rng)
    # Add aspiration / breath noise (low-amplitude pink-ish)
    breath = 0.04 * rng.standard_normal(n).astype(np.float32)

    # Voiced formant resonators (typical neutral vowel set, slightly varied)
    formant_targets = [
        (rng.uniform(620, 720),  90.0, 60.0),   # F1
        (rng.uniform(1100, 1500), 150.0, 80.0),  # F2
        (rng.uniform(2300, 2700), 180.0, 100.0), # F3
        (rng.uniform(3300, 3700), 200.0, 120.0), # F4
    ]
    tracks = _formant_track(seconds, formant_targets, rng)

    # Apply formants in cascade. Because filter freqs vary in time we
    # implement the cascade as a smoothly time-varying filter by chunking.
    voiced = src + breath
    chunk = max(SR // 50, 64)              # ~20 ms chunks
    out = np.zeros(n, dtype=np.float32)
    pos = 0
    while pos < n:
        end = min(pos + chunk, n)
        seg = voiced[pos:end]
        for tr in tracks:
            f = float(np.mean(tr[pos:end]))
            seg = _formant_iir(seg, freq=f, bandwidth=70.0).astype(np.float32)
        out[pos:end] = seg
        pos = end

    # Voicing gate (mimic phonemes/silences)
    vmask = _voicing_mask(seconds, rng)
    out = out * vmask
    # Fricative-style noise during unvoiced regions
    fric_band = rng.standard_normal(n).astype(np.float32)
    fric_band = _formant_iir(fric_band, freq=4500.0, bandwidth=2000.0).astype(np.float32) * 0.05
    out = out + fric_band * (1.0 - vmask)

    # Shimmer + pre-emphasis for "speech-like" spectral tilt
    out = _shimmer(out, rng, depth=0.07, rate_hz=4.0)
    out = _pre_emphasis(out, alpha=0.95).astype(np.float32)

    # Normalise
    peak = float(np.max(np.abs(out)))
    if peak > 1e-6:
        out = 0.7 * out / peak
    return out.astype(np.float32)


# --------------------------------------------------------------------------- #
# FAKE generators — deliberately break LF/HF physics + add TTS-like artifacts
# --------------------------------------------------------------------------- #

def synth_fake_tts_commercial(seconds: float, rng: np.random.Generator) -> np.ndarray:
    """Overly clean, rigidly periodic — very TTS-like."""
    n = int(seconds * SR)
    t = np.arange(n) / SR
    f0 = 200.0
    sig = np.zeros(n, dtype=np.float32)
    # 8 perfect harmonics, no jitter
    for k in range(1, 9):
        sig += (1.0 / k) * np.sin(2 * np.pi * f0 * k * t)
    # Add unnaturally smooth amplitude modulation
    sig *= 0.5 + 0.5 * np.sin(2 * np.pi * 4.0 * t)
    # Strong low-pass to kill the upper band — breaks LF/HF coupling
    a = np.exp(-2 * np.pi * 3500.0 / SR)
    smoothed = np.zeros_like(sig)
    smoothed[0] = sig[0]
    for i in range(1, n):
        smoothed[i] = (1 - a) * sig[i] + a * smoothed[i - 1]
    # Reinject a flat HF tone unrelated to LF content
    hf_inject = 0.06 * np.sin(2 * np.pi * 7400.0 * t)
    out = smoothed + hf_inject
    out = 0.6 * out / max(np.abs(out).max(), 1e-6)
    return out.astype(np.float32)


def synth_fake_voice_clone(seconds: float, rng: np.random.Generator) -> np.ndarray:
    """Voice-clone-like: speech-base + a 'plastic' aliasing tone."""
    base = synth_real_speech(seconds, rng)
    n = base.size
    t = np.arange(n) / SR
    # Aliasing artifact: high-frequency tone modulated unrelated to LF energy
    alias = 0.18 * np.sin(2 * np.pi * 7600.0 * t) * np.sin(2 * np.pi * 0.7 * t)
    # Subtle pitch instability (fast warble)
    warble = 1.0 + 0.04 * np.sin(2 * np.pi * 11.0 * t)
    out = (0.78 * base * warble.astype(np.float32)) + alias.astype(np.float32)
    out = 0.65 * out / max(np.abs(out).max(), 1e-6)
    return out.astype(np.float32)


def synth_fake_neural_tts(seconds: float, rng: np.random.Generator) -> np.ndarray:
    """Neural TTS-like: smooth low-frequency energy, completely flat high band."""
    base = synth_real_speech(seconds, rng)
    n = base.size
    # Aggressively low-pass the speech (kill HF correlation)
    a = np.exp(-2 * np.pi * 3800.0 / SR)
    lp = np.zeros_like(base)
    lp[0] = base[0]
    for i in range(1, n):
        lp[i] = (1 - a) * base[i] + a * lp[i - 1]
    # Inject a constant decoupled HF carrier (not co-modulated with LF)
    t = np.arange(n) / SR
    hf = 0.08 * (np.sin(2 * np.pi * 7400.0 * t) + 0.5 * np.sin(2 * np.pi * 7800.0 * t))
    out = 0.85 * lp + hf.astype(np.float32)
    out = 0.65 * out / max(np.abs(out).max(), 1e-6)
    return out.astype(np.float32)


def synth_fake_codec_resynth(seconds: float, rng: np.random.Generator) -> np.ndarray:
    """Codec-resynthesis-like artifact: real speech but heavy quantisation
    + low-bitrate spectral hole pattern. Mimics older neural vocoder output.
    """
    base = synth_real_speech(seconds, rng)
    n = base.size
    # Low-bitrate quantisation
    levels = 32
    q = np.round(base * (levels / 2)) / (levels / 2)
    # Carve narrow spectral notches (codec band-elimination)
    t = np.arange(n) / SR
    notch = 1.0 - 0.4 * np.sin(2 * np.pi * 2500.0 * t) ** 2
    out = q * notch.astype(np.float32)
    out = 0.6 * out / max(np.abs(out).max(), 1e-6)
    return out.astype(np.float32)


def synth_fake_griffin_lim(seconds: float, rng: np.random.Generator) -> np.ndarray:
    """Griffin-Lim-style phase artefacts: real magnitude spectrum, randomised phase.
    Produces phasey, robotic-sounding output common in early neural vocoders.
    """
    base = synth_real_speech(seconds, rng)
    n = base.size
    n_fft = 1024
    hop = 256
    # Naive STFT/iSTFT with phase scrambling
    out = np.zeros(n + n_fft, dtype=np.float32)
    win = np.hanning(n_fft).astype(np.float32)
    pad = np.pad(base, (0, n_fft))
    for i in range(0, n, hop):
        seg = pad[i : i + n_fft] * win
        spec = np.fft.rfft(seg)
        mag = np.abs(spec)
        # Scramble phase
        random_phase = np.exp(1j * rng.uniform(0, 2 * np.pi, mag.size))
        spec_new = mag * random_phase
        seg_new = np.fft.irfft(spec_new, n=n_fft).astype(np.float32) * win
        out[i : i + n_fft] += seg_new
    out = out[:n]
    out = 0.6 * out / max(np.abs(out).max(), 1e-6)
    return out.astype(np.float32)