Volver al índice
docs/voice-studio/studio.py
"""Estudio de voz local: grabaciones, importaciones y Piper opcional.
Los filtros reproducen el preset de generate-parasite-voice.py. Ninguna frase
ni grabación se envía a un servicio externo. FFmpeg recibe argumentos fijos.
"""
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse
import argparse
import importlib.util
import json
import math
import os
import shutil
import subprocess
import sys
import tempfile
import threading
import uuid
import webbrowser
# Rutas y límites. Los audios de trabajo desaparecen al cerrar el estudio.
HERE = Path(__file__).resolve().parent
TOOLS = Path.home() / "AppData/Local/CodexTools"
VOICE_ROOT = TOOLS / "RapsVoice"
FFMPEG = TOOLS / "ffmpeg/ffmpeg-9.0-full_build-shared/bin/ffmpeg.exe"
FFPROBE = FFMPEG.with_name("ffprobe.exe")
ENGINE = HERE / "generate-parasite-voice.py"
if not ENGINE.exists():
ENGINE = HERE.parent / "generate-parasite-voice.py"
MAX_UPLOAD = 32 * 1024 * 1024
MAX_SECONDS = 180
NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)
# .opus usa el contenedor Ogg; se permite su demultiplexor y no protocolos de red.
AUDIO_FORMATS = "wav,mp3,ogg,mov,mp4,m4a,3gp,3g2,mj2,matroska,webm,flac,aac"
def run_ffmpeg(*args, timeout=150):
subprocess.run([str(FFMPEG), "-nostdin", "-y", "-v", "error",
"-threads", "1", "-filter_threads", "1", "-filter_complex_threads", "1",
*map(str, args)], check=True, capture_output=True, timeout=timeout,
creationflags=NO_WINDOW)
def duration(path):
result = subprocess.run([str(FFPROBE), "-v", "error", "-protocol_whitelist", "file,pipe",
"-format_whitelist", AUDIO_FORMATS, "-show_entries",
"format=duration", "-of", "json", str(path)],
check=True, capture_output=True, text=True, timeout=12,
creationflags=NO_WINDOW)
value = float(json.loads(result.stdout)["format"]["duration"])
if not math.isfinite(value) or value <= 0 or value > MAX_SECONDS:
raise ValueError("Usá un audio de hasta tres minutos.")
return value
# Tres capas: voz grave, segunda garganta y aire. El preset inicial conserva
# las proporciones del audio de la criatura que ya utiliza el mod.
def creature_filter(pitch=.77, texture=1.0, echo=1.0):
throat_pitch = .67 * pitch / .77
return (
f"[0:a]rubberband=pitch={pitch:.6f}:formant=shifted,"
"highpass=f=62,lowpass=f=6800,asplit=3[main][throat][air];"
"[main]acompressor=threshold=.18:ratio=3:attack=12:release=100,volume=.87[m];"
f"[throat]rubberband=pitch={throat_pitch:.6f}:formant=shifted,lowpass=f=820,"
f"volume={.22 * texture:.6f},adelay=21:all=1[t];"
"[air]highpass=f=1300,lowpass=f=4600,tremolo=f=31:d=.55,"
f"volume={.14 * texture:.6f},adelay=47:all=1[a];"
"[m][t][a]amix=inputs=3:normalize=0,"
f"aecho=.8:.6:127|269:{.12 * echo:.6f}|{.055 * echo:.6f},"
"alimiter=limit=.92:level=false"
)
def settings(raw):
if not isinstance(raw, dict) or set(raw) - {"pitch", "texture", "echo"}:
raise ValueError("Ajustes inválidos.")
limits = {"pitch": (.60, .95, .77), "texture": (0, 1.5, 1), "echo": (0, 1.5, 1)}
result = {}
for key, (low, high, default) in limits.items():
value = raw.get(key, default)
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError("Ajustes inválidos.")
if not math.isfinite(value) or not low <= value <= high:
raise ValueError("El ajuste está fuera de su rango.")
result[key] = value
return result
def render(source, folder, options):
# Las tomas WebM del micrófono suelen omitir su duración. Decodificar primero
# produce un WAV medible; el segundo adicional detecta tomas demasiado largas.
decoded = folder / "entrada-normalizada.wav"
run_ffmpeg("-protocol_whitelist", "file,pipe", "-format_whitelist", AUDIO_FORMATS,
"-i", source, "-t", MAX_SECONDS + 1, "-ac", "1", "-ar", "32000",
"-c:a", "pcm_s16le", decoded)
seconds = duration(decoded)
# WAV conserva calidad para edición; OGG entra en Minecraft; MP3 se comparte.
wav = folder / "criatura.wav"
run_ffmpeg("-protocol_whitelist", "file,pipe", "-format_whitelist", AUDIO_FORMATS,
"-i", decoded, "-filter_complex", creature_filter(**options),
"-ac", "1", "-ar", "32000", "-c:a", "pcm_s16le", wav)
run_ffmpeg("-i", wav, "-c:a", "libvorbis", "-q:a", "5", folder / "criatura.ogg")
run_ffmpeg("-i", wav, "-c:a", "libmp3lame", "-q:a", "3", folder / "criatura.mp3")
return {"seconds": round(seconds, 2), "settings": options}
# Piper se carga sólo al pedir texto, en un proceso que termina tras la frase.
def synthesis_available():
return ENGINE.exists() and (VOICE_ROOT / "es_ES-carlfm-x_low.onnx").exists() \
and importlib.util.find_spec("piper") is not None
def synthesize_file(text_file, target):
# Evitar que la carga inicial de Piper abra un grupo de hilos por cada núcleo.
import onnxruntime
original_session = onnxruntime.InferenceSession
def single_thread_session(*args, **kwargs):
if kwargs.get("sess_options") is None:
options = onnxruntime.SessionOptions()
options.intra_op_num_threads = 1
options.inter_op_num_threads = 1
kwargs["sess_options"] = options
return original_session(*args, **kwargs)
onnxruntime.InferenceSession = single_thread_session
spec = importlib.util.spec_from_file_location("creature_voice_engine", ENGINE)
engine = importlib.util.module_from_spec(spec)
spec.loader.exec_module(engine)
engine.synthesize(Path(text_file).read_text(encoding="utf-8"), Path(target), 1.12)
class Studio(ThreadingHTTPServer):
daemon_threads = True
request_queue_size = 4
def __init__(self, address):
self.temp = tempfile.TemporaryDirectory(prefix="raps-voz-")
self.work = Path(self.temp.name)
self.busy = threading.Lock()
self.synthesis = synthesis_available()
super().__init__(address, Handler)
def server_close(self):
super().server_close()
self.temp.cleanup()
class Handler(BaseHTTPRequestHandler):
def send_bytes(self, data, content_type, status=200, download=None):
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(data)))
self.send_header("Cache-Control", "no-store")
self.send_header("X-Content-Type-Options", "nosniff")
if download:
self.send_header("Content-Disposition", 'attachment; filename="' + download + '"')
self.end_headers()
self.wfile.write(data)
def send_json(self, value, status=200):
self.send_bytes(json.dumps(value, ensure_ascii=False).encode("utf-8"),
"application/json; charset=utf-8", status)
def same_origin(self):
port = self.server.server_port
return self.headers.get("Origin") in (None, f"http://127.0.0.1:{port}", f"http://localhost:{port}")
def do_GET(self):
route = urlparse(self.path).path
if route == "/api/health":
self.send_json({"studio": "raps-voice", "synthesis": self.server.synthesis})
return
files = {"/": ("index.html", "text/html; charset=utf-8"),
"/app.js": ("app.js", "text/javascript; charset=utf-8")}
if route in files:
name, mime = files[route]
self.send_bytes((HERE / name).read_bytes(), mime)
return
parts = route.strip("/").split("/")
if len(parts) == 3 and parts[0] == "audio" and len(parts[1]) == 32 \
and all(c in "0123456789abcdef" for c in parts[1]) \
and parts[2] in {"criatura.wav", "criatura.ogg", "criatura.mp3"}:
path = self.server.work / parts[1] / parts[2]
if path.is_file():
types = {"wav": "audio/wav", "ogg": "audio/ogg", "mp3": "audio/mpeg"}
self.send_bytes(path.read_bytes(), types[path.suffix[1:]],
download=path.name if urlparse(self.path).query == "download" else None)
return
self.send_json({"error": "No se encontró el archivo."}, 404)
def do_POST(self):
if not self.same_origin():
self.send_json({"error": "Origen inválido."}, 403)
return
route = urlparse(self.path).path
if route == "/api/shutdown":
self.send_json({"closed": True})
threading.Thread(target=self.server.shutdown, daemon=True).start()
return
if route not in {"/api/process", "/api/synthesize"}:
self.send_json({"error": "Ruta inválida."}, 404)
return
if not self.server.busy.acquire(blocking=False):
self.send_json({"error": "Hay otro audio en preparación. Esperá a que termine."}, 409)
return
folder = None
completed = False
try:
self.connection.settimeout(15)
size = int(self.headers.get("Content-Length", "0"))
if size <= 0 or size > MAX_UPLOAD:
raise ValueError("Usá un archivo de hasta 32 MB.")
folder_id = uuid.uuid4().hex
folder = self.server.work / folder_id
folder.mkdir()
if route == "/api/process":
options = settings(json.loads(self.headers.get("X-Voice-Settings", "{}")))
source = folder / "entrada.audio"
source.write_bytes(self.rfile.read(size))
else:
if not self.server.synthesis:
raise ValueError("La síntesis local no está instalada.")
if size > 4096:
raise ValueError("La frase es demasiado larga.")
body = json.loads(self.rfile.read(size))
text = body.get("text")
if not isinstance(text, str) or not text.strip() or len(text) > 400 \
or any(ord(c) < 32 and c not in "\n\t" for c in text):
raise ValueError("Escribí una frase de hasta 400 caracteres.")
options = settings(body.get("settings", {}))
text_file = folder / "frase.txt"
text_file.write_text(text, encoding="utf-8")
source = folder / "entrada.wav"
subprocess.run([sys.executable, "-X", "utf8", str(HERE / "studio.py"),
"--synthesize", str(text_file), str(source)], check=True,
timeout=180, capture_output=True, creationflags=NO_WINDOW,
env={**os.environ, "OMP_NUM_THREADS": "1", "OPENBLAS_NUM_THREADS": "1"})
result = render(source, folder, options)
# No se acumulan tomas indefinidamente: conservar las últimas ocho.
old = sorted((p for p in self.server.work.iterdir() if p.is_dir()),
key=lambda p: p.stat().st_mtime)
for stale in old[:-8]:
shutil.rmtree(stale)
result["files"] = {ext: f"/audio/{folder_id}/criatura.{ext}" for ext in ("wav", "ogg", "mp3")}
completed = True
self.send_json(result)
except (ValueError, TypeError, KeyError, json.JSONDecodeError):
self.send_json({"error": "Revisá el archivo, la frase y los ajustes. Usá audio de hasta tres minutos."}, 400)
except subprocess.TimeoutExpired:
self.send_json({"error": "El audio tardó demasiado. Probá una toma más corta."}, 504)
except (subprocess.CalledProcessError, OSError):
self.send_json({"error": "No se pudo preparar este audio. Probá WAV, MP3, OGG, OPUS o WebM."}, 422)
finally:
if folder is not None and not completed:
shutil.rmtree(folder, ignore_errors=True)
self.server.busy.release()
def log_message(self, fmt, *args):
# No registrar frases, grabaciones ni contenido de archivos.
pass
def main():
if len(sys.argv) == 4 and sys.argv[1] == "--synthesize":
synthesize_file(sys.argv[2], sys.argv[3])
return
parser = argparse.ArgumentParser()
parser.add_argument("--port", type=int, default=8778)
parser.add_argument("--open", action="store_true")
args = parser.parse_args()
if not FFMPEG.is_file() or not FFPROBE.is_file():
raise SystemExit("No se encontró FFmpeg instalado.")
server = Studio(("127.0.0.1", args.port))
if args.open:
webbrowser.open(f"http://127.0.0.1:{server.server_port}/")
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
if __name__ == "__main__":
main()