# ===================================================================== # ПАТЧ 1. Совместимость fairseq2 с Python 3.11 # ===================================================================== import collections.abc import typing collections.abc.Callable = typing.Callable # ===================================================================== # ПАТЧ 2. Отключаем CUDA полностью. # ===================================================================== import os os.environ["CUDA_VISIBLE_DEVICES"] = "" import torch # ===================================================================== # ПАТЧ 3. Заглушки CUDA-API. # ===================================================================== class _DummyStream: def wait_stream(self, other): return None def record_event(self, event=None): return None def synchronize(self): return None def query(self): return True class _DummyEvent: def wait(self, stream=None): return None def record(self, stream=None): return None def synchronize(self): return None def query(self): return True class _DummyStreamContext: def __init__(self, stream=None): self.stream = stream def __enter__(self): return self.stream def __exit__(self, *args): return False class _DummyDeviceContext: def __init__(self, device=None): self.device = device def __enter__(self): return self def __exit__(self, *args): return False class _DummyDeviceProperties: name = "CPU" major = 0 minor = 0 total_memory = 64 * 1024 ** 3 multi_processor_count = 0 torch.cuda.Stream = _DummyStream torch.cuda.Event = _DummyEvent torch.cuda.stream = lambda stream=None: _DummyStreamContext(stream) torch.cuda.device = _DummyDeviceContext torch.cuda.current_stream = lambda device=None: _DummyStream() torch.cuda.default_stream = lambda device=None: _DummyStream() torch.cuda.is_available = lambda: False torch.cuda.current_device = lambda: 0 torch.cuda.device_count = lambda: 0 torch.cuda.set_device = lambda *a, **kw: None torch.cuda.init = lambda *a, **kw: None torch.cuda.memory_stats = lambda *a, **kw: {} torch.cuda.mem_get_info = lambda *a, **kw: (100, 100) torch.cuda.empty_cache = lambda *a, **kw: None torch.cuda.synchronize = lambda *a, **kw: None torch.cuda.ipc_collect = lambda *a, **kw: None torch.cuda.manual_seed = lambda *a, **kw: None torch.cuda.manual_seed_all = lambda *a, **kw: None torch.cuda.get_device_properties = lambda *a, **kw: _DummyDeviceProperties() # ===================================================================== # ПАТЧ 3.5. Принудительно грузим чекпоинты на CPU. # ===================================================================== _orig_torch_load = torch.load def _remap_map_location(ml): if ml is None: return "cpu" if isinstance(ml, str) and ml.startswith("cuda"): return "cpu" if isinstance(ml, torch.device) and ml.type == "cuda": return "cpu" return ml def _patched_torch_load(*args, **kwargs): if len(args) >= 2: args = (args[0], _remap_map_location(args[1])) + args[2:] else: kwargs["map_location"] = _remap_map_location( kwargs.get("map_location") ) return _orig_torch_load(*args, **kwargs) torch.load = _patched_torch_load # ===================================================================== # Обычные импорты. # ===================================================================== import gradio as gr from playdiffusion import PlayDiffusion, InpaintInput, TTSInput, RVCInput # ===================================================================== # ПАТЧ 4. Модели в float32. # ===================================================================== _orig_load_preset = PlayDiffusion.load_preset def _patched_load_preset(self, **kwargs): preset = _orig_load_preset(self, **kwargs) if str(self.device).startswith("cpu"): preset["speech_tokenizer"]["dtype"] = torch.float32 preset["vocoder"]["dtype"] = torch.float32 return preset PlayDiffusion.load_preset = _patched_load_preset # ===================================================================== # ПАТЧ 5. Исправление handle_word_times. # Оригинальный метод теряет поля start/end у слов, если jiwer возвращает # chunk.type не как строку "equal" (а как enum или с другим значением). # Из-за этого в calculate_audio_token_syllable_ratio получается 0, # n_frames = 0, и вокодер падает на пустом тензоре. # # Патч: сохраняем тайминги для equal-чанков и делаем быстрый путь для # случая, когда слова из ASR уже совпадают с нормализованным input_text. # ===================================================================== _orig_handle_word_times = PlayDiffusion.handle_word_times def _chunk_type_str(chunk): """Приводим chunk.type к строке независимо от того, enum это или str.""" t = chunk.type if hasattr(t, "value"): t = t.value return str(t).lower() def _strip_word(w): """Убираем пунктуацию по краям слова.""" return w.strip(".,!?;:\"'()[]{}").lower() def _patched_handle_word_times(self, input_word_times, input_text): import jiwer # та же нормализация, что и в оригинале input_text_clean = "".join( c for c in input_text if c.isalnum() or c.isspace() or c == "'" or c == "-" ) print(f"Text for word times alignment: {input_text_clean}") if not isinstance(input_word_times, list): raise ValueError("input_word_times must be a list") for word in input_word_times: if not isinstance(word, dict): raise ValueError("input_word_times must be a list of dictionaries") if word.get("word") is None or not isinstance(word["word"], str): raise ValueError( "input_word_times must contain a 'word' key with string value" ) word_times = [ { "word": word["word"].lower(), "start": word.get("start"), "end": word.get("end"), } for word in input_word_times ] timed_words = [entry["word"] for entry in word_times] refs = input_text_clean.lower().split() # -------- БЫСТРЫЙ ПУТЬ -------- # Если нормализованные слова ASR совпадают с нормализованными словами # input_text, то jiwer не нужен — просто сохраняем тайминги как есть. stripped_timed = [_strip_word(w) for w in timed_words] stripped_refs = [_strip_word(w) for w in refs] if stripped_refs == stripped_timed: print(f"Word times already aligned (fast path, " f"{len(word_times)} words)") for i, w in enumerate(word_times): print(f" {i}: {w}") return word_times # -------- МЕДЛЕННЫЙ ПУТЬ: jiwer с корректной обработкой equal -------- word_times_align = jiwer.process_words( input_text_clean.lower(), " ".join(timed_words) ) last_hyp_index = 0 word_times_mod = [] for chunk in word_times_align.alignments[0]: if _chunk_type_str(chunk) == "equal": # сохраняем слова из гипотезы вместе с их таймингами for i in range(chunk.hyp_start_idx, chunk.hyp_end_idx): if i < len(word_times): word_times_mod.append(word_times[i]) last_hyp_index = chunk.hyp_end_idx else: # слова до чанка — оставляем как есть for i in range(last_hyp_index, chunk.hyp_start_idx): if i < len(word_times): word_times_mod.append(word_times[i]) # слова из reference — без таймингов for i in range(chunk.ref_start_idx, chunk.ref_end_idx): word = word_times_align.references[0][i] word_times_mod.append({"word": word}) last_hyp_index = chunk.hyp_end_idx # хвост for i in range(last_hyp_index, len(word_times)): word_times_mod.append(word_times[i]) print(f"Word times (after alignment, {len(word_times_mod)} words):") for i, w in enumerate(word_times_mod): print(f" {i}: {w}") return word_times_mod PlayDiffusion.handle_word_times = _patched_handle_word_times # ===================================================================== # Создаём inpainter на CPU. # ===================================================================== print("Инициализация PlayDiffusion на CPU...") inpainter = PlayDiffusion(device="cpu") print("PlayDiffusion готов.") # ===================================================================== # ASR на локальном faster-whisper (без API-ключа). # Модель: large-v3-turbo — самая быстрая и точная на CPU. # ===================================================================== _whisper_model = None WHISPER_MODEL_SIZE = "large-v3-turbo" # ~1.6 GB в int8 WHISPER_COMPUTE_TYPE = "int8" # обязателен для CPU def get_whisper_model(): global _whisper_model if _whisper_model is None: print(f"Загрузка faster-whisper '{WHISPER_MODEL_SIZE}' на CPU...") from faster_whisper import WhisperModel _whisper_model = WhisperModel( WHISPER_MODEL_SIZE, device="cpu", compute_type=WHISPER_COMPUTE_TYPE, ) print("faster-whisper готов.") return _whisper_model def run_asr(audio): if audio is None: raise gr.Error("Сначала загрузите аудио.") model = get_whisper_model() segments, info = model.transcribe( audio, word_timestamps=True, language="en", beam_size=5, ) full_text_parts = [] word_times = [] for segment in segments: full_text_parts.append(segment.text.strip()) if segment.words: for w in segment.words: word = w.word.strip() if not word: continue word_times.append({ "word": word, "start": float(w.start), "end": float(w.end), }) text = " ".join(full_text_parts).strip() return text, text, word_times # ===================================================================== # Обработчики PlayDiffusion. # ===================================================================== def run_inpainter(input_text, output_text, word_times, audio, num_steps, init_temp, init_diversity, guidance, rescale, topk, use_manual_ratio, audio_token_syllable_ratio): if audio is None: raise gr.Error("Сначала загрузите аудио.") if not input_text or not input_text.strip(): raise gr.Error("Сначала запустите ASR — поле «Текст из ASR» пустое.") if not output_text or not output_text.strip(): raise gr.Error("Введите желаемый выходной текст.") if word_times is None: raise gr.Error( "Нет таймингов слов. Сначала нажмите «Запустить ASR»." ) if not use_manual_ratio: audio_token_syllable_ratio = None return inpainter.inpaint(InpaintInput( input_text=input_text, output_text=output_text, input_word_times=word_times, audio=audio, num_steps=num_steps, init_temp=init_temp, init_diversity=init_diversity, guidance=guidance, rescale=rescale, topk=topk, audio_token_syllable_ratio=audio_token_syllable_ratio )) def run_inpainter_tts(input_text, voice_audio, num_steps, init_temp, init_diversity, guidance, rescale, topk, use_manual_ratio, audio_token_syllable_ratio): if not input_text or not input_text.strip(): raise gr.Error("Введите текст для TTS.") if voice_audio is None: raise gr.Error("Загрузите голос для TTS.") if not use_manual_ratio: audio_token_syllable_ratio = None return inpainter.tts(TTSInput( output_text=input_text, voice=voice_audio, num_steps=num_steps, init_temp=init_temp, init_diversity=init_diversity, guidance=guidance, rescale=rescale, topk=topk, audio_token_syllable_ratio=audio_token_syllable_ratio )) def toggle_ratio_input(use_manual): return gr.update(visible=use_manual, interactive=use_manual) def create_advanced_options_accordion(): with gr.Accordion("Advanced options", open=False): num_steps_slider = gr.Slider(1, 100, 30, step=1, label="number of sampling steps codebook") init_temp_slider = gr.Slider(0.5, 10, 1, step=0.1, label="Initial temperature") init_diversity_slider = gr.Slider(0, 10, 1, step=0.1, label="Initial diversity") guidance_slider = gr.Slider(0, 10, 0.5, step=0.1, label="guidance") rescale_slider = gr.Slider(0, 1, 0.7, step=0.1, label="guidance rescale factor") topk_slider = gr.Slider(1, 10000, 25, step=1, label="sampling from top-k logits") gr.Markdown("#### Audio Token Syllable Ratio") gr.Markdown("*Автоматический расчёт (рекомендуется) даёт лучший " "результат в большинстве случаев.*") use_manual_ratio = gr.Checkbox( label="Использовать ручной audio token syllable ratio", value=False ) audio_token_syllable_ratio = gr.Number( label="Audio token syllable ratio (вручную)", value=12.5, precision=2, minimum=5.0, maximum=25.0, visible=False, interactive=False ) use_manual_ratio.change( toggle_ratio_input, inputs=[use_manual_ratio], outputs=[audio_token_syllable_ratio] ) return (num_steps_slider, init_temp_slider, init_diversity_slider, guidance_slider, rescale_slider, topk_slider, use_manual_ratio, audio_token_syllable_ratio) def speech_rvc(rvc_source_speech, rvc_target_voice): if rvc_source_speech is None: raise gr.Error("Загрузите исходную речь.") if rvc_target_voice is None: raise gr.Error("Загрузите целевой голос.") return inpainter.rvc(RVCInput( source_speech=rvc_source_speech, target_voice=rvc_target_voice )) if __name__ == '__main__': with gr.Blocks(analytics_enabled=False, title="PlayDiffusion (CPU)") as demo: gr.Markdown("## PlayDiffusion (CPU-режим)") gr.Markdown("⚠️ **Генерация на CPU идёт очень медленно** — " "от нескольких минут на один фрагмент.") gr.Markdown("ASR работает локально через `faster-whisper` " "(`large-v3-turbo`, int8). API-ключ не нужен.") with gr.Tab("Inpaint"): gr.Markdown("### 1. Загрузите аудио и нажмите «Запустить ASR».") gr.Markdown("### 2. Отредактируйте «Желаемый выходной текст».") gr.Markdown("### 3. Нажмите «Запустить Inpainter».") gr.Markdown("### Модель рассчитана на английский язык.") inpaint_advanced_options = create_advanced_options_accordion() with gr.Row(): audio_input = gr.Audio( label="Аудио для изменения", sources=["upload", "microphone"], type="filepath" ) with gr.Row(): asr_submit = gr.Button("Запустить ASR") with gr.Row(): with gr.Column(): text_input = gr.Textbox(label="Текст из ASR", interactive=False) text_output = gr.Textbox(label="Желаемый выходной текст") with gr.Column(): word_times = gr.JSON(label="Тайминги слов из ASR") with gr.Row(): inpainter_submit = gr.Button("Запустить Inpainter") with gr.Row(): audio_output = gr.Audio(label="Выходное аудио") asr_submit.click( run_asr, inputs=[audio_input], outputs=[text_input, text_output, word_times] ) inpainter_submit.click( run_inpainter, inputs=[text_input, text_output, word_times, audio_input] + list(inpaint_advanced_options), outputs=[audio_output] ) with gr.Tab("Text to Speech"): gr.Markdown("### Синтез речи из текста") tts_advanced_options = create_advanced_options_accordion() tts_text = gr.Textbox( label="Текст для TTS", placeholder="Введите текст для синтеза речи", lines=2 ) tts_voice = gr.Audio( label="Голос для TTS", sources=["upload", "microphone"], type="filepath" ) tts_submit = gr.Button("Синтезировать речь") tts_output = gr.Audio(label="Сгенерированная речь") tts_submit.click( run_inpainter_tts, inputs=[tts_text, tts_voice] + list(tts_advanced_options), outputs=[tts_output] ) with gr.Tab("Voice Conversion"): gr.Markdown("### Конвертация голоса в реальном времени " "(лучше всего работает для английского)") rvc_source_speech = gr.Audio( label="Исходная речь", sources=["upload", "microphone"], type="filepath" ) rvc_target_voice = gr.Audio( label="Целевой голос", sources=["upload", "microphone"], type="filepath" ) rvc_submit = gr.Button("Конвертировать голос") rvc_output = gr.Audio(label="Сконвертированная речь") rvc_submit.click( speech_rvc, inputs=[rvc_source_speech, rvc_target_voice], outputs=[rvc_output] ) demo.launch(share=False)