diff --git a/README.md b/README.md index 816871b..dc9be53 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,33 @@ The orchestrator runs on any cheap VPS. The GPU backend is a separate service th | GCP (a2/a3) | A100, H100 | ~$2-$4/hr | | Self-hosted 8× MI300X | MI300X (192GB each) | ~$16-$24/hr (node) | +## Deploy the GPU Backend + +Video generation needs a GPU backend — the open-source Wan server in +[`src/gpu_backend/`](src/gpu_backend/). It exposes one base64 HTTP contract +(`POST /generate/t2v|i2v|flf2v`, `GET /health`) that the pipeline consumes, +whether it runs on Modal, RunPod, a self-hosted box, or your own cloud. + +> A provider token or API key **is not enough on its own** — the backend has to +> be *deployed* into your account first. That deploy is what produces the +> endpoint URL your jobs route to. + +```bash +pip install "flow[gpu]" modal && modal token new # one-time Modal auth + +# Deploy the backend as a NAMED instance (deploy several — A100 pool, H100 pool…) +flow deploy modal --name flow-gpu-a100 --gpu A100-80GB +# → prints the base URL, e.g. https://--flow-gpu-a100.modal.run +``` + +Put that URL in `[gpu_backend].url`. Defaults live in the `[deploy]` section of +your config; CLI flags override them. `flow deploy aws` / `flow deploy gcp` print +the concrete manual steps (first-class automation is on the roadmap — they never +fake a deployment). + +**Self-hosted / RunPod:** run the same FastAPI app directly +(`uvicorn gpu_backend.server:app`, GPU host) and point `[gpu_backend].url` at it. + ## Text-to-Speech & Voice Cloning Narration is produced by a pluggable TTS layer. Pick a provider in config: diff --git a/config/config.example.toml b/config/config.example.toml index c1d6170..7698e17 100644 --- a/config/config.example.toml +++ b/config/config.example.toml @@ -23,11 +23,25 @@ model = "gpt-4o-mini" [gpu_backend] # Supported: modal, runpod, self-hosted provider = "modal" -url = "" # URL of your GPU backend inference server +url = "" # base URL of your deployed GPU backend (e.g. https://you--flow-gpu-backend.modal.run) api_key = "" resolution = "480p" # 480p or 720p clip_duration = 5 # seconds per clip +# `flow deploy` defaults — deploy the GPU backend as a NAMED instance. +# A token alone can't generate; you must deploy the open-source backend into +# your account first, which yields the `url` above. Deploy several named +# instances (e.g. an A100 pool and an H100 pool) and point `url` at one. +# flow deploy modal --name flow-gpu-a100 --gpu A100-80GB +[deploy] +provider = "modal" # modal (automated) | aws | gcp (scaffolded) +name = "flow-gpu-backend" # instance name -> Modal app + subdomain label +gpu = "A100-80GB" # e.g. A100-80GB, H100 +model_t2v = "Wan-AI/Wan2.2-T2V-A14B-Diffusers" +model_i2v = "Wan-AI/Wan2.2-I2V-A14B-Diffusers" +region = "" # AWS/GCP only +scaledown_window = 300 # seconds idle before Modal scales to zero + [tts] # Text-to-speech provider. Built-ins: edge (free, online, no cloning), # miso (MisoTTS — natural voice + one-shot cloning, needs a GPU). diff --git a/src/flow/cli.py b/src/flow/cli.py index b72b80c..c2f9d8a 100644 --- a/src/flow/cli.py +++ b/src/flow/cli.py @@ -99,5 +99,65 @@ def mcp( run(host=host, port=port) +@app.command() +def deploy( + provider: str = typer.Argument( + None, help="Target provider: modal | aws | gcp (default from config)" + ), + name: str = typer.Option(None, help="Instance name — deploy several, named"), + gpu: str = typer.Option(None, help="GPU type, e.g. A100-80GB / H100"), + model_t2v: str = typer.Option(None, help="T2V model id"), + model_i2v: str = typer.Option(None, help="I2V model id"), + region: str = typer.Option(None, help="Region (AWS/GCP)"), + config_path: str = typer.Option("config/config.toml", help="Path to config file"), +) -> None: + """Deploy the GPU backend to a provider as a named compute instance. + + A token alone can't generate — this deploys the open-source Wan backend into + your account and reports the endpoint URL to register. CLI flags override the + deploy section of your config file. + """ + from flow.config import load_config + from flow.deploy import DeploySpec, available_providers, get_deployer + + d = load_config(config_path).deploy + provider = provider or d.provider + spec = DeploySpec( + name=name or d.name, + gpu=gpu or d.gpu, + model_t2v=model_t2v or d.model_t2v, + model_i2v=model_i2v or d.model_i2v, + region=region or d.region, + scaledown_window=d.scaledown_window, + ) + + try: + deployer = get_deployer(provider) + except ValueError: + console.print( + f"[red]Unknown provider '{provider}'[/] — " + f"available: {', '.join(available_providers())}" + ) + raise typer.Exit(1) from None + + console.print( + f"[bold green]Flow[/] — deploying '{spec.name}' to " + f"[cyan]{provider}[/] (gpu={spec.gpu})" + ) + result = deployer.deploy(spec) + + if result.status == "deployed": + console.print(f"[bold green]✓[/] deployed [bold]{result.name}[/]") + if result.endpoint_url: + console.print(f" endpoint: [cyan]{result.endpoint_url}[/]") + console.print(f" {result.detail}") + elif result.status == "manual_required": + console.print(f"[yellow]⚠ manual steps required[/] for {provider}:") + console.print(f" {result.detail}") + else: + console.print(f"[red]✗ deploy failed:[/] {result.detail}") + raise typer.Exit(1) + + if __name__ == "__main__": app() diff --git a/src/flow/config.py b/src/flow/config.py index 74682f2..19bf560 100644 --- a/src/flow/config.py +++ b/src/flow/config.py @@ -75,6 +75,18 @@ class BillingConfig(BaseModel): can_generate: bool = True +class DeployConfig(BaseModel): + """Defaults for ``flow deploy`` (deploy the GPU backend as a named instance).""" + + provider: str = "modal" # modal, aws, gcp + name: str = "flow-gpu-backend" # instance name (deploy several, named) + gpu: str = "A100-80GB" + model_t2v: str = "Wan-AI/Wan2.2-T2V-A14B-Diffusers" + model_i2v: str = "Wan-AI/Wan2.2-I2V-A14B-Diffusers" + region: str = "" # AWS/GCP + scaledown_window: int = 300 # seconds idle before Modal scales to zero + + class Config(BaseModel): llm: LLMConfig = LLMConfig() gpu_backend: GPUBackendConfig = GPUBackendConfig() @@ -84,6 +96,7 @@ class Config(BaseModel): agent: AgentConfig = AgentConfig() mcp: MCPConfig = MCPConfig() billing: BillingConfig = BillingConfig() + deploy: DeployConfig = DeployConfig() output_dir: str = "storage/outputs" aspect_ratio: str = "9:16" # 9:16, 16:9 generation_mode: str = "sequential" # sequential, parallel_flf2v, pipelined_flf2v diff --git a/src/flow/deploy/__init__.py b/src/flow/deploy/__init__.py new file mode 100644 index 0000000..a83e2f3 --- /dev/null +++ b/src/flow/deploy/__init__.py @@ -0,0 +1,20 @@ +"""Deployers for the GPU backend (Modal automated; AWS/GCP scaffolded).""" + +from __future__ import annotations + +from flow.deploy import cloud, modal_deploy # noqa: F401 — register deployers +from flow.deploy.base import ( + Deployer, + DeployResult, + DeploySpec, + available_providers, + get_deployer, +) + +__all__ = [ + "DeploySpec", + "DeployResult", + "Deployer", + "get_deployer", + "available_providers", +] diff --git a/src/flow/deploy/base.py b/src/flow/deploy/base.py new file mode 100644 index 0000000..ca0df23 --- /dev/null +++ b/src/flow/deploy/base.py @@ -0,0 +1,74 @@ +"""Deployers — push the GPU backend to a provider as a NAMED instance. + +A token/API key alone can't generate video: the open-source GPU backend +(``gpu_backend/modal_server.py``) must be *deployed* into the target account, +which produces the thing jobs actually route to — an **endpoint URL**. These +deployers do that deployment, parameterized by a :class:`DeploySpec` (from CLI +args and/or the config file), so you can stand up several named instances +(e.g. an A100 pool and an H100 pool) and register their URLs. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass, field + + +@dataclass +class DeploySpec: + """What to deploy, and how. Defaults mirror the config example.""" + + name: str = "flow-gpu-backend" + gpu: str = "A100-80GB" + model_t2v: str = "Wan-AI/Wan2.2-T2V-A14B-Diffusers" + model_i2v: str = "Wan-AI/Wan2.2-I2V-A14B-Diffusers" + region: str = "" + scaledown_window: int = 300 + extra: dict = field(default_factory=dict) + + +@dataclass +class DeployResult: + """Outcome of a deploy. ``status`` is honest about what happened.""" + + name: str + provider: str + status: str # "deployed" | "manual_required" | "failed" + endpoint_url: str = "" + detail: str = "" + + @property + def ok(self) -> bool: + return self.status == "deployed" + + +class Deployer(ABC): + """Interface for provider deployers.""" + + provider: str = "base" + + @abstractmethod + def deploy(self, spec: DeploySpec) -> DeployResult: + """Deploy the GPU backend; return where it landed (or honest guidance).""" + ... + + +_REGISTRY: dict[str, type[Deployer]] = {} + + +def register(cls: type[Deployer]) -> type[Deployer]: + _REGISTRY[cls.provider] = cls + return cls + + +def get_deployer(provider: str) -> Deployer: + cls = _REGISTRY.get(provider) + if cls is None: + raise ValueError( + f"unknown provider {provider!r}; available: {sorted(_REGISTRY)}" + ) + return cls() + + +def available_providers() -> list[str]: + return sorted(_REGISTRY) diff --git a/src/flow/deploy/cloud.py b/src/flow/deploy/cloud.py new file mode 100644 index 0000000..4c506a4 --- /dev/null +++ b/src/flow/deploy/cloud.py @@ -0,0 +1,48 @@ +"""AWS and GCP deployers — honest scaffolds. + +These are **not** automated yet, and they do not pretend to be: they return a +``manual_required`` result with the concrete steps to stand up the GPU backend +on that cloud. First-class AWS/GCP automation is on the roadmap. No fake +success. +""" + +from __future__ import annotations + +from flow.deploy.base import Deployer, DeployResult, DeploySpec, register + +_BASE64_CONTRACT = ( + "exposing POST /generate/t2v|i2v|flf2v and GET /health (base64 contract), " + "then register the endpoint URL as a named compute instance" +) + + +@register +class AWSDeployer(Deployer): + provider = "aws" + + def deploy(self, spec: DeploySpec) -> DeployResult: + return DeployResult( + spec.name, self.provider, "manual_required", + detail=( + "AWS deploy is not automated yet. Manual path: build the GPU " + "backend image (Dockerfile.gpu), push to ECR, and deploy as a " + "SageMaker async endpoint or an ECS/EKS GPU service " + f"{_BASE64_CONTRACT}. Region: " + (spec.region or "us-east-1") + "." + ), + ) + + +@register +class GCPDeployer(Deployer): + provider = "gcp" + + def deploy(self, spec: DeploySpec) -> DeployResult: + return DeployResult( + spec.name, self.provider, "manual_required", + detail=( + "GCP deploy is not automated yet. Manual path: build the GPU " + "backend image (Dockerfile.gpu), push to Artifact Registry, and " + "deploy to Cloud Run (GPU) or GKE " + f"{_BASE64_CONTRACT}. Region: " + (spec.region or "us-central1") + "." + ), + ) diff --git a/src/flow/deploy/modal_deploy.py b/src/flow/deploy/modal_deploy.py new file mode 100644 index 0000000..ffd24f0 --- /dev/null +++ b/src/flow/deploy/modal_deploy.py @@ -0,0 +1,77 @@ +"""Modal deployer — runs ``modal deploy`` on the GPU backend as a named app. + +Real and automated: validates the modal CLI, passes the spec through as env +(read by ``gpu_backend/modal_server.py`` at deploy time), runs the deploy, and +reports the resulting ``*.modal.run`` base URL. +""" + +from __future__ import annotations + +import os +import re +import shutil +import subprocess +from pathlib import Path + +from flow.deploy.base import Deployer, DeployResult, DeploySpec, register + +_MODAL_URL = re.compile(r"https://[^\s\"']+\.modal\.run") + + +@register +class ModalDeployer(Deployer): + provider = "modal" + + def _server_path(self) -> Path: + """Locate the packaged GPU backend module (works installed or in-repo).""" + import gpu_backend + + return Path(gpu_backend.__file__).parent / "modal_server.py" + + def deploy(self, spec: DeploySpec) -> DeployResult: + if shutil.which("modal") is None: + return DeployResult( + spec.name, self.provider, "failed", + detail="modal CLI not found. Install + auth: " + "`pip install modal && modal token new`.", + ) + server = self._server_path() + if not server.exists(): + return DeployResult( + spec.name, self.provider, "failed", + detail=f"GPU backend module not found at {server}", + ) + + env = { + **os.environ, + "FLOW_GPU_APP_NAME": spec.name, + "FLOW_GPU_TYPE": spec.gpu, + "FLOW_MODEL_T2V": spec.model_t2v, + "FLOW_MODEL_I2V": spec.model_i2v, + "FLOW_GPU_SCALEDOWN": str(spec.scaledown_window), + } + try: + proc = subprocess.run( + ["modal", "deploy", str(server)], + env=env, capture_output=True, text=True, timeout=1800, + ) + except (subprocess.TimeoutExpired, OSError) as err: + return DeployResult(spec.name, self.provider, "failed", detail=str(err)) + + combined = (proc.stdout or "") + (proc.stderr or "") + if proc.returncode != 0: + return DeployResult( + spec.name, self.provider, "failed", + detail=combined.strip()[-800:] or "modal deploy failed", + ) + + urls = _MODAL_URL.findall(combined) + url = urls[0] if urls else "" + return DeployResult( + spec.name, self.provider, "deployed", endpoint_url=url, + detail=( + "Deployed. Register this base URL as a named compute instance." + if url else + "Deployed; run `modal app list` to find the base URL." + ), + ) diff --git a/src/flow/generator.py b/src/flow/generator.py index bd50a46..6b0fd46 100644 --- a/src/flow/generator.py +++ b/src/flow/generator.py @@ -150,15 +150,21 @@ def _generate_single( resp.raise_for_status() result = resp.json() - # Save clip + # Save clip — the backend may return base64 inline (serverless, e.g. + # Modal) or a URL (a file-serving backend). Support both so there is one + # consumer contract regardless of where the backend is deployed. clip_path = self.clip_dir / f"scene_{scene_id:03d}.mp4" - video_data = self._download(result["video_url"]) + video_data = self._fetch_asset(result, "video") + if video_data is None: + raise RuntimeError( + "GPU backend returned no video (expected 'video_b64' or 'video_url')" + ) clip_path.write_bytes(video_data) - # Save last frame for next scene's conditioning + # Save last frame for next scene's conditioning. last_frame_path_out = self.clip_dir / f"scene_{scene_id:03d}_last.png" - if "last_frame_url" in result: - frame_data = self._download(result["last_frame_url"]) + frame_data = self._fetch_asset(result, "last_frame") + if frame_data is not None: last_frame_path_out.write_bytes(frame_data) else: self._extract_last_frame(clip_path, last_frame_path_out) @@ -177,6 +183,29 @@ def _download(self, url: str) -> bytes: resp.raise_for_status() return resp.content + def _fetch_asset(self, result: dict, name: str) -> bytes | None: + """Materialise an output asset from the backend response. + + Backends may return the asset inline as base64 (``_b64`` — the + norm for serverless backends like Modal, whose containers are ephemeral) + or as a URL (``_url`` — a file-serving backend). Relative URLs are + resolved against the backend base URL. Returns ``None`` when neither key + is present (e.g. no last frame). + """ + b64 = result.get(f"{name}_b64") + if b64: + import base64 + + return base64.b64decode(b64) + url = result.get(f"{name}_url") + if url: + full = ( + url if url.startswith("http") + else f"{self.backend_url.rstrip('/')}{url}" + ) + return self._download(full) + return None + def _encode_image(self, path: str) -> str: """Encode image to base64 for API payload.""" import base64 diff --git a/src/gpu_backend/modal_server.py b/src/gpu_backend/modal_server.py index 71bf4ec..051ef75 100644 --- a/src/gpu_backend/modal_server.py +++ b/src/gpu_backend/modal_server.py @@ -1,21 +1,45 @@ -"""GPU Backend — Wan 2.2 inference server for Modal deployment.""" +"""GPU Backend — Wan 2.2 inference server for Modal deployment. + +Deploys as a **single FastAPI ASGI app** with path-based routes +(``POST /generate/t2v|i2v|flf2v``, ``GET /health``) returning **base64** — the +same contract the self-hosted ``server.py`` and the ``flow.Generator`` consumer +use. One deployment = one base URL; no per-endpoint subdomains to stitch +together. + +Parameterized via env so you can deploy several **named** instances into one +account (e.g. an A100 pool and an H100 pool, or one per tenant):: + + FLOW_GPU_APP_NAME=flow-gpu-a100 FLOW_GPU_TYPE=A100-80GB \ + modal deploy src/gpu_backend/modal_server.py + +The deployed base URL is ``https://--.modal.run``. Point +``[gpu_backend].url`` (or a named compute instance) at it. +""" from __future__ import annotations import base64 import io +import os import tempfile import uuid from pathlib import Path import modal -app = modal.App("flow-gpu-backend") +# --- Parameterization (read at deploy/import time) --------------------------- +APP_NAME = os.environ.get("FLOW_GPU_APP_NAME", "flow-gpu-backend") +GPU_TYPE = os.environ.get("FLOW_GPU_TYPE", "A100-80GB") +MODEL_T2V = os.environ.get("FLOW_MODEL_T2V", "Wan-AI/Wan2.2-T2V-A14B-Diffusers") +MODEL_I2V = os.environ.get("FLOW_MODEL_I2V", "Wan-AI/Wan2.2-I2V-A14B-Diffusers") +SCALEDOWN_WINDOW = int(os.environ.get("FLOW_GPU_SCALEDOWN", "300")) + +app = modal.App(APP_NAME) -# Model weights volume (persistent storage for downloaded weights) -volume = modal.Volume.from_name("flow-model-weights", create_if_missing=True) +# Model weights volume (persistent storage for downloaded weights), namespaced +# per instance so named deployments don't clobber each other. +volume = modal.Volume.from_name(f"{APP_NAME}-weights", create_if_missing=True) -# Container image with all dependencies image = ( modal.Image.debian_slim(python_version="3.12") .pip_install( @@ -31,47 +55,30 @@ .apt_install("ffmpeg") ) -# Separate, lighter image for voice cloning (no diffusers; Coqui XTTS-v2) -voice_image = ( - modal.Image.debian_slim(python_version="3.12") - .apt_install("ffmpeg") - .pip_install("coqui-tts", "fastapi[standard]", "torch>=2.6.0") - .env({"COQUI_TOS_AGREED": "1"}) -) - -MODEL_T2V = "Wan-AI/Wan2.2-T2V-A14B-Diffusers" -MODEL_I2V = "Wan-AI/Wan2.2-I2V-A14B-Diffusers" -MODEL_VACE = "Wan-AI/Wan2.1-VACE-14B-diffusers" - @app.cls( image=image, - gpu="A100-80GB", + gpu=GPU_TYPE, volumes={"/models": volume}, timeout=900, - scaledown_window=300, + scaledown_window=SCALEDOWN_WINDOW, ) class WanServer: - """Wan 2.2 inference server running on Modal.""" + """Wan 2.2 inference server — serves one base64 HTTP contract on Modal.""" @modal.enter() def load_models(self): - """Load models on container startup.""" + """Load the T2V pipeline on container startup; I2V/VACE load lazily.""" import torch from diffusers import WanPipeline self.device = "cuda" self.dtype = torch.float16 - - # Load T2V pipeline (primary — used for most generation) self.t2v = WanPipeline.from_pretrained( - MODEL_T2V, - torch_dtype=self.dtype, - cache_dir="/models", + MODEL_T2V, torch_dtype=self.dtype, cache_dir="/models" ).to(self.device) - - # I2V loaded lazily on first use (can't fit both in 80GB simultaneously) self._i2v = None + self._vace = None @property def i2v(self): @@ -79,207 +86,116 @@ def i2v(self): import torch from diffusers import WanPipeline - # Offload T2V to CPU to make room + # Offload T2V to CPU to make room (both can't fit in 80GB at once). self.t2v.to("cpu") torch.cuda.empty_cache() - self._i2v = WanPipeline.from_pretrained( - MODEL_I2V, - torch_dtype=self.dtype, - cache_dir="/models", + MODEL_I2V, torch_dtype=self.dtype, cache_dir="/models" ).to(self.device) return self._i2v - @modal.fastapi_endpoint(method="POST", label="t2v") - def generate_t2v_endpoint(self, body: dict) -> dict: - """HTTP endpoint: text-to-video.""" - return self._generate_t2v( - prompt=body["prompt"], - resolution=body.get("resolution", "480p"), - duration=body.get("duration", 5), - ) - - @modal.fastapi_endpoint(method="POST", label="i2v") - def generate_i2v_endpoint(self, body: dict) -> dict: - """HTTP endpoint: image-to-video.""" - return self._generate_i2v( - prompt=body["prompt"], - first_frame_b64=body["first_frame_b64"], - resolution=body.get("resolution", "480p"), - duration=body.get("duration", 5), - ) - - @modal.fastapi_endpoint(method="POST", label="flf2v") - def generate_flf2v_endpoint(self, body: dict) -> dict: - """HTTP endpoint: first-last-frame to video (scene chaining). - - Conditions on a first frame (and optional last frame) for temporal - continuity. Uses the I2V pipeline with the first frame as the anchor. - """ - return self._generate_i2v( - prompt=body["prompt"], - first_frame_b64=body["first_frame_b64"], - resolution=body.get("resolution", "480p"), - duration=body.get("duration", 5), - ) - - @modal.fastapi_endpoint(method="POST", label="vace") - def generate_vace_endpoint(self, body: dict) -> dict: - """HTTP endpoint: Wan VACE — reference/edit/compose to video. - - Best-effort: lazily loads WanVACEPipeline. Isolated from t2v/i2v so a - VACE/diffusers version mismatch never affects the proven paths. - """ - try: - return self._generate_vace( + @modal.asgi_app(label=APP_NAME) + def web(self): + """Single FastAPI app: path-based routes, base64 responses.""" + from fastapi import FastAPI, HTTPException + + api = FastAPI(title="Flow GPU Backend", version="0.2.0") + + @api.get("/health") + def health() -> dict: + return { + "status": "ok", + "model": "Wan2.2-14B", + "app": APP_NAME, + "endpoints": ["t2v", "i2v", "flf2v"], + } + + @api.post("/generate/t2v") + def t2v(body: dict) -> dict: + if not body.get("prompt"): + raise HTTPException(400, "prompt is required") + return self._generate_t2v( + prompt=body["prompt"], + resolution=body.get("resolution", "480p"), + duration=body.get("duration", 5), + ) + + @api.post("/generate/i2v") + def i2v(body: dict) -> dict: + first = body.get("first_frame") or body.get("first_frame_b64") + if not body.get("prompt") or not first: + raise HTTPException(400, "prompt and first_frame are required") + return self._generate_i2v( prompt=body["prompt"], - reference_b64=body.get("reference_b64") or body.get("first_frame_b64"), + first_frame_b64=first, resolution=body.get("resolution", "480p"), duration=body.get("duration", 5), ) - except Exception as e: # noqa: BLE001 - return {"error": f"VACE unavailable: {e}"} - @modal.fastapi_endpoint(method="GET", label="health") - def health_endpoint(self) -> dict: - """Health check endpoint.""" - return {"status": "ok", "model": "Wan2.2-14B", "endpoints": ["t2v", "i2v", "flf2v", "vace"]} + # FLF2V conditions on a first frame (last-frame anchoring is best-effort + # via the I2V pipeline) — same handler, kept as a distinct route so the + # contract matches the t2v/i2v/flf2v vocabulary callers expect. + api.post("/generate/flf2v")(i2v) + + return api def _generate_t2v( - self, - prompt: str, - resolution: str = "480p", - duration: int = 5, + self, prompt: str, resolution: str = "480p", duration: int = 5, num_inference_steps: int = 30, ) -> dict: - """Generate video from text prompt.""" import torch height, width = _resolution_to_dims(resolution) - num_frames = duration * 16 # 16 fps - with torch.inference_mode(): output = self.t2v( - prompt=prompt, - height=height, - width=width, - num_frames=num_frames, - num_inference_steps=num_inference_steps, + prompt=prompt, height=height, width=width, + num_frames=duration * 16, num_inference_steps=num_inference_steps, guidance_scale=5.0, ) - return self._save_output(output) def _generate_i2v( - self, - prompt: str, - first_frame_b64: str, - resolution: str = "480p", - duration: int = 5, - num_inference_steps: int = 30, + self, prompt: str, first_frame_b64: str, resolution: str = "480p", + duration: int = 5, num_inference_steps: int = 30, ) -> dict: - """Generate video from first frame + prompt.""" import torch from PIL import Image height, width = _resolution_to_dims(resolution) - num_frames = duration * 16 - - # Decode first frame image_data = base64.b64decode(first_frame_b64) first_frame = Image.open(io.BytesIO(image_data)).convert("RGB") first_frame = first_frame.resize((width, height)) - with torch.inference_mode(): output = self.i2v( - prompt=prompt, - image=first_frame, - height=height, - width=width, - num_frames=num_frames, - num_inference_steps=num_inference_steps, + prompt=prompt, image=first_frame, height=height, width=width, + num_frames=duration * 16, num_inference_steps=num_inference_steps, guidance_scale=5.0, ) - - return self._save_output(output) - - def _generate_vace( - self, - prompt: str, - reference_b64: str | None, - resolution: str = "480p", - duration: int = 5, - num_inference_steps: int = 30, - ) -> dict: - """Generate/edit video with Wan VACE (reference-driven composition).""" - import torch - from diffusers import WanVACEPipeline - from PIL import Image - - if getattr(self, "_vace", None) is None: - self.t2v.to("cpu") - if self._i2v is not None: - self._i2v.to("cpu") - torch.cuda.empty_cache() - self._vace = WanVACEPipeline.from_pretrained( - MODEL_VACE, torch_dtype=self.dtype, cache_dir="/models" - ).to(self.device) - - height, width = _resolution_to_dims(resolution) - num_frames = duration * 16 - kwargs = dict( - prompt=prompt, height=height, width=width, - num_frames=num_frames, num_inference_steps=num_inference_steps, - ) - if reference_b64: - ref = ( - Image.open(io.BytesIO(base64.b64decode(reference_b64))) - .convert("RGB") - .resize((width, height)) - ) - kwargs["reference_images"] = [ref] - with torch.inference_mode(): - output = self._vace(**kwargs) return self._save_output(output) def _save_output(self, output) -> dict: - """Save generated video and extract last frame.""" + """Save generated video + last frame and return them as base64.""" from diffusers.utils import export_to_video from PIL import Image job_id = str(uuid.uuid4()) tmp_dir = Path(tempfile.mkdtemp()) - - # Save video video_path = tmp_dir / f"{job_id}.mp4" export_to_video(output.frames[0], str(video_path), fps=16) - # Extract last frame last_frame = output.frames[0][-1] if not isinstance(last_frame, Image.Image): last_frame = Image.fromarray(last_frame) last_frame_path = tmp_dir / f"{job_id}_last.png" last_frame.save(str(last_frame_path)) - # Encode outputs as base64 - video_b64 = base64.b64encode( - video_path.read_bytes() - ).decode() - last_frame_b64 = base64.b64encode( - last_frame_path.read_bytes() - ).decode() - return { "job_id": job_id, - "video_b64": video_b64, - "last_frame_b64": last_frame_b64, + "video_b64": base64.b64encode(video_path.read_bytes()).decode(), + "last_frame_b64": base64.b64encode(last_frame_path.read_bytes()).decode(), } def _resolution_to_dims(resolution: str) -> tuple[int, int]: """Convert resolution string to (height, width).""" - resolutions = { - "480p": (480, 832), - "720p": (720, 1280), - } - return resolutions.get(resolution, (480, 832)) + return {"480p": (480, 832), "720p": (720, 1280)}.get(resolution, (480, 832)) diff --git a/src/gpu_backend/server.py b/src/gpu_backend/server.py index 9b65cbc..4416ffe 100644 --- a/src/gpu_backend/server.py +++ b/src/gpu_backend/server.py @@ -38,8 +38,8 @@ class I2VRequest(BaseModel): class GenerationResponse(BaseModel): job_id: str - video_url: str - last_frame_url: str + video_b64: str + last_frame_b64: str @app.on_event("startup") @@ -123,13 +123,19 @@ async def generate_i2v(req: I2VRequest): def _save_and_respond(output) -> GenerationResponse: - """Save output and return response with file URLs.""" + """Save output and return it as base64 (matches the Modal backend contract). + + Serverless/ephemeral containers can't reliably serve files back, so the + unified contract returns the assets inline as base64 — the ``flow.Generator`` + consumer decodes them directly. + """ + import tempfile + from diffusers.utils import export_to_video from PIL import Image job_id = str(uuid.uuid4()) - output_dir = Path("/tmp/flow_outputs") - output_dir.mkdir(exist_ok=True) + output_dir = Path(tempfile.mkdtemp(prefix="flow_outputs_")) # Save video video_path = output_dir / f"{job_id}.mp4" @@ -144,8 +150,8 @@ def _save_and_respond(output) -> GenerationResponse: return GenerationResponse( job_id=job_id, - video_url=f"/files/{job_id}.mp4", - last_frame_url=f"/files/{job_id}_last.png", + video_b64=base64.b64encode(video_path.read_bytes()).decode(), + last_frame_b64=base64.b64encode(last_frame_path.read_bytes()).decode(), ) diff --git a/tests/test_deploy.py b/tests/test_deploy.py new file mode 100644 index 0000000..f43f367 --- /dev/null +++ b/tests/test_deploy.py @@ -0,0 +1,41 @@ +"""Tests for the GPU-backend deployers (no GPU / no cloud account needed).""" + +from flow.deploy import DeploySpec, available_providers, get_deployer +from flow.deploy.modal_deploy import ModalDeployer + + +def test_providers_registered(): + assert set(available_providers()) >= {"modal", "aws", "gcp"} + + +def test_unknown_provider_raises(): + import pytest + + with pytest.raises(ValueError): + get_deployer("nope") + + +def test_modal_server_module_is_packaged(): + # The deployer must be able to locate the GPU backend it deploys. + path = ModalDeployer()._server_path() + assert path.name == "modal_server.py" + assert path.exists() + + +def test_modal_deploy_fails_cleanly_without_cli(monkeypatch): + # No modal CLI installed -> honest failure, not a crash or fake success. + monkeypatch.setattr("flow.deploy.modal_deploy.shutil.which", lambda _: None) + res = ModalDeployer().deploy(DeploySpec(name="flow-gpu-test")) + assert res.status == "failed" + assert res.ok is False + assert "modal CLI not found" in res.detail + + +def test_aws_gcp_are_honest_scaffolds(): + for provider in ("aws", "gcp"): + res = get_deployer(provider).deploy(DeploySpec(name=f"{provider}-inst")) + # Honest: declares manual steps, never claims a deployment happened. + assert res.status == "manual_required" + assert res.ok is False + assert res.endpoint_url == "" + assert "not automated yet" in res.detail diff --git a/tests/test_engine.py b/tests/test_engine.py index a2830e3..bb170c0 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -67,3 +67,44 @@ def test_generate_keyframe_delegates(monkeypatch, tmp_path): assert out == tmp_path / "kf.png" assert seen["prompt"] == "library under stars, closing frame" assert seen["path"] == tmp_path / "kf.png" + + +def test_fetch_asset_decodes_base64(): + """Serverless backends (e.g. Modal) return inline base64.""" + import base64 + + g = Generator(Config()) + payload = b"\x00\x01video-bytes" + result = {"video_b64": base64.b64encode(payload).decode()} + assert g._fetch_asset(result, "video") == payload + + +def test_fetch_asset_downloads_url(monkeypatch): + """A file-serving backend returns a URL; absolute URLs download as-is.""" + g = Generator(Config()) + seen = {} + + def fake_download(url): + seen["url"] = url + return b"downloaded" + + monkeypatch.setattr(g, "_download", fake_download) + out = g._fetch_asset({"video_url": "https://cdn.example/clip.mp4"}, "video") + assert out == b"downloaded" + assert seen["url"] == "https://cdn.example/clip.mp4" + + +def test_fetch_asset_resolves_relative_url(monkeypatch): + """Relative URLs resolve against the backend base URL.""" + cfg = Config() + cfg.gpu_backend.url = "https://gpu.example/" + g = Generator(cfg) + seen = {} + monkeypatch.setattr(g, "_download", lambda url: seen.setdefault("url", url) or b"x") + g._fetch_asset({"last_frame_url": "/files/abc_last.png"}, "last_frame") + assert seen["url"] == "https://gpu.example/files/abc_last.png" + + +def test_fetch_asset_missing_returns_none(): + g = Generator(Config()) + assert g._fetch_asset({}, "last_frame") is None