diff --git a/ttstoy/webui/app.py b/ttstoy/webui/app.py index b1502cd..b98c372 100644 --- a/ttstoy/webui/app.py +++ b/ttstoy/webui/app.py @@ -651,68 +651,228 @@ def api_mc_voices(): @app.route("/api/mc/tts", methods=["POST"]) def api_mc_tts(): - """Generate TTS audio. Returns WAV bytes. - JSON body: {"text": "...", "mode": "chatterbox|dectalk|morshu", "voice": "Emily.wav"} + """Generate TTS audio with full ttstoy features. Returns WAV bytes. + JSON body: {"text": "...", "mode": "chatterbox", "voice": "Emily.wav"} + + Supports: + - SFX emojis inline (🎉, 💀, etc.) → interleaved SFX audio + - Voice switching: [dectalk|Harry]hello [chatterbox|Emily]world + - Chatterbox turbo tokens: [laugh], [cough], etc. (passed through to model) """ + import io + import re + import wave + import random + import subprocess as _sp + data = request.get_json(force=True) if request.is_json else {} text = (data.get("text") or "").strip() if not text: return jsonify({"error": "Missing text"}), 400 - mode = data.get("mode", "chatterbox") - voice = data.get("voice", "Emily.wav") + default_mode = data.get("mode", "chatterbox") + default_voice = data.get("voice", "Emily.wav") + gcfg = get_global_config() + # --- SFX setup --- + sfx_root = Path(__file__).resolve().parent.parent / "sfx" + emoji_map_path = sfx_root / "emoji_map.json" try: - if mode == "chatterbox": + emoji_map = json.load(open(emoji_map_path)) if emoji_map_path.exists() else {} + except Exception: + emoji_map = {} + + EMOJI_RE = re.compile( + r"(|:[A-Za-z0-9_]+:|[\U0001F300-\U0001FAFF\U00002600-\U000026FF\U00002700-\U000027BF][\uFE0E\uFE0F]?(?:\u200D[\U0001F300-\U0001FAFF\U00002600-\U000026FF\U00002700-\U000027BF][\uFE0E\uFE0F]?)*)" + ) + VOICE_SWITCH_RE = re.compile(r"\[([a-zA-Z]+)(?:\|([^\]]+))?\]") + VALID_MODES = {"chatterbox", "clone", "dectalk", "morshu", "vox"} + # Turbo tokens — these look like voice switches but should pass through to chatterbox + TURBO_TOKENS = {"laugh", "chuckle", "sigh", "gasp", "cough", "cry", "groan", "yawn", "sniff"} + + def normalize_trigger(token): + if token.startswith("<") and token.endswith(">"): + parts = token.strip("<>").split(":") + if len(parts) == 3: + return f":{parts[1]}:" + return token + + def get_sfx_file(trigger): + norm = normalize_trigger(trigger) + folder_name = emoji_map.get(norm) + if not folder_name: + return None + folder = sfx_root / folder_name + files = sorted(folder.glob("*.mp3")) + sorted(folder.glob("*.wav")) + sorted(folder.glob("*.ogg")) + return random.choice(files) if files else None + + def generate_for_mode(seg_text, mode, voice): + """Generate WAV bytes for a text segment with given mode/voice.""" + if mode == "chatterbox" or mode == "clone": chatterbox_url = get_chatterbox_url() payload = { - "text": text, - "voice_mode": "predefined", - "predefined_voice_id": voice, + "text": seg_text, + "voice_mode": "clone" if mode == "clone" else "predefined", + ("reference_audio_filename" if mode == "clone" else "predefined_voice_id"): voice or default_voice, "output_format": "wav", "split_text": False, } r = requests.post(f"{chatterbox_url}/tts", json=payload, timeout=120) r.raise_for_status() - return r.content, 200, {"Content-Type": "audio/wav"} - - elif mode == "clone": - chatterbox_url = get_chatterbox_url() - payload = { - "text": text, - "voice_mode": "clone", - "reference_audio_filename": voice, - "output_format": "wav", - "split_text": False, - } - r = requests.post(f"{chatterbox_url}/tts", json=payload, timeout=120) - r.raise_for_status() - return r.content, 200, {"Content-Type": "audio/wav"} - + return r.content elif mode == "dectalk": dectalk_url = get_dectalk_url() - r = requests.get(f"{dectalk_url}/say", params={"text": text}, timeout=30) + # Resolve voice name to DECTalk command + dectalk_voices = {"paul": "[:np]", "betty": "[:nb]", "harry": "[:nh]", "frank": "[:nf]", + "dennis": "[:nd]", "kit": "[:nk]", "ursula": "[:nu]", "rita": "[:nr]", "wendy": "[:nw]"} + prefix = dectalk_voices.get((voice or "").lower(), "") + r = requests.get(f"{dectalk_url}/say", params={"text": f"{prefix}{seg_text}"}, timeout=30) r.raise_for_status() - return r.content, 200, {"Content-Type": "audio/wav"} - + return r.content elif mode == "morshu": - r = requests.get("http://127.0.0.1:33002/say", params={"text": text}, timeout=60) + r = requests.get("http://127.0.0.1:33002/say", params={"text": seg_text}, timeout=60) r.raise_for_status() - return r.content, 200, {"Content-Type": "audio/wav"} - + return r.content elif mode == "vox": - gcfg = get_global_config() pack = gcfg.get("vox_pack", "vox") - r = requests.get("http://127.0.0.1:33003/say", params={"text": text, "pack": pack}, timeout=60) + r = requests.get("http://127.0.0.1:33003/say", params={"text": seg_text, "pack": pack}, timeout=60) r.raise_for_status() - return r.content, 200, {"Content-Type": "audio/wav"} - + return r.content else: - return jsonify({"error": f"Unknown mode: {mode}"}), 400 + raise RuntimeError(f"Unknown mode: {mode}") - except requests.exceptions.ConnectionError: - return jsonify({"error": f"{mode} server not reachable"}), 502 + def sfx_to_wav(sfx_path): + """Convert any audio file to WAV bytes via ffmpeg.""" + result = _sp.run( + ["ffmpeg", "-y", "-i", str(sfx_path), "-ac", "1", "-ar", "24000", "-f", "wav", "-"], + capture_output=True + ) + if result.returncode != 0: + return None + return result.stdout + + def ensure_wav_format(audio_bytes): + """Normalize audio bytes to consistent WAV format (mono 24kHz 16-bit) via ffmpeg.""" + result = _sp.run( + ["ffmpeg", "-y", "-i", "pipe:0", "-ac", "1", "-ar", "24000", "-sample_fmt", "s16", "-f", "wav", "pipe:1"], + input=audio_bytes, capture_output=True + ) + return result.stdout if result.returncode == 0 else audio_bytes + + def concat_wavs(wav_list): + """Concatenate multiple WAV byte arrays into one.""" + if len(wav_list) == 1: + return wav_list[0] + # Use ffmpeg to concat + import tempfile, shutil + tmp_dir = tempfile.mkdtemp() + try: + paths = [] + for i, wav_bytes in enumerate(wav_list): + p = os.path.join(tmp_dir, f"{i:03}.wav") + with open(p, "wb") as f: + f.write(wav_bytes) + paths.append(p) + list_file = os.path.join(tmp_dir, "list.txt") + with open(list_file, "w") as f: + for p in paths: + f.write(f"file '{p}'\n") + out_path = os.path.join(tmp_dir, "out.wav") + _sp.run(["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", list_file, + "-ac", "1", "-ar", "24000", "-sample_fmt", "s16", out_path], + capture_output=True, check=True) + with open(out_path, "rb") as f: + return f.read() + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + try: + # Step 1: Split on voice switch tags (but NOT turbo tokens) + voice_segments = [] + last_end = 0 + for m in VOICE_SWITCH_RE.finditer(text): + tag_mode = m.group(1).lower() + tag_voice = (m.group(2) or "").strip() + # Skip turbo tokens — they stay in the text + if tag_mode in TURBO_TOKENS: + continue + if tag_mode not in VALID_MODES: + continue + before = text[last_end:m.start()] + if before: + if not voice_segments: + voice_segments.append((default_mode, default_voice, before)) + else: + prev = voice_segments[-1] + voice_segments[-1] = (prev[0], prev[1], prev[2] + before) + voice_segments.append((tag_mode, tag_voice or default_voice, "")) + last_end = m.end() + tail = text[last_end:] + if tail: + if not voice_segments: + voice_segments.append((default_mode, default_voice, tail)) + else: + prev = voice_segments[-1] + voice_segments[-1] = (prev[0], prev[1], prev[2] + tail) + if not voice_segments: + voice_segments = [(default_mode, default_voice, text)] + + # Step 2: For each voice segment, split on SFX emojis and generate + all_wav_parts = [] + for seg_mode, seg_voice, seg_text in voice_segments: + seg_text = seg_text.strip() + if not seg_text: + continue + + # Split segment into TTS and SFX parts + parts = [] + cursor = 0 + text_buf = [] + for match in EMOJI_RE.finditer(seg_text): + if match.start() > cursor: + text_buf.append(seg_text[cursor:match.start()]) + token = match.group(0) + if normalize_trigger(token) in emoji_map: + if text_buf: + parts.append(("tts", "".join(text_buf))) + text_buf = [] + parts.append(("sfx", token)) + else: + text_buf.append(token) + cursor = match.end() + if cursor < len(seg_text): + text_buf.append(seg_text[cursor:]) + if text_buf: + parts.append(("tts", "".join(text_buf))) + + if not parts: + parts = [("tts", seg_text)] + + # Generate audio for each part + for kind, value in parts: + if kind == "tts": + value = value.strip() + if not value: + continue + wav = generate_for_mode(value, seg_mode, seg_voice) + all_wav_parts.append(ensure_wav_format(wav)) + else: + sfx_file = get_sfx_file(value) + if sfx_file: + wav = sfx_to_wav(sfx_file) + if wav: + all_wav_parts.append(wav) + + if not all_wav_parts: + return jsonify({"error": "No audio generated"}), 400 + + final_wav = concat_wavs(all_wav_parts) if len(all_wav_parts) > 1 else all_wav_parts[0] + return final_wav, 200, {"Content-Type": "audio/wav"} + + except requests.exceptions.ConnectionError as e: + return jsonify({"error": f"TTS server not reachable: {e}"}), 502 except Exception as e: + log.exception(f"mc/tts error: {e}") return jsonify({"error": str(e)}), 500