From 2d50679ec06d3e2341deda54c4be8053db34b5fb Mon Sep 17 00:00:00 2001 From: Mike Teehan Date: Sun, 31 May 2026 05:54:51 -0400 Subject: [PATCH] Initial commit: Revids REST API for LTX-2.3 image-to-video generation - FastAPI with job polling: POST /generate, GET /jobs/{id}, /download, /health - SQLite job persistence (aiosqlite), background task processing - LTX-2.3 DistilledPipeline via native ltx-pipelines (editable install from submodule) - Configurable model paths, LoRA support, FP8 quantization via env vars - Single-concurrency GPU lock for safe inference - LTX-2 as git submodule under libs/ --- .gitignore | 34 +++++++++ .gitmodules | 3 + README.md | 150 +++++++++++++++++++++++++++++++++++++++ app/__init__.py | 0 app/config.py | 42 +++++++++++ app/database.py | 118 +++++++++++++++++++++++++++++++ app/main.py | 133 +++++++++++++++++++++++++++++++++++ app/models.py | 43 ++++++++++++ app/service.py | 177 +++++++++++++++++++++++++++++++++++++++++++++++ libs/.gitkeep | 0 libs/LTX-2 | 1 + requirements.txt | 7 ++ 12 files changed, 708 insertions(+) create mode 100644 .gitignore create mode 100644 .gitmodules create mode 100644 README.md create mode 100644 app/__init__.py create mode 100644 app/config.py create mode 100644 app/database.py create mode 100644 app/main.py create mode 100644 app/models.py create mode 100644 app/service.py create mode 100644 libs/.gitkeep create mode 160000 libs/LTX-2 create mode 100644 requirements.txt diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a46bef9 --- /dev/null +++ b/.gitignore @@ -0,0 +1,34 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +dist/ +build/ +*.egg +.eggs/ + +# Virtual env +.venv/ +venv/ + +# Environment +.env +.env.* + +# IDE +.vscode/ +.idea/ + +# OS +.DS_Store + +# Models & outputs +models/ +videos/ +*.safetensors + +# Database +app/jobs.db + +# LTX-2 submodule deps (uv creates this) +libs/LTX-2/.venv/ diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..22c6d2a --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "libs/LTX-2"] + path = libs/LTX-2 + url = https://github.com/Lightricks/LTX-2.git diff --git a/README.md b/README.md new file mode 100644 index 0000000..bff1368 --- /dev/null +++ b/README.md @@ -0,0 +1,150 @@ +# Revids + +REST API for generating video from images using [LTX-2.3](https://docs.ltx.video/open-source-model/integration-tools/pytorch-api). + +## Quick Start + +```bash +# 1. Clone the repo (including submodule) +git clone --recursive revids +cd revids + +# 2. Create venv and install LTX packages (editable, from submodule) +python -m venv .venv && source .venv/bin/activate +pip install -e libs/LTX-2/packages/ltx-core -e libs/LTX-2/packages/ltx-pipelines + +# 3. Install API deps +pip install -r requirements.txt + +# 4. Download LTX-2.3 model weights +huggingface-cli download Lightricks/LTX-2.3 \ + --include "ltx-2.3-22b-distilled-1.1.safetensors" \ + --local-dir models/ + +# 5. Download Gemma/GPT-4o text encoder +huggingface-cli download Lightricks/LTX-2 \ + --local-dir models/gpt-4o-5805-ava-gguf-model + +# 6. Download spatial upsampler (optional, for higher-res output) +huggingface-cli download Lightricks/LTX-2 \ + --include "ltx-2.3-22b-spatial-upscaler.safetensors" \ + --local-dir models/ + +# 7. Run +uvicorn app.main:app --reload +``` + +The API will be available at `http://localhost:8000`. Interactive docs at `http://localhost:8000/docs`. + +## API Endpoints + +### Submit a Job +```bash +curl -X POST http://localhost:8000/generate \ + -F "image=@photo.jpg" \ + -F "prompt=A cinematic pan across the landscape" \ + -F "width=768" \ + -F "height=512" \ + -F "num_frames=65" \ + -F "fps=24.0" +``` + +Response: +```json +{"job_id": "abc123def456", "status": "pending"} +``` + +### Check Job Status +```bash +curl http://localhost:8000/jobs/abc123def456 +``` + +Response: +```json +{ + "job_id": "abc123def456", + "status": "completed", + "prompt": "A cinematic pan across the landscape", + "params": {"width": 768, "height": 512, "num_frames": 65, "fps": 24.0}, + "error": null, + "created_at": "2025-01-01T00:00:00+00:00", + "completed_at": "2025-01-01T00:01:30+00:00" +} +``` + +### Download Video (`.mp4`) +```bash +curl http://localhost:8000/jobs/abc123def456/download -o output.mp4 +``` + +### Delete Job (removes DB record + video file) +```bash +curl -X DELETE http://localhost:8000/jobs/abc123def456 +``` + +### Health Check +```bash +curl http://localhost:8000/health +``` +Returns GPU availability and device name. + +## Generation Constraints + +- **Width/Height**: must be divisible by 32 (min 256, max 2048) +- **num_frames**: must follow `8n+1` pattern (9, 17, ..., 65, 97, 121, 161, 257, ...) +- **fps**: 0 < fps <= 60 + +## Configuration + +All settings in `app/config.py` can be overridden via `.env` or env vars prefixed with `REVIDS_`: + +| Variable | Default | Description | +|---|---|---| +| `REVIDS_HOST` | `0.0.0.0` | Bind address | +| `REVIDS_PORT` | `8000` | Port | +| `REVIDS_RELOAD` | `true` | uvicorn auto-reload | +| `REVIDS_LTX_DISTILLED_CHECKPOINT` | `models/ltx-2.3-22b-distilled-1.1.safetensors` | Distilled model path | +| `REVIDS_LTX_GEMMA_ROOT` | `models/gpt-4o-5805-ava-gguf-model` | Gemma text encoder path | +| `REVIDS_LTX_SPATIAL_UPSAMPLER` | *(none)* | Spatial upsampler model path | +| `REVIDS_LTX_QUANTIZATION` | `fp8_cast` | Quantization mode (`fp8_cast`, `fp8_scaled_mm`, or unset for bfloat16) | +| `REVIDS_LTX_LORAS` | *(empty)* | LoRA paths, colon-separated: `"models/lora1.safetensors:1.0,models/lora2.safetensors:0.5"` | +| `REVIDS_MAX_CONCURRENT_JOBS` | `1` | GPU concurrency | + +## Project Structure + +``` +revids/ + app/ + main.py # FastAPI app + routes + config.py # Settings (pydantic-settings) + models.py # Request/response schemas + service.py # LTX pipeline wrapper + job processing + database.py # SQLite job store + libs/ + LTX-2/ # LTX-2 git submodule + requirements.txt + videos/ # Generated video output (gitignored) +``` + +## Hardware Requirements + +- NVIDIA GPU with CUDA (>= 24GB VRAM for bfloat16, ~14GB with FP8) +- ~30GB disk for model weights + +## Memory Optimization + +```bash +export PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True +``` + +Set `REVIDS_LTX_QUANTIZATION=fp8_cast` (~40% VRAM reduction) or `fp8_scaled_mm` (Hopper GPUs only). + +## LoRA Support + +Configure LoRAs via environment variable: + +```bash +export REVIDS_LTX_LORAS="models/my-style.safetensors:1.0,models/my-motion.safetensors:0.5" +``` + +Or set as a comma-separated list of `path:scale` entries. diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..78e7a14 --- /dev/null +++ b/app/config.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +import os +from typing import Optional + +from pydantic_settings import BaseSettings + + +class Settings(BaseSettings): + app_name: str = "Revids" + host: str = "0.0.0.0" + port: int = 8000 + reload: bool = True + + # LTX-2.3 model paths (update after downloading models) + ltx_distilled_checkpoint: str = "models/ltx-2.3-22b-distilled-1.1.safetensors" + ltx_gemma_root: str = "models/gpt-4o-5805-ava-gguf-model" + ltx_spatial_upsampler: str | None = None + ltx_device: str = "cuda" + ltx_quantization: str = "fp8_cast" + + # LoRA paths (list of "path:scale" strings, env: REVIDS_LTX_LORAS="models/my-lora.safetensors:1.0") + ltx_loras: list[str] = [] + + video_output_dir: str = os.path.join(os.path.dirname(os.path.dirname(__file__)), "videos") + + # DB + db_path: str = os.path.join(os.path.dirname(__file__), "jobs.db") + + # Generation defaults + default_frames: int = 65 + default_fps: float = 24.0 + default_width: int = 768 + default_height: int = 512 + + # Concurrency + max_concurrent_jobs: int = 1 + + model_config = {"env_prefix": "REVIDS_", "env_file": ".env"} + + +settings = Settings() diff --git a/app/database.py b/app/database.py new file mode 100644 index 0000000..cbf3cdf --- /dev/null +++ b/app/database.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import os +from datetime import datetime, timezone +from typing import Optional + +import aiosqlite +from pydantic import BaseModel + + +VALID_FRAME_COUNTS = {8 * n + 1 for n in range(1, 129)} +VALID_STATUSES = {"pending", "processing", "completed", "failed"} + + +class JobRecord(BaseModel): + id: str + status: str = "pending" + prompt: Optional[str] = None + params: dict | None = None + error: Optional[str] = None + created_at: str + completed_at: Optional[str] = None + + +class JobDB: + def __init__(self, db_path: str) -> None: + self.db_path = db_path + + async def init(self) -> None: + os.makedirs(os.path.dirname(self.db_path) or ".", exist_ok=True) + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + """ + CREATE TABLE IF NOT EXISTS jobs ( + id TEXT PRIMARY KEY, + status TEXT NOT NULL DEFAULT 'pending', + prompt TEXT, + params TEXT, + error TEXT, + created_at TEXT NOT NULL, + completed_at TEXT + ) + """ + ) + await db.commit() + + async def create_job(self, job: JobRecord) -> None: + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + "INSERT INTO jobs (id, status, prompt, params, created_at) VALUES (?, ?, ?, ?, ?)", + ( + job.id, + job.status, + job.prompt, + _json_encode(job.params), + job.created_at, + ), + ) + await db.commit() + + async def get_job(self, job_id: str) -> JobRecord | None: + async with aiosqlite.connect(self.db_path) as db: + db.row_factory = aiosqlite.Row + async with db.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)) as cursor: + row = await cursor.fetchone() + if row is None: + return None + return JobRecord( + id=row["id"], + status=row["status"], + prompt=row["prompt"], + params=_json_decode(row["params"]), + error=row["error"], + created_at=row["created_at"], + completed_at=row["completed_at"], + ) + + async def update_status( + self, + job_id: str, + status: str, + error: Optional[str] = None, + ) -> None: + now = datetime.now(timezone.utc).isoformat() + async with aiosqlite.connect(self.db_path) as db: + if status in {"completed", "failed"}: + await db.execute( + "UPDATE jobs SET status = ?, error = ?, completed_at = ? WHERE id = ?", + (status, error, now, job_id), + ) + else: + await db.execute( + "UPDATE jobs SET status = ?, error = ? WHERE id = ?", + (status, error, job_id), + ) + await db.commit() + + async def delete_job(self, job_id: str) -> bool: + async with aiosqlite.connect(self.db_path) as db: + cur = await db.execute("DELETE FROM jobs WHERE id = ?", (job_id,)) + await db.commit() + return cur.rowcount > 0 + + +def _json_encode(obj: dict | None) -> str | None: + if obj is None: + return None + import json + + return json.dumps(obj) + + +def _json_decode(s: str | None) -> dict | None: + if s is None: + return None + import json + + return json.loads(s) diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..780dde0 --- /dev/null +++ b/app/main.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import logging +from contextlib import asynccontextmanager +from typing import AsyncGenerator + +from fastapi import FastAPI, File, Form, HTTPException, UploadFile +from fastapi.responses import FileResponse, JSONResponse + +from app.config import settings +from app.models import JobSubmitResponse, JobStatusResponse +from app.service import get_service + +logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") +logger = logging.getLogger(__name__) + + +@asynccontextmanager +async def lifespan(app: FastAPI) -> AsyncGenerator: + logger.info("Starting Revids API...") + await get_service() + yield + logger.info("Shutting down Revids API.") + + +app = FastAPI( + title=settings.app_name, + version="0.1.0", + lifespan=lifespan, +) + + +@app.get("/health") +async def health() -> JSONResponse: + import torch + + gpu_available = torch.cuda.is_available() + return JSONResponse({ + "status": "ok", + "gpu": gpu_available, + "gpu_name": torch.cuda.get_device_name(0) if gpu_available else None, + }) + + +@app.post("/generate", status_code=201) +async def generate( + image: UploadFile = File(..., description="Input image (PNG, JPG, etc.)"), + prompt: str | None = Form(None), + width: int = Form(768), + height: int = Form(512), + num_frames: int = Form(65), + fps: float = Form(24.0), + seed: int | None = Form(None), +) -> JobSubmitResponse: + from PIL import Image + + img_bytes = await image.read() + pil_image = Image.open(img_bytes).convert("RGB") + + validate_frame_count(num_frames) + + svc = await get_service() + job = await svc.submit_job(pil_image, { + "prompt": prompt, + "width": width, + "height": height, + "num_frames": num_frames, + "fps": fps, + "seed": seed, + }) + return JobSubmitResponse(job_id=job.id) + + +@app.get("/jobs/{job_id}") +async def get_job(job_id: str) -> JobStatusResponse: + svc = await get_service() + job = await svc.get_job(job_id) + if job is None: + raise HTTPException(status_code=404, detail="Job not found") + return JobStatusResponse( + job_id=job.id, + status=job.status, + prompt=job.prompt, + params=job.params, + error=job.error, + created_at=job.created_at, + completed_at=job.completed_at, + ) + + +@app.get("/jobs/{job_id}/download") +async def download_video(job_id: str) -> FileResponse: + svc = await get_service() + job = await svc.get_job(job_id) + if job is None: + raise HTTPException(status_code=404, detail="Job not found") + if job.status != "completed": + raise HTTPException(status_code=404, detail="Video not ready") + + video_path = await svc.get_video_path(job_id) + if video_path is None: + raise HTTPException(status_code=404, detail="Video file missing") + return FileResponse(video_path, media_type="video/mp4", filename=f"{job_id}.mp4") + + +@app.delete("/jobs/{job_id}") +async def delete_job(job_id: str) -> dict: + svc = await get_service() + job = await svc.get_job(job_id) + if job is None: + raise HTTPException(status_code=404, detail="Job not found") + deleted, _ = await svc.delete_job(job_id) + if not deleted: + raise HTTPException(status_code=404, detail="Job not found") + return {"deleted": job_id} + + +def validate_frame_count(num_frames: int) -> None: + valid = {8 * n + 1 for n in range(1, 129)} + if num_frames not in valid: + candidates = sorted(f for f in valid if abs(f - num_frames) <= 8) + raise HTTPException( + status_code=400, + detail=f"num_frames must be 8n+1. Got {num_frames}. Closest valid: {candidates[:4]}", + ) + + +@app.exception_handler(HTTPException) +async def http_exception_handler(_, exc: HTTPException) -> JSONResponse: + return JSONResponse( + status_code=exc.status_code, + content={"error": exc.detail}, + ) diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..21c327d --- /dev/null +++ b/app/models.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import uuid +from typing import Optional + +from pydantic import BaseModel, Field + + +# --- Request --- + +class GenerateRequest(BaseModel): + prompt: Optional[str] = Field(None, description="Optional text prompt to guide generation") + width: int = Field(768, ge=256, le=2048, multiple_of=32, description="Video width (must be divisible by 32)") + height: int = Field(512, ge=256, le=2048, multiple_of=32, description="Video height (must be divisible by 32)") + num_frames: int = Field(65, ge=9, le=257, description="Frame count (must be 8n+1)") + fps: float = Field(24.0, gt=0, le=60, description="Frames per second") + seed: Optional[int] = Field(None, description="Random seed for reproducibility") + + +# --- Job Status --- + +class JobStatusResponse(BaseModel): + job_id: str + status: str # pending, processing, completed, failed + prompt: Optional[str] = None + params: Optional[dict] = None + error: Optional[str] = None + created_at: Optional[str] = None + completed_at: Optional[str] = None + + +# --- Submission Response --- + +class JobSubmitResponse(BaseModel): + job_id: str + status: str = "pending" + message: str = "Job submitted. Poll GET /jobs/{job_id} for status." + + +# --- Error --- + +class ErrorResponse(BaseModel): + error: str diff --git a/app/service.py b/app/service.py new file mode 100644 index 0000000..a89df7d --- /dev/null +++ b/app/service.py @@ -0,0 +1,177 @@ +from __future__ import annotations + +import asyncio +import os +import uuid +from datetime import datetime, timezone +from typing import Optional + +import torch +from ltx_pipelines import DistilledPipeline +from PIL import Image + +from app.config import settings +from app.database import JobDB, JobRecord, VALID_FRAME_COUNTS + + +class LtxService: + def __init__(self, db: JobDB) -> None: + self.db = db + self.pipeline: Optional[DistilledPipeline] = None + self._load_lock = asyncio.Lock() + self._job_semaphore = asyncio.Semaphore(settings.max_concurrent_jobs) + self._video_dir = settings.video_output_dir + + async def load_pipeline(self) -> None: + async with self._load_lock: + if self.pipeline is not None: + return + lora_config = self._parse_lora_paths(settings.ltx_loras) + self.pipeline = await asyncio.to_thread( + DistilledPipeline, + distilled_checkpoint_path=settings.ltx_distilled_checkpoint, + gemma_root=settings.ltx_gemma_root, + spatial_upsampler_path=settings.ltx_spatial_upsampler, + loras=lora_config, + device=settings.ltx_device, + quantization=settings.ltx_quantization, + ) + + @staticmethod + def _parse_lora_paths(lora_specs: list[str]) -> list[dict]: + result = [] + for spec in lora_specs: + if ":" in spec: + path, scale = spec.rsplit(":", 1) + result.append({"path": path.strip(), "scale": float(scale.strip())}) + else: + result.append({"path": spec.strip(), "scale": 1.0}) + return result + + async def submit_job(self, image: Image.Image, request_params: dict) -> JobRecord: + job_id = uuid.uuid4().hex[:12] + now = datetime.now(timezone.utc).isoformat() + prompt = request_params.get("prompt") + width = request_params.get("width", settings.default_width) + height = request_params.get("height", settings.default_height) + num_frames = request_params.get("num_frames", settings.default_frames) + fps = request_params.get("fps", settings.default_fps) + seed = request_params.get("seed") + + if num_frames not in VALID_FRAME_COUNTS: + candidates = sorted( + f for f in VALID_FRAME_COUNTS if abs(f - num_frames) <= 8 + ) + raise ValueError( + f"num_frames must be 8n+1. Got {num_frames}. " + f"Closest valid: {candidates[:4]}" + ) + + job = JobRecord( + id=job_id, + status="pending", + prompt=prompt, + params={ + "width": width, + "height": height, + "num_frames": num_frames, + "fps": fps, + "seed": seed, + }, + created_at=now, + ) + await self.db.create_job(job) + asyncio.create_task(self._process_job(job_id, image, job.params)) + return job + + async def _process_job( + self, + job_id: str, + image: Image.Image, + params: dict, + ) -> None: + async with self._job_semaphore: + try: + await self.db.update_status(job_id, "processing") + await self.load_pipeline() + + prompt = params.get("prompt", "") or "" + width = params.get("width", settings.default_width) + height = params.get("height", settings.default_height) + num_frames = params.get("num_frames", settings.default_frames) + fps = params.get("fps", settings.default_fps) + seed = params.get("seed") if params.get("seed") is not None else 42 + + if image.mode != "RGB": + image = image.convert("RGB") + image = image.resize((width, height), Image.LANCZOS) + + output = await asyncio.to_thread( + self.pipeline, + prompt=prompt, + images=image, + width=width, + height=height, + num_frames=num_frames, + frame_rate=fps, + seed=seed, + ) + + video_path = os.path.join(self._video_dir, f"{job_id}.mp4") + os.makedirs(self._video_dir, exist_ok=True) + await asyncio.to_thread(self._save_video, output, video_path, fps) + await self.db.update_status(job_id, "completed") + + except Exception as e: + await self.db.update_status(job_id, "failed", error=str(e)) + + @staticmethod + def _save_video(output: any, path: str, fps: float) -> None: + import imageio.v3 as iio + + if isinstance(output, torch.Tensor): + frames = output.cpu().numpy() + elif isinstance(output, dict) and "video" in output: + v = output["video"] + frames = v.cpu().numpy() if isinstance(v, torch.Tensor) else v + else: + frames = output + if isinstance(frames, torch.Tensor): + frames = frames.cpu().numpy() + + if frames.ndim == 4: + frames = frames[0] + if frames.max() <= 1.0: + frames = (frames * 255).astype("uint8") + elif frames.dtype != "uint8": + frames = frames.clip(0, 255).astype("uint8") + + iio.imwrite(path, frames, fps=float(fps), codec="libx264") + + async def get_job(self, job_id: str) -> JobRecord | None: + return await self.db.get_job(job_id) + + async def delete_job(self, job_id: str) -> tuple[bool, str | None]: + video_path = os.path.join(self._video_dir, f"{job_id}.mp4") + deleted = await self.db.delete_job(job_id) + if os.path.exists(video_path): + os.remove(video_path) + return deleted, job_id + + async def get_video_path(self, job_id: str) -> str | None: + video_path = os.path.join(self._video_dir, f"{job_id}.mp4") + if os.path.exists(video_path): + return video_path + return None + + +_service: Optional[LtxService] = None + + +async def get_service() -> LtxService: + global _service + if _service is None: + db = JobDB(settings.db_path) + await db.init() + _service = LtxService(db) + return _service diff --git a/libs/.gitkeep b/libs/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/libs/LTX-2 b/libs/LTX-2 new file mode 160000 index 0000000..d605370 --- /dev/null +++ b/libs/LTX-2 @@ -0,0 +1 @@ +Subproject commit d6053703e00195bc668cbd1d5eda9dc0b2e7b74a diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..bffbc50 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +fastapi>=0.115.0 +uvicorn[standard]>=0.33.0 +python-multipart>=0.0.12 +Pillow>=11.0.0 +aiosqlite>=0.20.0 +pydantic-settings>=2.6.0 +imageio[ffmpeg]>=2.34.0