steve-adminandClaude Sonnet 5
057e728091
Real reference clips are often cheap-mic/phone-call quality. Adds AuK's dedicated Speech Enhancement task (denoise+dereverb, or a telephone-effect-specific variant) as a new dropdown, applied first in the edit chain - before emotion/pitch/speed/custom - since character edits on noisy source audio just carry the noise through. Found useful while iterating on real character voices for a sibling project (the-service-station) using this exact tool. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Y1oQ4h4WE4R9KRvMQLr5Ey
358 lines
13 KiB
Python
358 lines
13 KiB
Python
"""
|
|
Pipeline studio: Demucs (strip) -> AuK (edit voice, with preview/undo) ->
|
|
MOSS-TTS (generate speech in the edited voice).
|
|
|
|
Orchestrates across the three existing isolated venvs (demucs/, auk/, tts/)
|
|
via subprocess, rather than merging their conflicting dependencies
|
|
(MOSS-TTS needs transformers 5.0.0, AuK needs 4.57.6) into one environment.
|
|
|
|
Run with: webui/.venv/bin/python webui/app.py
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import gradio as gr
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
DEMUCS_DIR = REPO_ROOT / "demucs"
|
|
AUK_DIR = REPO_ROOT / "auk"
|
|
TTS_DIR = REPO_ROOT / "tts"
|
|
|
|
DEMUCS_PYTHON = DEMUCS_DIR / ".venv" / "bin" / "python3"
|
|
AUK_INFER = AUK_DIR / ".venv" / "bin" / "auk-infer"
|
|
AUK_CKPT = AUK_DIR / "ckpts" / "AuK" / "auk_base.safetensors"
|
|
AUK_QWEN_PATH = AUK_DIR / "ckpts" / "Qwen2.5-Omni-3B"
|
|
TTS_PYTHON = TTS_DIR / ".venv" / "bin" / "python3"
|
|
TTS_GENERATE = TTS_DIR / "generate.py"
|
|
|
|
WORKDIR = Path(tempfile.mkdtemp(prefix="audio-demo-webui-"))
|
|
|
|
EMOTIONS = ["happy", "angry", "sad", "fearful", "surprised", "disgusted", "calm", "excited"]
|
|
PITCH_OPTIONS = ["-3", "-2", "-1", "+1", "+2", "+3"]
|
|
SPEED_OPTIONS = ["0.5", "0.75", "1.25", "1.5", "2.0"]
|
|
LANGUAGES = [
|
|
"English", "Spanish", "French", "German", "Portuguese", "Italian",
|
|
"Chinese", "Japanese", "Korean", "Russian",
|
|
]
|
|
|
|
COMFYUI_BASE = "http://localhost:8188"
|
|
|
|
|
|
def _comfyui_queue_empty() -> bool:
|
|
try:
|
|
with urllib.request.urlopen(f"{COMFYUI_BASE}/queue", timeout=3) as resp:
|
|
data = json.loads(resp.read())
|
|
return not data.get("queue_running") and not data.get("queue_pending")
|
|
except (urllib.error.URLError, OSError):
|
|
return True # ComfyUI not reachable -> nothing to worry about
|
|
|
|
|
|
def _free_comfyui_memory() -> None:
|
|
if not _comfyui_queue_empty():
|
|
return
|
|
try:
|
|
req = urllib.request.Request(
|
|
f"{COMFYUI_BASE}/free",
|
|
data=json.dumps({"unload_models": True, "free_memory": True}).encode(),
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
urllib.request.urlopen(req, timeout=5)
|
|
except (urllib.error.URLError, OSError):
|
|
pass
|
|
|
|
|
|
def _run(cmd: list[str], cwd: Path, env: dict | None = None) -> None:
|
|
result = subprocess.run(
|
|
cmd, cwd=cwd, env=env, capture_output=True, text=True, timeout=1800
|
|
)
|
|
if result.returncode != 0:
|
|
tail = "\n".join(result.stderr.strip().splitlines()[-25:])
|
|
raise gr.Error(f"Command failed ({cmd[0]}):\n{tail}")
|
|
|
|
|
|
def _ensure_wav(path: str) -> str:
|
|
"""AuK's loader (librosa/soundfile) can't read compressed containers like
|
|
.m4a/.mp3 - only demucs's loader handles those. Normalize once, here, at
|
|
the point a reference is set, so everything downstream is always wav."""
|
|
if Path(path).suffix.lower() == ".wav":
|
|
return path
|
|
out_path = WORKDIR / f"converted_{int(time.time() * 1000)}.wav"
|
|
result = subprocess.run(
|
|
["ffmpeg", "-nostdin", "-y", "-i", path, "-ar", "44100", "-ac", "1", str(out_path)],
|
|
capture_output=True, text=True, timeout=120,
|
|
)
|
|
if result.returncode != 0:
|
|
tail = "\n".join(result.stderr.strip().splitlines()[-15:])
|
|
raise gr.Error(f"Could not convert '{path}' to wav:\n{tail}")
|
|
return str(out_path)
|
|
|
|
|
|
def _ld_library_path_for_venv(venv_dir: Path) -> str:
|
|
site_packages = next(venv_dir.glob(".venv/lib/python*/site-packages"))
|
|
nvidia_dir = site_packages / "nvidia"
|
|
if not nvidia_dir.is_dir():
|
|
return ""
|
|
lib_dirs = sorted(p for p in nvidia_dir.glob("*/lib") if p.is_dir())
|
|
return ":".join(str(p) for p in lib_dirs)
|
|
|
|
|
|
TTS_LD_LIBRARY_PATH = _ld_library_path_for_venv(TTS_DIR)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage 1: input / demucs stripping
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def strip_vocals(input_audio: str):
|
|
if not input_audio:
|
|
raise gr.Error("Upload or record some audio first.")
|
|
_free_comfyui_memory()
|
|
out_dir = WORKDIR / f"demucs_{int(time.time())}"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
_run(
|
|
[
|
|
str(DEMUCS_PYTHON), "-m", "demucs",
|
|
"--two-stems=vocals", "-n", "htdemucs",
|
|
"-o", str(out_dir), input_audio,
|
|
],
|
|
cwd=DEMUCS_DIR,
|
|
)
|
|
stem_name = Path(input_audio).stem
|
|
vocals = out_dir / "htdemucs" / stem_name / "vocals.wav"
|
|
no_vocals = out_dir / "htdemucs" / stem_name / "no_vocals.wav"
|
|
return str(vocals), str(no_vocals)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage 2: AuK voice editing, with history for undo/reset
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def use_as_reference(audio_path: str, history: list):
|
|
if not audio_path:
|
|
raise gr.Error("No audio to use yet.")
|
|
audio_path = _ensure_wav(audio_path)
|
|
history = [audio_path]
|
|
return audio_path, history, _history_label(history)
|
|
|
|
|
|
def _history_label(history: list) -> str:
|
|
if not history:
|
|
return "No reference set yet."
|
|
return f"Edit history: {len(history) - 1} edit(s) applied on top of the original reference."
|
|
|
|
|
|
ENHANCE_INSTRUCTIONS = {
|
|
"none": None,
|
|
"General (denoise + dereverb)": (
|
|
"Preserve all speakers, remove noise and reverberation, and output "
|
|
"clean speech of the same length."
|
|
),
|
|
"Telephone/call recording": (
|
|
"Repair the telephone effect and restore natural, clear speech."
|
|
),
|
|
}
|
|
|
|
|
|
def apply_edit(
|
|
current_ref: str,
|
|
history: list,
|
|
enhance: str,
|
|
emotion: str,
|
|
pitch: str,
|
|
speed: str,
|
|
custom_instruction: str,
|
|
):
|
|
if not current_ref:
|
|
raise gr.Error("Set a voice reference first (upload, strip with Demucs, or use a previous edit).")
|
|
|
|
instructions = []
|
|
# Enhancement goes first when picked - clean the recording up before any
|
|
# character/accent editing, not after (garbage in, garbage timbre-edited).
|
|
if enhance and ENHANCE_INSTRUCTIONS.get(enhance):
|
|
instructions.append(ENHANCE_INSTRUCTIONS[enhance])
|
|
if emotion and emotion != "none":
|
|
instructions.append(f"Change the emotion to {emotion}.")
|
|
if pitch and pitch != "none":
|
|
direction = "Raise" if pitch.startswith("+") else "Lower"
|
|
instructions.append(f"{direction} the pitch by {pitch.lstrip('+-')} semitone(s).")
|
|
if speed and speed != "none":
|
|
instructions.append(f"Adjust the speech speed to {speed}x.")
|
|
if custom_instruction and custom_instruction.strip():
|
|
instructions.append(custom_instruction.strip())
|
|
|
|
if not instructions:
|
|
raise gr.Error("Pick at least one edit (enhance / emotion / pitch / speed / custom instruction).")
|
|
|
|
_free_comfyui_memory()
|
|
ref = current_ref
|
|
out_dir = WORKDIR / f"auk_{int(time.time())}"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
for i, instruction in enumerate(instructions):
|
|
out_path = out_dir / f"edit_{i}.wav"
|
|
cmd = [
|
|
str(AUK_INFER),
|
|
"--ckpt", str(AUK_CKPT),
|
|
"--qwen_path", str(AUK_QWEN_PATH),
|
|
"--audio", ref,
|
|
"--instruction", instruction,
|
|
"--output", str(out_path),
|
|
"--cpu_offload",
|
|
]
|
|
_run(cmd, cwd=AUK_DIR)
|
|
ref = str(out_path)
|
|
|
|
history = history + [ref]
|
|
return ref, history, _history_label(history)
|
|
|
|
|
|
def undo_edit(history: list):
|
|
if len(history) <= 1:
|
|
raise gr.Error("Nothing to undo.")
|
|
history = history[:-1]
|
|
return history[-1], history, _history_label(history)
|
|
|
|
|
|
def reset_to_original(history: list):
|
|
if not history:
|
|
raise gr.Error("No reference set yet.")
|
|
history = [history[0]]
|
|
return history[0], history, _history_label(history)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stage 3: MOSS-TTS generation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def generate_speech(reference: str, text: str, language: str):
|
|
if not reference:
|
|
raise gr.Error("Set a voice reference first (upload, strip, and/or edit it above).")
|
|
if not text or not text.strip():
|
|
raise gr.Error("Enter some text to synthesize.")
|
|
|
|
_free_comfyui_memory()
|
|
out_path = WORKDIR / f"tts_{int(time.time())}.wav"
|
|
env = None
|
|
if TTS_LD_LIBRARY_PATH:
|
|
env = os.environ.copy()
|
|
env["LD_LIBRARY_PATH"] = TTS_LD_LIBRARY_PATH
|
|
|
|
_run(
|
|
[
|
|
str(TTS_PYTHON), str(TTS_GENERATE),
|
|
"--text", text,
|
|
"--reference", reference,
|
|
"--language", language,
|
|
"--output", str(out_path),
|
|
],
|
|
cwd=TTS_DIR,
|
|
env=env,
|
|
)
|
|
return str(out_path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# UI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
with gr.Blocks(title="Audio-demo pipeline studio") as demo:
|
|
gr.Markdown(
|
|
"# Pipeline studio\n"
|
|
"Strip a voice with **Demucs**, edit it with **AuK**, then generate new "
|
|
"speech in that voice with **MOSS-TTS**. Each stage shells out to its own "
|
|
"isolated venv, so this box's GPU is the real constraint — heavy edits "
|
|
"(long clips) can OOM AuK; keep reference clips short (under ~25s)."
|
|
)
|
|
|
|
reference_state = gr.State(None)
|
|
history_state = gr.State([])
|
|
|
|
with gr.Row():
|
|
with gr.Column():
|
|
gr.Markdown("## 1. Input")
|
|
input_audio = gr.Audio(label="Upload or record", type="filepath")
|
|
strip_btn = gr.Button("Strip vocals with Demucs")
|
|
with gr.Row():
|
|
vocals_out = gr.Audio(label="Vocals (isolated)", type="filepath")
|
|
no_vocals_out = gr.Audio(label="Background (no vocals)", type="filepath")
|
|
with gr.Row():
|
|
use_vocals_btn = gr.Button("Use vocals as reference")
|
|
use_no_vocals_btn = gr.Button("Use background as reference")
|
|
use_original_btn = gr.Button("Use uploaded audio as-is")
|
|
|
|
with gr.Column():
|
|
gr.Markdown("## 2. Voice editor (AuK)")
|
|
current_ref_audio = gr.Audio(label="Current reference", type="filepath", interactive=False)
|
|
history_label = gr.Markdown("No reference set yet.")
|
|
enhance_dd = gr.Dropdown(
|
|
list(ENHANCE_INSTRUCTIONS.keys()), value="none",
|
|
label="Enhance (denoise/dereverb/restore) - runs first, before character edits",
|
|
)
|
|
emotion_dd = gr.Dropdown(["none"] + EMOTIONS, value="none", label="Emotion")
|
|
pitch_dd = gr.Dropdown(["none"] + PITCH_OPTIONS, value="none", label="Pitch (semitones)")
|
|
speed_dd = gr.Dropdown(["none"] + SPEED_OPTIONS, value="none", label="Speed")
|
|
custom_instruction = gr.Textbox(
|
|
label="Custom instruction (advanced, optional)",
|
|
placeholder='e.g. "Lower the pitch very slightly."',
|
|
)
|
|
gr.Markdown(
|
|
"*Note: pitch editing has been unreliable in testing — even "
|
|
"\"1 semitone\" has overshot to ~4+ semitones on some voices. "
|
|
"Preview before committing.*"
|
|
)
|
|
apply_btn = gr.Button("Apply edit(s)", variant="primary")
|
|
with gr.Row():
|
|
undo_btn = gr.Button("Undo last edit")
|
|
reset_btn = gr.Button("Reset to original")
|
|
|
|
with gr.Column():
|
|
gr.Markdown("## 3. Generate (MOSS-TTS)")
|
|
gen_text = gr.Textbox(label="Text to speak", lines=6)
|
|
gen_language = gr.Dropdown(LANGUAGES, value="English", label="Language")
|
|
generate_btn = gr.Button("Generate speech", variant="primary")
|
|
generated_audio = gr.Audio(label="Generated speech", type="filepath")
|
|
|
|
strip_btn.click(strip_vocals, inputs=[input_audio], outputs=[vocals_out, no_vocals_out])
|
|
use_vocals_btn.click(
|
|
use_as_reference, inputs=[vocals_out, history_state],
|
|
outputs=[current_ref_audio, history_state, history_label],
|
|
)
|
|
use_no_vocals_btn.click(
|
|
use_as_reference, inputs=[no_vocals_out, history_state],
|
|
outputs=[current_ref_audio, history_state, history_label],
|
|
)
|
|
use_original_btn.click(
|
|
use_as_reference, inputs=[input_audio, history_state],
|
|
outputs=[current_ref_audio, history_state, history_label],
|
|
)
|
|
apply_btn.click(
|
|
apply_edit,
|
|
inputs=[current_ref_audio, history_state, enhance_dd, emotion_dd, pitch_dd, speed_dd, custom_instruction],
|
|
outputs=[current_ref_audio, history_state, history_label],
|
|
)
|
|
undo_btn.click(
|
|
undo_edit, inputs=[history_state],
|
|
outputs=[current_ref_audio, history_state, history_label],
|
|
)
|
|
reset_btn.click(
|
|
reset_to_original, inputs=[history_state],
|
|
outputs=[current_ref_audio, history_state, history_label],
|
|
)
|
|
generate_btn.click(
|
|
generate_speech, inputs=[current_ref_audio, gen_text, gen_language],
|
|
outputs=[generated_audio],
|
|
)
|
|
|
|
if __name__ == "__main__":
|
|
demo.launch(server_name="0.0.0.0", server_port=7860)
|