| from __future__ import annotations |
|
|
| import json |
| import math |
| import os |
| import shutil |
| import subprocess |
| import sys |
| import time |
| import uuid |
| import wave |
| from pathlib import Path |
| from typing import Any |
|
|
| try: |
| import matplotlib |
|
|
| matplotlib.use("Agg") |
| except ImportError: |
| pass |
|
|
| import gradio as gr |
| import requests |
| import spaces |
| import torch |
| import websocket |
|
|
| from scripts.bootstrap_comfy import patch_melband_loader, validate_melband_model |
| from scripts.workflow_client import load_workflow, patch_voicegate_workflow |
|
|
|
|
| ROOT = Path(__file__).resolve().parent |
| COMFY_DIR = ROOT / "ComfyUI" |
| COMFY_INPUT_DIR = COMFY_DIR / "input" |
| COMFY_LOG = Path("/tmp/voicegate_comfy_gradio.log") |
| COMFY_URL = "http://127.0.0.1:8188" |
| COMFY_HOST = "127.0.0.1" |
| COMFY_PORT = "8188" |
|
|
| COMFY_PROCESS: subprocess.Popen | None = None |
| PREPARE_PROCESS: subprocess.Popen | None = None |
| BOOTSTRAPPED = False |
| MODELS_VALIDATED = False |
| BOOTSTRAP_LOG = Path("/tmp/voicegate_bootstrap.log") |
| USER_OUTPUT_DIR = ROOT / "user_outputs" |
| REQUIRED_MODEL_PATHS = [ |
| COMFY_DIR / "models" / "diffusion_models" / "MelBandRoFormer_comfy" / "MelBandRoformer_fp32.safetensors", |
| COMFY_DIR / "models" / "voxcpm" / "VoxCPM2" / "model.safetensors", |
| COMFY_DIR / "models" / "voxcpm" / "VoxCPM2" / "audiovae.pth", |
| COMFY_DIR / "models" / "Qwen3-ASR" / "Qwen3-ASR-1.7B", |
| COMFY_DIR / "models" / "Qwen3-ASR" / "Qwen3-ForcedAligner-0.6B", |
| ] |
| TARGET_LANGUAGES = [ |
| "Arabic", |
| "Burmese", |
| "Chinese", |
| "Danish", |
| "Dutch", |
| "English", |
| "Finnish", |
| "French", |
| "German", |
| "Greek", |
| "Hebrew", |
| "Hindi", |
| "Indonesian", |
| "Italian", |
| "Japanese", |
| "Khmer", |
| "Korean", |
| "Lao", |
| "Malay", |
| "Norwegian", |
| "Polish", |
| "Portuguese", |
| "Russian", |
| "Spanish", |
| "Swahili", |
| "Swedish", |
| "Tagalog", |
| "Thai", |
| "Turkish", |
| "Vietnamese", |
| ] |
| VG_PRIMARY = "#6366c7" |
| VG_WAVEFORM = "#98a2b3" |
|
|
| VOICEGATE_WAVEFORM_OPTIONS = gr.WaveformOptions( |
| waveform_color=VG_WAVEFORM, |
| waveform_progress_color=VG_PRIMARY, |
| ) |
|
|
| APP_CSS = """ |
| :root { |
| --vg-primary: #6366c7; |
| --vg-primary-dark: #5255b5; |
| --vg-ink: #171827; |
| --vg-muted: #667085; |
| --vg-line: #eceef5; |
| --vg-soft: #f6f7fb; |
| --vg-radius: 8px; |
| --vg-radius-sm: 6px; |
| } |
| :root:root:root:root main { |
| max-width: 1160px; |
| margin-left: auto !important; |
| margin-right: auto !important; |
| } |
| :root:root:root:root .gradio-container { |
| overflow: unset; |
| } |
| .voicegate-shell { |
| gap: 16px; |
| } |
| .voicegate-card { |
| background: #ffffff; |
| border: 1px solid var(--vg-line); |
| border-radius: var(--vg-radius) !important; |
| padding: 12px; |
| box-shadow: none; |
| overflow: hidden; |
| } |
| |
| /* Gradio may attach elem_classes to an outer wrapper while the visible block is a |
| child element. Apply the same rounded corner to both so the final rendered card |
| never appears square. */ |
| .voicegate-card.block, |
| .voicegate-card > .block, |
| .voicegate-card > div, |
| .voicegate-card > div > .block { |
| border-radius: var(--vg-radius) !important; |
| overflow: hidden; |
| } |
| .voicegate-intro { |
| margin: 10px 0 12px; |
| padding: 18px; |
| border-color: rgba(99, 102, 199, 0.24); |
| background: linear-gradient(180deg, #ffffff 0%, #f8f8ff 100%); |
| } |
| .voicegate-kicker { |
| color: var(--vg-primary); |
| font-size: 12px; |
| font-weight: 700; |
| letter-spacing: 0; |
| text-transform: uppercase; |
| } |
| .voicegate-intro h1 { |
| margin: 6px 0 8px; |
| color: var(--vg-ink); |
| font-size: 30px; |
| line-height: 1.12; |
| letter-spacing: 0; |
| } |
| .voicegate-intro p { |
| max-width: none; |
| width: 100%; |
| margin: 0; |
| color: var(--vg-muted); |
| font-size: 14px; |
| line-height: 1.6; |
| } |
| .voicegate-link-row { |
| display: flex; |
| flex-wrap: wrap; |
| gap: 8px; |
| margin-top: 14px; |
| } |
| .voicegate-link-row a { |
| display: inline-flex; |
| min-height: 34px; |
| align-items: center; |
| justify-content: center; |
| border: 1px solid rgba(99, 102, 199, 0.34); |
| border-radius: var(--vg-radius-sm); |
| padding: 6px 12px; |
| color: var(--vg-primary) !important; |
| background: #ffffff; |
| font-size: 13px; |
| font-weight: 650; |
| text-decoration: none; |
| } |
| .voicegate-link-row a:hover { |
| border-color: var(--vg-primary); |
| background: #f4f4ff; |
| } |
| .voicegate-link-row a.voicegate-github { |
| border-color: var(--vg-primary); |
| background: var(--vg-primary); |
| color: #ffffff !important; |
| } |
| .voicegate-link-row a.voicegate-github:hover { |
| border-color: var(--vg-primary-dark); |
| background: var(--vg-primary-dark); |
| } |
| .voicegate-card-label { |
| display: inline-flex; |
| align-items: center; |
| margin: 0 0 10px; |
| border-radius: var(--vg-radius-sm); |
| padding: 5px 8px; |
| background: #ececf1; |
| color: var(--vg-ink); |
| font-size: 12px; |
| font-weight: 700; |
| letter-spacing: 0; |
| text-transform: uppercase; |
| } |
| .voicegate-card-label .voicegate-tag { |
| margin-left: 8px; |
| border-radius: 999px; |
| padding: 2px 7px; |
| color: var(--vg-primary); |
| background: #ffffff; |
| font-size: 12px; |
| font-weight: 700; |
| text-transform: none; |
| } |
| |
| /* Keep only the outer VoiceGate card. Gradio generates many nested blocks/forms; |
| these rules prevent each nested wrapper from drawing another visible box. */ |
| .voicegate-card .block, |
| .voicegate-card .form, |
| .voicegate-card .panel, |
| .voicegate-card .accordion, |
| .voicegate-card .tabs, |
| .voicegate-card .tabitem { |
| border: 0 !important; |
| box-shadow: none !important; |
| background: transparent !important; |
| } |
| .voicegate-card .block { |
| padding-left: 0 !important; |
| padding-right: 0 !important; |
| } |
| .voicegate-card textarea, |
| .voicegate-card input, |
| .voicegate-card select { |
| border: 0 !important; |
| box-shadow: none !important; |
| } |
| .voicegate-card textarea { |
| font-size: 13px; |
| } |
| |
| /* Match FaceFusion-like softly rounded inner controls without adding extra boxes. */ |
| .voicegate-card input, |
| .voicegate-card textarea, |
| .voicegate-card select, |
| .voicegate-card button, |
| .voicegate-card .wrap, |
| .voicegate-card .container, |
| .voicegate-card .input-container, |
| .voicegate-card .dropdown-arrow, |
| .voicegate-card details, |
| .voicegate-card details > summary { |
| border-radius: var(--vg-radius-sm) !important; |
| } |
| |
| /* Rounded corners for visible component cards such as Upload audio and Target language. |
| Gradio applies elem_classes to a wrapper, so radius must also be pushed into |
| the rendered block and its inner containers. */ |
| .voicegate-control-card, |
| .voicegate-control-card.block, |
| .voicegate-control-card > .block, |
| .voicegate-control-card > div, |
| .voicegate-control-card > div > .block, |
| .voicegate-control-card .wrap, |
| .voicegate-control-card .container, |
| .voicegate-control-card .input-container { |
| border-radius: var(--vg-radius) !important; |
| overflow: hidden !important; |
| } |
| |
| .voicegate-control-card .block, |
| .voicegate-control-card .form { |
| border-radius: var(--vg-radius) !important; |
| } |
| |
| .voicegate-control-card input, |
| .voicegate-control-card textarea, |
| .voicegate-control-card select, |
| .voicegate-control-card button { |
| border-radius: var(--vg-radius-sm) !important; |
| } |
| |
| /* Rounded accordion cards: Advanced audio cleanup, Subtitle preview, and Log. |
| Keep them visually light, but give the expanded sections the same soft radius as |
| Upload audio and Target language. */ |
| .voicegate-accordion-card, |
| .voicegate-accordion-card.block, |
| .voicegate-accordion-card > .block, |
| .voicegate-accordion-card > div, |
| .voicegate-accordion-card > div > .block, |
| .voicegate-accordion-card details { |
| border-radius: var(--vg-radius) !important; |
| overflow: hidden !important; |
| } |
| |
| .voicegate-accordion-card details { |
| border: 1px solid var(--vg-line) !important; |
| background: #ffffff !important; |
| box-shadow: none !important; |
| } |
| |
| .voicegate-accordion-card details > summary { |
| border-radius: var(--vg-radius) var(--vg-radius) 0 0 !important; |
| padding: 10px 12px !important; |
| background: var(--vg-soft) !important; |
| box-shadow: none !important; |
| } |
| |
| .voicegate-accordion-card details:not([open]) > summary { |
| border-radius: var(--vg-radius) !important; |
| } |
| |
| .voicegate-accordion-card details[open] > summary { |
| border-bottom: 1px solid var(--vg-line) !important; |
| } |
| |
| /* The content rendered inside an open accordion can have its own Gradio wrappers. |
| Round those wrappers too so textboxes/sliders do not look square inside. */ |
| .voicegate-accordion-card .block, |
| .voicegate-accordion-card .form, |
| .voicegate-accordion-card .wrap, |
| .voicegate-accordion-card .container, |
| .voicegate-accordion-card .input-container, |
| .voicegate-accordion-card textarea, |
| .voicegate-accordion-card input, |
| .voicegate-accordion-card select { |
| border-radius: var(--vg-radius-sm) !important; |
| } |
| |
| /* Full-width primary action without an extra gr.Group wrapper. */ |
| .voicegate-run-button, |
| .voicegate-run-button button, |
| button.voicegate-run-button { |
| width: 100%; |
| } |
| .voicegate-run-button button.primary, |
| .voicegate-run-button .primary, |
| button.voicegate-run-button.primary { |
| background: var(--vg-primary) !important; |
| border-color: var(--vg-primary) !important; |
| color: #ffffff !important; |
| } |
| .voicegate-run-button button.primary:hover, |
| .voicegate-run-button .primary:hover, |
| button.voicegate-run-button.primary:hover { |
| background: var(--vg-primary-dark) !important; |
| border-color: var(--vg-primary-dark) !important; |
| } |
| .voicegate-downloads { |
| gap: 10px; |
| } |
| .voicegate-downloads button, |
| .voicegate-downloads a { |
| width: 100%; |
| } |
| .voicegate-status textarea { |
| font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; |
| font-size: 12px; |
| } |
| :root:root:root:root input[type="range"] { |
| accent-color: var(--vg-primary); |
| } |
| :root:root:root:root input[type="range"]::-moz-range-thumb, |
| :root:root:root:root input[type="range"]::-webkit-slider-thumb { |
| background: var(--vg-primary); |
| box-shadow: none; |
| } |
| :root:root:root:root .tab-container button.selected, |
| :root:root:root:root button[role="tab"][aria-selected="true"] { |
| color: var(--vg-primary); |
| border-color: var(--vg-primary); |
| } |
| :root:root:root:root footer { |
| display: none; |
| } |
| @media (max-width: 760px) { |
| .voicegate-intro h1 { |
| font-size: 26px; |
| } |
| .voicegate-link-row a { |
| flex: 1 1 46%; |
| } |
| } |
| """ |
|
|
| def gpu_status_lines() -> list[str]: |
| lines = ["VoiceGate GPU status"] |
| lines.append(f"torch={torch.__version__}") |
| lines.append(f"cuda_available={torch.cuda.is_available()}") |
| lines.append(f"cuda_device_count={torch.cuda.device_count()}") |
| if torch.cuda.is_available(): |
| props = torch.cuda.get_device_properties(0) |
| lines.append(f"device_name={torch.cuda.get_device_name(0)}") |
| lines.append(f"total_memory_gb={props.total_memory / 1024**3:.2f}") |
| return lines |
|
|
|
|
| def voicegate_theme() -> gr.Theme: |
| primary = gr.themes.Color( |
| name="voicegate", |
| c50="#f5f5ff", |
| c100="#ececff", |
| c200="#dadaff", |
| c300="#b8b9fb", |
| c400="#9193ee", |
| c500="#6366c7", |
| c600="#5255b5", |
| c700="#444695", |
| c800="#393b78", |
| c900="#313262", |
| c950="#1f2040", |
| ) |
| return gr.themes.Base( |
| primary_hue=primary, |
| secondary_hue=gr.themes.colors.neutral, |
| radius_size=gr.themes.sizes.radius_md, |
| font=[gr.themes.GoogleFont("Open Sans"), "ui-sans-serif", "system-ui", "sans-serif"], |
| ).set( |
| background_fill_primary="*neutral_100", |
| background_fill_secondary="*neutral_50", |
| block_background_fill="white", |
| block_border_width="0", |
| block_label_background_fill="*neutral_100", |
| block_label_border_width="none", |
| block_label_margin="0.5rem", |
| block_label_radius="*radius_sm", |
| block_label_text_color="*neutral_700", |
| block_label_text_size="*text_sm", |
| block_label_text_weight="600", |
| block_padding="0.5rem", |
| border_color_primary="transparent", |
| button_primary_background_fill="*primary_500", |
| button_primary_background_fill_hover="*primary_600", |
| button_primary_text_color="white", |
| input_background_fill="*neutral_50", |
| shadow_drop="none", |
| slider_color="*primary_500", |
| ) |
|
|
|
|
| def wait_for_comfy(timeout: float = 180) -> dict[str, Any]: |
| deadline = time.time() + timeout |
| last_error = "" |
| while time.time() < deadline: |
| try: |
| response = requests.get(f"{COMFY_URL}/system_stats", timeout=5) |
| if response.ok: |
| return response.json() |
| last_error = f"HTTP {response.status_code}: {response.text[:300]}" |
| except requests.RequestException as exc: |
| last_error = repr(exc) |
| time.sleep(2) |
| raise RuntimeError(f"ComfyUI did not become ready: {last_error}") |
|
|
|
|
| def run_bootstrap(lines: list[str], *, allow_heavy: bool = True) -> None: |
| global BOOTSTRAPPED |
|
|
| if BOOTSTRAPPED and (COMFY_DIR / "main.py").exists(): |
| lines.append("bootstrap=already_done") |
| return |
| if (COMFY_DIR / "main.py").exists() and (COMFY_DIR / "custom_nodes").exists(): |
| if not allow_heavy: |
| lines.append("bootstrap=existing_comfyui") |
| BOOTSTRAPPED = True |
| return |
|
|
| started = time.time() |
| lines.append("bootstrap=starting") |
| command = [sys.executable, str(ROOT / "scripts" / "bootstrap_comfy.py")] |
| result = subprocess.run( |
| command, |
| cwd=ROOT, |
| text=True, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, |
| timeout=900, |
| ) |
| lines.append(f"bootstrap_returncode={result.returncode}") |
| lines.append(f"bootstrap_elapsed_sec={time.time() - started:.1f}") |
| if result.returncode != 0: |
| lines.append("bootstrap_tail:") |
| lines.extend(result.stdout.splitlines()[-80:]) |
| raise RuntimeError("bootstrap_comfy.py failed") |
| BOOTSTRAPPED = True |
|
|
|
|
| def missing_required_models() -> list[Path]: |
| global MODELS_VALIDATED |
|
|
| missing = [path for path in REQUIRED_MODEL_PATHS if not path.exists()] |
| if missing: |
| MODELS_VALIDATED = False |
| return missing |
| if not MODELS_VALIDATED: |
| melband_valid, _reason = validate_melband_model(verify_hash=True) |
| if not melband_valid: |
| return [REQUIRED_MODEL_PATHS[0]] |
| MODELS_VALIDATED = True |
| return [] |
|
|
|
|
| def ensure_runtime_assets(lines: list[str]) -> None: |
| missing = missing_required_models() |
| if not missing: |
| lines.append("models=ready") |
| return |
|
|
| lines.append("models=missing") |
| lines.extend(f"missing_model={path}" for path in missing) |
| started = time.time() |
| command = [sys.executable, str(ROOT / "scripts" / "bootstrap_comfy.py"), "--with-models"] |
| result = subprocess.run( |
| command, |
| cwd=ROOT, |
| text=True, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT, |
| timeout=1800, |
| ) |
| lines.append(f"model_prepare_returncode={result.returncode}") |
| lines.append(f"model_prepare_elapsed_sec={time.time() - started:.1f}") |
| if result.returncode != 0: |
| lines.append("model_prepare_tail:") |
| lines.extend(result.stdout.splitlines()[-100:]) |
| raise RuntimeError("Could not prepare required VoiceGate models.") |
| remaining = missing_required_models() |
| if remaining: |
| lines.append("models_still_missing:") |
| lines.extend(str(path) for path in remaining) |
| raise RuntimeError("Required VoiceGate models are still missing after preparation.") |
| lines.append("models=ready_after_prepare") |
|
|
|
|
| def ensure_comfy(lines: list[str], *, timeout: float = 240) -> dict[str, Any]: |
| global COMFY_PROCESS |
|
|
| if PREPARE_PROCESS is not None: |
| returncode = PREPARE_PROCESS.poll() |
| if returncode is None: |
| raise RuntimeError("Runtime preparation is still running. Check Prepare Status first.") |
| if returncode != 0: |
| raise RuntimeError(f"Runtime preparation failed with return code {returncode}.") |
|
|
| run_bootstrap(lines, allow_heavy=False) |
| patch_melband_loader() |
|
|
| try: |
| stats = wait_for_comfy(timeout=5) |
| lines.append("comfy=already_running") |
| return stats |
| except RuntimeError: |
| pass |
|
|
| log = COMFY_LOG.open("ab") |
| command = [ |
| sys.executable, |
| "main.py", |
| "--listen", |
| COMFY_HOST, |
| "--port", |
| COMFY_PORT, |
| ] |
| COMFY_PROCESS = subprocess.Popen( |
| command, |
| cwd=COMFY_DIR, |
| stdout=log, |
| stderr=subprocess.STDOUT, |
| ) |
| lines.append(f"comfy_started_pid={COMFY_PROCESS.pid}") |
| try: |
| return wait_for_comfy(timeout=timeout) |
| except Exception: |
| lines.append("comfy_log_tail:") |
| if COMFY_LOG.exists(): |
| lines.extend(COMFY_LOG.read_text(encoding="utf-8", errors="replace").splitlines()[-120:]) |
| raise |
|
|
|
|
| def write_sine_wav(filename: str, *, seconds: float = 1.0, frequency: float = 440.0) -> str: |
| COMFY_INPUT_DIR.mkdir(parents=True, exist_ok=True) |
| path = COMFY_INPUT_DIR / filename |
| sample_rate = 16000 |
| total = int(sample_rate * seconds) |
| amplitude = 0.2 |
| with wave.open(str(path), "wb") as file: |
| file.setnchannels(1) |
| file.setsampwidth(2) |
| file.setframerate(sample_rate) |
| for index in range(total): |
| value = int(32767 * amplitude * math.sin(2 * math.pi * frequency * index / sample_rate)) |
| file.writeframesraw(value.to_bytes(2, byteorder="little", signed=True)) |
| return filename |
|
|
|
|
| def submit_prompt(workflow: dict[str, Any], *, client_id: str | None = None) -> str: |
| response = requests.post( |
| f"{COMFY_URL}/prompt", |
| json={"prompt": workflow, "client_id": client_id or str(uuid.uuid4())}, |
| timeout=120, |
| ) |
| if not response.ok: |
| raise RuntimeError(f"/prompt failed HTTP {response.status_code}: {response.text[:2000]}") |
| return response.json()["prompt_id"] |
|
|
|
|
| def execute_prompt_with_timing(workflow: dict[str, Any], *, timeout: float) -> tuple[str, dict[str, Any], list[str]]: |
| client_id = str(uuid.uuid4()) |
| websocket_url = f"ws://{COMFY_HOST}:{COMFY_PORT}/ws?clientId={client_id}" |
| ws = websocket.create_connection(websocket_url, timeout=30) |
| prompt_id = submit_prompt(workflow, client_id=client_id) |
| started = time.time() |
| deadline = started + timeout |
| current_node: str | None = None |
| current_started = 0.0 |
| node_durations: dict[str, float] = {} |
| node_order: list[str] = [] |
| event_lines = [f"prompt_id={prompt_id}", "node_timing=started"] |
|
|
| def close_current_node(now: float) -> None: |
| nonlocal current_node, current_started |
| if current_node is not None: |
| node_durations[current_node] = node_durations.get(current_node, 0.0) + max(0.0, now - current_started) |
| current_node = None |
| current_started = 0.0 |
|
|
| try: |
| while time.time() < deadline: |
| ws.settimeout(max(1.0, min(10.0, deadline - time.time()))) |
| try: |
| message = ws.recv() |
| except websocket.WebSocketTimeoutException: |
| continue |
| if isinstance(message, bytes): |
| message = message.decode("utf-8", errors="replace") |
| try: |
| payload = json.loads(message) |
| except json.JSONDecodeError: |
| continue |
| event_type = payload.get("type") |
| data = payload.get("data") or {} |
| if data.get("prompt_id") not in (None, prompt_id): |
| continue |
|
|
| now = time.time() |
| if event_type == "executing": |
| close_current_node(now) |
| node = data.get("node") |
| if node is None: |
| continue |
| current_node = str(node) |
| current_started = now |
| if current_node not in node_order: |
| node_order.append(current_node) |
| elif event_type == "execution_success": |
| close_current_node(now) |
| event_lines.append(f"websocket_elapsed_sec={now - started:.1f}") |
| break |
| elif event_type == "execution_error": |
| close_current_node(now) |
| event_lines.append("websocket_execution_error:") |
| event_lines.append(json.dumps(data, ensure_ascii=False, indent=2)[:4000]) |
| break |
| else: |
| close_current_node(time.time()) |
| raise TimeoutError(f"Timed out waiting for prompt {prompt_id}") |
| finally: |
| ws.close() |
|
|
| history = wait_for_history(prompt_id, timeout=30) |
| timed_nodes = sorted( |
| ((node_id, node_durations.get(node_id, 0.0)) for node_id in node_order), |
| key=lambda item: item[1], |
| reverse=True, |
| ) |
| if timed_nodes: |
| event_lines.append("node_timing_top:") |
| for node_id, seconds in timed_nodes[:20]: |
| class_type = workflow.get(node_id, {}).get("class_type", "unknown") |
| event_lines.append(f"{node_id} {class_type}: {seconds:.1f}s") |
| return prompt_id, history, event_lines |
|
|
|
|
| def wait_for_history(prompt_id: str, timeout: float = 1200) -> dict[str, Any]: |
| deadline = time.time() + timeout |
| while time.time() < deadline: |
| response = requests.get(f"{COMFY_URL}/history/{prompt_id}", timeout=30) |
| response.raise_for_status() |
| payload = response.json() |
| if prompt_id in payload: |
| return payload[prompt_id] |
| time.sleep(2) |
| raise TimeoutError(f"Timed out waiting for prompt {prompt_id}") |
|
|
|
|
| def history_summary(history: dict[str, Any]) -> list[str]: |
| lines = [] |
| status = history.get("status", {}) |
| lines.append(f"status_str={status.get('status_str')}") |
| lines.append(f"completed={status.get('completed')}") |
| messages = status.get("messages") or [] |
| errors = [message for message in messages if isinstance(message, list) and message[0] == "execution_error"] |
| if errors: |
| lines.append("errors:") |
| lines.append(json.dumps(errors, ensure_ascii=False, indent=2)[:4000]) |
|
|
| outputs = history.get("outputs", {}) |
| output_files = [] |
| for node_output in outputs.values(): |
| for key in ("audio", "images", "gifs"): |
| for item in node_output.get(key, []) or []: |
| filename = item.get("filename") |
| subfolder = item.get("subfolder") |
| if subfolder: |
| output_files.append(f"{subfolder}/{filename}") |
| elif filename: |
| output_files.append(filename) |
| if output_files: |
| lines.append("outputs:") |
| lines.extend(output_files) |
| text_outputs = [] |
| for node_output in outputs.values(): |
| for key in ("text", "string"): |
| values = node_output.get(key, []) or [] |
| if isinstance(values, str): |
| values = [values] |
| text_outputs.extend(str(value) for value in values) |
| if text_outputs: |
| lines.append("text_outputs:") |
| for value in text_outputs: |
| lines.append(value[:2000]) |
| return lines |
|
|
|
|
| def first_output_audio_path(history: dict[str, Any]) -> str | None: |
| outputs = history.get("outputs", {}) |
| for node_output in outputs.values(): |
| for item in node_output.get("audio", []) or []: |
| filename = item.get("filename") |
| if not filename: |
| continue |
| subfolder = item.get("subfolder") or "" |
| path = COMFY_DIR / "output" / subfolder / filename |
| if path.exists(): |
| return str(path) |
| return None |
|
|
|
|
| def text_outputs_for_node(history: dict[str, Any], node_id: str) -> list[str]: |
| node_output = (history.get("outputs", {}) or {}).get(node_id, {}) |
| values: list[str] = [] |
| for key in ("text", "string"): |
| raw_values = node_output.get(key, []) or [] |
| if isinstance(raw_values, str): |
| raw_values = [raw_values] |
| values.extend(str(value) for value in raw_values if str(value).strip()) |
| return values |
|
|
|
|
| def write_srt_file(prefix: str, name: str, text: str) -> str | None: |
| if not text.strip(): |
| return None |
| USER_OUTPUT_DIR.mkdir(parents=True, exist_ok=True) |
| path = USER_OUTPUT_DIR / f"{prefix}_{name}.srt" |
| path.write_text(text, encoding="utf-8") |
| return str(path) |
|
|
|
|
| def melband_workflow(audio_filename: str, prefix: str) -> dict[str, Any]: |
| return { |
| "1": { |
| "class_type": "LoadAudio", |
| "inputs": {"audio": audio_filename, "audioUI": ""}, |
| }, |
| "2": { |
| "class_type": "MelBandRoFormerModelLoader", |
| "inputs": {"model_name": "MelBandRoFormer_comfy/MelBandRoformer_fp32.safetensors"}, |
| }, |
| "3": { |
| "class_type": "MelBandRoFormerSampler", |
| "inputs": {"model": ["2", 0], "audio": ["1", 0]}, |
| }, |
| "4": { |
| "class_type": "SaveAudioMP3", |
| "inputs": { |
| "filename_prefix": f"audio/{prefix}_vocals", |
| "quality": "V0", |
| "audioUI": "", |
| "audio": ["3", 0], |
| }, |
| }, |
| "5": { |
| "class_type": "SaveAudioMP3", |
| "inputs": { |
| "filename_prefix": f"audio/{prefix}_instruments", |
| "quality": "V0", |
| "audioUI": "", |
| "audio": ["3", 1], |
| }, |
| }, |
| } |
|
|
|
|
| def voxcpm_tts_workflow(prefix: str) -> dict[str, Any]: |
| return { |
| "1": { |
| "class_type": "RunningHub_VoxCPM_LoadModel", |
| "inputs": {"model_name": "VoxCPM2", "optimize": False, "lora_name": "None"}, |
| }, |
| "2": { |
| "class_type": "RunningHub_VoxCPM_Generate", |
| "inputs": { |
| "model": ["1", 0], |
| "control_instruction": "清晰自然的中文女声", |
| "text": "你好，VoiceGate GPU 语音合成测试。", |
| "cfg_value": 2.0, |
| "inference_steps": 4, |
| "seed": 20260605, |
| "ultimate_clone": False, |
| "reference_audio_text": "", |
| "normalize_text": False, |
| "denoise_reference": False, |
| "max_len": 512, |
| "retry_badcase": True, |
| }, |
| }, |
| "3": { |
| "class_type": "SaveAudioMP3", |
| "inputs": { |
| "filename_prefix": f"audio/{prefix}", |
| "quality": "V0", |
| "audioUI": "", |
| "audio": ["2", 0], |
| }, |
| }, |
| } |
|
|
|
|
| def copy_audio_to_comfy_input(audio_path: str | Path, prefix: str) -> str: |
| source = Path(audio_path) |
| if not source.exists(): |
| raise FileNotFoundError(f"Uploaded audio does not exist: {source}") |
| suffix = source.suffix or ".wav" |
| filename = f"{prefix}_{uuid.uuid4().hex[:8]}{suffix}" |
| COMFY_INPUT_DIR.mkdir(parents=True, exist_ok=True) |
| shutil.copyfile(source, COMFY_INPUT_DIR / filename) |
| return filename |
|
|
|
|
| def asr_workflow(audio_filename: str, prefix: str) -> dict[str, Any]: |
| return { |
| "1": { |
| "class_type": "LoadAudio", |
| "inputs": {"audio": audio_filename, "audioUI": ""}, |
| }, |
| "2": { |
| "class_type": "VoiceBridgeASRLoader", |
| "inputs": { |
| "repo_id": "Qwen/Qwen3-ASR-1.7B", |
| "source": "HuggingFace", |
| "precision": "bf16", |
| "attention": "sdpa", |
| "max_new_tokens": 256, |
| "forced_aligner": "Qwen/Qwen3-ForcedAligner-0.6B", |
| "local_model_path_asr": "", |
| "local_model_path_fa": "", |
| }, |
| }, |
| "3": { |
| "class_type": "VoiceBridgeASRTranscribe", |
| "inputs": { |
| "model_key": ["2", 0], |
| "audio": ["1", 0], |
| "language": "auto", |
| "context": "", |
| "return_timestamps": True, |
| }, |
| }, |
| "4": { |
| "class_type": "GenerateSRT", |
| "inputs": { |
| "forced_aligns": ["3", 0], |
| "text": ["3", 1], |
| "language": ["3", 2], |
| "save_srt": True, |
| "filename_prefix": f"VoiceBridge/{prefix}", |
| }, |
| }, |
| "5": { |
| "class_type": "easy showAnything", |
| "inputs": { |
| "text": "", |
| "anything": ["4", 0], |
| }, |
| }, |
| } |
|
|
|
|
| def full_voicegate_workflow( |
| audio_filename: str, |
| prefix: str, |
| target_language: str, |
| *, |
| tts_trim_start: float, |
| ) -> dict[str, Any]: |
| workflow = load_workflow() |
| return patch_voicegate_workflow( |
| workflow, |
| audio_filename=audio_filename, |
| target_language=target_language, |
| api_key=os.environ.get("DEEPSEEK_API_KEY"), |
| api_baseurl=os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com"), |
| llm_model=os.environ.get("DEEPSEEK_MODEL", "deepseek-v4-flash"), |
| job_id=prefix, |
| tts_trim_start=tts_trim_start, |
| ) |
|
|
|
|
| def run_full_voicegate( |
| audio_path: str | None, |
| target_language: str, |
| *, |
| tts_trim_start: float = 0.0, |
| timeout: float = 880, |
| ) -> dict[str, Any]: |
| lines = gpu_status_lines() |
| started = time.time() |
| trim_start = min(1.0, max(0.0, float(tts_trim_start))) |
| if not audio_path: |
| raise ValueError("Please upload an audio file before running VoiceGate.") |
| if not os.environ.get("DEEPSEEK_API_KEY"): |
| raise RuntimeError("DEEPSEEK_API_KEY is not configured in the Space.") |
| ensure_runtime_assets(lines) |
| ensure_comfy(lines) |
| prefix = f"full_{uuid.uuid4().hex[:8]}" |
| audio_filename = copy_audio_to_comfy_input(audio_path, prefix) |
| lines.append(f"input_audio={audio_filename}") |
| lines.append(f"target_language={target_language}") |
| lines.append(f"tts_trim_start={trim_start}") |
| prompt = full_voicegate_workflow( |
| audio_filename, |
| prefix, |
| target_language or "English", |
| tts_trim_start=trim_start, |
| ) |
| _prompt_id, history, timing_lines = execute_prompt_with_timing(prompt, timeout=timeout) |
| lines.extend(timing_lines) |
| lines.extend(history_summary(history)) |
| output_audio = first_output_audio_path(history) |
| source_subtitle = "\n\n".join(text_outputs_for_node(history, "61")) |
| translated_subtitle = "\n\n".join(text_outputs_for_node(history, "179") or text_outputs_for_node(history, "107")) |
| source_subtitle_file = write_srt_file(prefix, "source", source_subtitle) |
| translated_subtitle_file = write_srt_file(prefix, "translated", translated_subtitle) |
| if output_audio: |
| lines.append(f"output_audio_path={output_audio}") |
| if source_subtitle_file: |
| lines.append(f"source_subtitle_file={source_subtitle_file}") |
| if translated_subtitle_file: |
| lines.append(f"translated_subtitle_file={translated_subtitle_file}") |
| lines.append(f"elapsed_sec={time.time() - started:.1f}") |
| return { |
| "lines": lines, |
| "audio": output_audio, |
| "source_subtitle": source_subtitle, |
| "translated_subtitle": translated_subtitle, |
| "source_subtitle_file": source_subtitle_file, |
| "translated_subtitle_file": translated_subtitle_file, |
| } |
|
|
|
|
| def prepare_runtime() -> str: |
| global PREPARE_PROCESS |
|
|
| lines = ["VoiceGate runtime preparation"] |
| if PREPARE_PROCESS is not None and PREPARE_PROCESS.poll() is None: |
| lines.append(f"prepare=already_running pid={PREPARE_PROCESS.pid}") |
| return "\n".join(lines) |
| BOOTSTRAP_LOG.parent.mkdir(parents=True, exist_ok=True) |
| log = BOOTSTRAP_LOG.open("ab") |
| command = [sys.executable, str(ROOT / "scripts" / "bootstrap_comfy.py"), "--with-models"] |
| PREPARE_PROCESS = subprocess.Popen( |
| command, |
| cwd=ROOT, |
| stdout=log, |
| stderr=subprocess.STDOUT, |
| ) |
| lines.append(f"prepare=started pid={PREPARE_PROCESS.pid}") |
| lines.append(f"log={BOOTSTRAP_LOG}") |
| return "\n".join(lines) |
|
|
|
|
| def prepare_status() -> str: |
| global BOOTSTRAPPED |
|
|
| lines = ["VoiceGate runtime preparation status"] |
| if PREPARE_PROCESS is None: |
| lines.append("prepare=not_started") |
| else: |
| returncode = PREPARE_PROCESS.poll() |
| if returncode is None: |
| lines.append(f"prepare=running pid={PREPARE_PROCESS.pid}") |
| else: |
| lines.append(f"prepare=finished returncode={returncode}") |
| if returncode == 0 and (COMFY_DIR / "main.py").exists(): |
| BOOTSTRAPPED = True |
| lines.append(f"comfy_dir_exists={(COMFY_DIR / 'main.py').exists()}") |
| if BOOTSTRAP_LOG.exists(): |
| lines.append("bootstrap_log_tail:") |
| lines.extend(BOOTSTRAP_LOG.read_text(encoding="utf-8", errors="replace").splitlines()[-80:]) |
| return "\n".join(lines) |
|
|
|
|
| @spaces.GPU(duration=60) |
| def gpu_smoke_test() -> str: |
| lines = gpu_status_lines() |
| if torch.cuda.is_available(): |
| tensor = torch.arange(16, device="cuda:0", dtype=torch.float32) |
| result = (tensor * 2).sum().item() |
| torch.cuda.synchronize() |
| lines.append(f"tensor_result={result}") |
| lines.append(f"memory_reserved_mb={torch.cuda.memory_reserved(0) / 1024**2:.2f}") |
| return "\n".join(lines) |
|
|
|
|
| @spaces.GPU(duration=900) |
| def comfy_runtime_test() -> str: |
| lines = gpu_status_lines() |
| started = time.time() |
| try: |
| stats = ensure_comfy(lines) |
| lines.append(f"comfy_ready=true") |
| lines.append(f"comfy_elapsed_sec={time.time() - started:.1f}") |
| lines.append("system_stats:") |
| lines.append(json.dumps(stats, ensure_ascii=False, indent=2)[:4000]) |
| except Exception as exc: |
| lines.append(f"error={type(exc).__name__}: {exc}") |
| return "\n".join(lines) |
|
|
|
|
| @spaces.GPU(duration=1200) |
| def melband_gpu_test() -> str: |
| lines = gpu_status_lines() |
| started = time.time() |
| try: |
| ensure_comfy(lines) |
| audio_filename = write_sine_wav(f"voicegate_melband_{uuid.uuid4().hex[:8]}.wav") |
| prefix = f"melband_gpu_{uuid.uuid4().hex[:8]}" |
| prompt_id = submit_prompt(melband_workflow(audio_filename, prefix)) |
| lines.append(f"prompt_id={prompt_id}") |
| history = wait_for_history(prompt_id) |
| lines.extend(history_summary(history)) |
| lines.append(f"elapsed_sec={time.time() - started:.1f}") |
| except Exception as exc: |
| lines.append(f"error={type(exc).__name__}: {exc}") |
| if COMFY_LOG.exists(): |
| lines.append("comfy_log_tail:") |
| lines.extend(COMFY_LOG.read_text(encoding="utf-8", errors="replace").splitlines()[-120:]) |
| return "\n".join(lines) |
|
|
|
|
| @spaces.GPU(duration=1200) |
| def voxcpm_tts_gpu_test() -> str: |
| lines = gpu_status_lines() |
| started = time.time() |
| try: |
| ensure_comfy(lines) |
| prefix = f"voxcpm_tts_gpu_{uuid.uuid4().hex[:8]}" |
| prompt_id = submit_prompt(voxcpm_tts_workflow(prefix)) |
| lines.append(f"prompt_id={prompt_id}") |
| history = wait_for_history(prompt_id, timeout=1200) |
| lines.extend(history_summary(history)) |
| lines.append(f"elapsed_sec={time.time() - started:.1f}") |
| except Exception as exc: |
| lines.append(f"error={type(exc).__name__}: {exc}") |
| if COMFY_LOG.exists(): |
| lines.append("comfy_log_tail:") |
| lines.extend(COMFY_LOG.read_text(encoding="utf-8", errors="replace").splitlines()[-160:]) |
| return "\n".join(lines) |
|
|
|
|
| @spaces.GPU(duration=900) |
| def asr_gpu_test(audio_path: str | None) -> str: |
| lines = gpu_status_lines() |
| started = time.time() |
| try: |
| if not audio_path: |
| raise ValueError("Please upload an audio file before running ASR.") |
| ensure_comfy(lines) |
| prefix = f"asr_gpu_{uuid.uuid4().hex[:8]}" |
| audio_filename = copy_audio_to_comfy_input(audio_path, prefix) |
| lines.append(f"input_audio={audio_filename}") |
| prompt_id = submit_prompt(asr_workflow(audio_filename, prefix)) |
| lines.append(f"prompt_id={prompt_id}") |
| history = wait_for_history(prompt_id, timeout=900) |
| lines.extend(history_summary(history)) |
| lines.append(f"elapsed_sec={time.time() - started:.1f}") |
| except Exception as exc: |
| lines.append(f"error={type(exc).__name__}: {exc}") |
| if COMFY_LOG.exists(): |
| lines.append("comfy_log_tail:") |
| lines.extend(COMFY_LOG.read_text(encoding="utf-8", errors="replace").splitlines()[-180:]) |
| return "\n".join(lines) |
|
|
|
|
| @spaces.GPU(duration=900) |
| def full_voicegate_gpu_test(audio_path: str | None, target_language: str, tts_trim_start: float) -> str: |
| try: |
| result = run_full_voicegate(audio_path, target_language, tts_trim_start=tts_trim_start, timeout=880) |
| lines = result["lines"] |
| except Exception as exc: |
| lines = gpu_status_lines() |
| lines.append(f"error={type(exc).__name__}: {exc}") |
| if COMFY_LOG.exists(): |
| lines.append("comfy_log_tail:") |
| lines.extend(COMFY_LOG.read_text(encoding="utf-8", errors="replace").splitlines()[-220:]) |
| return "\n".join(lines) |
|
|
|
|
| @spaces.GPU(duration=900) |
| def voicegate_user_run(audio_path: str | None, target_language: str, tts_trim_start: float) -> tuple[ |
| str | None, |
| str, |
| str | None, |
| str | None, |
| str, |
| str, |
| ]: |
| try: |
| result = run_full_voicegate( |
| audio_path, |
| target_language, |
| tts_trim_start=tts_trim_start, |
| timeout=880, |
| ) |
| lines = result["lines"] |
| output_audio = result["audio"] |
| if not output_audio: |
| lines.append("warning=No output audio file was found in ComfyUI history.") |
| return ( |
| output_audio, |
| "\n".join(lines), |
| result["source_subtitle_file"], |
| result["translated_subtitle_file"], |
| result["source_subtitle"], |
| result["translated_subtitle"], |
| ) |
| except Exception as exc: |
| lines = gpu_status_lines() |
| lines.append(f"error={type(exc).__name__}: {exc}") |
| if COMFY_LOG.exists(): |
| lines.append("comfy_log_tail:") |
| lines.extend(COMFY_LOG.read_text(encoding="utf-8", errors="replace").splitlines()[-160:]) |
| return None, "\n".join(lines), None, None, "", "" |
|
|
|
|
| with gr.Blocks(title="VoiceGate", fill_width=True) as demo: |
| with gr.Tab("Translate"): |
| gr.HTML( |
| """ |
| <section class="voicegate-card voicegate-intro"> |
| <div class="voicegate-kicker">ComfyUI workflow · multilingual dubbing</div> |
| <h1>VoiceGate</h1> |
| <p> |
| VoiceGate transforms speech clips into precisely time-aligned multilingual dubbing. Each sentence is |
| automatically matched to the original speech timestamp, so the generated voice follows the source |
| rhythm and stays synchronized with the subtitles and video timeline. The pipeline combines ASR, |
| LLM translation, multilingual TTS, SRT-based audio alignment, and ambience preservation to produce |
| natural translated dubbing while keeping the original pacing and background atmosphere. Runtime is |
| usually close to the uploaded audio duration. |
| </p> |
| <div class="voicegate-link-row"> |
| <a class="voicegate-github" href="https://github.com/YanTianlong-01/VoiceGate" target="_blank">GitHub source</a> |
| <a href="https://www.runninghub.ai/ai-detail/2062442306350964737?inviteCode=rh-v1455" target="_blank">Online app - audio</a> |
| <a href="https://www.runninghub.ai/ai-detail/2062446982618238978?inviteCode=rh-v1455" target="_blank">Online app - video</a> |
| <a href="https://www.runninghub.ai/post/2062432233125928961?inviteCode=rh-v1455" target="_blank">ComfyUI workflow - audio</a> |
| <a href="https://www.runninghub.ai/post/2062445363042283522?inviteCode=rh-v1455" target="_blank">ComfyUI workflow - video</a> |
| </div> |
| </section> |
| """ |
| ) |
| with gr.Row(elem_classes=["voicegate-shell"]): |
| with gr.Column(scale=4, min_width=300): |
| with gr.Blocks(elem_classes=["voicegate-card"]): |
| gr.HTML('<div class="voicegate-card-label">Input <span class="voicegate-tag">required</span></div>') |
| user_audio = gr.Audio( |
| label="Upload audio", |
| type="filepath", |
| elem_classes=["voicegate-control-card"], |
| waveform_options=VOICEGATE_WAVEFORM_OPTIONS, |
| ) |
| user_target_language = gr.Dropdown( |
| label="Target language", |
| choices=TARGET_LANGUAGES, |
| value="English", |
| elem_classes=["voicegate-control-card"], |
| ) |
| with gr.Accordion("Advanced audio cleanup", open=False, elem_classes=["voicegate-accordion-card"]): |
| user_tts_trim_start = gr.Slider( |
| label="TTS segment trim start", |
| minimum=0.0, |
| maximum=1.0, |
| value=0.0, |
| step=0.05, |
| info=( |
| "Skips the first n seconds of each generated TTS segment. " |
| "Use this to remove short noises that may appear at the beginning of generated speech segments." |
| ), |
| ) |
| user_run = gr.Button( |
| "Generate translated dubbing", |
| variant="primary", |
| elem_classes=["voicegate-run-button"], |
| ) |
| with gr.Column(scale=8, min_width=420): |
| with gr.Blocks(elem_classes=["voicegate-card"]): |
| gr.HTML('<div class="voicegate-card-label">Output <span class="voicegate-tag">audio + subtitles</span></div>') |
| user_output_audio = gr.Audio( |
| label="Translated dubbing audio", |
| type="filepath", |
| elem_classes=["voicegate-control-card"], |
| waveform_options=VOICEGATE_WAVEFORM_OPTIONS, |
| ) |
| with gr.Row(elem_classes=["voicegate-downloads"]): |
| user_source_file = gr.DownloadButton("Download original subtitles", size="sm") |
| user_translated_file = gr.DownloadButton("Download translated subtitles", size="sm") |
| with gr.Accordion("Subtitle preview", open=True, elem_classes=["voicegate-accordion-card"]): |
| with gr.Row(): |
| user_source_text = gr.Textbox(label="Original subtitles", lines=8) |
| user_translated_text = gr.Textbox(label="Translated subtitles", lines=8) |
| with gr.Blocks(elem_classes=["voicegate-card"]): |
| with gr.Accordion("Log", open=True, elem_classes=["voicegate-accordion-card"]): |
| user_status = gr.Textbox(label="Status", lines=12, elem_classes=["voicegate-status"]) |
| user_run.click( |
| fn=voicegate_user_run, |
| inputs=[user_audio, user_target_language, user_tts_trim_start], |
| outputs=[ |
| user_output_audio, |
| user_status, |
| user_source_file, |
| user_translated_file, |
| user_source_text, |
| user_translated_text, |
| ], |
| ) |
|
|
|
|
|
|
| if __name__ == "__main__": |
| demo.launch(theme=voicegate_theme(), css=APP_CSS) |
|
|
