#!/usr/bin/env python3 """ Voice Casting Server — Qwen3 TTS Preview Lightweight HTTP server for the voice casting page. Serves the HTML and provides API endpoints for Qwen3-TTS previews (built-in speakers + cloned voices). Endpoints: GET / — Serve voice_casting.html GET /api/speakers — List built-in Qwen3 speakers GET /api/cloned-voices — List cloned voice configs from cloned_voices/ POST /api/preview — Generate TTS audio, return base64 WAV Usage: python3 voice_casting_server.py # defaults python3 voice_casting_server.py --port 8421 # custom port python3 voice_casting_server.py --model Qwen/Qwen3-TTS-12Hz-0.6B-CustomVoice The model loads once on startup and stays in memory for fast previews. """ import argparse import base64 import io import json import os import sys from http.server import HTTPServer, BaseHTTPRequestHandler from pathlib import Path from urllib.parse import urlparse # Lazy-loaded on startup _model = None _model_name = None _cloned_voices_dir = None DEFAULT_MODEL = "Qwen/Qwen3-TTS-12Hz-0.6B-CustomVoice" DEFAULT_PORT = 8421 HTML_FILE = Path(__file__).parent / "voice_casting.html" def load_model(model_name): """Load Qwen3 TTS model (one-time).""" global _model, _model_name print(f"Loading Qwen3 TTS model: {model_name}...") from qwen_tts import Qwen3TTSModel _model = Qwen3TTSModel.from_pretrained(model_name) _model_name = model_name print(f"Model loaded. Speakers: {_model.get_supported_speakers() or '(none — base/voice_design model)'}") def get_speakers(): """Return list of built-in speakers from the loaded model.""" if _model is None: return [] speakers = _model.get_supported_speakers() return speakers if speakers else [] def get_cloned_voices(): """Scan cloned_voices/ directory for voice configs.""" if _cloned_voices_dir is None or not _cloned_voices_dir.is_dir(): return [] voices = [] for f in sorted(_cloned_voices_dir.glob("*.json")): try: with open(f) as fh: cfg = json.load(fh) cfg.setdefault("name", f.stem) cfg.setdefault("type", "cloned") voices.append(cfg) except (json.JSONDecodeError, IOError): pass return voices TTS_TIMEOUT_SECONDS = 60 def generate_preview(speaker, text, instruct=""): """Generate audio via Qwen3 TTS, return (wav_bytes, sample_rate). Times out after TTS_TIMEOUT_SECONDS to prevent hung model from blocking the server indefinitely. """ import signal import soundfile as sf import numpy as np if _model is None: raise RuntimeError("Model not loaded") def _timeout_handler(signum, frame): raise TimeoutError(f"TTS generation timed out after {TTS_TIMEOUT_SECONDS}s") old_handler = signal.signal(signal.SIGALRM, _timeout_handler) signal.alarm(TTS_TIMEOUT_SECONDS) try: wavs, sr = _model.generate_custom_voice( text=text, speaker=speaker, language="English", instruct=instruct if instruct else None, ) finally: signal.alarm(0) signal.signal(signal.SIGALRM, old_handler) buf = io.BytesIO() sf.write(buf, wavs[0], sr, format="WAV") buf.seek(0) return buf.read(), sr class VoiceCastingHandler(BaseHTTPRequestHandler): """HTTP request handler for voice casting server.""" def log_message(self, format, *args): # Quieter logs — just method and path print(f" {args[0]}") def _cors_headers(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") def _json_response(self, data, status=200): body = json.dumps(data).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self._cors_headers() self.send_header("Content-Length", len(body)) self.end_headers() self.wfile.write(body) def do_OPTIONS(self): self.send_response(204) self._cors_headers() self.end_headers() def do_GET(self): path = urlparse(self.path).path if path in ("/", "/index.html", "/voice_casting.html"): self._serve_html() elif path == "/api/speakers": self._api_speakers() elif path == "/api/cloned-voices": self._api_cloned_voices() else: self.send_error(404) def do_POST(self): path = urlparse(self.path).path if path == "/api/preview": self._api_preview() else: self.send_error(404) def _serve_html(self): if not HTML_FILE.exists(): self.send_error(404, f"HTML file not found: {HTML_FILE}") return content = HTML_FILE.read_bytes() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", len(content)) self.end_headers() self.wfile.write(content) def _api_speakers(self): speakers = get_speakers() self._json_response({ "speakers": speakers, "model": _model_name or "", }) def _api_cloned_voices(self): voices = get_cloned_voices() self._json_response({"voices": voices}) def _api_preview(self): content_len = int(self.headers.get("Content-Length", 0)) if content_len == 0: self._json_response({"error": "Empty request body"}, 400) return try: body = json.loads(self.rfile.read(content_len)) except json.JSONDecodeError: self._json_response({"error": "Invalid JSON"}, 400) return speaker = body.get("speaker", "") text = body.get("text", "") instruct = body.get("instruct", "") if not speaker or not text: self._json_response({"error": "Missing 'speaker' or 'text'"}, 400) return try: wav_bytes, sr = generate_preview(speaker, text, instruct) audio_b64 = base64.b64encode(wav_bytes).decode("ascii") self._json_response({ "audio": audio_b64, "sample_rate": sr, "format": "wav", }) except Exception as e: self._json_response({"error": str(e)}, 500) def main(): global _cloned_voices_dir parser = argparse.ArgumentParser(description="Voice Casting Server — Qwen3 TTS") parser.add_argument("--port", type=int, default=DEFAULT_PORT) parser.add_argument("--model", default=DEFAULT_MODEL) parser.add_argument("--cloned-voices", default=None, help="Directory with cloned voice JSON configs") parser.add_argument("--no-model", action="store_true", help="Start server without loading model (for HTML dev)") args = parser.parse_args() if args.cloned_voices: _cloned_voices_dir = Path(args.cloned_voices) else: # Default: cloned_voices/ next to this script _cloned_voices_dir = Path(__file__).parent / "cloned_voices" if not args.no_model: load_model(args.model) else: print("Starting without model (--no-model). Preview will fail.") server = HTTPServer(("127.0.0.1", args.port), VoiceCastingHandler) print(f"\nReady at http://127.0.0.1:{args.port}") print(f"Serving: {HTML_FILE}") print("Press Ctrl+C to stop.\n") try: server.serve_forever() except KeyboardInterrupt: print("\nShutting down.") server.server_close() if __name__ == "__main__": main()