""" Sulphur — Image to Video (HF Spaces). POC-proven pattern: subprocess is pre-spawned from the main Gradio process via multiprocessing.Process (NOT subprocess.Popen — ZeroGPU only grants GPU access to children spawned through the patched multiprocessing API). @spaces.GPU(duration=1) signals CUDA initialisation; the worker retains the context for the full generation. Billing cost: 1 second per generation regardless of inference time. """ import json import multiprocessing import os import shutil import tempfile import threading import time from pathlib import Path import gradio as gr import spaces from generate import _worker_entrypoint _HF_TOKEN = os.environ.get("HF_TOKEN") _PERSISTENT = Path("/data") if Path("/data").exists() else Path(tempfile.gettempdir()) WAN2GP_ROOT = _PERSISTENT / "Wan2GP" CKPTS_DIR = WAN2GP_ROOT / "ckpts" LORAS_DIR = WAN2GP_ROOT / "loras" / "ltx2" FINETUNES_DIR = WAN2GP_ROOT / "finetunes" # Propagate to spawned worker processes os.environ["WAN2GP_ROOT"] = str(WAN2GP_ROOT) SULPHUR_ASSETS = [ ("SulphurAI/Sulphur-2-base", "sulphur_distil_bf16.safetensors", CKPTS_DIR), ] LTX_ASSETS = [ ("SulphurAI/Sulphur-2-base", "experimental/sulphur_experimental_lora_v1.safetensors", LORAS_DIR), ("DeepBeepMeep/LTX-2", "ltx-2.3-22b-distilled-lora-384.safetensors", LORAS_DIR), ("DeepBeepMeep/LTX-2", "ltx-2.3-22b_vae.safetensors", CKPTS_DIR), ("DeepBeepMeep/LTX-2", "ltx-2.3-22b_text_embedding_projection.safetensors", CKPTS_DIR), ("DeepBeepMeep/LTX-2", "ltx-2.3-22b_embeddings_connector.safetensors", CKPTS_DIR), ] SULPHUR_FINETUNE = { "model": { "name": "Sulphur 2 Base", "visible": True, "architecture": "ltx2_22B", "parent_model_type": "ltx2_22B", "description": "LTX-2.3 fine-tuned i2v. Distilled checkpoint.", "URLs": [str(CKPTS_DIR / "sulphur_distil_bf16.safetensors")], "preload_URLs": [], }, "num_inference_steps": 8, "video_length": 81, "resolution": "832x480", "guidance_scale": 5.0, "alt_guidance_scale": 5.0, } _setup_lock = threading.Lock() _setup_done = False def _download(repo_id, filename, dest_dir): from huggingface_hub import hf_hub_download dest_dir.mkdir(parents=True, exist_ok=True) dest = dest_dir / Path(filename).name if dest.exists(): print(f"[download] cached: {dest.name}") return print(f"[download] {repo_id}/{filename}") hf_hub_download(repo_id=repo_id, filename=filename, local_dir=str(dest_dir), token=_HF_TOKEN) downloaded = dest_dir / filename if downloaded.exists() and not dest.exists(): shutil.move(str(downloaded), str(dest)) def setup(): global _setup_done with _setup_lock: if _setup_done: return _setup_done = True if not (WAN2GP_ROOT / "shared" / "api.py").exists(): WAN2GP_ROOT.mkdir(parents=True, exist_ok=True) print("[setup] Cloning Wan2GP...") import subprocess subprocess.run( ["git", "clone", "--depth=1", "https://github.com/deepbeepmeep/Wan2GP.git", str(WAN2GP_ROOT)], check=True, ) for repo, fname, dest in SULPHUR_ASSETS + LTX_ASSETS: _download(repo, fname, dest) _gemma_folder = "gemma-3-12b-it-qat-q4_0-unquantized" _gemma_file = f"{_gemma_folder}_quanto_bf16_int8.safetensors" gemma_dest = CKPTS_DIR / _gemma_folder / _gemma_file if not gemma_dest.exists(): from huggingface_hub import hf_hub_download print("[download] Gemma text encoder...") hf_hub_download( repo_id="DeepBeepMeep/LTX-2", filename=f"{_gemma_folder}/{_gemma_file}", local_dir=str(CKPTS_DIR), token=_HF_TOKEN, ) else: print("[download] cached: Gemma text encoder") FINETUNES_DIR.mkdir(parents=True, exist_ok=True) (FINETUNES_DIR / "sulphur_2_base.json").write_text( json.dumps(SULPHUR_FINETUNE, indent=2) ) print("[setup] Done.") setup() RESOLUTIONS = ["832x480", "480x832", "640x640", "1024x576", "576x1024"] @spaces.GPU(duration=1) def _signal_cuda_init(signal_path): """Acquire a 1-second GPU lease just long enough for the worker to init CUDA.""" Path(signal_path).write_text("go") time.sleep(0.8) def generate_video(image, prompt, resolution, steps, guidance_scale, frames, seed): if image is None: raise gr.Error("Please upload an image.") if not prompt.strip(): raise gr.Error("Please enter a prompt.") signal_path = tempfile.mktemp(suffix=".signal") result_path = tempfile.mktemp(suffix=".json") out_dir = tempfile.mkdtemp() out_file = os.path.join(out_dir, "output.mp4") # Use multiprocessing.Process — ZeroGPU patches this to grant GPU access to children. # subprocess.Popen does NOT get GPU access (not patched by the spaces library). ctx = multiprocessing.get_context("spawn") proc = ctx.Process( target=_worker_entrypoint, args=(image, prompt, out_file, "sulphur-2", int(steps), float(guidance_scale), int(frames), resolution, int(seed), signal_path, result_path), daemon=False, ) proc.start() log_lines = ["Worker spawned, acquiring GPU lease..."] yield None, "\n".join(log_lines) # 1-second lease — signals worker to init CUDA, then expires _signal_cuda_init(signal_path) log_lines.append("GPU lease expired. Worker retains CUDA context and is generating...") yield None, "\n".join(log_lines) last_log_len = 0 deadline = time.monotonic() + 600 # 10 min hard timeout while time.monotonic() < deadline: time.sleep(2) if os.path.exists(result_path): try: with open(result_path) as f: data = json.load(f) new_entries = data.get("log", [])[last_log_len:] if new_entries: log_lines.extend(new_entries) last_log_len += len(new_entries) yield None, "\n".join(log_lines[-40:]) if data.get("done"): break except Exception: pass else: yield None, "\n".join(log_lines) proc.join(timeout=10) if proc.is_alive(): proc.terminate() for path in (signal_path, result_path): try: os.unlink(path) except Exception: pass if os.path.exists(out_file): final = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) shutil.copy2(out_file, final.name) yield final.name, "\n".join(log_lines) + "\n\n[DONE]" else: yield None, "\n".join(log_lines) + "\n\n[ERROR] No output file produced." with gr.Blocks(title="Sulphur — Image to Video") as demo: gr.Markdown("# Sulphur — Image to Video\nUsing Experimental LoRA v1 + Distillation LoRA") with gr.Row(): with gr.Column(scale=1): image_in = gr.Image(type="filepath", label="Input Image") prompt_in = gr.Textbox(label="Prompt", placeholder="Describe the motion…", lines=3) with gr.Accordion("Advanced", open=False): resolution_dd = gr.Dropdown(RESOLUTIONS, value="832x480", label="Resolution") steps_sl = gr.Slider(1, 50, value=8, step=1, label="Steps") guidance_sl = gr.Slider(1.0, 10.0, value=5.0, step=0.5, label="Guidance Scale") frames_sl = gr.Slider(17, 257, value=81, step=8, label="Frames") seed_num = gr.Number(value=-1, label="Seed (-1 = random)", precision=0) run_btn = gr.Button("Generate", variant="primary") with gr.Column(scale=1): video_out = gr.Video(label="Output Video") log_out = gr.Textbox(label="Log", lines=10, interactive=False) run_btn.click( fn=generate_video, inputs=[image_in, prompt_in, resolution_dd, steps_sl, guidance_sl, frames_sl, seed_num], outputs=[video_out, log_out], ) if __name__ == "__main__": demo.launch(theme=gr.themes.Soft())