#!/usr/bin/env python3 """Muse の返答を Irodori-TTS で読み上げる、Linux向けの簡素なパイプライン。 設計（朗読用の Koebuta から、会話用に削ぎ落としたもの）: - 話者は一人。声はユーザーが用意した参照WAVで固定する（クローン）。 - 声のキャプションは任意。あればそのまま全発話の土台として適用するだけ。 サンプル生成やキャプション探索はしない。 - 入力は「台本テキスト」。発話の塊ごとに感情IDと演技メモが付いている （付け方は SPEAK_PROMPT_JA.md を話す側LLMへ渡す）。素の文章でも可 （その場合は neutral で自動分割する）。 - 処理は上から下へ一方向： 短文分割 → 短すぎる断片は次に溜める → 感情の絵文字と演技を付けて逐次合成 → 順に再生／保存。 - 読み（tts_text）の指定は持たない。会話文は難読語を前提としないため。 使い方: python3 speak_pipeline.py --ref voice.wav [--caption-file caption.txt] \ [--input script.txt] [--out DIR] [--play] [--seed 314] [--steps 40] 入力がなければ標準入力から読む。--play は ffplay があれば合成しながら再生する。 """ from __future__ import annotations import argparse import queue import re import subprocess import sys import threading import unicodedata from pathlib import Path # Irodori-TTS の版固定（Koebuta ノートブック版と同じ検証済みの値） MODEL_ID = 'Aratako/Irodori-TTS-v4.1-Small' MODEL_REVISION = '2b28324dc263ed5e6638b3cf3dd94c82ead07b4b' CODEC_ID = 'Aratako/Semantic-DACVAE-Japanese-32dim' CODEC_REVISION = '47376ee24834d7a05a48ebabfe3cde29b3c5e214' EMOTIONS = dict(zip( ['neutral', 'joy', 'anger', 'sad', 'kind', 'tease', 'anxiety', 'tension', 'surprise', 'tired', 'whisper', 'giggle', 'fear', 'sleepy', 'resolve'], ['', '😆', '😠', '😭', '🫶', '😏', '🥺', '😰', '😲', '😮‍💨', '👂', '🤭', '😱', '😪', '💥'])) MAX_CHARS = 80 # 1回の合成の目安（これより長い塊は文単位で割る） MIN_CHARS = 10 # これより短い断片は次の塊に溜める（極短文は合成が不安定なため） PAUSE_MS = 280 # 塊と塊の間の無音 HEADER = re.compile(r'^\[\s*([a-z]+)\s*(?:/\s*(.*?))?\]\s*$') def parse_script(text: str) -> list[dict]: """台本テキストを発話の塊に分ける。見出し行なしの本文は neutral。""" chunks, current = [], None for line in text.splitlines(): m = HEADER.match(line.strip()) if m and m[1] in EMOTIONS: if current: chunks.append(current) current = {'emotion': m[1], 'acting': (m[2] or '').strip(), 'text': ''} elif line.strip() == '': if current: chunks.append(current) current = None else: if current is None: current = {'emotion': 'neutral', 'acting': '', 'text': ''} current['text'] += line.strip() if current: chunks.append(current) return [c for c in chunks if c['text'].strip()] def split_text(text: str, limit: int = MAX_CHARS) -> list[str]: """文の区切りを保ちながら limit 文字以内に割る（無損失）。""" units = re.findall(r'[^\n。！？!?]*[。！？!?][」』”"）)]*|[^\n。！？!?]*\n+|[^\n。！？!?]+', text) if ''.join(units) != text: raise ValueError('分割で本文を保持できませんでした') parts, current = [], '' for unit in units: if current and (len(current) + len(unit) > limit or current.endswith('\n')): parts.append(current) current = '' while len(unit) > limit: if current: parts.append(current) current = '' positions = [m.end() for m in re.finditer(r'[、，,；;：:\s]+', unit[:limit]) if m.end() >= limit // 3 and unit[:m.end()].strip()] cut = positions[-1] if positions else limit while cut > 1 and (unicodedata.combining(unit[cut]) or unit[cut] in ('\ufe0f', '\u200d') or unit[cut - 1] == '\u200d'): cut -= 1 parts.append(unit[:cut]) unit = unit[cut:] current += unit if current: parts.append(current) clean, pending = [], '' for part in parts: if not part.strip() and clean: clean[-1] += part elif not part.strip(): pending += part else: clean.append(pending + part) pending = '' return [p.strip() for p in clean if p.strip()] def shape_chunks(chunks: list[dict]) -> list[dict]: """長すぎる塊は割り、短すぎる断片は次の塊の先頭に溜める。""" flat = [] for c in chunks: pieces = split_text(c['text']) if len(c['text']) > MAX_CHARS else [c['text'].strip()] for piece in pieces: if piece: flat.append({'emotion': c['emotion'], 'acting': c['acting'], 'text': piece}) shaped, buffer = [], None for c in flat: if buffer is not None: c = {'emotion': buffer['emotion'], 'acting': buffer['acting'], 'text': buffer['text'] + c['text']} buffer = None if len(c['text']) < MIN_CHARS: buffer = c else: shaped.append(c) if buffer is not None: if shaped: shaped[-1] = {**shaped[-1], 'text': shaped[-1]['text'] + buffer['text']} else: shaped.append(buffer) return shaped def build_requests(chunks: list[dict], ref_wav: str, base_caption: str, seed: int, steps: int) -> list[dict]: """合成リクエストの列を作る。キャプション = 土台（あれば）＋演技メモ。""" requests = [] for c in chunks: caption = '\n'.join(p for p in [base_caption.strip(), c['acting'].strip()] if p) requests.append({ 'text': EMOTIONS[c['emotion']] + c['text'], 'caption': caption or None, 'ref_wav': ref_wav, 'no_ref': False, 'max_ref_seconds': 30, 'num_steps': steps, 'seed': seed, }) return requests class Speaker: """Irodori ランタイムを一度だけ読み込んで使い回す。""" def __init__(self, irodori_dir: str | None = None): if irodori_dir: sys.path.insert(0, irodori_dir) from huggingface_hub import hf_hub_download from irodori_tts.inference_runtime import InferenceRuntime, RuntimeKey ckpt = hf_hub_download(MODEL_ID, 'model.safetensors', revision=MODEL_REVISION) codec = hf_hub_download(CODEC_ID, 'weights.pth', revision=CODEC_REVISION) self.runtime = InferenceRuntime.from_key(RuntimeKey( checkpoint=ckpt, codec_repo=codec, model_device='cuda', codec_device='cuda', model_precision='fp32', codec_precision='fp32')) def synthesize(self, request: dict): import numpy as np from irodori_tts.inference_runtime import SamplingRequest result = self.runtime.synthesize(SamplingRequest(**request)) audio = result.audios[0].detach().cpu().float().view(-1).numpy() if not audio.size or not np.isfinite(audio).all(): raise RuntimeError('空または異常な音声が生成されました') return audio, int(result.sample_rate) def play_worker(play_queue: queue.Queue): """ffplay で順に再生する（合成と並行）。""" while True: path = play_queue.get() if path is None: break subprocess.run(['ffplay', '-nodisp', '-autoexit', '-loglevel', 'quiet', str(path)], check=False) play_queue.task_done() def run(script_text: str, ref_wav: str, base_caption: str = '', out_dir: str = 'speech_out', seed: int = 314, steps: int = 40, play: bool = False, irodori_dir: str | None = None) -> Path: import numpy as np import soundfile as sf chunks = shape_chunks(parse_script(script_text)) if not chunks: raise ValueError('読み上げる本文がありません') requests = build_requests(chunks, ref_wav, base_caption, seed, steps) out = Path(out_dir) out.mkdir(parents=True, exist_ok=True) speaker = Speaker(irodori_dir) play_queue: queue.Queue | None = None thread = None if play: play_queue = queue.Queue() thread = threading.Thread(target=play_worker, args=(play_queue,), daemon=True) thread.start() pieces, sample_rate = [], None for i, request in enumerate(requests, 1): audio, sample_rate = speaker.synthesize(request) part_path = out / f'chunk_{i:03d}.wav' sf.write(str(part_path), audio, sample_rate, subtype='PCM_16') print(f'[{i}/{len(requests)}] {request["text"][:30]}', file=sys.stderr) pieces.append(audio) if play_queue is not None: play_queue.put(part_path) merged = [] for i, audio in enumerate(pieces): merged.append(audio) if i < len(pieces) - 1: merged.append(np.zeros(int(sample_rate * PAUSE_MS / 1000), dtype='float32')) full_path = out / 'speech.wav' sf.write(str(full_path), np.concatenate(merged), sample_rate, subtype='PCM_16') if play_queue is not None: play_queue.put(None) if thread: thread.join() return full_path def main() -> None: ap = argparse.ArgumentParser(description='Muse の返答を Irodori-TTS で読み上げる') ap.add_argument('--ref', required=True, help='参照WAV（声を固定するクローン元）') ap.add_argument('--caption-file', help='声のキャプション文章のファイル（任意。そのまま適用する）') ap.add_argument('--input', help='台本テキストのファイル（省略時は標準入力）') ap.add_argument('--out', default='speech_out', help='出力フォルダ') ap.add_argument('--seed', type=int, default=314) ap.add_argument('--steps', type=int, default=40, choices=(20, 40, 60)) ap.add_argument('--play', action='store_true', help='ffplay で合成しながら再生する') ap.add_argument('--irodori-dir', help='Irodori-TTS のチェックアウト場所（pip導入済みなら不要）') args = ap.parse_args() script_text = Path(args.input).read_text(encoding='utf-8') if args.input else sys.stdin.read() base_caption = Path(args.caption_file).read_text(encoding='utf-8') if args.caption_file else '' full = run(script_text, args.ref, base_caption, args.out, args.seed, args.steps, args.play, args.irodori_dir) print(str(full)) if __name__ == '__main__': main()