Spaces:
Running on Zero
Running on Zero
| """Workflow Studio: local conditioning and generation in one Space and API call.""" | |
| from __future__ import annotations | |
| import base64 | |
| import faulthandler | |
| import gc | |
| import hashlib | |
| import json | |
| import os | |
| import shutil | |
| import signal | |
| import sys | |
| import threading | |
| import time | |
| import uuid | |
| from pathlib import Path | |
| # Import before anything that can initialize CUDA. | |
| try: | |
| import spaces | |
| except ImportError: | |
| spaces = None | |
| import gradio as gr | |
| from canvas_bridge import (get_oauth_available, get_oauth_scopes, get_token, make_api_fn, make_call_fn, | |
| make_media_fn, make_palette_fn) | |
| import media_tools | |
| import worker_probe | |
| from workflow_templates import function_palette, list_templates, list_workflow_templates | |
| from preset_core import list_presets, plan_reference_request, plan_request | |
| from studio_service import (initialization_blockers, reservation_seconds, | |
| conditioning_reservation_seconds, validate_local_conditioning, | |
| conditioning_text_rows_bound, workflow_reservation_seconds, | |
| reference_conditioning_seconds, reference_generation_seconds, | |
| reference_rows, reference_text_rows_bound, reference_workflow_seconds, | |
| COLD_DOWNLOAD_ALLOWANCE_S, REFERENCE_IMAGE_SHORT_EDGE) | |
| MODEL_REPO = "MiniMaxAI/MiniMax-H3" | |
| MODEL_REVISION = "42ed227ee7df40d41602854ae760620d6eb651fe" | |
| ADAPTER_REVISIONS = { | |
| "larryvrh/MiniMax-H3-Turbo-Lora": "43a74557ac3f6539db8e0f2a959d03feb7a81480", | |
| "lightx2v/Minimax-h3-Turbo": "30f485ed874aa44c8183348ad361c5eabad318da", | |
| } | |
| OUTPUT_DIR = Path(os.environ.get("WORKFLOW_OUTPUT_DIR", "outputs")).resolve() | |
| MODEL_CACHE_DIR = Path(os.environ.get("WORKFLOW_MODEL_CACHE", ".workflow-models")).resolve() | |
| MODEL_STORE = None | |
| LOAD_ERROR = None | |
| RUN_LOCK = threading.Lock() | |
| def blockers() -> list[str]: | |
| result = initialization_blockers(os.environ) | |
| if not os.environ.get("SPACE_ID") or spaces is None: | |
| result.append("hosted_zerogpu_required") | |
| if LOAD_ERROR: | |
| result.append(LOAD_ERROR) | |
| return result | |
| def health() -> dict: | |
| """Readiness is not an inference or quality claim.""" | |
| reasons = blockers() | |
| return { | |
| "schema_version": 2, "status": "ready" if MODEL_STORE is not None and not reasons else "not_ready", | |
| "runtime_initialized": MODEL_STORE is not None, | |
| "model_loaded": False, "model_loading": "staged_local", "blockers": reasons, | |
| "execution": "eager", "compiled": False, "inference_verified": False, | |
| "model_repo": MODEL_REPO, "model_revision": MODEL_REVISION, | |
| "conditioning_transport": "in_process", | |
| "workflow_components": ["qwen3_vl_conditioning", "h3_fl2va_generation"], | |
| "busy": RUN_LOCK.locked(), | |
| } | |
| def catalog() -> dict: | |
| return {"schema_version": 2, "presets": list_presets(), "runtime": health()} | |
| def plan(preset_id: str, prompt: str, canvas: str = "", duration: float = 5, | |
| seed: int = 42, has_first: bool = False, has_last: bool = False) -> dict: | |
| try: | |
| return plan_request(preset_id, prompt, canvas, duration, seed, has_first, has_last) | |
| except ValueError as error: | |
| # Planner errors are fixed validation messages, safe to expose without | |
| # enabling verbose provider/runtime exception reporting. | |
| raise gr.Error(str(error)) from None | |
| def initialize() -> None: | |
| global MODEL_STORE, LOAD_ERROR | |
| if blockers(): | |
| return | |
| try: | |
| # Import actual runtime dependencies, but never download or pack weights | |
| # during startup. Each stage owns its own bounded, temporary checkpoint. | |
| from h3_split_blocks import MiniMaxH3ConditionerBlocks, MiniMaxH3GeneratorBlocks | |
| from local_runtime import StageStore | |
| os.environ.setdefault("HF_XET_CHUNK_CACHE_SIZE_BYTES", "0") | |
| MODEL_STORE = StageStore(MODEL_CACHE_DIR, MODEL_REPO, MODEL_REVISION) | |
| except Exception as error: | |
| LOAD_ERROR = "runtime_init_failed_" + type(error).__name__ | |
| print(LOAD_ERROR, flush=True) | |
| # ZeroGPU reuses an idle worker process for the next call of the same GPU | |
| # function. Every GPU job ends its worker shortly after returning, and the | |
| # parent holds RUN_LOCK until then, so each job forks a fresh worker with no | |
| # pooled connections, model references or staged files left by an earlier job. | |
| _PARENT_PID = os.getpid() | |
| _WORKER_RETIRE_DELAY_S = 1.0 | |
| # Only hosted ZeroGPU forks workers; elsewhere nothing needs to wait or watch. | |
| _HOSTED_ZEROGPU = spaces is not None and bool(os.environ.get("SPACE_ID")) | |
| _WORKER_RETIRE_HOLD_S = 2.5 if _HOSTED_ZEROGPU else 0 | |
| # While a GPU call is pending the parent logs its workers' /proc state and | |
| # SIGUSR1-dumps a stalled one (worker_probe). | |
| _WORKER_WATCH_S = 45 if _HOSTED_ZEROGPU else 0 | |
| def _suppress_core_dumps(): | |
| """A crashed worker otherwise spends its GPU allocation writing a core dump. | |
| A shot-2 worker crashed seconds into a job and sat in vfs_coredump for ~300 s | |
| until ZeroGPU aborted it (2026-10-23). Limit 1, not 0: the kernel also skips | |
| piped core handlers (core_pattern "|...") at exactly 1. | |
| """ | |
| import resource | |
| _, hard = resource.getrlimit(resource.RLIMIT_CORE) | |
| limit = 1 if hard == resource.RLIM_INFINITY or hard >= 1 else 0 | |
| resource.setrlimit(resource.RLIMIT_CORE, (limit, hard)) | |
| return limit | |
| if _HOSTED_ZEROGPU: | |
| # Forked workers inherit all three: a fatal signal prints every Python | |
| # thread's stack, SIGUSR1 does the same on demand, and no core is written. | |
| faulthandler.enable(file=sys.stderr, all_threads=True) | |
| faulthandler.register(signal.SIGUSR1, file=sys.stderr, all_threads=True) | |
| print(f"core_dump_limit={_suppress_core_dumps()}", flush=True) | |
| def _retire_worker(): | |
| """In a forked ZeroGPU worker, exit just after the result is handed back.""" | |
| if os.getpid() == _PARENT_PID: | |
| return | |
| def exit_soon(): | |
| time.sleep(_WORKER_RETIRE_DELAY_S) | |
| os._exit(0) | |
| threading.Thread(target=exit_soon, daemon=True).start() | |
| def _sleep(seconds): | |
| time.sleep(seconds) | |
| def _workflow_reservation(plan_data, image_path, last_image_path): | |
| return workflow_reservation_seconds(plan_data, bool(image_path), bool(last_image_path)) | |
| def _encode_gpu(plan_data, image_path, last_image_path, model_path): | |
| from local_runtime import encode_local | |
| return encode_local(plan_data, image_path, last_image_path, model_path) | |
| def _execute_gpu(plan_data, embeds, tags, image_path, last_image_path, model_path, adapter_path): | |
| from local_runtime import generate_local | |
| return generate_local(plan_data, embeds, tags, image_path, last_image_path, model_path, adapter_path) | |
| def _workflow_gpu(plan_data, image_path, last_image_path): | |
| """One allocation owns both downloads and model stages, never the parent lock.""" | |
| try: | |
| return _workflow_stages(plan_data, image_path, last_image_path) | |
| finally: | |
| _retire_worker() | |
| def _workflow_stages(plan_data, image_path, last_image_path): | |
| started = time.monotonic() | |
| has_first, has_last = bool(image_path), bool(last_image_path) | |
| rows_bound = conditioning_text_rows_bound(plan_data, has_first, has_last) | |
| metrics = { | |
| "gpu_allocations": 1, "downloads_inside_gpu_allocation": True, | |
| "cold_download_allowance_s": COLD_DOWNLOAD_ALLOWANCE_S, | |
| "conditioning_reservation_s": conditioning_reservation_seconds(plan_data, has_first, has_last), | |
| "generation_reservation_s": reservation_seconds(plan_data, rows_bound), | |
| } | |
| print("workflow_stage=conditioning_download", flush=True) | |
| with MODEL_STORE.stage("conditioner") as model_path: | |
| downloaded = time.monotonic() | |
| metrics["conditioning_download_s"] = round(downloaded - started, 3) | |
| print("workflow_stage=conditioning_gpu", flush=True) | |
| embeds, tags, height, width, num_frames = _encode_gpu( | |
| plan_data, image_path, last_image_path, str(model_path)) | |
| metrics["conditioning_gpu_s"] = round(time.monotonic() - downloaded, 3) | |
| validate_local_conditioning(plan_data, embeds, tags, height, width, num_frames) | |
| metrics["conditioning_elapsed_s"] = round(time.monotonic() - started, 3) | |
| if int(tags.numel()) > rows_bound: | |
| raise ValueError("local conditioning exceeds the preallocated text-row bound") | |
| metrics["generation_actual_reservation_s"] = reservation_seconds(plan_data, int(tags.numel())) | |
| generation_started = time.monotonic() | |
| print("workflow_stage=generator_download", flush=True) | |
| with MODEL_STORE.stage("generator") as model_path: | |
| downloaded = time.monotonic() | |
| metrics["generator_download_s"] = round(downloaded - generation_started, 3) | |
| adapter_path = None | |
| if plan_data["adapter"] != "off": | |
| from huggingface_hub import hf_hub_download | |
| from h3_adapter_spec import PROFILES | |
| profile = PROFILES[plan_data["adapter"]] | |
| adapter_path = hf_hub_download( | |
| profile.repo, profile.file, revision=ADAPTER_REVISIONS[profile.repo], | |
| token=False, local_dir=Path(model_path) / "adapters") | |
| execution_started = time.monotonic() | |
| metrics["adapter_download_s"] = round(execution_started - downloaded, 3) | |
| print("workflow_stage=generator_gpu", flush=True) | |
| frames, audio, sr = _execute_gpu(plan_data, embeds, tags, image_path, | |
| last_image_path, str(model_path), adapter_path) | |
| metrics["generation_gpu_s"] = round(time.monotonic() - execution_started, 3) | |
| metrics["generation_elapsed_s"] = round(time.monotonic() - generation_started, 3) | |
| # GPU-body elapsed excludes scheduler/worker IPC overhead, and is not billing. | |
| metrics["gpu_elapsed_s"] = round(time.monotonic() - started, 3) | |
| return frames, audio, sr, metrics | |
| def _reference_reservation(plan_data, references, metas): | |
| return reference_workflow_seconds(plan_data, metas) | |
| def _encode_reference_gpu(plan_data, references, model_path): | |
| from local_runtime import encode_local_ref2va | |
| return encode_local_ref2va(plan_data, references, model_path, REFERENCE_IMAGE_SHORT_EDGE) | |
| def _execute_reference_gpu(plan_data, embeds, tags, references, model_path): | |
| from local_runtime import generate_local_ref2va | |
| return generate_local_ref2va(plan_data, embeds, tags, references, model_path, REFERENCE_IMAGE_SHORT_EDGE) | |
| def _reference_gpu(plan_data, references, metas): | |
| """Ref2VA in one allocation: conditioner stage, then the transformer_ref stage.""" | |
| try: | |
| return _reference_stages(plan_data, references, metas) | |
| finally: | |
| _retire_worker() | |
| def _reference_stages(plan_data, references, metas): | |
| started = time.monotonic() | |
| rows_bound = reference_text_rows_bound(plan_data, metas) | |
| ref_rows = reference_rows(metas, plan_data["num_frames"]) | |
| metrics = { | |
| "gpu_allocations": 1, "downloads_inside_gpu_allocation": True, | |
| "cold_download_allowance_s": COLD_DOWNLOAD_ALLOWANCE_S, "reference_rows": ref_rows, | |
| "conditioning_reservation_s": reference_conditioning_seconds(plan_data), | |
| "generation_reservation_s": reference_generation_seconds(plan_data, rows_bound, ref_rows), | |
| } | |
| print("workflow_stage=conditioning_download", flush=True) | |
| with MODEL_STORE.stage("conditioner") as model_path: | |
| downloaded = time.monotonic() | |
| metrics["conditioning_download_s"] = round(downloaded - started, 3) | |
| print("workflow_stage=conditioning_gpu", flush=True) | |
| embeds, tags, height, width, num_frames = _encode_reference_gpu(plan_data, references, str(model_path)) | |
| metrics["conditioning_gpu_s"] = round(time.monotonic() - downloaded, 3) | |
| validate_local_conditioning(plan_data, embeds, tags, height, width, num_frames) | |
| metrics["conditioning_elapsed_s"] = round(time.monotonic() - started, 3) | |
| # Calibration evidence: the real conditioning rows against the reserved bound. | |
| metrics["text_rows"] = int(tags.numel()) | |
| print(f"reference_rows text={metrics['text_rows']} bound={rows_bound} reference={ref_rows}", flush=True) | |
| if metrics["text_rows"] > rows_bound: | |
| raise ValueError("reference conditioning exceeds the preallocated text-row bound; " | |
| "use fewer or shorter references") | |
| generation_started = time.monotonic() | |
| print("workflow_stage=generator_download", flush=True) | |
| with MODEL_STORE.stage("generator_ref") as model_path: | |
| downloaded = time.monotonic() | |
| metrics["generator_download_s"] = round(downloaded - generation_started, 3) | |
| print("workflow_stage=generator_gpu", flush=True) | |
| frames, audio, sr = _execute_reference_gpu(plan_data, embeds, tags, references, str(model_path)) | |
| metrics["generation_gpu_s"] = round(time.monotonic() - downloaded, 3) | |
| metrics["generation_elapsed_s"] = round(time.monotonic() - generation_started, 3) | |
| metrics["gpu_elapsed_s"] = round(time.monotonic() - started, 3) | |
| return frames, audio, sr, metrics | |
| def _encode_output(frames, audio, sr, target): | |
| from diffusers.utils import encode_video | |
| encode_video(frames, fps=24, output_path=str(target), audio=audio, audio_sample_rate=sr) | |
| if spaces is not None and os.environ.get("SPACE_ID"): | |
| _workflow_gpu = spaces.GPU(duration=_workflow_reservation, size="xlarge")(_workflow_gpu) | |
| _reference_gpu = spaces.GPU(duration=_reference_reservation, size="xlarge")(_reference_gpu) | |
| # ZeroGPU schedules with the caller's X-IP-Token. A page left open for hours, or | |
| # a stale client, carries an expired one that the scheduler rejects with a 401. | |
| _IDENTITY_MIN_TTL_S = 3 | |
| _IDENTITY_EXPIRED = ("zerogpu_session_expired; reload the page (or reconnect your API client) " | |
| "and run again. Nothing was submitted.") | |
| def _identity_ttl(): | |
| """Seconds left on the caller's ZeroGPU identity token, or None without one.""" | |
| from gradio.context import LocalContext | |
| request = LocalContext.request.get(None) | |
| token = (getattr(request, "headers", None) or {}).get("x-ip-token") | |
| if not token: | |
| return None | |
| try: | |
| payload = token.split(".")[1] | |
| claims = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4))) | |
| return float(claims["exp"]) - time.time() | |
| except Exception: | |
| return None | |
| def _admit(): | |
| """Readiness and caller identity, before the run lock or any allocation.""" | |
| if MODEL_STORE is None or blockers(): | |
| raise gr.Error("runtime_not_ready: " + ", ".join(blockers() or ["runtime_not_initialized"])) | |
| ttl = _identity_ttl() | |
| print(f"zerogpu_identity_ttl_s={'none' if ttl is None else round(ttl)}", flush=True) | |
| if ttl is not None and ttl < _IDENTITY_MIN_TTL_S: | |
| raise gr.Error(_IDENTITY_EXPIRED) | |
| def _run_locked(plan_data, reserve, run_gpu): | |
| """Serialize one GPU workflow, encode its MP4 on CPU, and report it.""" | |
| if not RUN_LOCK.acquire(blocking=False): | |
| raise gr.Error("runtime_busy; another job is running and nothing was submitted; " | |
| "retry after it finishes") | |
| gpu_finished = None | |
| try: | |
| OUTPUT_DIR.mkdir(parents=True, exist_ok=True, mode=0o700) | |
| if shutil.disk_usage(OUTPUT_DIR).free < 3 * 1024**3: | |
| raise gr.Error("insufficient_output_storage") | |
| # Retain only our own expired outputs. Active output paths never preexist. | |
| for old in OUTPUT_DIR.glob("workflow-*.mp4"): | |
| if old.is_file() and not old.is_symlink() and time.time() - old.stat().st_mtime > 7200: | |
| old.unlink() | |
| started = time.monotonic() | |
| # Admission precedes all downloads. Schedule once while the original | |
| # request's proxy token is fresh; cold downloads consume GPU allocation. | |
| reservation = reserve() | |
| watch_stop = threading.Event() | |
| if _WORKER_WATCH_S: | |
| threading.Thread(target=worker_probe.watch, args=(watch_stop, _WORKER_WATCH_S), | |
| name="worker-probe", daemon=True).start() | |
| # The worker is forked inside run_gpu() and must never garbage-collect | |
| # parent objects: a shot-2 worker segfaulted in its first collection, | |
| # freeing garbage the parent still held from the previous job (2026-10-23). | |
| # Collect here, then freeze what is left out of every later collection. | |
| gc.collect() | |
| gc.freeze() | |
| try: | |
| frames, audio, sr, metrics = run_gpu() | |
| finally: | |
| gc.unfreeze() | |
| watch_stop.set() | |
| gpu_finished = time.monotonic() | |
| print("workflow_stage=output_encoding", flush=True) | |
| encoding_started = time.monotonic() | |
| target = OUTPUT_DIR / f"workflow-{uuid.uuid4().hex}.mp4" | |
| _encode_output(frames, audio, sr, target) | |
| metrics["output_encoding_s"] = round(time.monotonic() - encoding_started, 3) | |
| target.chmod(0o600) | |
| with target.open("rb") as handle: | |
| digest = hashlib.file_digest(handle, "sha256").hexdigest() | |
| report = { | |
| **plan_data, "schema_version": 2, "status": "generated_unreviewed", "model_revision": MODEL_REVISION, | |
| "conditioning_transport": "in_process", "model_loading": "staged_local", | |
| **metrics, | |
| "elapsed_s": round(time.monotonic() - started, 3), "reservation_s": reservation, | |
| "output_sha256": digest, "output_bytes": target.stat().st_size, | |
| "compiled": False, "quality_reviewed": False, | |
| } | |
| return str(target), report | |
| except gr.Error as error: | |
| # gr.Error subclasses ValueError, so it must be handled first. | |
| if "Expired ZeroGPU proxy token" in str(error.message): | |
| raise gr.Error(_IDENTITY_EXPIRED) from None | |
| raise | |
| except ValueError as error: | |
| # Validation messages contain no credentials; remote/provider exceptions | |
| # are deliberately not copied into the public API response. | |
| raise gr.Error(str(error)) from None | |
| except Exception as error: | |
| print("generation_error_" + type(error).__name__, flush=True) | |
| raise gr.Error("generation_outcome_unknown; reconcile before retrying") from None | |
| finally: | |
| # Let the retiring worker exit before another job can be scheduled. | |
| if gpu_finished is not None and _WORKER_RETIRE_HOLD_S: | |
| remaining = _WORKER_RETIRE_HOLD_S - (time.monotonic() - gpu_finished) | |
| if remaining > 0: | |
| _sleep(remaining) | |
| RUN_LOCK.release() | |
| def generate(preset_id, prompt, image_path=None, last_image_path=None, canvas="", duration=5, | |
| seed=42): | |
| plan_data = plan(preset_id, prompt, canvas, duration, seed, bool(image_path), bool(last_image_path)) | |
| _admit() | |
| return _run_locked(plan_data, lambda: _workflow_reservation(plan_data, image_path, last_image_path), | |
| lambda: _workflow_gpu(plan_data, image_path, last_image_path)) | |
| REFERENCE_VIDEO_SECONDS = (2.0, 15.0) | |
| def _media_seconds(stream, container): | |
| import av | |
| if stream.duration is not None and stream.time_base is not None: | |
| return float(stream.duration * stream.time_base) | |
| return None if container.duration is None else container.duration / av.time_base | |
| def _probe_reference(kind, path): | |
| """Metadata the Ref2VA budget needs; no decoding of pixels or samples.""" | |
| if kind == "image": | |
| from PIL import Image | |
| with Image.open(path) as image: | |
| width, height = image.size | |
| return {"kind": "image", "width": width, "height": height} | |
| import av | |
| with av.open(path) as container: | |
| video = container.streams.video[0] if container.streams.video else None | |
| audio = container.streams.audio[0] if container.streams.audio else None | |
| audio_seconds = _media_seconds(audio, container) if audio else None | |
| if kind == "audio": | |
| if not audio_seconds: | |
| raise ValueError("the reference audio has no audio stream") | |
| return {"kind": "audio", "seconds": audio_seconds} | |
| if video is None: | |
| raise ValueError("the reference video has no video stream") | |
| seconds = _media_seconds(video, container) or 0.0 | |
| low, high = REFERENCE_VIDEO_SECONDS | |
| if not low <= seconds <= high: | |
| raise ValueError(f"the reference video must be {low:g}-{high:g} seconds long") | |
| return {"kind": "video", "width": video.width, "height": video.height, | |
| "seconds": seconds, "audio_seconds": audio_seconds} | |
| def reference_to_video(prompt, image_1=None, image_2=None, image_3=None, audio=None, video=None, | |
| canvas="", duration=5, seed=42): | |
| """Ref2VA: images, one soundtrack and one video read in that order, then the prompt.""" | |
| references = [("image", path) for path in (image_1, image_2, image_3) if path] | |
| references += [(kind, path) for kind, path in (("audio", audio), ("video", video)) if path] | |
| try: | |
| plan_data = plan_reference_request(prompt, [kind for kind, _ in references], canvas, duration, seed) | |
| except ValueError as error: | |
| raise gr.Error(str(error)) from None | |
| try: | |
| metas = [_probe_reference(kind, path) for kind, path in references] | |
| except ValueError as error: | |
| raise gr.Error(str(error)) from None | |
| except Exception: | |
| raise gr.Error("a reference file could not be read") from None | |
| _admit() | |
| return _run_locked(plan_data, lambda: _reference_reservation(plan_data, references, metas), | |
| lambda: _reference_gpu(plan_data, references, metas)) | |
| # Card backgrounds for the canvas Templates modal, one per model family. | |
| _CARD_GRADIENTS = { | |
| "base": ("#2dd4bf", "radial-gradient(ellipse at 15% 85%, #2dd4bf 0%, #0d9488 40%, transparent 70%), " | |
| "radial-gradient(ellipse at 80% 10%, #38bdf8 0%, #0284c7 45%, transparent 70%), #0f766e"), | |
| "keyframes": ("#a3e635", "radial-gradient(ellipse at 20% 20%, #a3e635 0%, #65a30d 40%, transparent 70%), " | |
| "radial-gradient(ellipse at 85% 80%, #34d399 0%, #059669 45%, transparent 70%), #166534"), | |
| "larry": ("#c084fc", "radial-gradient(ellipse at 15% 80%, #c084fc 0%, #7c3aed 40%, transparent 70%), " | |
| "radial-gradient(ellipse at 85% 15%, #f472b6 0%, #db2777 45%, transparent 70%), #4c1d95"), | |
| "lightx": ("#fb923c", "radial-gradient(ellipse at 80% 85%, #fbbf24 0%, #d97706 40%, transparent 70%), " | |
| "radial-gradient(ellipse at 15% 15%, #fb923c 0%, #ea580c 45%, transparent 70%), #7c2d12"), | |
| "ref2va": ("#f472b6", "radial-gradient(ellipse at 20% 80%, #f472b6 0%, #be185d 40%, transparent 70%), " | |
| "radial-gradient(ellipse at 80% 15%, #fb7185 0%, #e11d48 45%, transparent 70%), #831843"), | |
| "extend": ("#818cf8", "radial-gradient(ellipse at 15% 20%, #818cf8 0%, #4f46e5 40%, transparent 70%), " | |
| "radial-gradient(ellipse at 85% 85%, #22d3ee 0%, #0891b2 45%, transparent 70%), #312e81"), | |
| } | |
| _WORKFLOW_CARD_TEXT = { | |
| "ref2va-images": ("Reference to video · 2 images", "Ref2VA · images then prompt · budget checked on Run"), | |
| "extend-clip": ("Extend a clip · 2 shots", "Text → last frame → image to video → join · 2 GPU runs"), | |
| } | |
| _CARD_MODES = {"t2va": "Text to video", "i2va": "First frame to video", "fl2va": "First + last frame to video"} | |
| def template_cards() -> list[dict]: | |
| """Templates-modal cards: each preset's local H3 workflow with its admission budget, | |
| then the Ref2VA and multi-node workflows.""" | |
| presets = {preset["id"]: preset for preset in list_presets()} | |
| cards = [] | |
| for entry in list_templates(): | |
| preset, mode = presets[entry["preset_id"]], entry["mode"] | |
| family = ("keyframes" if mode != "t2va" else "base" if preset["adapter"] == "off" | |
| else preset["adapter"].split("-")[0]) | |
| accent, gradient = _CARD_GRADIENTS[family] | |
| admission = entry["admission"] | |
| budget = (f"{admission['reservation_seconds']} s GPU reserved" if admission["allowed"] | |
| else "Blocked at defaults") | |
| name = preset["label"] if mode == "t2va" else f"{preset['label']} · {_CARD_MODES[mode].split(' to ')[0]}" | |
| cards.append({ | |
| "id": entry["id"], "name": name, "category": _CARD_MODES[mode], | |
| "description": f"{_CARD_MODES[mode]} · {preset['default_canvas'].split(' · ')[0]} · {budget}", | |
| "accent": accent, "gradient": gradient, "workflow": entry["graph"], | |
| }) | |
| for entry in list_workflow_templates(): | |
| accent, gradient = _CARD_GRADIENTS[entry["mode"]] | |
| name, description = _WORKFLOW_CARD_TEXT[entry["id"]] | |
| cards.append({"id": entry["id"], "name": name, "category": "Custom workflow", | |
| "description": description, "accent": accent, "gradient": gradient, | |
| "workflow": entry["graph"]}) | |
| return cards | |
| initialize() | |
| CARDS = template_cards() | |
| # Gradio appends launch(js=...) verbatim as a <script>, so call the function. | |
| CANVAS_JS = (f"({Path(__file__).with_name('canvas_launch.js').read_text().strip()})" | |
| f"({json.dumps(CARDS).replace('</', '<' + chr(92) + '/')});") | |
| with gr.Blocks(title="Workflow Studio", analytics_enabled=False, fill_width=True, fill_height=True) as demo: | |
| # Build the stable API first so the canvas invokes its exact preprocessing, | |
| # execution context and output registration, without remote inference nodes. | |
| gr.api(catalog, api_name="presets", queue=False) | |
| gr.api(plan, api_name="plan", queue=False) | |
| first = gr.Image(label="First frame", type="filepath", visible=False) | |
| last = gr.Image(label="Last frame", type="filepath", visible=False) | |
| video = gr.Video(label="Generated output", visible=False) | |
| result = gr.JSON(label="Run metadata", visible=False) | |
| # Preserve raw API scalar types. UI Textbox/Number preprocessing otherwise | |
| # coerces numeric prompts and rounds fractional or boolean seeds. Keep normal | |
| # Gradio file preprocessing so upload/cache path protections remain enabled. | |
| api_defaults = {"preset_id": "stable", "canvas": "", "duration": 5, "seed": 42} | |
| # JSON components parse string values as JSON; encode scalar defaults once | |
| # so the public schema retains the same optional arguments as the UI route. | |
| api_scalars = {name: gr.JSON(value=json.dumps(api_defaults.get(name)), label=name, | |
| visible=False) for name in ( | |
| "preset_id", "prompt", "canvas", "duration", "seed", | |
| )} | |
| api_run = gr.Button(visible=False) | |
| api_run.click(generate, [api_scalars["preset_id"], api_scalars["prompt"], first, last, | |
| api_scalars["canvas"], api_scalars["duration"], api_scalars["seed"]], | |
| [video, result], api_name="generate", concurrency_limit=None) | |
| reference_images = [gr.Image(label=f"Reference image {n}", type="filepath", visible=False) | |
| for n in (1, 2, 3)] | |
| reference_audio = gr.Audio(label="Reference audio", type="filepath", visible=False) | |
| reference_video = gr.Video(label="Reference video", visible=False) | |
| reference_defaults = {"canvas": "", "duration": 5, "seed": 42} | |
| reference_scalars = {name: gr.JSON(value=json.dumps(reference_defaults.get(name)), label=name, | |
| visible=False) for name in ("prompt", "canvas", "duration", "seed")} | |
| gr.Button(visible=False).click( | |
| reference_to_video, | |
| [reference_scalars["prompt"], *reference_images, reference_audio, reference_video, | |
| reference_scalars["canvas"], reference_scalars["duration"], reference_scalars["seed"]], | |
| [video, result], api_name="reference_to_video", concurrency_limit=None) | |
| # The canvas has its own sign-in button; this hidden one makes Gradio mount | |
| # the /login/huggingface OAuth routes it uses. | |
| gr.LoginButton(visible=False) | |
| runtime = gr.JSON(visible=False) | |
| gr.Button(visible=False).click(health, outputs=runtime, api_name="health", queue=False) | |
| def endpoint_index(api_name): | |
| return next(i for i, fn in demo.fns.items() if fn.api_name == api_name) | |
| def queued_node(name, run): | |
| async def node(args_json: str, request: gr.Request) -> str: | |
| return await run([name, args_json], request) | |
| node.__name__ = f"predict_fn_{name}" | |
| return node | |
| # Gradio 6.28 canvases submit fn nodes to /predict_fn_<name> through the | |
| # queue, which carries the viewer's ZeroGPU identity. No route waits behind a | |
| # GPU job (a wait can outlive that identity); RUN_LOCK answers "busy". | |
| gpu_nodes = { | |
| "generate_video": make_call_fn(demo, endpoint_index("generate")), | |
| "reference_to_video": make_call_fn(demo, endpoint_index("reference_to_video"), | |
| "reference_to_video", 9, (1, 2, 3, 4, 5)), | |
| } | |
| for name, run in gpu_nodes.items(): | |
| gr.api(queued_node(name, run), api_name=f"predict_fn_{name}", | |
| api_visibility="undocumented", concurrency_limit=None) | |
| for name, tool, arity, mime_type in (("last_frame", media_tools.last_frame, 1, "image/png"), | |
| ("first_frame", media_tools.first_frame, 1, "image/png"), | |
| ("concat_videos", media_tools.concat_videos, 2, "video/mp4")): | |
| gr.api(make_media_fn(tool, arity, mime_type), api_name=f"predict_fn_{name}", | |
| api_visibility="undocumented", concurrency_limit=2) | |
| # Full-page canvas, as in gr.Workflow; the Templates modal lists CARDS. | |
| workflow_canvas = gr.WorkflowCanvas( | |
| value=json.dumps(CARDS[0]["workflow"]), elem_id="workflow-canvas", | |
| server_functions=[get_token, get_oauth_available, get_oauth_scopes, | |
| make_palette_fn(function_palette()), make_api_fn(demo, [ | |
| ("/generate", "Generate video"), ("/reference_to_video", "Reference to video"), | |
| ("/plan", "Plan a request"), | |
| ("/presets", "Presets and readiness"), ("/health", "Runtime health")])]) | |
| if __name__ == "__main__": | |
| demo.queue(default_concurrency_limit=1, max_size=4).launch( | |
| server_name=os.environ.get("GRADIO_SERVER_NAME", "127.0.0.1"), | |
| server_port=int(os.environ.get("GRADIO_SERVER_PORT", "7860")), | |
| show_error=False, max_file_size="1gb", footer_links=["api"], | |
| js=CANVAS_JS, | |
| css=Path(__file__).with_name("canvas.css").read_text(), | |
| blocked_paths=[str(Path.home() / ".cache"), str(MODEL_CACHE_DIR), | |
| str(Path(__file__).resolve().parent / "tests")], | |
| ) | |