706 lines
24 KiB
Python
706 lines
24 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Post-processing audio effects keyed off speaker vocal_traits.
|
|
|
|
Applies signal-processing effects (bitcrushing, gating, filtering, noise
|
|
injection) to generated TTS WAVs so that speakers described as "broken",
|
|
"distorted", "choppy", etc. actually sound that way. TTS models cannot
|
|
produce these effects from text alone — they require DSP.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
from scipy.signal import butter, lfilter
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TraitEffectMap = dict[str, list[tuple[str, dict[str, Any]]]]
|
|
|
|
|
|
def _bitcrusher(
|
|
audio: np.ndarray, sr: int, bits: int = 6, sample_rate_reduction: int = 1
|
|
) -> np.ndarray:
|
|
"""Reduce bit depth and optionally downsample for digital distortion."""
|
|
max_val = float(2 ** (bits - 1))
|
|
crushed = np.round(audio * max_val) / max_val
|
|
if sample_rate_reduction > 1:
|
|
crushed = crushed[::sample_rate_reduction]
|
|
crushed = np.interp(
|
|
np.linspace(0, len(crushed) - 1, len(audio)),
|
|
np.arange(len(crushed)),
|
|
crushed,
|
|
)
|
|
return np.clip(crushed, -1.0, 1.0).astype(np.float32)
|
|
|
|
|
|
def _hard_clip(audio: np.ndarray, _sr: int, threshold: float = 0.3) -> np.ndarray:
|
|
"""Hard clip above threshold for aggressive distortion."""
|
|
clipped = np.clip(audio, -threshold, threshold) / threshold
|
|
return clipped.astype(np.float32)
|
|
|
|
|
|
def _tremolo(audio: np.ndarray, sr: int, rate: float = 8.0, depth: float = 0.7) -> np.ndarray:
|
|
"""Amplitude modulation for choppy/stutter effect."""
|
|
t = np.linspace(0, len(audio) / sr, len(audio), dtype=np.float32)
|
|
mod = 1.0 - depth * 0.5 * (1.0 + np.sin(2.0 * np.pi * rate * t))
|
|
return (audio * mod).astype(np.float32)
|
|
|
|
|
|
def _random_gating(
|
|
audio: np.ndarray, sr: int, gap_prob: float = 0.15,
|
|
min_gap: int = 0, max_gap: int = 0,
|
|
) -> np.ndarray:
|
|
"""Randomly zero out short segments for broken/choppy effect."""
|
|
if min_gap == 0:
|
|
min_gap = int(sr * 0.02)
|
|
if max_gap == 0:
|
|
max_gap = int(sr * 0.08)
|
|
out = audio.copy()
|
|
i = 0
|
|
rng = np.random.default_rng(42)
|
|
while i < len(out):
|
|
if rng.random() < gap_prob:
|
|
gap_len = int(rng.integers(min_gap, max(1, max_gap)))
|
|
end = min(i + gap_len, len(out))
|
|
out[i:end] = 0.0
|
|
i = end
|
|
else:
|
|
i += int(sr * 0.01)
|
|
return out
|
|
|
|
|
|
def _bandpass(
|
|
audio: np.ndarray, sr: int, low_hz: float = 300.0,
|
|
high_hz: float = 3000.0, order: int = 4,
|
|
) -> np.ndarray:
|
|
"""Bandpass filter for muffled/indistinct quality."""
|
|
nyq = sr / 2.0
|
|
low = max(low_hz / nyq, 0.001)
|
|
high = min(high_hz / nyq, 0.999)
|
|
b, a = butter(order, [low, high], btype="band")
|
|
return lfilter(b, a, audio).astype(np.float32)
|
|
|
|
|
|
def _lowpass(audio: np.ndarray, sr: int, cutoff_hz: float = 2000.0, order: int = 4) -> np.ndarray:
|
|
"""Low-pass filter for muffled quality."""
|
|
nyq = sr / 2.0
|
|
c = min(cutoff_hz / nyq, 0.999)
|
|
b, a = butter(order, c, btype="low")
|
|
return lfilter(b, a, audio).astype(np.float32)
|
|
|
|
|
|
def _highpass(audio: np.ndarray, sr: int, cutoff_hz: float = 600.0, order: int = 2) -> np.ndarray:
|
|
"""High-pass to strip human chest/body resonance (syrinx: no ~100-600Hz body)."""
|
|
nyq = sr / 2.0
|
|
c = min(cutoff_hz / nyq, 0.999)
|
|
b, a = butter(order, c, btype="high")
|
|
return lfilter(b, a, audio).astype(np.float32)
|
|
|
|
|
|
def _tube_sat(audio: np.ndarray, sr: int, drive_db: float = 8.0, mix: float = 1.0) -> np.ndarray:
|
|
"""Soft-knee (tanh) saturation for harsh squawky harmonics without digital clipping."""
|
|
drive = 10.0 ** (drive_db / 20.0)
|
|
dry = audio.astype(np.float64)
|
|
wet = np.tanh(drive * dry) / np.tanh(drive)
|
|
out = mix * wet + (1.0 - mix) * dry
|
|
return out.astype(np.float32)
|
|
|
|
|
|
def _chorus(
|
|
audio: np.ndarray, sr: int, rate_hz: float = 1.8, depth: float = 0.25,
|
|
centre_delay_ms: float = 7.0, feedback: float = 0.15, mix: float = 0.35,
|
|
) -> np.ndarray:
|
|
"""Dual-modulated-tap chorus, emulating a syrinx's two independent membranes.
|
|
|
|
Two short delays sawing against each other produce the subtle
|
|
thickening/beating of a bird's dual sound source.
|
|
"""
|
|
n = len(audio)
|
|
if n < 2:
|
|
return audio.astype(np.float32)
|
|
t = np.arange(n, dtype=np.float64) / sr
|
|
base = centre_delay_ms / 1000.0 * sr
|
|
depth_samp = depth * base
|
|
d1 = base + depth_samp * np.sin(2.0 * np.pi * rate_hz * t)
|
|
d2 = base * 0.6 + depth_samp * np.sin(2.0 * np.pi * (rate_hz * 1.31) * t + 0.7)
|
|
|
|
idx = np.arange(n, dtype=np.float64)
|
|
read1 = np.clip(idx - d1, 0, n - 1)
|
|
read2 = np.clip(idx - d2, 0, n - 1)
|
|
tap1 = np.interp(read1, idx, audio).astype(np.float32)
|
|
tap2 = np.interp(read2, idx, audio).astype(np.float32)
|
|
|
|
wet = (0.7 * tap1 + 0.3 * tap2).astype(np.float64)
|
|
out = (1.0 - mix) * audio + mix * wet
|
|
if feedback > 0.0:
|
|
take = min(int(sr * 0.05), n)
|
|
if take < n:
|
|
fb = np.concatenate([np.zeros(take), out[: n - take]])
|
|
out = out + feedback * fb
|
|
peak = float(np.max(np.abs(out)))
|
|
if peak > 1.0:
|
|
out = out / peak
|
|
return out.astype(np.float32)
|
|
|
|
|
|
def _noise_inject(audio: np.ndarray, sr: int, amount: float = 0.03) -> np.ndarray:
|
|
"""Add white noise for raspy/grainy quality."""
|
|
rng = np.random.default_rng(42)
|
|
noise = rng.normal(0, amount, len(audio)).astype(np.float32)
|
|
return np.clip(audio + noise, -1.0, 1.0).astype(np.float32)
|
|
|
|
|
|
def _pitch_jitter(
|
|
audio: np.ndarray, sr: int, jitter_hz: float = 6.0, depth: float = 0.02
|
|
) -> np.ndarray:
|
|
"""Slight periodic pitch variation for creaky/raspy quality."""
|
|
t = np.linspace(0, len(audio) / sr, len(audio), dtype=np.float32)
|
|
mod = 1.0 + depth * np.sin(2.0 * np.pi * jitter_hz * t)
|
|
indices = np.clip(
|
|
np.round(np.cumsum(mod)).astype(int), 0, len(audio) - 1
|
|
)
|
|
return audio[indices].astype(np.float32)
|
|
|
|
|
|
def _ring_modulate(audio: np.ndarray, sr: int, freq: float = 40.0) -> np.ndarray:
|
|
"""Ring modulation for robotic/synthetic quality."""
|
|
t = np.linspace(0, len(audio) / sr, len(audio), dtype=np.float32)
|
|
return (audio * np.sin(2.0 * np.pi * freq * t)).astype(np.float32)
|
|
|
|
|
|
def _pitch_shift(
|
|
audio: np.ndarray, sr: int, semitones: float = 6.0
|
|
) -> np.ndarray:
|
|
"""Shift pitch (and formants) upward by resampling.
|
|
|
|
Raising pitch via resampling also raises the spectral envelope, shrinking
|
|
the apparent vocal tract (the classic "chipmunk" effect). Positive
|
|
values make voices sound smaller, faster and higher — the core of a
|
|
cartoon parrot/squawk voice. Negative values do the reverse.
|
|
"""
|
|
factor = 2.0 ** (semitones / 12.0)
|
|
n_out = max(1, int(len(audio) / factor))
|
|
idx = np.linspace(0.0, len(audio) - 1.0, n_out)
|
|
return np.interp(idx, np.arange(len(audio), dtype=np.float64), audio).astype(
|
|
np.float32
|
|
)
|
|
|
|
|
|
def _pitch_shift_pv(
|
|
audio: np.ndarray, sr: int, semitones: float = 7.0
|
|
) -> np.ndarray:
|
|
"""Duration-preserving pitch shift via the phase vocoder.
|
|
|
|
Unlike resampling (``_pitch_shift``), the speech rate stays the same
|
|
while pitch rises — so a parrot keeps its natural hopping delivery but
|
|
sounds higher. Falls back to resampling if ``librosa`` is unavailable.
|
|
"""
|
|
try:
|
|
import librosa # noqa: PLC0415
|
|
except ImportError: # pragma: no cover
|
|
return _pitch_shift(audio, sr, semitones)
|
|
shifted = librosa.effects.pitch_shift(y=audio, sr=sr, n_steps=semitones)
|
|
return shifted.astype(np.float32)
|
|
|
|
|
|
def _vibrato(
|
|
audio: np.ndarray, sr: int, rate: float = 20.0, depth: float = 0.15
|
|
) -> np.ndarray:
|
|
"""Rapid pitch wobble (warbling squawk) via time-domain resampling."""
|
|
t = np.linspace(0, len(audio) / sr, len(audio), dtype=np.float32)
|
|
mod = 1.0 + depth * np.sin(2.0 * np.pi * rate * t)
|
|
indices = np.clip(np.round(np.cumsum(mod)).astype(int), 0, len(audio) - 1)
|
|
return audio[indices].astype(np.float32)
|
|
|
|
|
|
def _time_stretch(audio: np.ndarray, sr: int, rate: float = 1.0) -> np.ndarray:
|
|
"""Pitch-preserving tempo change for emotion delivery.
|
|
|
|
librosa semantics: ``rate > 1`` speeds the signal up (shorter), ``rate < 1``
|
|
slows it down (longer). Falls back to resampling (which also shifts pitch)
|
|
when librosa is unavailable.
|
|
"""
|
|
try:
|
|
import librosa # noqa: PLC0415
|
|
except ImportError: # pragma: no cover
|
|
n_out = max(1, int(len(audio) * rate))
|
|
idx = np.linspace(0.0, len(audio) - 1.0, n_out)
|
|
return np.interp(idx, np.arange(len(audio), dtype=np.float64), audio).astype(
|
|
np.float32
|
|
)
|
|
return librosa.effects.time_stretch(y=audio, rate=rate).astype(np.float32)
|
|
|
|
|
|
def _gain_db(audio: np.ndarray, _sr: int, db: float = 0.0) -> np.ndarray:
|
|
"""Linear gain stage in decibels (positive = louder)."""
|
|
factor = 10.0 ** (db / 20.0)
|
|
return (audio * factor).astype(np.float32)
|
|
|
|
|
|
TRAIT_EFFECTS: TraitEffectMap = {
|
|
"broken": [
|
|
("bitcrusher", {"bits": 6, "sample_rate_reduction": 2}),
|
|
("random_gating", {"gap_prob": 0.2}),
|
|
],
|
|
"distorted": [
|
|
("bitcrusher", {"bits": 5, "sample_rate_reduction": 1}),
|
|
("hard_clip", {"threshold": 0.25}),
|
|
],
|
|
"raspy": [
|
|
("noise_inject", {"amount": 0.04}),
|
|
("pitch_jitter", {"jitter_hz": 7.0, "depth": 0.025}),
|
|
],
|
|
"grainy": [
|
|
("noise_inject", {"amount": 0.05}),
|
|
],
|
|
"choppy": [
|
|
("tremolo", {"rate": 10.0, "depth": 0.8}),
|
|
],
|
|
"stutter": [
|
|
("random_gating", {"gap_prob": 0.25, "min_gap": 500, "max_gap": 2000}),
|
|
],
|
|
"staccato": [
|
|
("tremolo", {"rate": 12.0, "depth": 0.6}),
|
|
],
|
|
"hard to understand": [
|
|
("bandpass", {"low_hz": 400.0, "high_hz": 2500.0}),
|
|
("noise_inject", {"amount": 0.02}),
|
|
],
|
|
"muffled": [
|
|
("lowpass", {"cutoff_hz": 1800.0}),
|
|
],
|
|
"indistinct": [
|
|
("bandpass", {"low_hz": 500.0, "high_hz": 2200.0}),
|
|
("noise_inject", {"amount": 0.03}),
|
|
],
|
|
"robotic": [
|
|
("ring_modulate", {"freq": 35.0}),
|
|
],
|
|
"creaky": [
|
|
("pitch_jitter", {"jitter_hz": 5.0, "depth": 0.03}),
|
|
("tremolo", {"rate": 6.0, "depth": 0.4}),
|
|
],
|
|
"whisper": [
|
|
("noise_inject", {"amount": 0.02}),
|
|
("bandpass", {"low_hz": 500.0, "high_hz": 4000.0}),
|
|
],
|
|
"metallic": [
|
|
("ring_modulate", {"freq": 50.0}),
|
|
("bandpass", {"low_hz": 800.0, "high_hz": 3500.0}),
|
|
],
|
|
"parrot": [
|
|
("pitch_shift_pv", {"semitones": 7.0}),
|
|
("bandpass", {"low_hz": 700.0, "high_hz": 4800.0}),
|
|
("tremolo", {"rate": 16.0, "depth": 0.32}),
|
|
("vibrato", {"rate": 24.0, "depth": 0.15}),
|
|
("hard_clip", {"threshold": 0.4}),
|
|
],
|
|
"squawky": [
|
|
("pitch_shift_pv", {"semitones": 7.0}),
|
|
("bandpass", {"low_hz": 700.0, "high_hz": 4800.0}),
|
|
("tremolo", {"rate": 16.0, "depth": 0.32}),
|
|
("vibrato", {"rate": 24.0, "depth": 0.15}),
|
|
("hard_clip", {"threshold": 0.4}),
|
|
],
|
|
}
|
|
|
|
EFFECT_FN = {
|
|
"bitcrusher": _bitcrusher,
|
|
"hard_clip": _hard_clip,
|
|
"tremolo": _tremolo,
|
|
"random_gating": _random_gating,
|
|
"bandpass": _bandpass,
|
|
"lowpass": _lowpass,
|
|
"highpass": _highpass,
|
|
"tube_sat": _tube_sat,
|
|
"chorus": _chorus,
|
|
"noise_inject": _noise_inject,
|
|
"pitch_jitter": _pitch_jitter,
|
|
"pitch_shift": _pitch_shift,
|
|
"pitch_shift_pv": _pitch_shift_pv,
|
|
"vibrato": _vibrato,
|
|
"ring_modulate": _ring_modulate,
|
|
"time_stretch": _time_stretch,
|
|
"gain_db": _gain_db,
|
|
}
|
|
|
|
# Named chains that expand to one or more effects with tuned parameters.
|
|
# ``squawk`` / ``parrot`` is the recipe for a high, warbling cartoon-parrot
|
|
# voice: duration-preserving pitch/formant lift, mid-high bandpass, fast
|
|
# flutter, vibrato and a touch of grit. No noise injection (that just adds
|
|
# hiss on an already bright band).
|
|
PRESET_EFFECTS: dict[str, list[tuple[str, dict[str, Any]]]] = {
|
|
"squawk": [
|
|
("pitch_shift_pv", {"semitones": 7.0}),
|
|
("bandpass", {"low_hz": 700.0, "high_hz": 4800.0}),
|
|
("tremolo", {"rate": 16.0, "depth": 0.32}),
|
|
("vibrato", {"rate": 24.0, "depth": 0.15}),
|
|
("hard_clip", {"threshold": 0.4}),
|
|
],
|
|
"parrot": [
|
|
("pitch_shift_pv", {"semitones": 7.0}),
|
|
("bandpass", {"low_hz": 700.0, "high_hz": 4800.0}),
|
|
("tremolo", {"rate": 16.0, "depth": 0.32}),
|
|
("vibrato", {"rate": 24.0, "depth": 0.15}),
|
|
("hard_clip", {"threshold": 0.4}),
|
|
],
|
|
# ``polly`` follows the avian-syrinx blueprint: strip human chest
|
|
# resonance, lift pitch/formants, fast throat flutter (~30 Hz), soft
|
|
# overdrive for harsh squawk harmonics, dual-membrane micro-chorus,
|
|
# then roll off hypersonic brittleness.
|
|
"polly": [
|
|
("highpass", {"cutoff_hz": 600.0}),
|
|
("pitch_shift_pv", {"semitones": 6.0}),
|
|
("tremolo", {"rate": 30.0, "depth": 0.30}),
|
|
("tube_sat", {"drive_db": 8.0, "mix": 1.0}),
|
|
("chorus", {"rate_hz": 1.8, "depth": 0.25, "centre_delay_ms": 7.0, "feedback": 0.15, "mix": 0.35}),
|
|
("lowpass", {"cutoff_hz": 6800.0}),
|
|
],
|
|
}
|
|
|
|
|
|
# Bump this when the emotion DSP recipe changes so existing fx'd audio is
|
|
# regenerated instead of reused.
|
|
EMOTION_FX_VERSION = 1
|
|
|
|
# Per-emotion exaggeration chains. The TTS voice clone carries the speaker's
|
|
# timbre but not emotion delivery, so each emotion is stamped onto the cloned
|
|
# audio deterministically (tempo, pitch, brightness, energy). ``neutral`` is
|
|
# intentionally empty so it stays the untouched baseline. Note librosa
|
|
# ``time_stretch`` semantics: rate > 1 = faster, rate < 1 = slower.
|
|
EMOTION_EFFECTS: dict[str, list[tuple[str, dict[str, Any]]]] = {
|
|
"neutral": [],
|
|
"angry": [
|
|
("time_stretch", {"rate": 1.08}),
|
|
("pitch_shift_pv", {"semitones": 1.5}),
|
|
("tube_sat", {"drive_db": 6.0, "mix": 0.5}),
|
|
("gain_db", {"db": 2.5}),
|
|
],
|
|
"sad": [
|
|
("time_stretch", {"rate": 0.89}),
|
|
("pitch_shift_pv", {"semitones": -1.5}),
|
|
("lowpass", {"cutoff_hz": 5000.0}),
|
|
("gain_db", {"db": -2.0}),
|
|
],
|
|
"happy": [
|
|
("time_stretch", {"rate": 1.06}),
|
|
("pitch_shift_pv", {"semitones": 1.0}),
|
|
("gain_db", {"db": 2.0}),
|
|
],
|
|
"fearful": [
|
|
("time_stretch", {"rate": 0.94}),
|
|
("pitch_shift_pv", {"semitones": 2.5}),
|
|
("tremolo", {"rate": 5.0, "depth": 0.25}),
|
|
("gain_db", {"db": 1.5}),
|
|
],
|
|
"surprised": [
|
|
("time_stretch", {"rate": 1.02}),
|
|
("pitch_shift_pv", {"semitones": 3.0}),
|
|
("gain_db", {"db": 1.5}),
|
|
],
|
|
"whisper": [
|
|
("lowpass", {"cutoff_hz": 4500.0}),
|
|
("gain_db", {"db": -7.0}),
|
|
("noise_inject", {"amount": 0.012}),
|
|
],
|
|
"pleading": [
|
|
("time_stretch", {"rate": 0.93}),
|
|
("lowpass", {"cutoff_hz": 5500.0}),
|
|
("gain_db", {"db": -1.5}),
|
|
],
|
|
"sarcastic": [
|
|
("time_stretch", {"rate": 1.04}),
|
|
("bandpass", {"low_hz": 400.0, "high_hz": 3200.0}),
|
|
("gain_db", {"db": 0.5}),
|
|
],
|
|
"authoritative": [
|
|
("time_stretch", {"rate": 1.04}),
|
|
("pitch_shift_pv", {"semitones": -1.0}),
|
|
("lowpass", {"cutoff_hz": 3500.0}),
|
|
("gain_db", {"db": 2.5}),
|
|
],
|
|
"thoughtful": [
|
|
("time_stretch", {"rate": 0.94}),
|
|
("gain_db", {"db": -1.0}),
|
|
],
|
|
"excited": [
|
|
("time_stretch", {"rate": 1.10}),
|
|
("pitch_shift_pv", {"semitones": 1.5}),
|
|
("tube_sat", {"drive_db": 4.0, "mix": 0.4}),
|
|
("gain_db", {"db": 3.0}),
|
|
],
|
|
"anxious": [
|
|
("time_stretch", {"rate": 1.05}),
|
|
("pitch_shift_pv", {"semitones": 1.0}),
|
|
("tremolo", {"rate": 6.0, "depth": 0.18}),
|
|
("gain_db", {"db": 0.5}),
|
|
],
|
|
"tired": [
|
|
("time_stretch", {"rate": 0.88}),
|
|
("pitch_shift_pv", {"semitones": -2.0}),
|
|
("lowpass", {"cutoff_hz": 4000.0}),
|
|
("gain_db", {"db": -3.0}),
|
|
],
|
|
"romantic": [
|
|
("time_stretch", {"rate": 0.95}),
|
|
("pitch_shift_pv", {"semitones": -0.5}),
|
|
("lowpass", {"cutoff_hz": 5200.0}),
|
|
("gain_db", {"db": -1.0}),
|
|
],
|
|
"disgusted": [
|
|
("time_stretch", {"rate": 0.92}),
|
|
("pitch_shift_pv", {"semitones": -1.0}),
|
|
("bandpass", {"low_hz": 350.0, "high_hz": 3000.0}),
|
|
],
|
|
"amused": [
|
|
("time_stretch", {"rate": 1.05}),
|
|
("pitch_shift_pv", {"semitones": 1.0}),
|
|
("tremolo", {"rate": 3.5, "depth": 0.12}),
|
|
("gain_db", {"db": 1.0}),
|
|
],
|
|
"nostalgic": [
|
|
("time_stretch", {"rate": 0.93}),
|
|
("pitch_shift_pv", {"semitones": -1.0}),
|
|
("lowpass", {"cutoff_hz": 4500.0}),
|
|
("gain_db", {"db": -1.0}),
|
|
],
|
|
}
|
|
|
|
|
|
def get_effects_for_traits(vocal_traits: list[str]) -> list[tuple[str, dict[str, Any]]]:
|
|
"""Map a list of vocal_traits to an ordered chain of (effect_name, params).
|
|
|
|
Returns an empty list when no traits match known effects.
|
|
"""
|
|
chain: list[tuple[str, dict[str, Any]]] = []
|
|
seen: set[str] = set()
|
|
for trait in vocal_traits:
|
|
key = trait.strip().lower()
|
|
for effect_name, params in TRAIT_EFFECTS.get(key, []):
|
|
if effect_name not in seen:
|
|
seen.add(effect_name)
|
|
chain.append((effect_name, dict(params)))
|
|
return chain
|
|
|
|
|
|
def apply_effects(audio: np.ndarray, sr: int, vocal_traits: list[str]) -> np.ndarray:
|
|
"""Apply DSP effects to audio based on vocal_traits.
|
|
|
|
Returns the processed audio. If no traits match known effects, returns
|
|
the original audio unchanged.
|
|
"""
|
|
chain = get_effects_for_traits(vocal_traits)
|
|
if not chain:
|
|
return audio
|
|
|
|
out = audio.copy()
|
|
for effect_name, params in chain:
|
|
fn = EFFECT_FN.get(effect_name)
|
|
if fn is None:
|
|
continue
|
|
out = fn(out, sr, **params)
|
|
peak = float(np.max(np.abs(out)))
|
|
if peak > 1.0:
|
|
out = out / peak
|
|
if peak < 1e-6:
|
|
logger.warning("Effect %s produced silent output", effect_name)
|
|
return audio
|
|
|
|
return out
|
|
|
|
|
|
def get_effects_by_name(effect_names: list[str]) -> list[tuple[str, dict[str, Any]]]:
|
|
"""Build an effect chain from explicit effect names (e.g. from UI selection).
|
|
|
|
Each name must be a preset in ``PRESET_EFFECTS`` or a key in
|
|
``EFFECT_FN``. Unknown names are skipped.
|
|
"""
|
|
chain: list[tuple[str, dict[str, Any]]] = []
|
|
seen: set[str] = set()
|
|
for name in effect_names:
|
|
key = name.strip().lower()
|
|
if key in PRESET_EFFECTS:
|
|
for eff, params in PRESET_EFFECTS[key]:
|
|
if eff not in seen:
|
|
seen.add(eff)
|
|
chain.append((eff, dict(params)))
|
|
elif key in EFFECT_FN and key not in seen:
|
|
seen.add(key)
|
|
chain.append((key, {}))
|
|
return chain
|
|
|
|
|
|
def apply_effects_by_name(
|
|
audio: np.ndarray, sr: int, effect_names: list[str]
|
|
) -> np.ndarray:
|
|
"""Apply DSP effects by explicit effect names (not trait-derived).
|
|
|
|
Uses tuned parameters for presets and defaults for individual effects.
|
|
"""
|
|
chain = get_effects_by_name(effect_names)
|
|
if not chain:
|
|
return audio
|
|
|
|
out = audio.copy()
|
|
for effect_name, params in chain:
|
|
fn = EFFECT_FN[effect_name]
|
|
out = fn(out, sr, **params)
|
|
peak = float(np.max(np.abs(out)))
|
|
if peak > 1.0:
|
|
out = out / peak
|
|
if peak < 1e-6:
|
|
logger.warning("Effect %s produced silent output", effect_name)
|
|
return audio
|
|
|
|
return out
|
|
|
|
|
|
def apply_effects_to_file_by_name(
|
|
wav_path: Path,
|
|
effect_names: list[str],
|
|
output_path: Path | None = None,
|
|
) -> bool:
|
|
"""Apply effects to a WAV file by explicit effect names.
|
|
|
|
Returns True if effects were applied, False if no effects matched.
|
|
"""
|
|
chain = get_effects_by_name(effect_names)
|
|
if not chain:
|
|
return False
|
|
|
|
import soundfile as sf
|
|
|
|
audio, sr = sf.read(str(wav_path), dtype="float32")
|
|
if audio.ndim > 1:
|
|
audio = audio.mean(axis=1)
|
|
|
|
processed = apply_effects_by_name(audio, sr, effect_names)
|
|
dest = output_path or wav_path
|
|
sf.write(str(dest), processed, sr, subtype="PCM_16")
|
|
return True
|
|
|
|
|
|
def apply_effects_to_file(
|
|
wav_path: Path,
|
|
vocal_traits: list[str],
|
|
output_path: Path | None = None,
|
|
) -> bool:
|
|
"""Apply effects to a WAV file in-place or to a new file.
|
|
|
|
Returns True if effects were applied, False if no traits matched.
|
|
"""
|
|
chain = get_effects_for_traits(vocal_traits)
|
|
if not chain:
|
|
return False
|
|
|
|
import soundfile as sf
|
|
|
|
audio, sr = sf.read(str(wav_path), dtype="float32")
|
|
if audio.ndim > 1:
|
|
audio = audio.mean(axis=1)
|
|
|
|
processed = apply_effects(audio, sr, vocal_traits)
|
|
dest = output_path or wav_path
|
|
sf.write(str(dest), processed, sr, subtype="PCM_16")
|
|
return True
|
|
|
|
|
|
def load_speaker_traits(speakers_path: Path) -> dict[str, list[str]]:
|
|
"""Load vocal_traits from speakers.json, keyed by primary_name.
|
|
|
|
If a speaker has ``distortion_effects`` set, those are used instead
|
|
of the implicit ``vocal_traits`` mapping.
|
|
"""
|
|
if not speakers_path.exists():
|
|
return {}
|
|
with open(speakers_path) as f:
|
|
data = json.load(f)
|
|
result: dict[str, list[str]] = {}
|
|
for s in data.get("speakers", []):
|
|
name = s.get("primary_name", "")
|
|
if not name:
|
|
continue
|
|
explicit = s.get("distortion_effects") or []
|
|
if explicit:
|
|
result[name] = list(explicit)
|
|
else:
|
|
traits = s.get("vocal_traits", [])
|
|
if traits:
|
|
result[name] = list(traits)
|
|
return result
|
|
|
|
|
|
def load_speaker_distortion_effects(speakers_path: Path) -> dict[str, list[str]]:
|
|
"""Load only explicit distortion_effects from speakers.json.
|
|
|
|
Returns a dict of speaker_name -> effect_names for speakers that have
|
|
``distortion_effects`` explicitly set. Speakers without that field
|
|
are omitted (they use implicit vocal_traits instead).
|
|
"""
|
|
if not speakers_path.exists():
|
|
return {}
|
|
with open(speakers_path) as f:
|
|
data = json.load(f)
|
|
result: dict[str, list[str]] = {}
|
|
for s in data.get("speakers", []):
|
|
name = s.get("primary_name", "")
|
|
effects = s.get("distortion_effects") or []
|
|
if name and effects:
|
|
result[name] = list(effects)
|
|
return result
|
|
|
|
|
|
def apply_emotion_effects(audio: np.ndarray, sr: int, emotion: str | None) -> np.ndarray:
|
|
"""Apply the emotion exaggeration chain for *emotion* to *audio*.
|
|
|
|
Unknown or empty emotions (e.g. ``neutral``) return the audio unchanged so
|
|
they remain the untouched baseline for comparison.
|
|
"""
|
|
chain = EMOTION_EFFECTS.get((emotion or "").strip().lower(), [])
|
|
if not chain:
|
|
return audio
|
|
|
|
out = audio.copy()
|
|
for effect_name, params in chain:
|
|
fn = EFFECT_FN.get(effect_name)
|
|
if fn is None:
|
|
continue
|
|
out = fn(out, sr, **params)
|
|
if out.size == 0:
|
|
return audio
|
|
peak = float(np.max(np.abs(out)))
|
|
if peak > 1.0:
|
|
out = out / peak
|
|
if float(np.max(np.abs(out))) < 1e-6:
|
|
logger.warning("Emotion effect %s silenced output", effect_name)
|
|
return audio
|
|
return out
|
|
|
|
|
|
def apply_emotion_effects_to_file(
|
|
wav_path: Path,
|
|
emotion: str | None,
|
|
output_path: Path | None = None,
|
|
) -> bool:
|
|
"""Apply the emotion exaggeration chain to a WAV file in-place (or to
|
|
``output_path``). Returns True when an effect chain was applied."""
|
|
if not EMOTION_EFFECTS.get((emotion or "").strip().lower()):
|
|
return False
|
|
|
|
import soundfile as sf
|
|
|
|
audio, sr = sf.read(str(wav_path), dtype="float32")
|
|
if audio.ndim > 1:
|
|
audio = audio.mean(axis=1)
|
|
|
|
processed = apply_emotion_effects(audio, sr, emotion)
|
|
dest = output_path or wav_path
|
|
sf.write(str(dest), processed, sr, subtype="PCM_16")
|
|
return True
|