from __future__ import annotations # spaces は必ず最初に import する(ZeroGPU ランタイム登録のため) try: import spaces ZERO_GPU = True except ImportError: ZERO_GPU = False class spaces: # type: ignore[no-redef] @staticmethod def GPU(duration=None): def decorator(fn): return fn return decorator import os, re, sys, traceback, tempfile from pathlib import Path import torch import torchaudio import soundfile as sf import gradio as gr # ── HF login ──────────────────────────────────────────────────────────────── _HF_TOKEN = os.environ.get("HF_TOKEN") if _HF_TOKEN: try: from huggingface_hub import login as hf_login hf_login(token=_HF_TOKEN, add_to_git_credential=False) print("[HF] logged in") except Exception as _e: print(f"[HF] login failed: {_e}") try: from liquid_audio import LFM2AudioModel, LFM2AudioProcessor, ChatState except ImportError as e: raise ImportError(f"liquid_audio not found: {e}") BASE_REPO = "LiquidAI/LFM2.5-Audio-1.5B-JP" OUTPUT_SAMPLE_RATE = 24_000 VOICE_PROMPTS = { "us_male": "Perform TTS. Use the US male voice.", "us_female": "Perform TTS. Use the US female voice.", "uk_male": "Perform TTS. Use the UK male voice.", "uk_female": "Perform TTS. Use the UK female voice.", } DEFAULT_VOICE = "us_female" ASR_SYSTEM_PROMPT = "Perform ASR." DEFAULT_REPLACEMENT = "にゃん" DEFAULT_TOXIC_WORDS: list[str] = [ "バカ","ばか","馬鹿","アホ","あほ","阿呆","ボケ","ぼけ","ドアホ","どあほ", "アホンダラ","あほんだら","タワケ","たわけ","アンポンタン","あんぽんたん", "マヌケ","まぬけ","間抜け","ノロマ","のろま","グズ","ぐず","ウスノロ","うすのろ", "ドンクサイ","どんくさい","タコ","でくのぼう","ポンコツ","ぽんこつ", "出来損ない","できそこない","欠陥品","けっかんひん","役立たず","やくたたず", "能なし","のうなし","ダメ人間","だめにんげん", "死","しね","死ねよ","死にさらせ","死ぬべき","消えろ","きえろ","消え失せろ", "きえうせろ","失せろ","うせろ","ぶっ殺","殺","ころす","殺すぞ","ころすぞ", "くたばれ","クタバレ","ぶっとばす","ぶん殴る","ぶんなぐる","地獄に落ちろ", "じごくにおちろ","黙れ","だまれ","出て行け","でていけ","引っ込め","ひっこめ", "晒す","さらす","晒してやる","特定した","とくていした", "クソ","くそ","糞","クソヤロウ","くそやろう","ゴミ","ごみ","ゴミクズ","ごみくず", "カス","かす","チンカス","ムカつく","むかつく","畜生","ちくしょう","外道","げどう", "人でなし","ひとでなし","虫けら","むしけら","ゴキブリ","ごきぶり", "お前","おまえ","テメエ","てめえ","キサマ","貴様","きさま","オメー","おめー", "コノヤロウ","この野郎","このやろう","野郎","やろう", "ブス","ぶす","不細工","ぶさいく","醜い","みにくい","デブ","でぶ","ブタ","豚","ぶた", "豚野郎","ぶたやろう","ガリ","がり","チビ","ちび","ハゲ","はげ", "ウザい","ウザ","うざい","キモい","キモ","きもい","最悪","目障り","めざわり", "邪魔","じゃま","邪魔者","じゃまもの","口答え","くちごたえ","舐めて","なめて", "ナメんな","なめんな","負け犬","まけいぬ","惨め","みじめ", "頭おかしい","あたまおかしい","頭悪い","あたまわるい", "ビッチ","びっち","ヤリマン","やりまん","淫乱","いんらん","売女","ばいた", "尻軽","しりがるい","女々しい","めめしい","男のくせに","おとこのくせに","女のくせに","おんなのくせに", "ホモ","ほも","オカマ","おかま","オナベ","おなべ","レズ","れず","変態","へんたい", "キチガイ","きちがい","気違い","キチ","きち","池沼","ちしょう","精薄","せいはく","文盲","もんもう", "チョン","ちょん","鮮人","せんじん","半島人","はんとうじん","チャンコロ","ちゃんころ", "支那","しな","土人","どじん","黒んぼ","くろんぼ","ニガー","にがー", "穢多","えた","非人","ひにん", ] # ══════════════════════════════════════════════════════════════ # Utilities (pure Python / CPU only — safe outside GPU scope) # ══════════════════════════════════════════════════════════════ class ToxicWordFilter: def __init__(self, toxic_words=DEFAULT_TOXIC_WORDS, replacement=DEFAULT_REPLACEMENT): self.replacement = replacement sorted_words = sorted(toxic_words, key=len, reverse=True) self._pattern = re.compile("|".join(re.escape(w) for w in sorted_words)) def filter(self, text: str): stats: dict[str, int] = {} def _replace(m): w = m.group(0); stats[w] = stats.get(w, 0) + 1; return self.replacement return self._pattern.sub(_replace, text), stats def _save_wav(path: Path, waveform: torch.Tensor, sr: int) -> None: path.parent.mkdir(parents=True, exist_ok=True) wav = waveform.cpu() while wav.ndim > 2: wav = wav.squeeze(0) if wav.ndim == 1: wav = wav.unsqueeze(0) try: torchaudio.save(str(path), wav, sr) except Exception as e: print(f"[save_wav] torchaudio failed ({e}), fallback to soundfile") data = wav.squeeze(0).numpy() if data.ndim == 2: data = data.T sf.write(str(path), data, sr) def _load_weights(model, path: Path) -> None: if path.suffix == ".safetensors": from safetensors.torch import load_file state = load_file(str(path), device="cpu") else: state = torch.load(str(path), map_location="cpu", weights_only=True) missing, unexpected = model.load_state_dict(state, strict=True) if missing: print(f" [warn] missing keys: {len(missing)}") if unexpected: print(f" [warn] unexpect keys: {len(unexpected)}") # ══════════════════════════════════════════════════════════════ # ZeroGPU — 3 separate @spaces.GPU functions # # ZeroGPU では: # - @spaces.GPU 関数はプリミティブ型 (str/int/float/list) のみ # 引数として受け取れる(カスタムオブジェクトはワーカー間で # シリアライズ不可のため RuntimeError になる) # - モデル/プロセッサはワーカー内で毎回生成する # - @torch.inference_mode はデコレータでなく with 文で使う # ══════════════════════════════════════════════════════════════ @spaces.GPU(duration=120) def _step_asr(audio_path: str, base_repo: str, max_new_tokens: int) -> str: """Step 1: load model+processor, run ASR, return transcript string.""" print(f"[ASR] device={('cuda' if torch.cuda.is_available() else 'cpu')}") dev = "cuda" if torch.cuda.is_available() else "cpu" processor = LFM2AudioProcessor.from_pretrained(base_repo) processor = processor.eval() model = LFM2AudioModel.from_pretrained(base_repo) model = model.to(device=dev, dtype=torch.float32).eval() with torch.inference_mode(): wav, sr = torchaudio.load(audio_path) wav = wav.mean(dim=0, keepdim=True) chat = ChatState(processor) chat.new_turn("system"); chat.add_text(ASR_SYSTEM_PROMPT); chat.end_turn() chat.new_turn("user"); chat.add_audio(wav, sr); chat.end_turn() chat.new_turn("assistant") tokens = [] for t in model.generate_sequential(**dict(chat), max_new_tokens=max_new_tokens): if t.numel() == 1: tokens.append(t) transcript = processor.text.decode(torch.cat(tokens)) if tokens else "" transcript = re.sub(r"<\|[^|]*\|>", "", transcript).strip() print(f"[ASR] result: 「{transcript}」") return transcript @spaces.GPU(duration=180) def _step_tts( filtered_text: str, base_repo: str, hub_id: str, max_new_tokens: int, audio_temperature: float, audio_top_k: int, ) -> str: """Step 2: load model+processor, run TTS, return output wav path.""" print(f"[TTS] device={('cuda' if torch.cuda.is_available() else 'cpu')}") dev = "cuda" if torch.cuda.is_available() else "cpu" processor = LFM2AudioProcessor.from_pretrained(base_repo) processor = processor.eval() model = LFM2AudioModel.from_pretrained(base_repo) model = model.to(device=dev, dtype=torch.float32).eval() if hub_id: import huggingface_hub from huggingface_hub import hf_hub_download repo_files = list(huggingface_hub.list_repo_files(hub_id)) fname = next((f for f in ["model.safetensors", "pytorch_model.bin"] if f in repo_files), None) if fname is None: raise FileNotFoundError(f"No weight file in '{hub_id}'") _load_weights(model, Path(hf_hub_download(hub_id, filename=fname))) model = model.to(device=dev, dtype=torch.float32).eval() with torch.inference_mode(): chat = ChatState(processor) chat.new_turn("system"); chat.add_text("Respond with interleaved text and audio."); chat.end_turn() chat.new_turn("user"); chat.add_text(f"次の文章をそのまま読み上げてください:\n{filtered_text}"); chat.end_turn() chat.new_turn("assistant") audio_out = [] for t in model.generate_interleaved( **dict(chat), max_new_tokens=max_new_tokens, audio_temperature=audio_temperature, audio_top_k=audio_top_k, ): if t.numel() != 1: audio_out.append(t) if not audio_out: print("[TTS] no audio tokens") return "" tokens_for_decode = audio_out[:-1] if len(audio_out) >= 2 else audio_out # processor.audio_detokenizer は CPU 上にあるため、 # audio_codes を CPU に移動してから decode する audio_codes = torch.stack(tokens_for_decode, 1).unsqueeze(0).cpu() with torch.inference_mode(): waveform = processor.decode(audio_codes) with tempfile.NamedTemporaryFile(suffix=".wav", delete=False, prefix="tts_") as tmp: out_path = tmp.name _save_wav(Path(out_path), waveform, OUTPUT_SAMPLE_RATE) print(f"[TTS] saved: {out_path}") return out_path # ══════════════════════════════════════════════════════════════ # Gradio pipeline (generator — yields intermediate updates) # ══════════════════════════════════════════════════════════════ def pipeline( audio_input, hub_id, base_repo, voice, extra_toxic, replacement, asr_max_tokens, tts_max_tokens, audio_temperature, audio_top_k, progress=gr.Progress(track_tqdm=True), ): if audio_input is None: gr.Warning("⚠️ 音声を入力してください。") yield None, "", "", "⚠️ 音声が入力されていません。" return base_repo = (base_repo or "").strip() or BASE_REPO hub_id = (hub_id or "").strip() extra_toxic = extra_toxic or "" replacement = (replacement or "").strip() or DEFAULT_REPLACEMENT asr_max_tokens = int(asr_max_tokens) tts_max_tokens = int(tts_max_tokens) audio_temperature = float(audio_temperature) audio_top_k = int(audio_top_k) logs = [] def log(msg): logs.append(msg); print(msg) # ── ASR ────────────────────────────────────────────────── log("🔄 [1/3] ASR 実行中…(GPU 確保 + モデルロード)") yield None, "", "", "\n".join(logs) try: transcript = _step_asr(audio_input, base_repo, asr_max_tokens) log(f"✅ ASR 完了: 「{transcript}」") yield None, transcript, "", "\n".join(logs) except Exception as e: err = f"❌ ASR エラー: {type(e).__name__}: {e}\n{traceback.format_exc()}" log(err); yield None, "", "", "\n".join(logs); return # ── Toxic filter ───────────────────────────────────────── log("🔄 [2/3] Toxic Word フィルタ適用中…") yield None, transcript, "", "\n".join(logs) try: extra_words = [w.strip() for w in extra_toxic.split(",") if w.strip()] tf = ToxicWordFilter(DEFAULT_TOXIC_WORDS + extra_words, replacement) filtered_text, stats = tf.filter(transcript) if stats: log("⚠️ フィルタ完了(置換あり): " + "、".join(f"「{w}」×{c}" for w,c in stats.items())) else: log("✅ フィルタ完了(toxic word なし)") yield None, transcript, filtered_text, "\n".join(logs) except Exception as e: err = f"❌ フィルタエラー: {type(e).__name__}: {e}\n{traceback.format_exc()}" log(err); yield None, transcript, "", "\n".join(logs); return # ── TTS ────────────────────────────────────────────────── log("🔄 [3/3] TTS 実行中…(GPU 確保 + モデルロード)") yield None, transcript, filtered_text, "\n".join(logs) try: result_wav = _step_tts( filtered_text, base_repo, hub_id, tts_max_tokens, audio_temperature, audio_top_k, ) if result_wav: log("✅ TTS 完了"); log("🎉 全パイプライン完了!") yield result_wav, transcript, filtered_text, "\n".join(logs) else: log("❌ TTS 失敗: 音声トークンが生成されませんでした") yield None, transcript, filtered_text, "\n".join(logs) except Exception as e: err = f"❌ TTS エラー: {type(e).__name__}: {e}\n{traceback.format_exc()}" log(err); yield None, transcript, filtered_text, "\n".join(logs) # ══════════════════════════════════════════════════════════════ # UI # ══════════════════════════════════════════════════════════════ CSS = """ body, .gradio-container { font-family:'Nunito','Hiragino Sans','Yu Gothic',sans-serif !important; background:linear-gradient(160deg,#daeeff 0%,#eaf5ff 45%,#d6ecff 100%) !important; min-height:100vh; } .hero-header { text-align:center; padding:2.5rem 1rem 1.8rem; background:linear-gradient(135deg,#5bb8f5 0%,#7ecef7 50%,#a8deff 100%); border-radius:24px; margin-bottom:1.5rem; box-shadow:0 8px 32px rgba(91,184,245,.35); position:relative; overflow:hidden; } .hero-header::before { content:''; position:absolute; top:-40%; left:-20%; width:60%; height:200%; background:radial-gradient(ellipse,rgba(255,255,255,.4) 0%,transparent 70%); pointer-events:none; } .hero-header h1 { font-size:2.4rem !important; font-weight:800 !important; color:#fff !important; text-shadow:0 2px 12px rgba(50,130,200,.35) !important; margin:0 0 .4rem !important; } .hero-header p { color:rgba(255,255,255,.95) !important; font-size:1.05rem !important; margin:0 !important; } .hero-emoji { font-size:1.1rem; opacity:.75; letter-spacing:.5rem; margin-top:.6rem; } .card { background:rgba(255,255,255,.78) !important; backdrop-filter:blur(12px); border-radius:20px !important; border:1.5px solid rgba(126,206,247,.5) !important; box-shadow:0 4px 24px rgba(91,184,245,.15) !important; padding:1.2rem 1.2rem .8rem !important; margin-bottom:1rem !important; } /* Gradio gr.Group / gr.Column / gr.Row inner wrappers — make transparent so .card bg shows through */ .card > .form, .card > div, .card .block, .card .gap, .card .gr-group, .card .gr-block, .card .gr-box, .card [data-node-id], .card .svelte-1gfkn6j, .card > .wrap { background:transparent !important; border:none !important; box-shadow:none !important; padding:0 !important; } .section-title { font-size:1rem !important; font-weight:700 !important; color:#3a87c8 !important; margin:0 0 .7rem !important; } #run-btn { background:linear-gradient(135deg,#4fb3f0 0%,#6ecbf7 50%,#9ddcff 100%) !important; border:none !important; border-radius:50px !important; color:#fff !important; font-size:1.15rem !important; font-weight:700 !important; letter-spacing:.04em; padding:.85rem 1.5rem !important; box-shadow:0 6px 20px rgba(79,179,240,.45) !important; transition:transform .18s,box-shadow .18s !important; width:100%; } #run-btn:hover { transform:translateY(-2px) scale(1.02) !important; box-shadow:0 10px 28px rgba(79,179,240,.6) !important; } #run-btn:active { transform:translateY(0) scale(.98) !important; } .gradio-accordion { background:rgba(255,255,255,.65) !important; border-radius:16px !important; border:1.5px solid rgba(126,206,247,.45) !important; margin-bottom:.7rem !important; overflow:hidden; } .gradio-accordion .label-wrap { background:linear-gradient(90deg,#d6eeff 0%,#c2e5ff 100%) !important; border-radius:14px 14px 0 0 !important; padding:.55rem 1rem !important; font-weight:700 !important; color:#3a87c8 !important; } /* ── Gradio 4 theme variables: eliminate all grey ── */ :root, .gradio-container { --block-background-fill: #eaf6ff !important; --block-border-color: #aad9f7 !important; --block-border-width: 1.5px !important; --block-label-background-fill: #d6eeff !important; --block-label-text-color: #3a87c8 !important; --block-shadow: 0 2px 8px rgba(91,184,245,.1) !important; --input-background-fill: #f0f9ff !important; --input-border-color: #aad9f7 !important; --input-border-width: 1.5px !important; --panel-background-fill: #eaf6ff !important; --panel-border-color: #aad9f7 !important; --background-fill-primary: #eaf6ff !important; --background-fill-secondary: #d6eeff !important; --border-color-primary: #aad9f7 !important; --border-color-accent: #5bb8f5 !important; --color-accent: #5bb8f5 !important; --color-accent-soft: #d6eeff !important; --neutral-50: #eaf6ff !important; --neutral-100: #d6eeff !important; --neutral-200: #aad9f7 !important; } /* ── Textbox actual input elements ── */ .gradio-textbox textarea, .gradio-textbox input, [data-testid="textbox"] textarea, [data-testid="textbox"] input { border-radius:12px !important; border:1.5px solid #aad9f7 !important; background:#f0f9ff !important; font-family:inherit !important; } .gradio-textbox textarea:focus, .gradio-textbox input:focus, [data-testid="textbox"] textarea:focus, [data-testid="textbox"] input:focus { border-color:#5bb8f5 !important; box-shadow:0 0 0 3px rgba(91,184,245,.2) !important; background:#ffffff !important; } /* ── output-box / log-box overrides ── */ .output-box textarea { background:#edf7ff !important; border:1.5px solid #aad9f7 !important; border-radius:12px !important; } .log-box textarea { background:#e4f3ff !important; border:1.5px solid #93d4fb !important; border-radius:12px !important; font-size:.82rem !important; font-family:'Courier New',monospace !important; white-space:pre-wrap !important; } input[type=range] { accent-color:#5bb8f5 !important; } /* ── Audio widget: override all grey surfaces ── */ .gradio-audio, .gradio-audio > div, [data-testid="audio"], [data-testid="audio"] > div, [data-testid="audio"] > div > div { border-radius:16px !important; border:1.5px solid #aad9f7 !important; background:#eaf6ff !important; } /* waveform / player bar */ .gradio-audio .waveform-container, .gradio-audio audio, [data-testid="audio"] .waveform-container, [data-testid="audio"] audio { background:#eaf6ff !important; border-radius:12px !important; } /* toolbar row (mic / upload buttons area) */ .gradio-audio .controls, .gradio-audio .icon-buttons, .gradio-audio .recording-toolbar, [data-testid="audio"] .controls, [data-testid="audio"] .icon-buttons, [data-testid="audio"] .recording-toolbar { background:#d6eeff !important; border-radius:0 0 14px 14px !important; } /* WaveSurfer canvas background */ .gradio-audio wave, [data-testid="audio"] wave { background:#eaf6ff !important; } /* grey block that wraps the whole upload/record area */ .gradio-audio .wrap, [data-testid="audio"] .wrap, .gradio-audio .empty, [data-testid="audio"] .empty { background:#eaf6ff !important; border-color:#aad9f7 !important; border-radius:14px !important; } /* label row above the widget */ .gradio-audio label span, [data-testid="audio"] label span { color:#3a87c8 !important; font-weight:700 !important; } .step-badge { display:inline-block; background:linear-gradient(90deg,#5bb8f5,#93d4fb); color:#fff; font-size:.72rem; font-weight:700; padding:.18em .7em; border-radius:20px; letter-spacing:.06em; margin-bottom:.3rem; } .output-box textarea { background:linear-gradient(135deg,rgba(255,255,255,.92) 0%,rgba(214,238,255,.65) 100%) !important; border:1.5px solid #aad9f7 !important; border-radius:12px !important; } .log-box textarea { background:linear-gradient(135deg,rgba(210,236,255,.75) 0%,rgba(235,248,255,.6) 100%) !important; border:1.5px solid #93d4fb !important; border-radius:12px !important; font-size:.82rem !important; font-family:'Courier New',monospace !important; white-space:pre-wrap !important; } footer { display:none !important; } """ with gr.Blocks(title="Everything becomes Nyan 🐱") as demo: gr.HTML("""

