Spaces:
Sleeping
Sleeping
| # ===================================================================== | |
| # ПАТЧ 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) |