🐱 Everything becomes "Nyan" 🐱

Automatically replaces inappropriate words with "にゃん" to make them cute ✨

🌸 ✨ 🐱 ⭐ 💙
""") with gr.Row(equal_height=False): with gr.Column(scale=5): with gr.Group(elem_classes="card"): gr.HTML('

🎙️ Input Audio

') audio_input = gr.Audio( label="Record or upload an audio file", sources=["microphone","upload"], type="filepath", ) run_btn = gr.Button("🌟 Start Conversion!", variant="primary", size="lg", elem_id="run-btn") with gr.Accordion("⚙️ Model Settings", open=False): base_repo_in = gr.Textbox(label="Base Model (HuggingFace Hub ID)", value=BASE_REPO) hub_id_in = gr.Textbox(label="Custom Model Hub ID (optional)", placeholder="your-username/your-model") with gr.Accordion("🔊 TTS Voice Settings", open=True): voice_in = gr.Dropdown(label="TTS Voice", choices=list(VOICE_PROMPTS.keys()), value=DEFAULT_VOICE) with gr.Accordion("🚫 Toxic Word Settings", open=False): with gr.Row(): extra_toxic_in = gr.Textbox(label="Additional Toxic Words (comma-separated)", placeholder="word1, word2") replacement_in = gr.Textbox(label="Replacement String", value=DEFAULT_REPLACEMENT) with gr.Accordion("🎛️ Generation Parameters", open=False): asr_max_tokens_in = gr.Slider(label="ASR Max Tokens", minimum=64, maximum=1024, step=64, value=512) tts_max_tokens_in = gr.Slider(label="TTS Max Tokens", minimum=64, maximum=2048, step=64, value=1024) audio_temp_in = gr.Slider(label="Audio Temperature", minimum=0.1, maximum=2.0, step=0.1, value=1.0) audio_topk_in = gr.Slider(label="Audio Top-K", minimum=1, maximum=128, step=1, value=4) with gr.Column(scale=5): with gr.Group(elem_classes="card"): gr.HTML('

🎵 Generated Audio

') audio_output = gr.Audio(label="", type="filepath", interactive=False) with gr.Group(elem_classes="card"): gr.HTML('

📝 Text Results

') with gr.Row(): with gr.Column(): gr.HTML('STEP 1 — ASR') transcript_out = gr.Textbox( label="Recognized Text (before filter)", lines=4, interactive=False, placeholder="Recognized text will appear here", elem_classes="output-box", ) with gr.Column(): gr.HTML('STEP 2 — Filtered') filtered_out = gr.Textbox( label="Filtered Text", lines=4, interactive=False, placeholder="Filtered text will appear here", elem_classes="output-box", ) with gr.Group(elem_classes="card"): gr.HTML('

📋 Processing Log

') status_out = gr.Textbox( label="", lines=10, interactive=False, placeholder="Logs will appear here after conversion starts…", elem_classes="log-box", ) with gr.Accordion("📋 Default Toxic Word List (click to expand)", open=False): gr.Markdown("The following words are replaced with 'にゃん' by default.\n\n" + " ".join(f"`{w}`" for w in DEFAULT_TOXIC_WORDS)) run_btn.click( fn=pipeline, inputs=[ audio_input, hub_id_in, base_repo_in, voice_in, extra_toxic_in, replacement_in, asr_max_tokens_in, tts_max_tokens_in, audio_temp_in, audio_topk_in, ], outputs=[audio_output, transcript_out, filtered_out, status_out], ) if __name__ == "__main__": demo.queue().launch( theme=gr.themes.Soft( primary_hue="sky", secondary_hue="blue", neutral_hue="sky", font=gr.themes.GoogleFont("Nunito"), ).set( block_background_fill="#eaf6ff", block_border_color="#aad9f7", block_label_background_fill="#d6eeff", input_background_fill="#f0f9ff", panel_background_fill="#eaf6ff", background_fill_primary="#eaf6ff", background_fill_secondary="#d6eeff", border_color_primary="#aad9f7", ), css=CSS, )