Skip to content

C
coderai
Commit 5abbb81d authored by Stefy Lanza (nextime / spora )'s avatar Stefy Lanza (nextime / spora )
Browse files
Options

compose: POST /v1/video/compose and POST /v1/files/upload (v0.2.22) 

CoderAI could make every piece of a short video and nothing turned the pieces
into a reel — assembly needs ffmpeg and two facts only the server has (how long
the narration it just synthesised is, and when each word falls), so a detached
client cannot do it. Requested by the Motus project over the HTTP API.

- codai/api/compose.py: scenes -> one uploadable video. Scene length = narration
  + padding (min_duration floor), visuals split it evenly; async job with
  stage/progress/message/warnings, cancel (the ffmpeg child is killed), a queue
  that reports its position, and a blocking form for tests. Narration, music and
  word timings come from this server's own endpoints, called on a loop this
  module owns — never the request's loop, which may be closed before the render
  needs it. 400 names the offending scene/field; 501 when a sub-feature has no
  model (music.generate, timing=stt, TTS for narration) or ffmpeg is missing.
- codai/compose/render.py: one normalised clip per visual (so a failed visual
  becomes a gradient and a warning instead of a lost render), concat or xfade
  that preserves the total duration, narration padded per scene and
  loudness-normalised, music looped/faded/ducked with sidechaincompress, burned
  captions + overlays with the final H.264 High/AAC/+faststart encode, thumbnail.
- codai/compose/captions.py: word timings from STT aligned onto the CALLER's
  words (a mis-heard word never reaches the screen) or estimated; line breaking;
  ASS with six presets inside a 10%/6% safe area; SRT/VTT with the same text.
- codai/compose/media.py: resolve a URL / /v1/files path / data URI / base64
  (never a server path), probe with ffprobe, resolve fonts through fontconfig.
- codai/api/uploads.py: POST /v1/files/upload, content-addressed by sha256 with
  GET /v1/files/blob/{hash} so a client can skip a re-upload; type decided by
  content, not by the client's claim.
- compose and uploads are pinned local in the capability gateway (a pod has
  ffmpeg but not this install's models or files); both advertised in the
  capability document; caption fonts (Montserrat/Inter/Noto CJK) added to the
  three image builds.
- docs/video-compose.md; tests/test_compose.py (30, real ffmpeg, no model/GPU).
Co-Authored-By: 's avatarClaude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01B7EnDW6RSd2nhEPZMzMkqi
parent 36ae2ba8
Hide whitespace changes
Inline Side-by-side
Showing with 3292 additions and 4 deletions
README.md
View file @ 5abbb81d
......@@ -205,6 +205,7 @@ Full guide: [`docs/cluster.md`](docs/cluster.md).
- **Character Profiles**: Apply up to 6 saved character profiles for visual consistency. Pipelines with IP-Adapter (AnimateDiff and other SD-based video models) are conditioned directly. For the rest — Wan, CogVideoX, LTX, SVD, i.e. most video models — CoderAI renders an **identity keyframe** on the image endpoint (which does have real IP-Adapter) and runs the video as `ti2v` from that frame, so `character_profiles` locks a face on every model. Controlled by `keyframe_identity` (`auto` / `always` / `never`) with optional `keyframe_model` and `keyframe_steps`; combine with a per-character LoRA for the strongest consistency
- **Environment Profiles**: Apply up to 6 saved environment profiles to condition the scene
- **Multi-Character Dialog**: Assign spoken lines to individual characters — each line picks a character profile, voice profile or TTS voice ID, and text; lines are **voice-cloned with F5-TTS** from the named profile, mixed with correct timing (sequential or manual), and lip-synced via Wav2Lip
- **Composition (reels)**: `POST /v1/video/compose` turns a list of **scenes** — narration plus visuals — into one finished, uploadable video: the scene's length comes from its narration, visuals are Ken-Burns'd or trimmed to fit, music is looped and ducked under the voice, captions are burned in with word timings (karaoke presets) inside the platforms' safe area, and a thumbnail, SRT/VTT and the narration track come back with it. Async jobs with stage/progress/cancel; a visual that fails to download becomes a gradient and a warning instead of a failed render. See [`docs/video-compose.md`](docs/video-compose.md)
- **Lip Sync**: Wav2Lip and SadTalker. The Docker/OCI image ships both (launcher shims plus a dedicated lip-sync venv; checkpoints download on first use). A from-source install has no shims, so CoderAI falls back to a self-managed Wav2Lip that fetches and patches itself on first use. Either way, if lip sync can't run the response carries a `warnings` entry saying the audio was muxed without mouth sync — rather than passing off an unsynced video as a success
### Audio
......@@ -802,6 +803,8 @@ persisted per `config_id`, so the configs don't overwrite each other.
| `POST /v1/embeddings` | Embeddings — text, image, multi-vector (dense/sparse/colbert), geolocation, VPR |
| `POST /v1/rerank` | Cross-encoder reranking of documents against a query |
| `GET /v1/files/{filename}` | Fetch a generated file |
| `POST /v1/files/upload` | Upload your own video/image/audio (content-addressed); returns a `/v1/files/...` URL to reference in later requests |
| `GET /v1/files/blob/{sha256}` | Does this install already hold these bytes? (skip a re-upload) |
### Image
......@@ -826,6 +829,8 @@ persisted per `config_id`, so the configs don't overwrite each other.
| Endpoint | Description |
|---|---|
| `POST /v1/video/generations` | Generate video (t2v/i2v/v2v/ti2v/interp) — supports `character_profiles`, `environment_profiles`, `dialogs` |
| `POST /v1/video/compose` | Compose scenes into a finished reel (narration timing, visuals, ducked music, burned captions, thumbnail) — async job |
| `GET /v1/video/compose/{id}` | Composition progress / result; `POST …/{id}/cancel` stops it |
| `POST /v1/video/upscale` | Upscale video |
| `POST /v1/video/subtitle` | Generate/burn subtitles |
| `POST /v1/video/interpolate` | Frame interpolation |
......
codai/__init__.py
View file @ 5abbb81d
......@@ -16,7 +16,7 @@
# Canonical product version for CoderAI — single source of truth. Both the API
# metadata and the admin web UI read from here.
__version__ = "0.2.21"
__version__ = "0.2.22"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
codai/api/app.py
View file @ 5abbb81d
......@@ -199,6 +199,8 @@ from codai.api.images import router as images_router
from codai.api.tts import router as tts_router
from codai.api.text import router as text_router
from codai.api.video import router as video_router
from codai.api.compose import router as compose_router
from codai.api.uploads import router as uploads_router
from codai.api.audio_gen import router as audio_gen_router
from codai.api.audio_stems import router as audio_stems_router
from codai.api.audio_clean import router as audio_clean_router
......@@ -691,6 +693,11 @@ app.include_router(images_router, tags=["Images"])
app.include_router(tts_router, tags=["Audio"])
app.include_router(text_router, tags=["Text"])
app.include_router(video_router, tags=["Video"])
# Composition and generic media uploads. Registered BEFORE the /v1/files/{filename}
# download route defined further down so /v1/files/upload and /v1/files/blob/{hash}
# are matched as themselves rather than as a filename.
app.include_router(compose_router, tags=["Video"])
app.include_router(uploads_router, tags=["Files"])
app.include_router(audio_gen_router, tags=["Audio"])
app.include_router(audio_stems_router, tags=["Audio"])
app.include_router(audio_clean_router, tags=["Audio"])
......
codai/api/compose.py 0 → 100644
View file @ 5abbb81d
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
"""``POST /v1/video/compose`` — N pieces in, one finished video out.
CoderAI could already make every piece of a short video; nothing turned the
pieces into a reel, because that step needs ffmpeg *and* facts only the server
has — how long the narration it just synthesised actually is, and when each
word falls. A client on another machine cannot know either, so composition has
to live here.
A request is an ordered list of **scenes**. Each scene has narration (text to
synthesise, or audio you already have) and one or more **visuals**; the scene
lasts as long as its narration plus its padding, and its visuals split that time
between them. Everything else — voice, music, captions, overlays, thumbnail — is
defaults you can override per request.
The work is queued and run one render at a time (a render is ffmpeg-bound and
two at once are slower than one after the other), reported through
``GET /v1/video/compose/{id}`` with a stage and a percentage, and stoppable with
``POST /v1/video/compose/{id}/cancel``. A visual that cannot be fetched or
decoded never fails the job: it becomes a gradient and a line in ``warnings``.
The pipeline itself is :mod:`codai.compose.render`; captions are
:mod:`codai.compose.captions`; this module is the API, the queue and the
orchestration between them.
Narration, music and word timings are taken from this server's own endpoints,
called in-process on a loop this module owns (never the loop of the request that
queued the job — that one belongs to one request and may be closed long before
the render needs it). In-process means they run in whichever engine the front
routed the compose request to, so on a multi-engine install the TTS/STT models
must be loadable there; ``docs/video-compose.md`` says so where an operator will
read it.
"""
import asyncio
import base64
import os
import shutil
import tempfile
import threading
import time
import uuid
from typing import Any, Dict, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, ConfigDict, Field
from codai.api.loras import _require_api_auth
from codai.compose import captions as cap
from codai.compose import media
from codai.compose import render as rnd
router = APIRouter()
def set_global_file_path(path: str) -> None:
"""Where the composed artefacts are written (``/v1/files`` serves them)."""
media.set_global_file_path(path)
_MAX_SCENES = 200
_MAX_VISUALS_PER_SCENE = 12
_JOB_HISTORY = 100
_STAGES = ("queued", "narration", "captions", "visuals", "music", "render", "thumbnail")
# =============================================================================
# Request
# =============================================================================
class Canvas(BaseModel):
width: int = 1080
height: int = 1920
fps: int = 30
background: str = "#000000"
model_config = ConfigDict(extra="allow")
class VoiceSpec(BaseModel):
engine: str = "tts" # tts → /v1/audio/speech, clone → /v1/audio/clone
model: Optional[str] = None
voice: Optional[str] = None
voice_name: Optional[str] = None # saved voice profile (engine=clone)
language: Optional[str] = None
speed: float = 1.0
model_config = ConfigDict(extra="allow")
class Visual(BaseModel):
type: str = "image" # video | image | gradient | color
src: Optional[str] = None
trim_start: float = 0.0
ken_burns: Optional[Any] = None
colors: Optional[List[str]] = None
color: Optional[str] = None
model_config = ConfigDict(extra="allow")
class Scene(BaseModel):
text: Optional[str] = None
audio: Optional[str] = None
duration: Optional[float] = None # required when there is no narration
min_duration: float = 0.0
padding: float = 0.15
visuals: List[Visual] = Field(default_factory=list)
voice: Optional[VoiceSpec] = None # per-scene override (a second speaker)
model_config = ConfigDict(extra="allow")
class MusicSpec(BaseModel):
src: Optional[str] = None
generate: Optional[Dict[str, Any]] = None
volume: float = 0.14
loop: bool = True
duck: bool = True
fade_in: float = 1.0
fade_out: float = 2.0
model_config = ConfigDict(extra="allow")
class CaptionSpec(BaseModel):
enabled: bool = True
timing: str = "auto" # auto | stt | estimate
stt_model: Optional[str] = None
preset: str = "karaoke"
font: str = "Montserrat"
font_scale: float = 1.0
color: str = "#FFFFFF"
highlight_color: str = "#FFE14D"
outline_color: str = "#000000"
position: str = "center"
max_words_per_line: int = 4
max_lines: int = 2
uppercase: bool = False
model_config = ConfigDict(extra="allow")
class Overlay(BaseModel):
type: str = "text" # text | image
text: Optional[str] = None
src: Optional[str] = None
position: str = "bottom-right"
opacity: float = 1.0
size: float = 0.025 # text height as a fraction of the canvas
width: float = 0.15 # image width as a fraction of the canvas
color: str = "#FFFFFF"
font: Optional[str] = None
model_config = ConfigDict(extra="allow")
class ThumbnailSpec(BaseModel):
at: float = 0.25
text: Optional[str] = None
text_color: str = "#FFFFFF"
highlight_color: str = "#FFE14D"
font: Optional[str] = None
model_config = ConfigDict(extra="allow")
class OutputSpec(BaseModel):
format: str = "mp4"
video_codec: str = "h264"
crf: int = 20
audio_bitrate: str = "192k"
model_config = ConfigDict(extra="allow")
class ComposeRequest(BaseModel):
scenes: List[Scene]
canvas: Canvas = Field(default_factory=Canvas)
voice: VoiceSpec = Field(default_factory=VoiceSpec)
visual_fit: str = "cover" # cover | contain | blur_pad
transition: Dict[str, Any] = Field(default_factory=lambda: {"type": "none",
"duration": 0.3})
music: Optional[MusicSpec] = None
captions: Optional[CaptionSpec] = None
overlays: List[Overlay] = Field(default_factory=list)
thumbnail: Optional[ThumbnailSpec] = None
output: OutputSpec = Field(default_factory=OutputSpec)
outputs: List[str] = Field(default_factory=lambda: ["video"])
is_async: bool = Field(default=True, alias="async")
model_config = ConfigDict(extra="allow", populate_by_name=True)
# =============================================================================
# Jobs
# =============================================================================
_jobs: Dict[str, dict] = {}
_jobs_lock = threading.Lock()
_cancels: Dict[str, threading.Event] = {}
_queue: List[str] = [] # job ids waiting, in order
_worker: Optional[threading.Thread] = None
_worker_wake = threading.Event()
_pending: Dict[str, tuple] = {} # job id -> (request, loop, base_url)
def _now() -> float:
return time.time()
def _new_job(job_id: str, scenes: int) -> dict:
rec = {"id": job_id, "status": "queued", "stage": "queued", "progress": 0.0,
"message": "queued", "warnings": [], "error": None, "result": None,
"scenes": scenes, "created_at": _now(), "updated_at": _now()}
with _jobs_lock:
_jobs[job_id] = rec
_prune_locked()
return dict(rec)
def _prune_locked() -> None:
done = [(jid, r) for jid, r in _jobs.items()
if r.get("status") in ("done", "failed", "cancelled")]
excess = len(done) - _JOB_HISTORY
if excess <= 0:
return
done.sort(key=lambda kv: kv[1].get("updated_at") or 0)
for jid, _ in done[:excess]:
_jobs.pop(jid, None)
_cancels.pop(jid, None)
def _update(job_id: str, **fields) -> None:
with _jobs_lock:
rec = _jobs.get(job_id)
if rec is None:
return
warn = fields.pop("warning", None)
if warn:
rec.setdefault("warnings", []).append(str(warn)[:300])
rec.update(fields)
rec["updated_at"] = _now()
def _get(job_id: str) -> Optional[dict]:
with _jobs_lock:
rec = _jobs.get(job_id)
return dict(rec) if rec else None
def _public(rec: dict) -> dict:
"""The document a client polls: the spec's fields, in the spec's order."""
return {"id": rec["id"], "status": rec["status"], "stage": rec.get("stage"),
"progress": round(float(rec.get("progress") or 0.0), 1),
"message": rec.get("message") or "", "warnings": rec.get("warnings") or [],
"error": rec.get("error"), "result": rec.get("result")}
def cancel_job(job_id: str) -> bool:
rec = _get(job_id)
if rec is None:
return False
if rec["status"] in ("done", "failed", "cancelled"):
return True
ev = _cancels.get(job_id)
if ev is not None:
ev.set()
with _jobs_lock:
_queue[:] = [j for j in _queue if j != job_id]
_update(job_id, status="cancelled", message="cancelled by the client",
stage=rec.get("stage"))
return True
def _queue_positions() -> None:
"""Tell every waiting job where it is in the line (Motus shows this)."""
with _jobs_lock:
waiting = list(_queue)
for pos, jid in enumerate(waiting, 1):
rec = _get(jid)
if rec and rec["status"] == "queued":
_update(jid, message=f"queued: {pos} ahead of this job"
if pos > 1 else "queued: next")
def _ensure_worker() -> None:
global _worker
if _worker is not None and _worker.is_alive():
return
_worker = threading.Thread(target=_worker_loop, name="compose-worker", daemon=True)
_worker.start()
def _worker_loop() -> None:
"""One render at a time, in submission order."""
while True:
_worker_wake.wait(timeout=5.0)
_worker_wake.clear()
while True:
with _jobs_lock:
job_id = _queue.pop(0) if _queue else None
if job_id is None:
break
args = _pending.pop(job_id, None)
rec = _get(job_id)
if args is None or rec is None or rec["status"] == "cancelled":
continue
_queue_positions()
req, base_url = args
try:
# The loop is started on demand: a job whose narration is supplied
# and whose captions are estimated never needs one.
_run_job(job_id, req, None, base_url)
except rnd.Cancelled:
_update(job_id, status="cancelled", message="cancelled by the client")
except Exception as exc: # never kill the worker
_update(job_id, status="failed", error=str(exc)[:500],
message=f"failed: {exc}"[:300])
def submit(req: ComposeRequest, base_url: str) -> dict:
job_id = "cmp_" + uuid.uuid4().hex[:10]
rec = _new_job(job_id, len(req.scenes))
_cancels[job_id] = threading.Event()
_pending[job_id] = (req, base_url)
with _jobs_lock:
_queue.append(job_id)
_ensure_worker()
_worker_wake.set()
_queue_positions()
return rec
# =============================================================================
# Validation
# =============================================================================
def _validate(req: ComposeRequest) -> None:
"""Reject a spec that cannot be rendered, naming what is wrong and where."""
if not req.scenes:
raise HTTPException(status_code=400, detail="scenes: at least one scene is required")
if len(req.scenes) > _MAX_SCENES:
raise HTTPException(status_code=400,
detail=f"scenes: at most {_MAX_SCENES} scenes per request")
c = req.canvas
if not (16 <= c.width <= 7680 and 16 <= c.height <= 7680):
raise HTTPException(status_code=400, detail="canvas: width/height out of range")
if c.width % 2 or c.height % 2:
raise HTTPException(status_code=400,
detail="canvas: width and height must be even (H.264)")
if not (1 <= c.fps <= 120):
raise HTTPException(status_code=400, detail="canvas.fps: must be 1..120")
if (req.visual_fit or "cover").lower() not in ("cover", "contain", "blur_pad"):
raise HTTPException(status_code=400,
detail="visual_fit: expected cover, contain or blur_pad")
ttype = str((req.transition or {}).get("type") or "none").lower()
if ttype not in ("none", "fade", "crossfade", "slide", "wipe", "dissolve"):
raise HTTPException(status_code=400, detail=(
"transition.type: expected none, fade, crossfade, slide, wipe or dissolve"))
for i, sc in enumerate(req.scenes):
has_narration = bool((sc.text or "").strip()) or bool(sc.audio)
if not has_narration and not (sc.duration and sc.duration > 0):
raise HTTPException(status_code=400, detail=(
f"scenes[{i}]: a scene with no text and no audio needs an explicit "
f"\"duration\""))
if not sc.visuals:
raise HTTPException(status_code=400,
detail=f"scenes[{i}].visuals: at least one visual is required")
if len(sc.visuals) > _MAX_VISUALS_PER_SCENE:
raise HTTPException(status_code=400, detail=(
f"scenes[{i}].visuals: at most {_MAX_VISUALS_PER_SCENE} per scene"))
for j, v in enumerate(sc.visuals):
vt = (v.type or "").lower()
if vt not in ("video", "image", "gradient", "color"):
raise HTTPException(status_code=400, detail=(
f"scenes[{i}].visuals[{j}].type: expected video, image, gradient "
f"or color, got {v.type!r}"))
if vt in ("video", "image") and not (v.src or "").strip():
raise HTTPException(status_code=400, detail=(
f"scenes[{i}].visuals[{j}].src: required for type {vt!r}"))
caps = req.captions
if caps and caps.enabled and (caps.timing or "auto").lower() not in ("auto", "stt", "estimate"):
raise HTTPException(status_code=400,
detail="captions.timing: expected auto, stt or estimate")
if caps and caps.enabled and (caps.preset or "karaoke").lower() not in cap.PRESETS:
raise HTTPException(status_code=400, detail=(
"captions.preset: expected one of " + ", ".join(sorted(cap.PRESETS))))
unknown = [o for o in req.outputs
if o not in ("video", "thumbnail", "srt", "vtt", "narration")]
if unknown:
raise HTTPException(status_code=400, detail=(
f"outputs: unknown artefact(s) {unknown}; expected video, thumbnail, "
f"srt, vtt, narration"))
if (req.output.format or "mp4").lower() not in ("mp4", "mov", "webm", "mkv"):
raise HTTPException(status_code=400,
detail="output.format: expected mp4, mov, webm or mkv")
if shutil.which("ffmpeg") is None:
raise HTTPException(status_code=501, detail=(
"composition needs ffmpeg on PATH and this install has none "
"(the published images ship it)"))
# Sub-features that need a model: answer 501 now rather than failing a job
# minutes later.
if req.music and req.music.generate and not _has_model("audio_gen"):
raise HTTPException(status_code=501, detail=(
"music.generate needs an audio-generation model (e.g. musicgen) configured "
"on this install; supply music.src instead"))
if caps and caps.enabled and (caps.timing or "auto").lower() == "stt" \
and not _has_model("audio", caps.stt_model):
raise HTTPException(status_code=501, detail=(
f"captions.timing=stt needs a speech-to-text model"
f"{' called ' + repr(caps.stt_model) if caps.stt_model else ''} on this "
f"install; use \"estimate\" or \"auto\""))
if _voice_needed(req) and not _voice_available(req):
raise HTTPException(status_code=501, detail=(
"narration needs a TTS model (engine=tts) or a saved voice profile "
"(engine=clone) on this install; pass per-scene \"audio\" instead"))
def _model_ids(mtype: str) -> List[str]:
try:
from codai.models.manager import multi_model_manager
return [m.id for m in multi_model_manager.list_models()
if (m.type or "") == mtype]
except Exception:
return []
def _has_model(mtype: str, name: str = "") -> bool:
ids = _model_ids(mtype)
if not ids:
return False
if not name:
return True
tail = name.rstrip("/").split("/")[-1].lower()
return any(name.lower() == i.lower() or i.rstrip("/").split("/")[-1].lower() == tail
for i in ids)
def _voice_needed(req: ComposeRequest) -> bool:
return any((sc.text or "").strip() and not sc.audio for sc in req.scenes)
def _voice_available(req: ComposeRequest) -> bool:
engines = {(sc.voice or req.voice).engine or "tts" for sc in req.scenes}
if "clone" in engines:
try:
from codai.api.voice_clone import _load_voice
names = {(sc.voice or req.voice).voice_name for sc in req.scenes
if ((sc.voice or req.voice).engine or "tts") == "clone"}
if not all(n and _load_voice(n) for n in names):
return False
except Exception:
return False
if engines - {"clone"}:
return _has_model("tts", req.voice.model or "")
return True
# =============================================================================
# Pieces: narration, music, word timings
# =============================================================================
#: Narration, music and speech recognition are async endpoints, and a render
#: runs in a worker thread. It must NOT borrow the event loop of the request
#: that queued the job: that loop belongs to one request and may be closed long
#: before the render reaches its narration (which is exactly what happens under
#: a test client, and under any server that finishes a request early). So this
#: module owns one long-lived loop in a thread of its own.
_loop: Optional[asyncio.AbstractEventLoop] = None
_loop_lock = threading.Lock()
def _media_loop() -> asyncio.AbstractEventLoop:
global _loop
with _loop_lock:
if _loop is not None and not _loop.is_closed():
return _loop
_loop = asyncio.new_event_loop()
threading.Thread(target=_loop.run_forever, name="compose-loop",
daemon=True).start()
return _loop
def _await(loop, coro, timeout: float = 1800.0):
"""Run one of the API's coroutines from the render thread."""
fut = asyncio.run_coroutine_threadsafe(coro, loop or _media_loop())
return fut.result(timeout=timeout)
def _b64_to_file(b64: str, path: str) -> str:
with open(path, "wb") as f:
f.write(base64.b64decode(b64))
return path
def _synth_scene(loop, text: str, voice: VoiceSpec, out_path: str) -> str:
"""One scene's narration as a wav file, through the ordinary endpoints."""
engine = (voice.engine or "tts").lower()
if engine == "clone":
from codai.api.voice_clone import VoiceCloneRequest, clone_voice
req = VoiceCloneRequest(text=text, voice_name=voice.voice_name,
speed=voice.speed or 1.0, language=voice.language)
res = _await(loop, clone_voice(req, None))
# /v1/audio/clone answers {created, data: [{url}|{b64_wav}], engine}: a
# url when this install has an output directory, inline audio otherwise.
item = ((res or {}).get("data") or [{}])[0]
blob = item.get("b64_wav") or (res or {}).get("audio")
if blob:
return _b64_to_file(blob, out_path)
url = item.get("url") or (res or {}).get("url")
if not url:
raise rnd.RenderError("voice cloning returned no audio")
local = media.local_path_for_files_url(url)
if local:
return local
with open(out_path, "wb") as f:
f.write(media.fetch(url))
return out_path
from codai.api.tts import TTSRequest, create_speech
req = TTSRequest(model=voice.model or "", input=text,
voice=voice.voice or "af_sarah", response_format="wav",
speed=voice.speed or 1.0)
if voice.language:
setattr(req, "language", voice.language)
if voice.voice_name:
req.voice_profile = voice.voice_name
res = _await(loop, create_speech(req, None))
blob = (res or {}).get("audio")
if not blob:
raise rnd.RenderError("TTS returned no audio")
return _b64_to_file(blob, out_path)
def _generate_music(loop, spec: dict, total: float, out_path: str) -> str:
from codai.api.audio_gen import AudioGenerationRequest, audio_generate
req = AudioGenerationRequest(
model=str(spec.get("model") or ""),
prompt=str(spec.get("prompt") or "calm background music"),
duration=float(spec.get("duration") or max(10.0, min(total, 120.0))),
response_format="b64_wav")
res = _await(loop, audio_generate(req, None))
data = (res or {}).get("data") or []
if not data:
raise rnd.RenderError("music generation returned nothing")
item = data[0]
blob = item.get("b64_wav") or item.get("b64_json")
if blob:
return _b64_to_file(blob, out_path)
url = item.get("url")
if not url:
raise rnd.RenderError("music generation returned no audio")
with open(out_path, "wb") as f:
f.write(media.fetch(url))
return out_path
def _stt_words(loop, audio_path: str, model: str, language: Optional[str]) -> List[dict]:
"""Word timestamps for a narration clip (only the timings are used)."""
from codai.api.transcriptions import _run_transcription
with open(audio_path, "rb") as f:
content = f.read()
res = _await(loop, _run_transcription(
content, model, language, None, "verbose_json", 0.0,
os.path.basename(audio_path), word_timestamps=True))
if hasattr(res, "body"): # a PlainTextResponse: no timings in it
return []
return list((res or {}).get("words") or [])
def _pick_stt_model(caps: CaptionSpec) -> Optional[str]:
if caps.stt_model and _has_model("audio", caps.stt_model):
return caps.stt_model
ids = _model_ids("audio")
if not ids:
return None
# Prefer something whisper-shaped: it is what reports word timings.
for want in ("whisper", "canary", "wav2vec"):
for i in ids:
if want in i.lower():
return i
return ids[0]
# =============================================================================
# The render
# =============================================================================
def _run_job(job_id: str, req: ComposeRequest, loop, base_url: str) -> None:
cancel = _cancels.get(job_id) or threading.Event()
work = tempfile.mkdtemp(prefix=f"coderai-compose-{job_id}-")
canvas = req.canvas
fit = (req.visual_fit or "cover").lower()
ttype = str((req.transition or {}).get("type") or "none").lower()
tdur = float((req.transition or {}).get("duration") or 0.3)
caps = req.captions
log_lines: List[str] = []
def log(msg: str) -> None:
log_lines.append(msg)
if len(log_lines) > 200:
del log_lines[:100]
def stage(name: str, progress: float, message: str) -> None:
if cancel.is_set():
raise rnd.Cancelled()
_update(job_id, stage=name, progress=progress, message=message[:300],
status="running")
def warn(msg: str) -> None:
_update(job_id, warning=msg)
print(f"[compose {job_id}] WARNING {msg}", flush=True)
try:
_update(job_id, status="running", stage="narration", progress=1.0,
message="starting")
# ---------------------------------------------------- 1. narration
n_scenes = len(req.scenes)
scene_audio: List[Optional[str]] = []
scene_dur: List[float] = []
for i, sc in enumerate(req.scenes):
stage("narration", 2.0 + 26.0 * i / max(1, n_scenes),
f"scene {i + 1}/{n_scenes}: narration")
apath = None
if sc.audio:
try:
apath = media.resolve(sc.audio, work, f"scene{i}-audio")
except media.MediaError as exc:
warn(f"scene {i}: audio could not be fetched ({exc}); "
f"synthesising from text instead")
if apath is None and (sc.text or "").strip():
apath = _synth_scene(loop, sc.text, sc.voice or req.voice,
os.path.join(work, f"scene{i}-voice.wav"))
dur = 0.0
if apath:
try:
dur = media.duration_of(apath)
except media.MediaError as exc:
warn(f"scene {i}: narration could not be probed ({exc})")
apath, dur = None, 0.0
if apath:
dur = dur + max(0.0, float(sc.padding or 0.0))
else:
dur = float(sc.duration or 0.0)
dur = max(dur, float(sc.min_duration or 0.0))
if dur <= 0:
dur = 3.0
warn(f"scene {i}: no narration and no duration; used 3.0s")
scene_audio.append(apath)
scene_dur.append(round(dur, 3))
total = round(sum(scene_dur), 3)
starts, t = [], 0.0
for d in scene_dur:
starts.append(round(t, 3))
t += d
# ---------------------------------------------------- 2. captions
events: List[dict] = []
if caps and caps.enabled:
stage("captions", 30.0, "timing the words")
timing = (caps.timing or "auto").lower()
model = _pick_stt_model(caps) if timing in ("auto", "stt") else None
if timing == "stt" and not model:
raise rnd.RenderError("captions.timing=stt but no speech model is configured")
if timing == "auto" and not model:
warn("no speech model configured; caption timings are estimated")
words: List[dict] = []
for i, sc in enumerate(req.scenes):
text = (sc.text or "").strip()
if not text:
continue
stt: List[dict] = []
if model and scene_audio[i]:
try:
stt = _stt_words(loop, scene_audio[i], model,
(sc.voice or req.voice).language)
except Exception as exc:
warn(f"scene {i}: word timestamps failed ({exc}); estimated instead")
span = max(0.2, scene_dur[i] - max(0.0, float(sc.padding or 0.0)))
words += (cap.align_words(text, stt, starts[i], span) if stt
else cap.estimate_words(text, starts[i], span))
events = cap.break_lines(words, caps.max_words_per_line, caps.max_lines)
if caps.font and not media.font_installed(caps.font):
warn(f"font {caps.font!r} is not installed; captions use the "
f"closest available family")
# ---------------------------------------------------- 3. visuals
clips: List[str] = []
clip_durs: List[float] = []
made = 0
total_visuals = sum(len(sc.visuals) for sc in req.scenes)
extra = tdur if ttype != "none" else 0.0
for i, sc in enumerate(req.scenes):
n = max(1, len(sc.visuals))
slot = scene_dur[i] / n
for j, v in enumerate(sc.visuals):
stage("visuals", 40.0 + 28.0 * made / max(1, total_visuals),
f"scene {i + 1}/{n_scenes}: visual {j + 1}/{n}")
spec = v.model_dump()
vt = (spec.get("type") or "image").lower()
if vt in ("video", "image"):
try:
spec["_path"] = media.resolve(spec.get("src") or "", work,
f"s{i}v{j}")
spec["_probe"] = media.probe(spec["_path"])
if vt == "video" and not spec["_probe"].get("has_video"):
raise media.MediaError("no video stream")
except (media.MediaError, OSError) as exc:
warn(f"scene {i} visual {j}: {exc}; used gradient")
spec = {"type": "gradient", "colors": None}
out = os.path.join(work, f"clip-{i}-{j}.mp4")
try:
rnd.build_visual(spec, out, width=canvas.width, height=canvas.height,
fps=canvas.fps, duration=slot + extra, fit=fit,
index=made, workdir=work, cancel=cancel, log=log)
except rnd.Cancelled:
raise
except Exception as exc:
warn(f"scene {i} visual {j}: render failed ({exc}); used gradient")
rnd.build_visual({"type": "gradient"}, out, width=canvas.width,
height=canvas.height, fps=canvas.fps,
duration=slot + extra, fit=fit, index=made,
workdir=work, cancel=cancel, log=log)
clips.append(out)
clip_durs.append(slot + extra)
made += 1
stage("visuals", 69.0, "joining the clips")
silent = os.path.join(work, "video.mp4")
if len(clips) == 1:
silent = clips[0]
elif ttype == "none":
rnd.concat_clips(clips, silent, work, cancel=cancel, log=log)
else:
try:
rnd.xfade_clips(clips, clip_durs, silent, kind=ttype, dur=tdur,
fps=canvas.fps, cancel=cancel, log=log)
except rnd.Cancelled:
raise
except Exception as exc:
warn(f"transition {ttype!r} failed ({exc}); joined without one")
rnd.concat_clips(clips, silent, work, cancel=cancel, log=log)
# ---------------------------------------------------- 4. audio
stage("music", 71.0, "building the soundtrack")
parts = []
for i, sc in enumerate(req.scenes):
p = os.path.join(work, f"narr-{i}.wav")
if scene_audio[i]:
rnd.pad_to(scene_audio[i], p, scene_dur[i], cancel=cancel, log=log)
else:
rnd.silence(p, scene_dur[i], cancel=cancel, log=log)
parts.append(p)
narration = rnd.concat_audio(parts, os.path.join(work, "narration.wav"), work,
cancel=cancel, log=log)
if any(scene_audio):
try:
narration = rnd.loudnorm(narration, os.path.join(work, "narration-ln.wav"),
cancel=cancel, log=log)
except rnd.Cancelled:
raise
except Exception as exc:
warn(f"loudness normalisation failed ({exc}); narration used as synthesised")
music_bed = None
if req.music:
src = None
if req.music.src:
try:
src = media.resolve(req.music.src, work, "music")
except media.MediaError as exc:
warn(f"music: {exc}; rendered without music")
elif req.music.generate:
stage("music", 73.0, "generating music")
try:
src = _generate_music(loop, req.music.generate, total,
os.path.join(work, "music-gen.wav"))
except Exception as exc:
warn(f"music generation failed ({exc}); rendered without music")
if src:
try:
music_bed = rnd.build_music_bed(
src, os.path.join(work, "music.wav"), total=total,
volume=req.music.volume, loop=req.music.loop,
fade_in=req.music.fade_in, fade_out=req.music.fade_out,
cancel=cancel, log=log)
except rnd.Cancelled:
raise
except Exception as exc:
warn(f"music could not be prepared ({exc}); rendered without music")
mixed = rnd.mix_narration_music(narration, music_bed,
os.path.join(work, "audio.wav"),
duck=bool(req.music and req.music.duck),
cancel=cancel, log=log)
# ---------------------------------------------------- 5. final render
stage("render", 76.0, "encoding")
ass_path = None
if events:
ass_path = os.path.join(work, "captions.ass")
with open(ass_path, "w", encoding="utf-8") as f:
f.write(cap.build_ass(
events, width=canvas.width, height=canvas.height,
font=caps.font, font_scale=caps.font_scale, colour=caps.color,
highlight=caps.highlight_color, outline_colour=caps.outline_color,
position=caps.position, preset=caps.preset,
uppercase=caps.uppercase))
overlays = []
for k, o in enumerate(req.overlays or []):
spec = o.model_dump()
if (spec.get("type") or "text").lower() == "image":
try:
spec["_path"] = media.resolve(spec.get("src") or "", work, f"ov{k}")
except media.MediaError as exc:
warn(f"overlay {k}: {exc}; skipped")
continue
overlays.append(spec)
fmt = (req.output.format or "mp4").lower()
final = os.path.join(work, f"{job_id}.{fmt}")
rnd.final_render(silent, mixed, final, width=canvas.width, height=canvas.height,
fps=canvas.fps, total=total, ass_file=ass_path,
overlays=overlays, crf=req.output.crf,
audio_bitrate=req.output.audio_bitrate,
video_codec=req.output.video_codec, cancel=cancel, log=log)
# ---------------------------------------------------- 6. artefacts
stage("thumbnail", 95.0, "artefacts")
want = set(req.outputs or ["video"])
out_dir = _output_dir()
result: Dict[str, Any] = {}
def publish(src: str, name: str) -> dict:
dst = os.path.join(out_dir, name)
shutil.copyfile(src, dst)
return {"url": _file_url(name, base_url), "path": f"/v1/files/{name}",
"size": os.path.getsize(dst)}
vid = publish(final, f"{job_id}.{fmt}")
info = media.probe(final)
vid.update({"duration": round(float(info.get("duration") or total), 3),
"width": info.get("width") or canvas.width,
"height": info.get("height") or canvas.height})
result["video"] = vid
if "thumbnail" in want:
th = req.thumbnail or ThumbnailSpec()
at = max(0.0, min(0.999, float(th.at if th.at is not None else 0.25))) * total
tpath = os.path.join(work, f"{job_id}.jpg")
try:
rnd.thumbnail(final, tpath, at=at, width=canvas.width,
height=canvas.height, text=th.text or "",
text_color=th.text_color, highlight_color=th.highlight_color,
font=th.font or "", cancel=cancel, log=log)
result["thumbnail"] = publish(tpath, f"{job_id}.jpg")
except rnd.Cancelled:
raise
except Exception as exc:
warn(f"thumbnail failed ({exc})")
if events and ("srt" in want or "vtt" in want):
paths = cap.write_files(events, work, job_id)
for kind in ("srt", "vtt"):
if kind in want:
result[kind] = publish(paths[kind], f"{job_id}.{kind}")
if "narration" in want:
result["narration"] = publish(narration, f"{job_id}_voice.wav")
result["scenes"] = [{"index": i, "start": starts[i],
"end": round(starts[i] + scene_dur[i], 3)}
for i in range(n_scenes)]
result["duration"] = total
_update(job_id, status="done", stage="thumbnail", progress=100.0,
message="done", result=result)
print(f"[compose {job_id}] done: {total:.1f}s, {n_scenes} scenes, "
f"{len(clips)} visuals", flush=True)
finally:
shutil.rmtree(work, ignore_errors=True)
def _output_dir() -> str:
d = media.files_dir()
if not d:
raise rnd.RenderError("this install has no output directory configured "
"(--file-path), so composed video cannot be served")
os.makedirs(d, exist_ok=True)
return d
def _file_url(name: str, base_url: str) -> str:
base = (base_url or "").rstrip("/")
return f"{base}/v1/files/{name}" if base else f"/v1/files/{name}"
# =============================================================================
# Endpoints
# =============================================================================
@router.post("/v1/video/compose", summary="Compose a finished video from scenes",
tags=["Video"])
async def compose_video(req: ComposeRequest, http_request: Request,
_auth=Depends(_require_api_auth)):
"""Assemble narration, visuals, music and captions into one uploadable video.
Each scene's length comes from its narration (synthesised here unless the
scene carries ``audio``) plus its ``padding``, and its visuals share that
time. With ``"async": true`` (the default) the call returns
``202 {id, status}`` and the client polls ``GET /v1/video/compose/{id}``;
with ``"async": false`` it blocks and returns the finished document, which is
convenient for tests and a bad idea for a real render.
A visual that cannot be fetched or decoded is replaced by a gradient and
reported in ``warnings`` — one bad stock clip does not lose a five-minute
render."""
_validate(req)
try:
from codai.api.urlutils import get_base_url
base_url = get_base_url(http_request)
except Exception:
base_url = ""
rec = submit(req, base_url)
if req.is_async:
from fastapi.responses import JSONResponse
return JSONResponse(status_code=202,
content={"id": rec["id"], "status": "queued"})
# Blocking: wait for the worker to finish this job.
job_id = rec["id"]
while True:
await asyncio.sleep(0.5)
cur = _get(job_id)
if cur is None:
raise HTTPException(status_code=500, detail="job disappeared")
if cur["status"] in ("done", "failed", "cancelled"):
if cur["status"] == "failed":
raise HTTPException(status_code=500,
detail=cur.get("error") or "composition failed")
return _public(cur)
@router.get("/v1/video/compose/{job_id}", summary="Composition job status", tags=["Video"])
async def compose_status(job_id: str, _auth=Depends(_require_api_auth)):
"""Progress, warnings and — when ``status`` is ``done`` — the artefact URLs."""
rec = _get(job_id)
if rec is None:
raise HTTPException(status_code=404, detail=f"unknown composition job {job_id!r}")
return _public(rec)
@router.post("/v1/video/compose/{job_id}/cancel", summary="Cancel a composition job",
tags=["Video"])
async def compose_cancel(job_id: str, _auth=Depends(_require_api_auth)):
"""Stop a queued or running job (the ffmpeg child is killed, not left to finish)."""
if not cancel_job(job_id):
raise HTTPException(status_code=404, detail=f"unknown composition job {job_id!r}")
return {"id": job_id, "status": "cancelled"}
@router.get("/v1/video/compose", summary="List composition jobs", tags=["Video"])
async def compose_list(_auth=Depends(_require_api_auth)):
"""Every job this front still remembers, newest first — for an admin page or
a client that lost a job id."""
with _jobs_lock:
recs = sorted(_jobs.values(), key=lambda r: r.get("created_at") or 0, reverse=True)
return {"jobs": [_public(r) for r in recs]}
codai/api/remote_gateway.py
View file @ 5abbb81d
......@@ -115,7 +115,16 @@ _SKIP_PATHS = ("/v1/chat/completions", "/v1/completions", "/v1/models",
# is only meaningful locally. Forwarded, it would report the pod's
# view of a config the pod does not have — the exact confusion it
# was added to end.
"/v1/models/test/state")
"/v1/models/test/state",
# This install's own media store: an upload must land where THIS
# front serves /v1/files from, and the existence check answers for
# this store. Forwarding either would put the bytes on a pod that
# is gone an hour later.
"/v1/files/upload", "/v1/files/blob/{hash}")
#: Paths that are this install's own, matched by prefix at run time (the literal
#: route templates above are what the coverage test reads).
_LOCAL_PREFIXES = ("/v1/files/upload", "/v1/files/blob")
#: Orchestration endpoints: a sequence of calls to other endpoints, not a model.
#: They are NEVER forwarded as a whole — the chain stays here and each step goes
......@@ -123,7 +132,11 @@ _SKIP_PATHS = ("/v1/chat/completions", "/v1/completions", "/v1/models",
#: every step is placed by its OWN model's configuration. Forwarding the
#: orchestration instead would move the logic to a pod and place every step by
#: that pod's catalogue, which is the opposite of what these are for.
_ORCHESTRATION_PREFIXES = ("/v1/pipelines", "/v1/characters", "/v1/environments")
#: ``/v1/video/compose`` is one of these: it calls speech, music and transcription
#: endpoints itself, and the visuals it was given may be files only this install
#: holds. Forwarded to a video pod it would find neither the models nor the files.
_ORCHESTRATION_PREFIXES = ("/v1/pipelines", "/v1/characters", "/v1/environments",
"/v1/video/compose")
#: `model` field in a multipart body, without paying for a full form parse.
_MULTIPART_MODEL = re.compile(
......@@ -505,7 +518,7 @@ def resolve_target(path: str, method: str, query: str, body: bytes,
return None
if not path.startswith("/v1/") or path in _SKIP_PATHS:
return None
if path.startswith(_ORCHESTRATION_PREFIXES):
if path.startswith(_ORCHESTRATION_PREFIXES) or path.startswith(_LOCAL_PREFIXES):
return None
# A generated file is fetched from whichever remote rendered it.
if path.startswith("/v1/files/"):
......
codai/api/uploads.py 0 → 100644
View file @ 5abbb81d
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
"""Generic media upload — ``POST /v1/files/upload``.
A client that is not on this machine has media of its own: the user's footage,
a logo, a music bed. It could inline all of it as base64 in every request, and
a 40 MB clip sent again for each of ten renders is exactly the kind of waste
that makes an API unusable. So: upload once, get a ``/v1/files/...`` URL, name
that URL in as many compose requests as you like.
Storage is content-addressed (sha256 of the bytes), which makes re-uploading
the same file free and makes ``GET /v1/files/blob/{hash}`` a useful question —
a client can ask before sending. This mirrors ``/v1/loras/upload`` for LoRA
weights; the difference is that these files are served back by
``/v1/files/{name}``, so they live in the ordinary output directory under their
hash, not in a private blob store.
"""
import os
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Request
from codai.api.loras import _require_api_auth
from codai.compose import media
router = APIRouter()
# The name a hash is stored under, e.g. up-3f9c…-<12 hex>.mp4. The prefix keeps
# uploads recognisable in the archive listing and in /v1/files URLs.
_PREFIX = "up-"
_MAX_BYTES = 2 << 30 # 2 GiB: a minute of 4K is ~400 MB
def _files_dir() -> str:
d = media.files_dir()
if not d:
raise HTTPException(status_code=501, detail=(
"this install has no output directory configured, so uploads cannot be "
"stored (start the server with --file-path / files.path)"))
os.makedirs(d, exist_ok=True)
return d
def _stored_name(hexhash: str, ext: str) -> str:
return f"{_PREFIX}{hexhash}{ext}"
def find_blob(hexhash: str) -> Optional[str]:
"""The stored path for a hash, whatever extension it landed under."""
h = (hexhash or "").strip().lower()
if h.startswith("sha256:"):
h = h[7:]
if not h or not all(c in "0123456789abcdef" for c in h) or len(h) != 64:
return None
try:
d = _files_dir()
except HTTPException:
return None
try:
for name in os.listdir(d):
if name.startswith(_PREFIX + h):
p = os.path.join(d, name)
if os.path.isfile(p):
return p
except OSError:
return None
return None
def store(data: bytes, filename: str = "", content_type: str = "") -> dict:
"""Put bytes in the content-addressed store; returns the response document."""
if not data:
raise HTTPException(status_code=400, detail="empty upload")
if len(data) > _MAX_BYTES:
raise HTTPException(status_code=413,
detail=f"upload larger than {_MAX_BYTES // (1 << 20)} MiB")
mime = media.sniff_mime(data, content_type)
kind = media.kind_of(mime)
if kind not in ("video", "image", "audio"):
raise HTTPException(status_code=415, detail=(
f"unsupported media type {mime!r}: uploads must be video "
f"(mp4/mov/webm/mkv), image (png/jpg/webp) or audio (mp3/wav/m4a/ogg/flac)"))
h = media.sha256_hex(data)
existing = find_blob(h)
if existing:
return {"id": f"sha256:{h}", "name": os.path.basename(existing),
"bytes": os.path.getsize(existing), "mime": mime, "kind": kind,
"existed": True}
name = _stored_name(h, media.ext_for(mime, filename))
path = os.path.join(_files_dir(), name)
tmp = path + ".part"
with open(tmp, "wb") as f:
f.write(data)
os.replace(tmp, path)
return {"id": f"sha256:{h}", "name": name, "bytes": len(data), "mime": mime,
"kind": kind, "existed": False}
def _with_urls(doc: dict, http_request: Optional[Request]) -> dict:
from codai.api.urlutils import build_file_url
name = doc.pop("name")
doc["path"] = f"/v1/files/{name}"
doc["url"] = build_file_url(name, http_request)
return doc
@router.post("/v1/files/upload", summary="Upload media (content-addressed)", tags=["Files"])
async def upload_file(request: Request, _auth=Depends(_require_api_auth)):
"""Store a media file and return the URL to reference it by.
Accepts the file as ``multipart/form-data`` (field ``file``), as JSON
``{"file": "<base64 or data: URI>"}``, or as a raw request body — the three
forms ``/v1/loras/upload`` takes. Returns ``{id, url, path, bytes, mime,
kind, existed}``; ``existed: true`` means these exact bytes were already
here and nothing was written. Video, image and audio only."""
ctype = request.headers.get("content-type", "")
data = b""
filename = ""
part_type = ""
if "multipart/form-data" in ctype:
form = await request.form()
up = form.get("file") or form.get("upload") or form.get("media")
if up is None:
raise HTTPException(status_code=400,
detail="multipart upload missing 'file' field")
if hasattr(up, "read"):
data = await up.read()
filename = getattr(up, "filename", "") or ""
part_type = getattr(up, "content_type", "") or ""
else:
data = bytes(up)
elif "application/json" in ctype:
body = await request.json()
blob = body.get("file") or body.get("data")
if not blob:
raise HTTPException(status_code=400,
detail="JSON upload missing 'file'/'data' (base64)")
filename = str(body.get("filename") or "")
try:
data = media.fetch(str(blob))
except media.MediaError as exc:
raise HTTPException(status_code=400, detail=str(exc))
else:
data = await request.body()
part_type = ctype
return _with_urls(store(data, filename, part_type), request)
@router.get("/v1/files/blob/{hash}", summary="Check an uploaded file exists", tags=["Files"])
async def file_blob_info(hash: str, request: Request, _auth=Depends(_require_api_auth)):
"""Does this install already hold these bytes? 200 with the URL, else 404.
Lets a client skip re-uploading: hash the file locally, ask, and only send
it when the answer is 404. ``hash`` is a hex sha256, with or without the
``sha256:`` prefix."""
p = find_blob(hash)
if not p:
raise HTTPException(status_code=404, detail="blob not found")
name = os.path.basename(p)
h = name[len(_PREFIX):].split(".")[0]
doc = {"id": f"sha256:{h}", "name": name, "bytes": os.path.getsize(p),
"mime": media.sniff_mime(open(p, "rb").read(32)), "exists": True}
return _with_urls(doc, request)
codai/broker/capabilities.py
View file @ 5abbb81d
......@@ -39,6 +39,13 @@ DEFAULT_STUDIO_ENDPOINTS = [
"v1/video/dub",
"v1/video/to3d",
"v1/video/from3d",
# Server-side composition (codai/api/compose.py) and the media upload it
# references: a client discovers these the same way it discovers the rest.
"v1/video/compose",
"v1/video/compose/{id}",
"v1/video/compose/{id}/cancel",
"v1/files/upload",
"v1/files/blob/{hash}",
]
......
codai/compose/__init__.py 0 → 100644
View file @ 5abbb81d
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
"""Server-side video composition — turning the pieces into a finished reel.
CoderAI can generate every *piece* of a short video (narration, stills, clips,
music) but a client that is not on this machine cannot assemble them: that
needs ffmpeg, the narration's exact duration, and word timings. This package
is that assembly, driven by ``POST /v1/video/compose``
(:mod:`codai.api.compose`):
* :mod:`codai.compose.media` — resolving a media reference (URL, ``/v1/files``
path, data URI, base64) to a local file, probing duration, finding fonts;
* :mod:`codai.compose.captions` — word timings (from STT or estimated) and the
ASS / SRT / VTT the renderer burns in and returns;
* :mod:`codai.compose.render` — the ffmpeg pipeline: one normalised clip per
visual, concat or crossfade, narration + ducked music, burned captions,
overlays, thumbnail.
Everything here is synchronous and side-effect-free apart from the files it
writes into the job's scratch directory, so the whole pipeline is testable
without a server and without a GPU.
"""
codai/compose/captions.py 0 → 100644
View file @ 5abbb81d
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
"""Captions: when each word is said, how the lines break, what libass draws.
Two rules from the spec shape all of this. **The text shown is the text the
caller gave** — speech recognition is used only to find out *when* each word is
spoken, never to rewrite it; a mis-heard word must not reach the screen. And
**the lines stay out of the platform's furniture**: a safe area of ~10% top and
bottom, ~6% at the sides, which is where TikTok, Reels and Shorts put their own
buttons.
Word timings come from one of two places:
* ``stt`` — word timestamps from a speech model transcribing the narration we
just synthesised, aligned onto the caller's words (the transcript's words are
matched positionally, so a mis-heard word still gets its slot);
* ``estimate`` — a duration per word proportional to its length, which is what
you get with no STT model configured and is good enough for a 4-words-a-line
reel.
"""
import os
import re
from typing import List, Optional
# Presets: the look, not the position. Each is a small set of ASS style values
# plus whether the active word is highlighted word-by-word.
PRESETS = {
"karaoke": {"karaoke": True, "bold": True, "outline": 4.0, "shadow": 1.0,
"back": 0.0, "scale": 1.00, "spacing": 0.0, "upper": False},
"bold": {"karaoke": True, "bold": True, "outline": 5.0, "shadow": 0.0,
"back": 0.0, "scale": 1.08, "spacing": 0.5, "upper": True},
"classic": {"karaoke": False, "bold": False, "outline": 2.5, "shadow": 1.0,
"back": 0.0, "scale": 0.92, "spacing": 0.0, "upper": False},
"boxed": {"karaoke": False, "bold": True, "outline": 0.0, "shadow": 0.0,
"back": 0.75, "scale": 0.95, "spacing": 0.0, "upper": False},
"neon": {"karaoke": True, "bold": True, "outline": 3.0, "shadow": 3.0,
"back": 0.0, "scale": 1.02, "spacing": 1.0, "upper": True},
"minimal": {"karaoke": False, "bold": False, "outline": 1.5, "shadow": 0.0,
"back": 0.0, "scale": 0.82, "spacing": 0.0, "upper": False},
}
# ASS alignment (numpad): 2 = bottom-centre, 5 = middle-centre, 8 = top-centre.
_ALIGN = {"bottom": 2, "center": 5, "middle": 5, "top": 8}
_WORD_RE = re.compile(r"\S+")
# ------------------------------------------------------------------- words
def split_words(text: str) -> List[str]:
return _WORD_RE.findall(text or "")
def estimate_words(text: str, start: float, duration: float) -> List[dict]:
"""Spread ``text``'s words over ``duration`` proportionally to their length.
Every word gets a floor of 90 ms plus time proportional to its characters,
so "a" does not flash by in one frame and a long word is not clipped."""
words = split_words(text)
if not words or duration <= 0:
return []
floor = min(0.09, duration / (len(words) * 2.0))
weights = [len(w) + 1.0 for w in words]
total_w = sum(weights)
spare = max(0.0, duration - floor * len(words))
out, t = [], start
for w, weight in zip(words, weights):
d = floor + spare * (weight / total_w)
out.append({"word": w, "start": round(t, 3), "end": round(t + d, 3)})
t += d
if out:
out[-1]["end"] = round(start + duration, 3)
return out
def _norm(word: str) -> str:
return re.sub(r"[^\w']+", "", (word or "").lower(), flags=re.UNICODE)
def align_words(text: str, stt_words: List[dict], start: float,
duration: float) -> List[dict]:
"""Put the caller's words on the STT words' clock.
The two lists rarely match one-to-one: a model hears "3" as "three", joins
"voice over" or drops a filler. So we walk both and consume STT words per
caller word, greedily matching on the normalised form and falling back to
proportional splitting for any caller word left without a timestamp. The
result always covers exactly ``[start, start+duration]`` and never contains
a word the caller did not write."""
words = split_words(text)
if not words:
return []
stt = [w for w in (stt_words or [])
if isinstance(w, dict) and (w.get("word") or "").strip()]
if not stt:
return estimate_words(text, start, duration)
# STT timestamps are relative to the clip we sent; shift onto the timeline.
times = []
for w in stt:
try:
s = float(w.get("start") or 0.0)
e = float(w.get("end") or s)
except (TypeError, ValueError):
continue
times.append({"t": _norm(str(w.get("word"))), "start": start + s,
"end": start + max(e, s)})
if not times:
return estimate_words(text, start, duration)
out: List[Optional[dict]] = [None] * len(words)
j = 0
for i, word in enumerate(words):
if j >= len(times):
break
target = _norm(word)
# Normally each caller word takes the next STT word, even when the model
# heard something else ("3" for "three") — a substitution still marks
# when that word was said. Two refinements:
hit = j
if target and times[j]["t"] != target:
# the model inserted a word the caller did not write ("um"): skip it
# when the real match is just behind it;
for k in range(j + 1, min(j + 3, len(times))):
if times[k]["t"] == target:
hit = k
break
end_idx = hit
acc = times[hit]["t"]
# one caller word may have been heard as several ("voiceover" → "voice
# over"): merge while the pieces still spell the caller's word.
while (end_idx + 1 < len(times) and target and acc != target
and target.startswith(acc)
and target.startswith(acc + times[end_idx + 1]["t"])):
end_idx += 1
acc += times[end_idx]["t"]
out[i] = {"word": word, "start": times[hit]["start"], "end": times[end_idx]["end"]}
j = end_idx + 1
# Fill the gaps (unmatched tails) by splitting the space around them.
last_end = start
for i, rec in enumerate(out):
if rec is not None:
last_end = max(last_end, rec["end"])
continue
# Find the next anchored word to know how much room there is.
nxt_start = None
for k in range(i + 1, len(out)):
if out[k] is not None:
nxt_start = out[k]["start"]
break
room_end = nxt_start if nxt_start is not None else start + duration
# The contiguous run of unanchored words starting here shares the space.
run = []
k = i
while k < len(out) and out[k] is None:
run.append(k)
k += 1
span = max(0.05, room_end - last_end)
filler = estimate_words(" ".join(words[r] for r in run), last_end, span)
for r, f in zip(run, filler):
out[r] = {"word": words[r], "start": f["start"], "end": f["end"]}
last_end = out[run[-1]]["end"] if run else last_end
fixed = [w for w in out if w]
# Monotonic, inside the scene, non-zero length.
prev = start
for w in fixed:
w["start"] = round(max(prev, min(w["start"], start + duration)), 3)
w["end"] = round(max(w["start"] + 0.04, min(w["end"], start + duration)), 3)
prev = w["end"]
if fixed:
fixed[-1]["end"] = round(min(start + duration, max(fixed[-1]["end"],
fixed[-1]["start"] + 0.04)), 3)
return fixed
# ------------------------------------------------------------------- lines
def break_lines(words: List[dict], max_words_per_line: int = 4,
max_lines: int = 2) -> List[dict]:
"""Group timed words into caption events of at most ``max_lines`` lines of
``max_words_per_line`` words, each with its own start/end."""
per_line = max(1, int(max_words_per_line or 4))
lines_max = max(1, int(max_lines or 1))
per_event = per_line * lines_max
out = []
for i in range(0, len(words), per_event):
chunk = words[i:i + per_event]
if not chunk:
continue
rows = [chunk[k:k + per_line] for k in range(0, len(chunk), per_line)]
out.append({
"start": chunk[0]["start"],
"end": chunk[-1]["end"],
"words": chunk,
"rows": [[w["word"] for w in row] for row in rows],
"text": " ".join(w["word"] for w in chunk),
})
return out
# --------------------------------------------------------------------- ASS
def _ass_colour(hex_colour: str, alpha: int = 0) -> str:
"""``#RRGGBB`` → ASS ``&HAABBGGRR``."""
h = (hex_colour or "").strip().lstrip("#")
if len(h) == 3:
h = "".join(c * 2 for c in h)
if len(h) != 6:
h = "FFFFFF"
r, g, b = h[0:2], h[2:4], h[4:6]
return f"&H{alpha:02X}{b}{g}{r}".upper()
def _ass_time(t: float) -> str:
t = max(0.0, float(t))
h = int(t // 3600)
m = int((t % 3600) // 60)
s = t % 60
return f"{h:d}:{m:02d}:{s:05.2f}"
def _ass_escape(text: str) -> str:
return (text or "").replace("\\", "\\\\").replace("{", "\\{").replace("}", "\\}")
def build_ass(events: List[dict], *, width: int, height: int, font: str = "Montserrat",
font_scale: float = 1.0, colour: str = "#FFFFFF",
highlight: str = "#FFE14D", outline_colour: str = "#000000",
position: str = "center", preset: str = "karaoke",
uppercase: bool = False) -> str:
"""The ASS subtitle file the renderer burns in.
Sizes are derived from the canvas height so the same request looks the same
at 1080x1920 and 720x1280; margins keep every line inside the safe area."""
p = PRESETS.get((preset or "karaoke").lower(), PRESETS["karaoke"])
upper = bool(uppercase) or p["upper"]
base = height * 0.052 * p["scale"] * float(font_scale or 1.0)
size = max(12, int(round(base)))
align = _ALIGN.get((position or "center").lower(), 5)
margin_v = int(round(height * 0.10))
margin_h = int(round(width * 0.06))
primary = _ass_colour(colour)
hi = _ass_colour(highlight)
outline = _ass_colour(outline_colour)
back = _ass_colour("#000000", alpha=int(round((1.0 - p["back"]) * 255)) if p["back"] else 255)
border_style = 3 if p["back"] else 1 # 3 = opaque box
head = [
"[Script Info]",
"ScriptType: v4.00+",
f"PlayResX: {int(width)}",
f"PlayResY: {int(height)}",
"WrapStyle: 2", # no automatic wrapping: we break lines
"ScaledBorderAndShadow: yes",
"YCbCr Matrix: TV.709",
"",
"[V4+ Styles]",
"Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, "
"BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, "
"BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding",
f"Style: cap,{font or 'DejaVu Sans'},{size},{primary},{hi},{outline},{back},"
f"{-1 if p['bold'] else 0},0,0,0,100,100,{p['spacing']},0,"
f"{border_style},{p['outline'] * size / 40.0:.2f},{p['shadow'] * size / 40.0:.2f},"
f"{align},{margin_h},{margin_h},{margin_v},1",
"",
"[Events]",
"Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text",
]
body = []
for ev in events:
rows = ev.get("rows") or [[w["word"] for w in ev.get("words", [])]]
if not p["karaoke"]:
text = "\\N".join(" ".join(_ass_escape(w.upper() if upper else w) for w in row)
for row in rows)
body.append(f"Dialogue: 0,{_ass_time(ev['start'])},{_ass_time(ev['end'])},"
f"cap,,0,0,0,,{text}")
continue
# Karaoke: one event per word, the whole line drawn each time with the
# active word in the highlight colour. Simpler than \k timings and it
# renders identically on every libass version.
flat = ev.get("words") or []
for idx, active in enumerate(flat):
parts = []
n = 0
for row in rows:
row_out = []
for w in row:
shown = _ass_escape(w.upper() if upper else w)
row_out.append(f"{{\\c{hi}}}{shown}{{\\c{primary}}}"
if n == idx else shown)
n += 1
parts.append(" ".join(row_out))
start = active["start"]
end = flat[idx + 1]["start"] if idx + 1 < len(flat) else ev["end"]
if end <= start:
end = start + 0.04
body.append(f"Dialogue: 0,{_ass_time(start)},{_ass_time(end)},"
f"cap,,0,0,0,,{{\\c{primary}}}" + "\\N".join(parts))
return "\n".join(head + body) + "\n"
# ---------------------------------------------------------------- SRT / VTT
def _srt_time(t: float) -> str:
t = max(0.0, float(t))
h = int(t // 3600)
m = int((t % 3600) // 60)
s = t % 60
return f"{h:02d}:{m:02d}:{s:06.3f}".replace(".", ",")
def _vtt_time(t: float) -> str:
return _srt_time(t).replace(",", ".")
def build_srt(events: List[dict]) -> str:
out = []
for i, ev in enumerate(events, 1):
text = "\n".join(" ".join(row) for row in (ev.get("rows") or [[ev.get("text", "")]]))
out.append(f"{i}\n{_srt_time(ev['start'])} --> {_srt_time(ev['end'])}\n{text}\n")
return "\n".join(out)
def build_vtt(events: List[dict]) -> str:
out = ["WEBVTT", ""]
for ev in events:
text = "\n".join(" ".join(row) for row in (ev.get("rows") or [[ev.get("text", "")]]))
out.append(f"{_vtt_time(ev['start'])} --> {_vtt_time(ev['end'])}\n{text}\n")
return "\n".join(out)
def write_files(events: List[dict], out_dir: str, stem: str) -> dict:
"""Write ``<stem>.srt`` / ``.vtt`` next to each other; returns their paths."""
os.makedirs(out_dir, exist_ok=True)
paths = {}
for kind, text in (("srt", build_srt(events)), ("vtt", build_vtt(events))):
p = os.path.join(out_dir, f"{stem}.{kind}")
with open(p, "w", encoding="utf-8") as f:
f.write(text)
paths[kind] = p
return paths
codai/compose/media.py 0 → 100644
View file @ 5abbb81d
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
"""Media in, facts out: resolving references, probing files, finding fonts.
A compose request names its media the same four ways every other CoderAI
endpoint accepts — an ``http(s)`` URL, a ``/v1/files/<name>`` path served by
this install, a ``data:`` URI, or raw base64 — and the renderer needs a local
path for each. It also needs two facts ffmpeg can tell us (how long a file is,
whether it has audio) and one fontconfig can (which file a font name means).
"""
import base64
import hashlib
import json
import mimetypes
import os
import re
import shutil
import subprocess
import urllib.parse
import urllib.request
from typing import Optional
_DATA_URI = re.compile(r"^data:([^;,]*)(;base64)?,", re.I)
# What an upload may be. Kept in step with the compose spec's visual/music types.
ALLOWED_MIME_EXT = {
"video/mp4": ".mp4", "video/quicktime": ".mov", "video/webm": ".webm",
"video/x-matroska": ".mkv", "video/mpeg": ".mpg",
"image/png": ".png", "image/jpeg": ".jpg", "image/webp": ".webp",
"audio/mpeg": ".mp3", "audio/mp3": ".mp3", "audio/wav": ".wav",
"audio/x-wav": ".wav", "audio/mp4": ".m4a", "audio/m4a": ".m4a",
"audio/aac": ".m4a", "audio/ogg": ".ogg", "audio/flac": ".flac",
"audio/x-flac": ".flac",
}
_MAGIC = (
(b"\x89PNG\r\n\x1a\n", "image/png"),
(b"\xff\xd8\xff", "image/jpeg"),
(b"RIFF", None), # WAV or WEBP — decided below
(b"OggS", "audio/ogg"),
(b"fLaC", "audio/flac"),
(b"\x1a\x45\xdf\xa3", "video/x-matroska"), # also webm
(b"ID3", "audio/mpeg"),
)
class MediaError(Exception):
"""A reference that cannot be turned into a local file."""
# ----------------------------------------------------------------- sniffing
def sniff_mime(data: bytes, hint: str = "") -> str:
"""The media type of these bytes: content first, caller's hint second.
Uploads arrive with a browser-supplied content type that is often wrong or
``application/octet-stream``; the first bytes are not."""
head = data[:16]
for magic, mime in _MAGIC:
if not head.startswith(magic):
continue
if magic == b"RIFF":
if data[8:12] == b"WEBP":
return "image/webp"
if data[8:12] == b"WAVE":
return "audio/wav"
continue
if magic == b"\x1a\x45\xdf\xa3":
return "video/webm" if b"webm" in data[:256].lower() else "video/x-matroska"
return mime
if data[4:8] == b"ftyp":
brand = data[8:12].lower()
if brand.startswith(b"qt"):
return "video/quicktime"
if brand in (b"m4a ", b"m4b "):
return "audio/m4a"
return "video/mp4"
# MP3 without an ID3 header starts with a frame sync.
if head[:2] in (b"\xff\xfb", b"\xff\xf3", b"\xff\xf2"):
return "audio/mpeg"
hint = (hint or "").split(";")[0].strip().lower()
return hint or "application/octet-stream"
def ext_for(mime: str, filename: str = "") -> str:
"""The extension to store a blob under."""
if mime in ALLOWED_MIME_EXT:
return ALLOWED_MIME_EXT[mime]
ext = os.path.splitext(filename or "")[1].lower()
if ext in set(ALLOWED_MIME_EXT.values()):
return ext
return mimetypes.guess_extension(mime or "") or ".bin"
def kind_of(mime: str) -> str:
"""``video`` | ``image`` | ``audio`` | ``other``."""
m = (mime or "").lower()
for k in ("video", "image", "audio"):
if m.startswith(k + "/"):
return k
return "other"
def sha256_hex(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
# ----------------------------------------------------------------- resolving
def _decode_data_uri(ref: str) -> bytes:
m = _DATA_URI.match(ref)
payload = ref[m.end():]
if m.group(2):
return base64.b64decode(payload)
return urllib.parse.unquote_to_bytes(payload)
#: Set by the server (codai/main.py) so composition can write where
#: ``/v1/files`` reads. Kept here rather than in each module so uploads,
#: composition and reference resolution always agree on one directory.
global_file_path: Optional[str] = None
def set_global_file_path(path: str) -> None:
global global_file_path
global_file_path = path or None
def files_dir() -> Optional[str]:
"""Where this install writes generated files (``/v1/files`` serves it).
Whoever set it first wins: the explicit setter, then the already-imported
api modules (never importing them — that would drag the whole server in)."""
if global_file_path:
return global_file_path
import sys
for mod in ("codai.api.app", "codai.api.video", "codai.api.images"):
m = sys.modules.get(mod)
d = getattr(m, "global_file_path", None) if m is not None else None
if d:
return d
return None
def _files_dir() -> Optional[str]:
return files_dir()
def local_path_for_files_url(ref: str) -> Optional[str]:
"""If ``ref`` points at a file this install already holds, its path.
Covers ``/v1/files/x.mp4`` and any absolute URL whose path ends that way —
a client that got a full URL back from an earlier call sends it verbatim,
and downloading our own file over the loopback would be silly (and fails
when the front is bound elsewhere)."""
if not ref:
return None
path = ref
if "://" in ref:
try:
path = urllib.parse.urlsplit(ref).path
except ValueError:
return None
if "/v1/files/" not in path:
return None
name = path.split("/v1/files/", 1)[1].split("/")[-1]
name = urllib.parse.unquote(name)
base = _files_dir()
if not (base and name):
return None
safe_base = os.path.realpath(base)
cand = os.path.realpath(os.path.join(base, name))
if not (cand == safe_base or cand.startswith(safe_base + os.sep)):
return None
return cand if os.path.isfile(cand) else None
def fetch(ref: str, timeout: float = 120.0, max_bytes: int = 2 << 30) -> bytes:
"""The bytes behind a media reference (URL / data URI / base64)."""
if not ref:
raise MediaError("empty media reference")
if _DATA_URI.match(ref):
return _decode_data_uri(ref)
if ref.startswith(("http://", "https://")):
req = urllib.request.Request(ref, headers={"User-Agent": "coderai/compose"})
with urllib.request.urlopen(req, timeout=timeout) as r:
data = r.read(max_bytes + 1)
if len(data) > max_bytes:
raise MediaError(f"{ref}: larger than {max_bytes} bytes")
return data
if ref.startswith(("/", "./")) or os.path.isabs(ref):
# A bare path is only ever ours: /v1/files/... (handled by the caller)
# or a local file, which we do NOT read — a client cannot name paths on
# this machine.
raise MediaError(f"{ref}: not a URL, a /v1/files path, or base64 data")
try:
return base64.b64decode(ref, validate=True)
except Exception:
raise MediaError("media reference is not a URL, a /v1/files path, a data: URI or base64")
def resolve(ref: str, workdir: str, name_hint: str = "media",
timeout: float = 120.0) -> str:
"""A media reference → a local file path, downloading into ``workdir`` when
the reference is remote or inline. Files this install already holds are used
where they are."""
local = local_path_for_files_url(ref)
if local:
return local
data = fetch(ref, timeout=timeout)
if not data:
raise MediaError(f"{name_hint}: empty media")
mime = sniff_mime(data)
os.makedirs(workdir, exist_ok=True)
out = os.path.join(workdir, f"{name_hint}-{sha256_hex(data)[:12]}{ext_for(mime)}")
if not os.path.isfile(out):
tmp = out + ".part"
with open(tmp, "wb") as f:
f.write(data)
os.replace(tmp, out)
return out
# ------------------------------------------------------------------ probing
def _ffprobe_bin() -> str:
return shutil.which("ffprobe") or "ffprobe"
def ffmpeg_bin() -> str:
return shutil.which("ffmpeg") or "ffmpeg"
def probe(path: str) -> dict:
"""``{duration, width, height, fps, has_audio, has_video}`` of a media file."""
cmd = [_ffprobe_bin(), "-v", "error", "-print_format", "json",
"-show_format", "-show_streams", path]
try:
r = subprocess.run(cmd, capture_output=True, timeout=120)
doc = json.loads(r.stdout or b"{}")
except Exception as exc:
raise MediaError(f"ffprobe failed on {os.path.basename(path)}: {exc}")
if not doc:
raise MediaError(f"ffprobe returned nothing for {os.path.basename(path)}"
f"{': ' + r.stderr.decode(errors='replace')[:200] if r.stderr else ''}")
out = {"duration": 0.0, "width": 0, "height": 0, "fps": 0.0,
"has_audio": False, "has_video": False}
try:
out["duration"] = float((doc.get("format") or {}).get("duration") or 0.0)
except (TypeError, ValueError):
pass
for st in doc.get("streams") or []:
ctype = st.get("codec_type")
if ctype == "video" and not out["has_video"]:
out["has_video"] = True
out["width"] = int(st.get("width") or 0)
out["height"] = int(st.get("height") or 0)
fr = st.get("avg_frame_rate") or st.get("r_frame_rate") or "0/1"
try:
num, _, den = fr.partition("/")
out["fps"] = (float(num) / float(den)) if float(den or 0) else 0.0
except (TypeError, ValueError, ZeroDivisionError):
out["fps"] = 0.0
if not out["duration"]:
try:
out["duration"] = float(st.get("duration") or 0.0)
except (TypeError, ValueError):
pass
elif ctype == "audio":
out["has_audio"] = True
if not out["duration"]:
try:
out["duration"] = float(st.get("duration") or 0.0)
except (TypeError, ValueError):
pass
# An animated GIF/PNG reports a video stream but no usable duration; treat a
# zero-duration video as a still (the renderer gives stills a Ken Burns move).
return out
def duration_of(path: str) -> float:
return float(probe(path).get("duration") or 0.0)
# -------------------------------------------------------------------- fonts
_FONT_CACHE: dict = {}
# Tried in order when a named font is not installed. DejaVu ships with the
# image's base system, so there is always a last resort.
_FALLBACKS = ("Montserrat", "Inter", "DejaVu Sans", "Noto Sans", "Liberation Sans", "Arial")
def font_file(name: str = "", bold: bool = True) -> Optional[str]:
"""The path of a font file for ``name`` (fontconfig), or a fallback.
Captions are burned by libass, which resolves families through fontconfig
itself, but ``drawtext`` (overlays, the thumbnail's hook text) needs a file,
and we want to know whether the asked-for family exists at all so a missing
font becomes a warning instead of a silently different look."""
key = (name or "", bool(bold))
if key in _FONT_CACHE:
return _FONT_CACHE[key]
fcmatch = shutil.which("fc-match")
out = None
names = [n for n in ([name] if name else []) + list(_FALLBACKS) if n]
if fcmatch:
for cand in names:
pattern = f"{cand}:style={'Bold' if bold else 'Regular'}"
try:
r = subprocess.run([fcmatch, "-f", "%{file}|%{family}", pattern],
capture_output=True, timeout=20)
got = (r.stdout or b"").decode(errors="replace").strip()
except Exception:
got = ""
if not got or "|" not in got:
continue
path, family = got.split("|", 1)
# fontconfig always answers *something*; only accept the answer when
# it is the family we asked for (substring, case-insensitive).
asked = cand.replace(" ", "").lower()
if path and os.path.isfile(path) and asked in family.replace(" ", "").lower():
out = path
break
if out is None:
for guess in ("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"
if bold else "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"):
if os.path.isfile(guess):
out = guess
break
_FONT_CACHE[key] = out
return out
def font_installed(name: str) -> bool:
"""Is this exact family available? (A caption font that is not gets a warning.)"""
if not name:
return True
return bool(font_file(name, bold=True)) and _FONT_CACHE.get((name, True)) is not None \
and _family_matches(name)
def _family_matches(name: str) -> bool:
fcmatch = shutil.which("fc-match")
if not fcmatch:
return False
try:
r = subprocess.run([fcmatch, "-f", "%{family}", name], capture_output=True, timeout=20)
fam = (r.stdout or b"").decode(errors="replace")
except Exception:
return False
return name.replace(" ", "").lower() in fam.replace(" ", "").lower()
codai/compose/render.py 0 → 100644
View file @ 5abbb81d
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
"""The ffmpeg pipeline: pieces in, one uploadable file out.
Shape of the render, and why it is shaped this way:
1. **one normalised clip per visual** — same size, same fps, same pixel format,
exact duration. Building intermediates rather than one enormous filter graph
costs a re-encode but buys three things that matter more: a visual that fails
to decode can be replaced by a gradient without losing the render, progress
is reportable per visual, and the ffmpeg command stays a length a human can
read in a log.
2. **join** — concat demuxer when there is no transition (stream copy, free), an
``xfade`` chain otherwise. Clips for a transition are built one transition
longer so the overlap is absorbed and the timeline still matches the audio.
3. **audio** — narration per scene padded to the scene's length and concatenated,
loudness-normalised; music looped, trimmed, faded and ducked under the
narration with ``sidechaincompress``; the two mixed.
4. **burn** — captions (libass) and overlays in one pass with the final encode:
H.264 High + AAC + ``+faststart``, which is what TikTok, Reels, Shorts and
Facebook accept without re-processing.
Every ffmpeg invocation goes through :func:`run` so a cancelled job kills the
child it is waiting on instead of finishing a five-minute encode nobody wants.
"""
import os
import shlex
import subprocess
import threading
from typing import Callable, List, Optional
from codai.compose import media
# A cheap, visually lossless intermediate: the final encode is the one that
# matters, and veryfast/crf 18 keeps a 20-visual reel from spending its time in
# x264 twice.
_INTER_ARGS = ["-c:v", "libx264", "-preset", "veryfast", "-crf", "18",
"-pix_fmt", "yuv420p", "-movflags", "+faststart"]
class Cancelled(Exception):
"""The job was cancelled while a child process was running."""
class RenderError(Exception):
"""ffmpeg refused to do something we asked for."""
def _thread_cap() -> str:
"""Leave the machine usable: never more than half its cores per child."""
try:
n = len(os.sched_getaffinity(0))
except AttributeError:
n = os.cpu_count() or 4
return str(max(1, min(8, n // 2 or 1)))
def run(args: List[str], *, cancel: Optional[threading.Event] = None,
timeout: float = 7200.0, log: Optional[Callable[[str], None]] = None) -> None:
"""Run one ffmpeg command, honouring cancellation while it runs."""
if cancel is not None and cancel.is_set():
raise Cancelled()
cmd = [media.ffmpeg_bin(), "-hide_banner", "-nostdin", "-y",
"-threads", _thread_cap()] + list(args)
if log:
log("ffmpeg " + " ".join(shlex.quote(a) for a in args[:40])
+ (" …" if len(args) > 40 else ""))
proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
waited = 0.0
while True:
try:
# A short poll so a cancel lands in a quarter of a second rather
# than at the end of a long encode.
_, err = proc.communicate(timeout=0.25)
break
except subprocess.TimeoutExpired:
waited += 0.25
if cancel is not None and cancel.is_set():
proc.kill()
proc.communicate()
raise Cancelled()
if waited > timeout:
proc.kill()
proc.communicate()
raise RenderError(f"ffmpeg timed out after {timeout:.0f}s")
if proc.returncode != 0:
tail = (err or b"").decode(errors="replace").strip().splitlines()[-12:]
raise RenderError("ffmpeg failed (rc=%d): %s" % (proc.returncode, " | ".join(tail)))
# ------------------------------------------------------------------- visuals
def fit_filter(width: int, height: int, fit: str) -> str:
"""Scale a source of any aspect to the canvas.
``cover`` crops to fill (what a reel wants), ``contain`` letterboxes onto
black, ``blur_pad`` letterboxes onto a blurred blow-up of the same frame —
the look every vertical repost of a horizontal clip uses."""
w, h = int(width), int(height)
fit = (fit or "cover").lower()
if fit == "contain":
return (f"scale={w}:{h}:force_original_aspect_ratio=decrease,"
f"pad={w}:{h}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1")
if fit == "blur_pad":
return (f"split=2[bg][fg];"
f"[bg]scale={w}:{h}:force_original_aspect_ratio=increase,"
f"crop={w}:{h},gblur=sigma=28[bgb];"
f"[fg]scale={w}:{h}:force_original_aspect_ratio=decrease[fgs];"
f"[bgb][fgs]overlay=(W-w)/2:(H-h)/2,setsar=1")
return (f"scale={w}:{h}:force_original_aspect_ratio=increase,"
f"crop={w}:{h},setsar=1")
def ken_burns_filter(width: int, height: int, fps: int, duration: float,
zoom_from: float = 1.0, zoom_to: float = 1.12,
pan: str = "auto", index: int = 0) -> str:
"""A slow zoom/pan over a still (``zoompan``).
The still is first blown up and cropped to the canvas aspect so the move
never reveals an edge, then zoompan walks the zoom from ``zoom_from`` to
``zoom_to`` across the clip. ``pan: auto`` alternates direction per image so
consecutive stills do not drift the same way."""
w, h = int(width), int(height)
frames = max(2, int(round(max(0.04, duration) * max(1, fps))))
z0 = max(1.0, float(zoom_from or 1.0))
z1 = max(1.0, float(zoom_to if zoom_to is not None else 1.12))
if abs(z1 - z0) < 1e-3:
z1 = z0 + 0.0001 # zoompan needs a direction
step = (z1 - z0) / frames
zexpr = (f"min(max(zoom,{z0:.5f})+{step:.7f},{max(z0, z1):.5f})" if step > 0
else f"max(max(zoom,{z0:.5f}){step:.7f},{min(z0, z1):.5f})")
p = (pan or "auto").lower()
if p == "auto":
p = ("right", "left", "up", "down")[index % 4]
# Pan expressions move the crop window across the zoomed frame.
if p == "left":
x, y = "(iw-iw/zoom)*(1-on/%d)" % frames, "(ih-ih/zoom)/2"
elif p == "right":
x, y = "(iw-iw/zoom)*(on/%d)" % frames, "(ih-ih/zoom)/2"
elif p == "up":
x, y = "(iw-iw/zoom)/2", "(ih-ih/zoom)*(1-on/%d)" % frames
elif p == "down":
x, y = "(iw-iw/zoom)/2", "(ih-ih/zoom)*(on/%d)" % frames
else:
x, y = "(iw-iw/zoom)/2", "(ih-ih/zoom)/2"
# Oversample before zoompan: it samples the INPUT, so a canvas-sized input
# visibly softens as it zooms in.
pre = (f"scale={w * 2}:{h * 2}:force_original_aspect_ratio=increase,"
f"crop={w * 2}:{h * 2}")
return (f"{pre},zoompan=z='{zexpr}':x='{x}':y='{y}':d={frames}:s={w}x{h}:fps={fps},"
f"setsar=1,format=yuv420p")
def gradient_png(path: str, width: int, height: int, colours: List[str]) -> str:
"""A vertical gradient as a PNG — the fallback for a visual that fails, and
a visual type of its own. Drawn with PIL rather than ffmpeg's ``gradients``
filter: one less filter whose options changed between ffmpeg versions."""
from PIL import Image
cols = [c for c in (colours or []) if isinstance(c, str) and c.strip()] or \
["#6D28D9", "#DB2777"]
def _rgb(c: str):
h = c.strip().lstrip("#")
if len(h) == 3:
h = "".join(x * 2 for x in h)
if len(h) != 6:
h = "6D28D9"
return tuple(int(h[i:i + 2], 16) for i in (0, 2, 4))
stops = [_rgb(c) for c in cols]
if len(stops) == 1:
stops = stops * 2
img = Image.new("RGB", (max(2, int(width)), max(2, int(height))))
px = img.load()
h = img.height
segs = len(stops) - 1
for y in range(h):
pos = (y / max(1, h - 1)) * segs
i = min(segs - 1, int(pos))
f = pos - i
a, b = stops[i], stops[i + 1]
row = (int(a[0] + (b[0] - a[0]) * f), int(a[1] + (b[1] - a[1]) * f),
int(a[2] + (b[2] - a[2]) * f))
for x in range(img.width):
px[x, y] = row
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
img.save(path)
return path
def build_visual(visual: dict, out_path: str, *, width: int, height: int, fps: int,
duration: float, fit: str = "cover", index: int = 0,
workdir: str = "", cancel: Optional[threading.Event] = None,
log: Optional[Callable[[str], None]] = None) -> None:
"""Render one visual to a clip of exactly ``duration`` seconds."""
vtype = (visual.get("type") or "").lower()
dur = max(0.08, float(duration))
common_out = ["-t", f"{dur:.3f}", "-r", str(fps), "-an"] + _INTER_ARGS + [out_path]
if vtype == "color":
colour = visual.get("color") or "#111827"
args = ["-f", "lavfi", "-i", f"color=c={colour}:s={width}x{height}:r={fps}",
"-vf", "format=yuv420p,setsar=1"] + common_out
run(args, cancel=cancel, log=log)
return
if vtype == "gradient":
png = gradient_png(os.path.join(workdir or ".", f"grad-{index}.png"),
width, height, visual.get("colors") or visual.get("colours"))
args = ["-loop", "1", "-i", png,
"-vf", ken_burns_filter(width, height, fps, dur, 1.0, 1.06, "auto", index)
] + common_out
run(args, cancel=cancel, log=log)
return
src = visual.get("_path")
if not src:
raise RenderError("visual has no resolved source")
if vtype == "image":
kb = visual.get("ken_burns")
if kb is False:
vf = fit_filter(width, height, fit) + ",format=yuv420p"
else:
kb = kb if isinstance(kb, dict) else {}
vf = ken_burns_filter(width, height, fps, dur,
kb.get("zoom_from", 1.0), kb.get("zoom_to", 1.12),
kb.get("pan", "auto"), index)
run(["-loop", "1", "-i", src, "-vf", vf] + common_out, cancel=cancel, log=log)
return
# video: seek, loop if it is shorter than its slot, normalise.
trim = max(0.0, float(visual.get("trim_start") or 0.0))
info = visual.get("_probe") or {}
src_dur = float(info.get("duration") or 0.0)
pre = []
if src_dur and (src_dur - trim) < dur - 0.02:
pre += ["-stream_loop", "-1"] # shorter than its slot → loop
if trim > 0:
pre += ["-ss", f"{trim:.3f}"]
vf = fit_filter(width, height, fit) + f",fps={fps},format=yuv420p"
run(pre + ["-i", src, "-vf", vf] + common_out, cancel=cancel, log=log)
# --------------------------------------------------------------------- join
def concat_clips(clips: List[str], out_path: str, workdir: str,
cancel: Optional[threading.Event] = None,
log: Optional[Callable[[str], None]] = None) -> None:
"""Join identically-encoded clips without re-encoding."""
lst = os.path.join(workdir, "concat.txt")
with open(lst, "w", encoding="utf-8") as f:
for c in clips:
f.write("file '%s'\n" % os.path.abspath(c).replace("'", "'\\''"))
run(["-f", "concat", "-safe", "0", "-i", lst, "-c", "copy", out_path],
cancel=cancel, log=log)
_XFADE = {"fade": "fadeblack", "crossfade": "fade", "slide": "slideleft",
"wipe": "wipeleft", "dissolve": "dissolve"}
def xfade_clips(clips: List[str], durations: List[float], out_path: str, *,
kind: str, dur: float, fps: int,
cancel: Optional[threading.Event] = None,
log: Optional[Callable[[str], None]] = None) -> None:
"""Join clips with a transition, keeping the total length intact.
Each clip was built ``dur`` seconds longer than its slot, so an ``xfade``
that consumes ``dur`` of overlap per join lands the timeline back where the
audio expects it."""
trans = _XFADE.get((kind or "crossfade").lower(), "fade")
args = []
for c in clips:
args += ["-i", c]
graph = []
prev = "0:v"
offset = 0.0
for i in range(1, len(clips)):
offset += max(0.05, durations[i - 1] - dur)
out = f"x{i}"
graph.append(f"[{prev}][{i}:v]xfade=transition={trans}:duration={dur:.3f}:"
f"offset={offset:.3f}[{out}]")
prev = out
graph.append(f"[{prev}]fps={fps},format=yuv420p[vout]")
run(args + ["-filter_complex", ";".join(graph), "-map", "[vout]"]
+ _INTER_ARGS + [out_path], cancel=cancel, log=log)
# -------------------------------------------------------------------- audio
def silence(path: str, duration: float, cancel=None, log=None) -> str:
run(["-f", "lavfi", "-i", "anullsrc=channel_layout=stereo:sample_rate=48000",
"-t", f"{max(0.02, duration):.3f}", "-c:a", "pcm_s16le", path],
cancel=cancel, log=log)
return path
def pad_to(src: str, out_path: str, duration: float, cancel=None, log=None) -> str:
"""One scene's narration, padded (or trimmed) to the scene's exact length."""
run(["-i", src, "-af",
f"aresample=48000,apad,atrim=0:{max(0.02, duration):.3f},asetpts=N/SR/TB",
"-ac", "2", "-ar", "48000", "-c:a", "pcm_s16le", out_path],
cancel=cancel, log=log)
return out_path
def concat_audio(parts: List[str], out_path: str, workdir: str,
cancel=None, log=None) -> str:
if len(parts) == 1:
run(["-i", parts[0], "-ac", "2", "-ar", "48000", "-c:a", "pcm_s16le", out_path],
cancel=cancel, log=log)
return out_path
lst = os.path.join(workdir, "concat-audio.txt")
with open(lst, "w", encoding="utf-8") as f:
for p in parts:
f.write("file '%s'\n" % os.path.abspath(p).replace("'", "'\\''"))
run(["-f", "concat", "-safe", "0", "-i", lst,
"-ac", "2", "-ar", "48000", "-c:a", "pcm_s16le", out_path], cancel=cancel, log=log)
return out_path
def loudnorm(src: str, out_path: str, lufs: float = -16.0, cancel=None, log=None) -> str:
"""Bring narration to about ``lufs`` so every reel is as loud as the last."""
run(["-i", src, "-af", f"loudnorm=I={lufs}:TP=-1.5:LRA=11",
"-ac", "2", "-ar", "48000", "-c:a", "pcm_s16le", out_path], cancel=cancel, log=log)
return out_path
def build_music_bed(src: str, out_path: str, *, total: float, volume: float = 0.14,
loop: bool = True, fade_in: float = 1.0, fade_out: float = 2.0,
cancel=None, log=None) -> str:
"""Music looped/trimmed to the video's length, faded at both ends."""
pre = ["-stream_loop", "-1"] if loop else []
fo_start = max(0.0, total - max(0.0, fade_out))
af = [f"aresample=48000", f"atrim=0:{max(0.1, total):.3f}", "asetpts=N/SR/TB",
f"volume={max(0.0, float(volume)):.4f}"]
if fade_in and fade_in > 0:
af.append(f"afade=t=in:st=0:d={float(fade_in):.3f}")
if fade_out and fade_out > 0:
af.append(f"afade=t=out:st={fo_start:.3f}:d={float(fade_out):.3f}")
run(pre + ["-i", src, "-af", ",".join(af), "-ac", "2", "-ar", "48000",
"-c:a", "pcm_s16le", out_path], cancel=cancel, log=log)
return out_path
def mix_narration_music(narration: str, music: Optional[str], out_path: str, *,
duck: bool = True, cancel=None, log=None) -> str:
"""Narration over music, with the music ducked under speech when asked."""
if not music:
run(["-i", narration, "-ac", "2", "-ar", "48000", "-c:a", "pcm_s16le", out_path],
cancel=cancel, log=log)
return out_path
if duck:
graph = ("[1:a]asplit=2[m1][m2];"
"[0:a]asplit=2[n1][n2];"
# n2 is only the sidechain key; m1 is what we hear.
"[m1][n2]sidechaincompress=threshold=0.05:ratio=8:attack=20:"
"release=400:makeup=1[mduck];"
"[n1][mduck]amix=inputs=2:duration=first:dropout_transition=0,"
"alimiter=limit=0.97[aout];"
"[m2]anullsink")
else:
graph = ("[0:a][1:a]amix=inputs=2:duration=first:dropout_transition=0,"
"alimiter=limit=0.97[aout]")
run(["-i", narration, "-i", music, "-filter_complex", graph, "-map", "[aout]",
"-ac", "2", "-ar", "48000", "-c:a", "pcm_s16le", out_path], cancel=cancel, log=log)
return out_path
# ------------------------------------------------------- overlays + captions
def _pos_expr(position: str, pad_x: str, pad_y: str) -> tuple:
"""Nine-box position → (x, y) expressions for overlay/drawtext."""
p = (position or "bottom-right").lower().replace("_", "-")
horiz = {"left": pad_x, "center": "(W-w)/2", "centre": "(W-w)/2",
"right": f"W-w-{pad_x}"}
vert = {"top": pad_y, "center": "(H-h)/2", "centre": "(H-h)/2",
"middle": "(H-h)/2", "bottom": f"H-h-{pad_y}"}
if "-" in p:
v, _, hh = p.partition("-")
else:
v, hh = ("center", p) if p in horiz else (p, "center")
return horiz.get(hh, horiz["right"]), vert.get(v, vert["bottom"])
def _esc_text(text: str) -> str:
"""Escape for drawtext's expression parser."""
return ((text or "")
.replace("\\", "\\\\").replace(":", "\\:").replace("'", "’")
.replace("%", "\\%").replace(",", "\\,").replace("[", "\\[")
.replace("]", "\\]").replace(";", "\\;"))
def final_render(video: str, audio: str, out_path: str, *, width: int, height: int,
fps: int, total: float, ass_file: Optional[str] = None,
overlays: Optional[List[dict]] = None, crf: int = 20,
audio_bitrate: str = "192k", video_codec: str = "h264",
fonts_dir: Optional[str] = None, cancel=None, log=None) -> None:
"""Burn captions and overlays and encode the deliverable."""
inputs = ["-i", video, "-i", audio]
img_overlays = [o for o in (overlays or [])
if (o.get("type") or "image").lower() == "image" and o.get("_path")]
for o in img_overlays:
inputs += ["-i", o["_path"]]
graph = [f"[0:v]fps={fps},format=yuv420p[base]"]
cur = "base"
if ass_file:
opts = f"filename={ass_file}"
if fonts_dir:
opts += f":fontsdir={fonts_dir}"
graph.append(f"[{cur}]ass={opts}[capd]")
cur = "capd"
# Images first (a logo belongs under text), then text.
for n, o in enumerate(img_overlays):
idx = 2 + n
w = max(0.01, min(1.0, float(o.get("width") or 0.15)))
alpha = max(0.0, min(1.0, float(o.get("opacity") if o.get("opacity") is not None else 1.0)))
x, y = _pos_expr(o.get("position"), f"{int(width * 0.05)}", f"{int(height * 0.04)}")
graph.append(f"[{idx}:v]scale={int(width * w)}:-1,format=rgba,"
f"colorchannelmixer=aa={alpha:.3f}[ov{n}]")
graph.append(f"[{cur}][ov{n}]overlay={x}:{y}[ovd{n}]")
cur = f"ovd{n}"
for n, o in enumerate(overlays or []):
if (o.get("type") or "image").lower() != "text":
continue
text = _esc_text(o.get("text") or "")
if not text:
continue
size = max(8, int(height * max(0.008, min(0.2, float(o.get("size") or 0.025)))))
alpha = max(0.0, min(1.0, float(o.get("opacity") if o.get("opacity") is not None else 1.0)))
colour = (o.get("color") or "#FFFFFF").lstrip("#")
x, y = _pos_expr(o.get("position"), f"{int(width * 0.05)}", f"{int(height * 0.04)}")
ff = media.font_file(o.get("font") or "", bold=True)
parts = [f"text='{text}'", f"fontsize={size}",
f"fontcolor=0x{colour}@{alpha:.3f}", f"x={x}", f"y={y}",
"borderw=%d" % max(1, size // 14), "bordercolor=black@0.6"]
if ff:
parts.insert(0, f"fontfile={ff}")
graph.append(f"[{cur}]drawtext={':'.join(parts)}[txt{n}]")
cur = f"txt{n}"
graph.append(f"[{cur}]null[vout]")
codec = {"h264": "libx264", "avc": "libx264", "h265": "libx265",
"hevc": "libx265"}.get((video_codec or "h264").lower(), "libx264")
enc = ["-c:v", codec, "-preset", "medium", "-crf", str(int(crf)),
"-pix_fmt", "yuv420p", "-movflags", "+faststart"]
if codec == "libx264":
enc += ["-profile:v", "high", "-level", "4.1"]
run(inputs + ["-filter_complex", ";".join(graph), "-map", "[vout]", "-map", "1:a"]
+ enc + ["-c:a", "aac", "-b:a", str(audio_bitrate or "192k"),
"-ar", "48000", "-ac", "2", "-shortest",
"-t", f"{max(0.1, total):.3f}", out_path],
cancel=cancel, log=log)
def thumbnail(video: str, out_path: str, *, at: float, width: int, height: int,
text: str = "", text_color: str = "#FFFFFF",
highlight_color: str = "#FFE14D", font: str = "",
cancel=None, log=None) -> None:
"""A frame from the finished video, optionally with the hook text on it."""
vf = ["format=yuv420p"]
if text:
ff = media.font_file(font or "Montserrat", bold=True)
size = int(height * 0.085)
# Break the hook into at most three lines of ~16 characters.
words, lines, cur = (text or "").split(), [], ""
for w in words:
if len(cur) + len(w) + 1 > 16 and cur:
lines.append(cur)
cur = w
else:
cur = f"{cur} {w}".strip()
if len(lines) == 3:
break
if cur and len(lines) < 3:
lines.append(cur)
block = len(lines)
for i, line in enumerate(lines):
y = f"(h-{block}*{int(size * 1.25)})/2+{i * int(size * 1.25)}"
parts = [f"text='{_esc_text(line)}'", f"fontsize={size}",
f"fontcolor=0x{(highlight_color if i == 0 else text_color).lstrip('#')}",
"x=(w-text_w)/2", f"y={y}",
f"borderw={max(2, size // 12)}", "bordercolor=black@0.85"]
if ff:
parts.insert(0, f"fontfile={ff}")
vf.append("drawtext=" + ":".join(parts))
run(["-ss", f"{max(0.0, at):.3f}", "-i", video, "-frames:v", "1",
"-vf", ",".join(vf), "-q:v", "3", out_path], cancel=cancel, log=log)
codai/main.py
View file @ 5abbb81d
......@@ -1281,6 +1281,11 @@ def main():
if global_file_path:
set_video_file_path(global_file_path)
# Composition writes its artefacts where /v1/files reads (codai/compose).
from codai.api.compose import set_global_file_path as set_compose_file_path
if global_file_path:
set_compose_file_path(global_file_path)
# Set audio_gen module global args
from codai.api.audio_gen import set_global_args as set_audiogen_global_args, set_global_file_path as set_audiogen_file_path
set_audiogen_global_args(global_args)
......
docs/README.md
View file @ 5abbb81d
......@@ -24,6 +24,7 @@ reference; these are the deep dives.
| Doc | Subject |
|---|---|
| [video-compose.md](video-compose.md) | `POST /v1/video/compose`: scenes → a finished, uploadable reel (narration timing, Ken Burns, ducked music, burned captions, thumbnail) and `/v1/files/upload` |
| [ocr.md](ocr.md) | The `/v1/ocr` document-transcription subsystem |
| [expressive-tts.md](expressive-tts.md) | Expressive text-to-speech |
| [zimage-lora-training.md](zimage-lora-training.md) | LoRA training for image models |
......
docs/video-compose.md 0 → 100644
View file @ 5abbb81d
# Composing a finished video — `POST /v1/video/compose`
CoderAI can generate every *piece* of a short video: narration
(`/v1/audio/speech`, `/v1/audio/clone`), stills (`/v1/images/generations`),
clips (`/v1/video/generations`), music (`/v1/audio/generate`). Nothing turned the
pieces into a reel, and a client on another machine cannot do it: assembly needs
ffmpeg and two facts only this server has — how long the narration it just
synthesised actually is, and when each word falls.
So composition happens here. You send an ordered list of **scenes**; you get back
an uploadable MP4, and optionally a thumbnail, subtitles and the narration track.
```
POST /v1/video/compose → 202 {"id": "cmp_…", "status": "queued"}
GET /v1/video/compose/{id} → status, stage, progress, warnings, result
POST /v1/video/compose/{id}/cancel → stop it (the ffmpeg child is killed)
GET /v1/video/compose → every job this front still remembers
POST /v1/files/upload → store your own footage/music once
GET /v1/files/blob/{sha256} → "do you already have these bytes?"
```
Everything is bearer-authenticated like the rest of the API, media fields accept
the usual four forms (a URL, a `/v1/files/...` path, a `data:` URI, raw base64),
and every artefact is served from `/v1/files/`.
## The shape of a request
```jsonc
{
"canvas": { "width": 1080, "height": 1920, "fps": 30 },
// Voice-over defaults for every scene that gives `text` without `audio`.
"voice": { "engine": "tts", "model": "kokoro", "voice": "af_sarah",
"language": "en-us", "speed": 1.0 },
"scenes": [
{
"text": "Octopuses have three hearts and blue blood.", // narration + caption
"audio": null, // pre-rendered narration instead; overrides TTS
"min_duration": 2.0, // floor, seconds
"padding": 0.15, // silence after the narration
"visuals": [ // split evenly across the scene, in order
{"type": "video", "src": "https://…/clip.mp4", "trim_start": 0.0},
{"type": "image", "src": "/v1/files/abc.png",
"ken_burns": {"zoom_from": 1.0, "zoom_to": 1.2, "pan": "auto"}},
{"type": "gradient", "colors": ["#6D28D9", "#DB2777"]},
{"type": "color", "color": "#111827"}
]
}
],
"visual_fit": "cover", // cover | contain | blur_pad
"transition": {"type": "none", "duration": 0.3}, // none|fade|crossfade|slide|wipe|dissolve
"music": {"src": "/v1/files/track.mp3", "volume": 0.14, "loop": true,
"duck": true, "fade_in": 1.0, "fade_out": 2.0},
"captions": {"enabled": true, "timing": "auto", "preset": "karaoke",
"font": "Montserrat", "position": "center",
"max_words_per_line": 4, "max_lines": 2},
"overlays": [{"type": "text", "text": "made with motus",
"position": "bottom-right", "opacity": 0.6, "size": 0.025},
{"type": "image", "src": "https://…/logo.png",
"position": "top-left", "width": 0.15}],
"thumbnail": {"at": 0.25, "text": "3 hearts?!"},
"output": {"format": "mp4", "video_codec": "h264", "crf": 20,
"audio_bitrate": "192k"},
"outputs": ["video", "thumbnail", "srt", "vtt", "narration"],
"async": true
}
```
A scene needs either narration (`text` and/or `audio`) or an explicit
`"duration"`, and at least one visual. Anything else has a default.
## What the server decides
**Timing.** A scene lasts *narration + padding*, at least `min_duration`; with no
narration it lasts its `duration`. The scene's visuals split that time equally,
in order. This is the whole reason composition is server-side: the client would
have to probe the audio it cannot see.
**Visuals.** Each is scaled to the canvas per `visual_fit` — `cover` crops to
fill (what a reel wants), `contain` letterboxes, `blur_pad` letterboxes onto a
blurred blow-up of the same frame — normalised to `canvas.fps` and `yuv420p`. A
video shorter than its slot loops; a longer one is trimmed from `trim_start`. A
still gets a slow Ken Burns move, alternating direction per image when
`pan: "auto"`. **A visual that cannot be fetched or decoded never fails the
job**: it becomes a gradient and a line in `warnings`.
**Audio.** Narration is synthesised per scene (so the timing is exact), padded,
concatenated and loudness-normalised to about −16 LUFS, which is why two reels
made a week apart are as loud as each other. Music is looped or trimmed to the
video's length, faded in and out, and — with `duck: true` — pushed under the
narration with a sidechain compressor. The mix is limited to −0.3 dBFS so no
platform re-encodes it for loudness.
**Captions.** Burned in, and **the text shown is the text you sent**. Speech
recognition is used only for *when* each word is spoken, never for what it says,
so a mis-heard word cannot reach the screen. `timing`:
* `auto` (default) — word timestamps from a speech model when one is configured,
estimated from word length otherwise (and a warning saying so);
* `stt` — require the model; `501` at submission time if there is none;
* `estimate` — never load a speech model.
Presets `karaoke`, `bold`, `neon` highlight the active word; `classic`, `boxed`,
`minimal` draw whole lines. Lines break at `max_words_per_line` and stay inside a
safe area of 10% top/bottom and 6% at the sides, clear of the TikTok/Reels/Shorts
furniture. The returned SRT/VTT carry the same text and the same timings.
**Output.** H.264 High + AAC 48 kHz stereo, `yuv420p`, `+faststart` — uploadable
to YouTube, TikTok, Instagram and Facebook as it is.
## Polling
```jsonc
{
"id": "cmp_8f2c1a",
"status": "running", // queued | running | done | failed | cancelled
"stage": "narration", // narration | captions | visuals | music | render | thumbnail
"progress": 42.0,
"message": "scene 3/8: narration",
"warnings": ["scene 2 visual 1: download failed, used gradient"],
"error": null,
"result": null
}
```
When `status` is `done`:
```jsonc
"result": {
"video": {"url": "https://…/v1/files/cmp_8f2c1a.mp4", "path": "/v1/files/cmp_8f2c1a.mp4",
"duration": 47.3, "width": 1080, "height": 1920, "size": 18234567},
"thumbnail": {"url": "…", "path": "/v1/files/cmp_8f2c1a.jpg", "size": 84213},
"srt": {"url": "…", "path": "/v1/files/cmp_8f2c1a.srt", "size": 1204},
"vtt": {"url": "…", "path": "/v1/files/cmp_8f2c1a.vtt", "size": 1250},
"narration": {"url": "…", "path": "/v1/files/cmp_8f2c1a_voice.wav", "size": 4500044},
"scenes": [{"index": 0, "start": 0.0, "end": 4.12}],
"duration": 47.3
}
```
`url` is absolute (built from the request that created the job) and `path` is the
relative form; either can be fetched, the relative one survives a change of
hostname.
Jobs **queue** and run one at a time — a render is ffmpeg-bound and two at once
finish later than one after the other — and a waiting job's `message` says how
many are ahead of it. Poll every 2–3 s. With `"async": false` the call blocks and
returns the finished document, which is useful in tests and a bad idea for a real
render.
Errors: `400` for an invalid spec, naming the scene and field
(`scenes[2].visuals[0].src: required for type 'video'`); `404` for an unknown job
id; `501` when a requested sub-feature has no model configured (`music.generate`
without an audio-generation model, `captions.timing: "stt"` without a speech
model, narration without a TTS model) or when the install has no ffmpeg.
## Uploading your own media
Inlining a 40 MB clip as base64 in every render is the kind of waste that makes
an API unusable, so upload it once:
```bash
curl -H "Authorization: Bearer $TOKEN" -F file=@footage.mp4 \
https://coderai.example/v1/files/upload
# {"id":"sha256:9f86d0…","url":"https://…/v1/files/up-9f86d0….mp4",
# "path":"/v1/files/up-9f86d0….mp4","bytes":41234567,"mime":"video/mp4",
# "kind":"video","existed":false}
```
Then name `path` (or `url`) as a visual's `src`, or as `music.src`. Storage is
content-addressed by sha256, so re-uploading the same bytes is free (`existed:
true`) and a client can ask first:
```bash
curl -H "Authorization: Bearer $TOKEN" https://coderai.example/v1/files/blob/9f86d0…
# 200 with the URL, or 404 — upload only on 404
```
The body may also be JSON (`{"file": "<base64 or data: URI>"}`) or the raw bytes
with a `Content-Type`. Video, image and audio only; the type is decided by the
content, not by the client's claim.
## Notes for operators
* **ffmpeg is required** (the published images have it, with libass for the
captions). Without it every compose request answers `501`.
* **Fonts**: captions ask for a family by name. The images ship Montserrat,
Inter, DejaVu and Noto (including CJK); a family that is not installed falls
back to the closest one and adds a warning, it does not fail.
* **Where the files go**: the ordinary output directory (`--file-path`), served
by `/v1/files/`, under normal retention. A composed video is typically read
once by the client and copied to its own storage.
* **Cost**: a render is CPU-bound. One job at a time, each ffmpeg child capped at
half the cores, so composition never starves the GPU engines.
* **Multi-engine installs**: composition runs in the engine the front routes it
to (the primary), and it synthesises narration and runs speech recognition
*in that process* — so the TTS and STT models it needs must be loadable there.
Leave them unpinned, or pin them to the primary; a model pinned to another
engine makes the job fail with that engine's own "model not allowed here"
message. Supplying each scene's `audio` and `captions.timing: "estimate"`
needs no model at all.
* **Jobs live in memory** (the last 100, plus whatever is running): a front
restart loses them, and a client polling a lost id gets `404` — treat that as
"restart the render", not as an error in the request.
## Code
| Piece | Where |
|---|---|
| Endpoints, validation, job queue, orchestration | `codai/api/compose.py` |
| Uploads (`/v1/files/upload`, `/v1/files/blob/{hash}`) | `codai/api/uploads.py` |
| Reference resolving, probing, fonts | `codai/compose/media.py` |
| Word timings, line breaking, ASS/SRT/VTT | `codai/compose/captions.py` |
| The ffmpeg pipeline | `codai/compose/render.py` |
| Tests (real ffmpeg, no model) | `tests/test_compose.py` |
packaging/linux/Dockerfile.oci
View file @ 5abbb81d
......@@ -169,6 +169,11 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
libvulkan1 \
lm-sensors \
espeak \
fonts-montserrat \
fonts-inter \
fonts-noto-core \
fonts-noto-cjk \
fonts-liberation \
mesa-vulkan-drivers \
pciutils \
openssl \
......
packaging/linux/Dockerfile.update
View file @ 5abbb81d
......@@ -65,4 +65,13 @@ RUN /opt/coderai/python/bin/python3 -m pip install --no-input --no-cache-dir "ze
RUN apt-get update && apt-get install -y --no-install-recommends openssh-client espeak \
&& rm -rf /var/lib/apt/lists/* || true
# Fonts for burned-in captions and overlays (codai/compose): Montserrat and
# Inter are what the caption presets ask for by name, Noto (with CJK) is what
# makes a non-Latin script render as glyphs instead of boxes. The base image
# carried DejaVu only, so a request for "Montserrat" silently got DejaVu.
RUN apt-get update && apt-get install -y --no-install-recommends \
fonts-montserrat fonts-inter fonts-noto-core fonts-noto-cjk fonts-liberation \
&& fc-cache -f >/dev/null 2>&1 \
&& rm -rf /var/lib/apt/lists/* || true
# ENTRYPOINT / EXPOSE / VOLUME / ENV / WORKDIR are inherited from the base image.
packaging/runpod/Dockerfile.capability-base-light
View file @ 5abbb81d
......@@ -27,6 +27,7 @@ ENV PYTHONUNBUFFERED=1 \
# at install time; libgl/libglib for opencv.
RUN apt-get update && apt-get install -y --no-install-recommends \
ffmpeg git curl ca-certificates libgl1 libglib2.0-0 \
fonts-montserrat fonts-inter fonts-noto-core fonts-noto-cjk \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /opt/coderai/app
......
tests/test_compose.py 0 → 100644
View file @ 5abbb81d
"""Server-side composition: ``POST /v1/video/compose`` and ``POST /v1/files/upload``.
These tests really run ffmpeg — a composition that "works" in mocks and produces
an unplayable file is worthless — but never a model: narration is supplied as
pre-made audio, captions are timed by estimation, and music is a sine wave. So
the whole pipeline (visual normalisation, concat and crossfade, narration
padding, ducking, burned captions, thumbnail, SRT/VTT) is exercised end to end
on any machine with ffmpeg and no GPU.
"""
import json
import os
import subprocess
import sys
import time
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from codai.compose import captions as cap
from codai.compose import media
from codai.compose import render as rnd
pytestmark = pytest.mark.skipif(media.ffmpeg_bin() is None
or subprocess.run(["which", "ffmpeg"],
capture_output=True).returncode != 0,
reason="composition needs ffmpeg on PATH")
# --------------------------------------------------------------- fixtures
def _tone(path, seconds=1.0, freq=440):
subprocess.run([media.ffmpeg_bin(), "-y", "-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", f"sine=frequency={freq}:duration={seconds}",
"-ac", "2", "-ar", "48000", str(path)], check=True)
return str(path)
def _clip(path, seconds=1.0, w=320, h=240, colour="red"):
subprocess.run([media.ffmpeg_bin(), "-y", "-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", f"color=c={colour}:s={w}x{h}:r=25:d={seconds}",
"-c:v", "libx264", "-pix_fmt", "yuv420p", str(path)], check=True)
return str(path)
def _uri(path):
"""A local fixture file as a client would send it: a data URI. Compose
deliberately refuses bare server paths, so tests use the real wire form."""
import base64
import mimetypes
mime = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
return f"data:{mime};base64," + base64.b64encode(Path(path).read_bytes()).decode()
def _png(path, w=64, h=64, colour=(200, 40, 90)):
from PIL import Image
Image.new("RGB", (w, h), colour).save(str(path))
return str(path)
@pytest.fixture
def files_dir(tmp_path, monkeypatch):
d = tmp_path / "out"
d.mkdir()
media.set_global_file_path(str(d))
yield d
media.set_global_file_path("")
# =================================================================== words
def test_estimated_word_timings_fill_the_scene_exactly():
words = cap.estimate_words("Octopuses have three hearts and blue blood.", 2.0, 4.0)
assert [w["word"] for w in words] == ["Octopuses", "have", "three", "hearts",
"and", "blue", "blood."]
assert words[0]["start"] == 2.0
assert words[-1]["end"] == pytest.approx(6.0, abs=0.01)
# Monotonic, nothing zero-length, longer words get more time than "and".
for a, b in zip(words, words[1:]):
assert a["end"] <= b["start"] + 1e-6 and a["end"] > a["start"]
assert (words[0]["end"] - words[0]["start"]) > (words[4]["end"] - words[4]["start"])
assert cap.estimate_words("", 0.0, 3.0) == []
def test_stt_timings_are_aligned_onto_the_callers_words():
# The model heard "3" as "three" and split "voiceover"; the caption must still
# show what the caller wrote, on the model's clock.
stt = [{"word": "octopuses", "start": 0.0, "end": 0.5},
{"word": "have", "start": 0.5, "end": 0.7},
{"word": "three", "start": 0.7, "end": 1.1},
{"word": "hearts", "start": 1.1, "end": 1.6}]
out = cap.align_words("Octopuses have 3 hearts", stt, 10.0, 2.0)
assert [w["word"] for w in out] == ["Octopuses", "have", "3", "hearts"]
assert out[0]["start"] == pytest.approx(10.0, abs=0.01)
assert out[2]["start"] == pytest.approx(10.7, abs=0.05) # "3" got "three"'s slot
assert out[-1]["end"] <= 12.0 + 1e-6
# No STT at all → estimation, never an empty caption track.
assert len(cap.align_words("two words here", [], 0.0, 1.5)) == 3
def test_lines_break_and_srt_matches_the_text():
words = cap.estimate_words("one two three four five six seven eight", 0.0, 8.0)
events = cap.break_lines(words, max_words_per_line=2, max_lines=2)
assert [ev["rows"] for ev in events] == [[["one", "two"], ["three", "four"]],
[["five", "six"], ["seven", "eight"]]]
assert events[0]["start"] == 0.0 and events[-1]["end"] == pytest.approx(8.0, abs=0.01)
srt = cap.build_srt(events)
assert srt.startswith("1\n00:00:00,000 --> ")
assert "one two\nthree four" in srt
vtt = cap.build_vtt(events)
assert vtt.startswith("WEBVTT") and "-->" in vtt and "," not in vtt.split("\n")[2]
def test_ass_is_karaoke_safe_area_and_never_leaks_the_stt_text():
words = cap.estimate_words("hello brave new world", 0.0, 4.0)
events = cap.break_lines(words, 2, 2)
ass = cap.build_ass(events, width=1080, height=1920, font="Montserrat",
colour="#FFFFFF", highlight="#FFE14D", position="center",
preset="karaoke")
assert "PlayResX: 1080" and "PlayResY: 1920" in ass
# Safe area: 10% vertical, 6% horizontal margins, centre alignment.
style = [l for l in ass.splitlines() if l.startswith("Style: cap,")][0]
assert style.endswith(",65,65,192,1") and ",5," in style # 6% sides, 10% vertical
# One event per word (karaoke) and the highlight colour appears in each.
dialogues = [l for l in ass.splitlines() if l.startswith("Dialogue:")]
assert len(dialogues) == 4
assert all("&H004DE1FF" in d for d in dialogues) # #FFE14D as ASS BGR
# classic preset: one event per line, no per-word highlight
plain = cap.build_ass(events, width=1080, height=1920, preset="classic")
# One event for the whole 2-line block, the rows joined with ASS's \N.
assert len([l for l in plain.splitlines() if l.startswith("Dialogue:")]) == 1
assert "hello brave\\Nnew world" in plain
def test_uppercase_and_box_presets_change_the_style_not_the_words():
events = cap.break_lines(cap.estimate_words("keep it real", 0.0, 2.0), 3, 1)
boxed = cap.build_ass(events, width=720, height=1280, preset="boxed", uppercase=True)
assert "KEEP IT REAL" in boxed
style = [l for l in boxed.splitlines() if l.startswith("Style: cap,")][0]
assert ",3," in style # BorderStyle 3 = opaque box
# =================================================================== media
def test_mime_sniffing_beats_the_clients_content_type(tmp_path):
png = Path(_png(tmp_path / "x.png"))
assert media.sniff_mime(png.read_bytes(), "application/octet-stream") == "image/png"
assert media.kind_of("image/png") == "image"
wav = Path(_tone(tmp_path / "a.wav", 0.2))
assert media.sniff_mime(wav.read_bytes()) == "audio/wav"
mp4 = Path(_clip(tmp_path / "v.mp4", 0.2))
assert media.sniff_mime(mp4.read_bytes()) == "video/mp4"
assert media.ext_for("video/mp4") == ".mp4"
assert media.kind_of("text/plain") == "other"
def test_references_resolve_from_data_uris_files_urls_and_base64(tmp_path, files_dir):
import base64
png = Path(_png(tmp_path / "y.png")).read_bytes()
work = str(tmp_path / "work")
p1 = media.resolve("data:image/png;base64," + base64.b64encode(png).decode(),
work, "a")
assert Path(p1).read_bytes() == png
p2 = media.resolve(base64.b64encode(png).decode(), work, "b")
assert Path(p2).read_bytes() == png
# A file this install already serves is used where it lies, not re-downloaded.
(files_dir / "held.png").write_bytes(png)
assert media.resolve("/v1/files/held.png", work, "c") == str(files_dir / "held.png")
assert media.resolve("http://elsewhere.invalid:1/v1/files/held.png", work, "d") \
== str(files_dir / "held.png")
# Path traversal through the files URL is refused.
assert media.local_path_for_files_url("/v1/files/../../etc/passwd") is None
with pytest.raises(media.MediaError):
media.resolve("/etc/passwd", work, "e")
with pytest.raises(media.MediaError):
media.resolve("not base64 at all !!", work, "f")
def test_probe_reports_what_the_renderer_needs(tmp_path):
mp4 = _clip(tmp_path / "p.mp4", seconds=1.5, w=640, h=360)
info = media.probe(mp4)
assert info["has_video"] and not info["has_audio"]
assert info["width"] == 640 and info["height"] == 360
assert info["duration"] == pytest.approx(1.5, abs=0.15)
assert info["fps"] == pytest.approx(25, abs=1)
wav = _tone(tmp_path / "p.wav", 0.7)
a = media.probe(wav)
assert a["has_audio"] and not a["has_video"]
assert media.duration_of(wav) == pytest.approx(0.7, abs=0.05)
with pytest.raises(media.MediaError):
media.probe(str(tmp_path / "nope.mp4"))
# ================================================================== render
def test_fit_filters_cover_contain_and_blur_pad():
assert "crop=1080:1920" in rnd.fit_filter(1080, 1920, "cover")
assert "pad=1080:1920" in rnd.fit_filter(1080, 1920, "contain")
blur = rnd.fit_filter(1080, 1920, "blur_pad")
assert "gblur" in blur and "overlay" in blur
assert rnd.fit_filter(100, 100, "nonsense") == rnd.fit_filter(100, 100, "cover")
def test_ken_burns_alternates_direction_and_respects_the_zoom_range():
a = rnd.ken_burns_filter(1080, 1920, 30, 2.0, 1.0, 1.2, "auto", index=0)
b = rnd.ken_burns_filter(1080, 1920, 30, 2.0, 1.0, 1.2, "auto", index=1)
assert "zoompan" in a and "d=60" in a and "s=1080x1920" in a
assert a != b # right, then left
assert "2160" in a # oversampled before zoompan
down = rnd.ken_burns_filter(720, 1280, 25, 1.0, 1.2, 1.0, "none", 0)
assert "max(" in down # zooming out
assert "zoompan" in rnd.ken_burns_filter(720, 1280, 25, 1.0, 1.0, 1.0, "up", 0)
def test_a_visual_of_each_type_becomes_a_clip_of_the_right_length(tmp_path):
work = tmp_path / "w"
work.mkdir()
cases = [
({"type": "color", "color": "#112233"}, 0.6),
({"type": "gradient", "colors": ["#6D28D9", "#DB2777"]}, 0.5),
({"type": "image", "_path": _png(tmp_path / "i.png", 200, 300)}, 0.8),
({"type": "video", "_path": _clip(tmp_path / "v.mp4", 2.0),
"_probe": None, "trim_start": 0.5}, 0.7),
]
for n, (spec, dur) in enumerate(cases):
if spec.get("type") == "video":
spec["_probe"] = media.probe(spec["_path"])
out = str(work / f"c{n}.mp4")
rnd.build_visual(spec, out, width=240, height=426, fps=24, duration=dur,
fit="cover", index=n, workdir=str(work))
info = media.probe(out)
assert info["width"] == 240 and info["height"] == 426
assert info["duration"] == pytest.approx(dur, abs=0.12), spec["type"]
assert not info["has_audio"]
def test_a_short_video_is_looped_to_fill_its_slot(tmp_path):
src = _clip(tmp_path / "short.mp4", seconds=0.4)
out = str(tmp_path / "looped.mp4")
rnd.build_visual({"type": "video", "_path": src, "_probe": media.probe(src)},
out, width=160, height=284, fps=24, duration=1.6,
workdir=str(tmp_path))
assert media.probe(out)["duration"] == pytest.approx(1.6, abs=0.12)
def test_clips_join_by_concat_and_by_crossfade_without_losing_time(tmp_path):
work = tmp_path / "j"
work.mkdir()
clips, durs = [], []
for i, colour in enumerate(("red", "green", "blue")):
p = str(work / f"{i}.mp4")
rnd.build_visual({"type": "color", "color": colour}, p, width=160, height=284,
fps=24, duration=1.0 + 0.3, workdir=str(work), index=i)
clips.append(p)
durs.append(1.3)
joined = str(work / "concat.mp4")
rnd.concat_clips(clips, joined, str(work))
assert media.probe(joined)["duration"] == pytest.approx(3.9, abs=0.15)
faded = str(work / "xfade.mp4")
rnd.xfade_clips(clips, durs, faded, kind="crossfade", dur=0.3, fps=24)
# 3 clips of 1.3 with 0.3 of overlap each → 3*1.0 + 0.3 of tail.
assert media.probe(faded)["duration"] == pytest.approx(3.3, abs=0.2)
def test_narration_is_padded_per_scene_and_music_is_ducked(tmp_path):
work = tmp_path / "a"
work.mkdir()
v1 = _tone(work / "v1.wav", 0.5, 300)
padded = rnd.pad_to(v1, str(work / "p1.wav"), 1.5)
assert media.duration_of(padded) == pytest.approx(1.5, abs=0.05)
sil = rnd.silence(str(work / "s.wav"), 0.8)
joined = rnd.concat_audio([padded, sil], str(work / "n.wav"), str(work))
assert media.duration_of(joined) == pytest.approx(2.3, abs=0.08)
normed = rnd.loudnorm(joined, str(work / "nl.wav"))
assert media.duration_of(normed) == pytest.approx(2.3, abs=0.15)
bed = rnd.build_music_bed(_tone(work / "m.wav", 0.6, 120), str(work / "bed.wav"),
total=2.3, volume=0.2, loop=True, fade_in=0.2, fade_out=0.4)
assert media.duration_of(bed) == pytest.approx(2.3, abs=0.08)
mixed = rnd.mix_narration_music(normed, bed, str(work / "mix.wav"), duck=True)
assert media.duration_of(mixed) == pytest.approx(2.3, abs=0.12)
alone = rnd.mix_narration_music(normed, None, str(work / "solo.wav"))
assert media.duration_of(alone) == pytest.approx(2.3, abs=0.12)
def test_the_final_render_is_uploadable_and_carries_captions_and_overlays(tmp_path):
work = tmp_path / "f"
work.mkdir()
silent = str(work / "v.mp4")
rnd.build_visual({"type": "gradient"}, silent, width=240, height=426, fps=24,
duration=2.0, workdir=str(work))
audio = rnd.mix_narration_music(_tone(work / "n.wav", 2.0), None, str(work / "a.wav"))
events = cap.break_lines(cap.estimate_words("burned in captions here", 0.0, 2.0), 2, 2)
ass = work / "c.ass"
ass.write_text(cap.build_ass(events, width=240, height=426, font="DejaVu Sans"))
out = str(work / "final.mp4")
rnd.final_render(silent, audio, out, width=240, height=426, fps=24, total=2.0,
ass_file=str(ass),
overlays=[{"type": "text", "text": "made with motus",
"position": "bottom-right", "opacity": 0.6, "size": 0.04}],
crf=28)
info = media.probe(out)
assert info["has_video"] and info["has_audio"]
assert info["duration"] == pytest.approx(2.0, abs=0.15)
# H.264 High + AAC + faststart is what the platforms take without re-encoding.
probe = subprocess.run(["ffprobe", "-v", "error", "-show_streams", "-show_format",
"-print_format", "json", out], capture_output=True)
doc = json.loads(probe.stdout)
vs = [s for s in doc["streams"] if s["codec_type"] == "video"][0]
aud = [s for s in doc["streams"] if s["codec_type"] == "audio"][0]
assert vs["codec_name"] == "h264" and vs["profile"] == "High"
assert vs["pix_fmt"] == "yuv420p"
assert aud["codec_name"] == "aac" and int(aud["sample_rate"]) == 48000
thumb = str(work / "t.jpg")
rnd.thumbnail(out, thumb, at=0.5, width=240, height=426, text="3 hearts?!")
assert os.path.getsize(thumb) > 500
def test_a_cancelled_render_kills_its_ffmpeg(tmp_path):
import threading
ev = threading.Event()
ev.set()
with pytest.raises(rnd.Cancelled):
rnd.run(["-f", "lavfi", "-i", "color=c=black:s=64x64:r=30:d=30",
"-c:v", "libx264", str(tmp_path / "never.mp4")], cancel=ev)
with pytest.raises(rnd.RenderError):
rnd.run(["-i", str(tmp_path / "missing.mp4"), str(tmp_path / "x.mp4")])
# ============================================================ the whole job
def _spec(tmp_path, **over):
voice = _tone(tmp_path / "vo1.wav", 1.2, 330)
voice2 = _tone(tmp_path / "vo2.wav", 0.8, 420)
spec = {
"canvas": {"width": 160, "height": 284, "fps": 24},
"scenes": [
{"text": "Octopuses have three hearts.", "audio": _uri(voice), "padding": 0.2,
"visuals": [{"type": "gradient", "colors": ["#6D28D9", "#DB2777"]},
{"type": "image", "src": _data_uri_png(tmp_path)}]},
{"text": "And blue blood.", "audio": _uri(voice2), "padding": 0.1,
"min_duration": 1.5,
"visuals": [{"type": "color", "color": "#111827"}]},
],
"captions": {"enabled": True, "timing": "estimate", "preset": "karaoke",
"font": "DejaVu Sans", "max_words_per_line": 2},
"overlays": [{"type": "text", "text": "motus", "position": "bottom-right"}],
"thumbnail": {"at": 0.3, "text": "3 hearts?!"},
"output": {"crf": 30},
"outputs": ["video", "thumbnail", "srt", "vtt", "narration"],
"async": True,
}
spec.update(over)
return spec
def _data_uri_png(tmp_path):
import base64
data = Path(_png(tmp_path / "scene.png", 120, 200)).read_bytes()
return "data:image/png;base64," + base64.b64encode(data).decode()
def _client(files_dir):
from fastapi import FastAPI
from fastapi.testclient import TestClient
from codai.api import compose as compose_mod
from codai.api import uploads as uploads_mod
app = FastAPI()
app.include_router(compose_mod.router)
app.include_router(uploads_mod.router)
return TestClient(app), compose_mod
def _wait(client, job_id, timeout=300):
deadline = time.time() + timeout
while time.time() < deadline:
r = client.get(f"/v1/video/compose/{job_id}")
assert r.status_code == 200
doc = r.json()
if doc["status"] in ("done", "failed", "cancelled"):
return doc
time.sleep(0.4)
raise AssertionError("composition did not finish in time")
def test_a_composition_job_produces_every_artefact(tmp_path, files_dir):
client, _ = _client(files_dir)
r = client.post("/v1/video/compose", json=_spec(tmp_path))
assert r.status_code == 202, r.text
job_id = r.json()["id"]
assert r.json()["status"] == "queued" and job_id.startswith("cmp_")
doc = _wait(client, job_id)
assert doc["status"] == "done", doc
assert doc["progress"] == 100.0 and doc["error"] is None
res = doc["result"]
# Scene timing is narration-driven: 1.2+0.2, then max(0.8+0.1, 1.5).
assert [s["start"] for s in res["scenes"]] == [0.0, pytest.approx(1.4, abs=0.1)]
assert res["scenes"][1]["end"] == pytest.approx(2.9, abs=0.15)
assert res["duration"] == pytest.approx(2.9, abs=0.15)
for key in ("video", "thumbnail", "srt", "vtt", "narration"):
assert key in res, key
name = res[key]["path"].split("/")[-1]
assert (files_dir / name).is_file()
assert res[key]["url"].endswith(res[key]["path"]) or res[key]["url"].startswith("/v1/")
vid = files_dir / res["video"]["path"].split("/")[-1]
info = media.probe(str(vid))
assert info["has_video"] and info["has_audio"]
assert (info["width"], info["height"]) == (160, 284)
assert info["duration"] == pytest.approx(2.9, abs=0.25)
assert res["video"]["duration"] == pytest.approx(info["duration"], abs=0.05)
# The captions carry the caller's text, not a transcription.
srt = (files_dir / res["srt"]["path"].split("/")[-1]).read_text()
assert "Octopuses have" in srt and "three hearts." in srt
assert "And blue" in srt and "blood." in srt # 2 words per line, as asked
# The download endpoint of the real app serves them by name (path shape).
assert res["video"]["path"].startswith("/v1/files/cmp_")
def test_the_blocking_form_returns_the_finished_document(tmp_path, files_dir):
client, _ = _client(files_dir)
spec = _spec(tmp_path, **{"async": False, "outputs": ["video"],
"captions": {"enabled": False}})
spec["scenes"] = spec["scenes"][:1]
r = client.post("/v1/video/compose", json=spec)
assert r.status_code == 200, r.text
doc = r.json()
assert doc["status"] == "done" and doc["result"]["video"]["size"] > 1000
assert "thumbnail" not in doc["result"] and "srt" not in doc["result"]
def test_a_broken_visual_becomes_a_gradient_and_a_warning(tmp_path, files_dir):
client, _ = _client(files_dir)
spec = _spec(tmp_path, captions={"enabled": False}, outputs=["video"])
spec["scenes"] = [{
"text": "one scene", "audio": _uri(_tone(tmp_path / "v3.wav", 0.6)),
"visuals": [{"type": "video", "src": "http://127.0.0.1:1/nope.mp4"},
{"type": "image", "src": "data:image/png;base64,bm90YW5pbWFnZQ=="}],
}]
doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
assert doc["status"] == "done", doc
assert len(doc["warnings"]) >= 2
assert any("gradient" in w for w in doc["warnings"])
vid = files_dir / doc["result"]["video"]["path"].split("/")[-1]
assert media.probe(str(vid))["has_video"]
def test_crossfade_and_music_paths_render(tmp_path, files_dir):
client, _ = _client(files_dir)
spec = _spec(tmp_path, captions={"enabled": False}, outputs=["video"],
transition={"type": "crossfade", "duration": 0.25},
music={"src": _uri(_tone(tmp_path / "bed.wav", 0.5, 110)), "volume": 0.2,
"duck": True, "loop": True, "fade_in": 0.2, "fade_out": 0.3})
doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
assert doc["status"] == "done", doc
vid = files_dir / doc["result"]["video"]["path"].split("/")[-1]
assert media.probe(str(vid))["has_audio"]
def test_cancel_stops_a_running_job(tmp_path, files_dir):
client, compose_mod = _client(files_dir)
# Long enough that cancellation lands mid-render.
spec = _spec(tmp_path, captions={"enabled": False}, outputs=["video"])
spec["canvas"] = {"width": 720, "height": 1280, "fps": 30}
spec["scenes"] = [{"text": "x", "audio": _uri(_tone(tmp_path / "long.wav", 8.0)),
"visuals": [{"type": "gradient"}]}]
job_id = client.post("/v1/video/compose", json=spec).json()["id"]
time.sleep(1.0)
r = client.post(f"/v1/video/compose/{job_id}/cancel")
assert r.status_code == 200 and r.json()["status"] == "cancelled"
doc = _wait(client, job_id, timeout=60)
assert doc["status"] == "cancelled"
assert client.post("/v1/video/compose/cmp_nope/cancel").status_code == 404
assert client.get("/v1/video/compose/cmp_nope").status_code == 404
def test_the_spec_is_validated_with_the_offending_field_named(tmp_path, files_dir):
client, _ = _client(files_dir)
def err(spec):
r = client.post("/v1/video/compose", json=spec)
assert r.status_code == 400, r.text
return r.json()["detail"]
assert "at least one scene" in err({"scenes": []})
assert "scenes[0].visuals" in err({"scenes": [{"text": "hi", "visuals": []}]})
assert "explicit" in err({"scenes": [{"visuals": [{"type": "color"}]}]})
d = err({"scenes": [{"text": "a", "visuals": [{"type": "hologram"}]}]})
assert "scenes[0].visuals[0].type" in d and "hologram" in d
assert "src" in err({"scenes": [{"text": "a", "visuals": [{"type": "video"}]}]})
good = {"scenes": [{"text": "a", "audio": _uri(_tone(tmp_path / "ok.wav", 0.3)),
"visuals": [{"type": "color"}]}]}
assert "even" in err({**good, "canvas": {"width": 101, "height": 100}})
assert "visual_fit" in err({**good, "visual_fit": "squeeze"})
assert "transition.type" in err({**good, "transition": {"type": "swirl"}})
assert "captions.preset" in err({**good, "captions": {"preset": "sparkles"}})
assert "outputs" in err({**good, "outputs": ["video", "gif"]})
assert "output.format" in err({**good, "output": {"format": "avi"}})
def test_missing_models_answer_501_not_a_failed_job(tmp_path, files_dir, monkeypatch):
client, compose_mod = _client(files_dir)
monkeypatch.setattr(compose_mod, "_model_ids", lambda mtype: [])
base = {"scenes": [{"text": "a", "audio": _uri(_tone(tmp_path / "m.wav", 0.3)),
"visuals": [{"type": "color"}]}]}
r = client.post("/v1/video/compose", json={**base, "music": {"generate": {"prompt": "x"}}})
assert r.status_code == 501 and "audio-generation model" in r.json()["detail"]
r = client.post("/v1/video/compose",
json={**base, "captions": {"enabled": True, "timing": "stt"}})
assert r.status_code == 501 and "speech-to-text" in r.json()["detail"]
# Text with no audio and no TTS model: 501 rather than a job that fails later.
r = client.post("/v1/video/compose",
json={"scenes": [{"text": "a", "visuals": [{"type": "color"}]}]})
assert r.status_code == 501 and "TTS model" in r.json()["detail"]
def test_narration_is_synthesised_through_the_tts_endpoint(tmp_path, files_dir, monkeypatch):
"""No TTS model here, so the endpoint is stubbed — what is tested is that
compose calls it once per scene with that scene's text and voice."""
import base64
client, compose_mod = _client(files_dir)
monkeypatch.setattr(compose_mod, "_model_ids",
lambda mtype: ["kokoro"] if mtype == "tts" else [])
calls = []
wav = Path(_tone(tmp_path / "synth.wav", 0.5)).read_bytes()
async def fake_speech(req, http_request=None):
calls.append({"input": req.input, "voice": req.voice, "model": req.model,
"speed": req.speed, "language": getattr(req, "language", None)})
return {"audio": base64.b64encode(wav).decode()}
import codai.api.tts as tts_mod
monkeypatch.setattr(tts_mod, "create_speech", fake_speech)
spec = {"canvas": {"width": 160, "height": 284, "fps": 24},
"voice": {"engine": "tts", "model": "kokoro", "voice": "af_sarah",
"language": "en-us", "speed": 1.1},
"captions": {"enabled": True, "timing": "estimate", "font": "DejaVu Sans"},
"outputs": ["video", "srt"],
"scenes": [{"text": "first line", "visuals": [{"type": "color"}]},
{"text": "second line", "visuals": [{"type": "gradient"}]}]}
doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
assert doc["status"] == "done", doc
assert [c["input"] for c in calls] == ["first line", "second line"]
assert calls[0]["voice"] == "af_sarah" and calls[0]["model"] == "kokoro"
assert calls[0]["speed"] == 1.1 and calls[0]["language"] == "en-us"
srt = (files_dir / doc["result"]["srt"]["path"].split("/")[-1]).read_text()
assert "first line" in srt and "second line" in srt
def test_stt_timings_are_used_when_a_speech_model_is_configured(tmp_path, files_dir,
monkeypatch):
client, compose_mod = _client(files_dir)
monkeypatch.setattr(compose_mod, "_model_ids",
lambda mtype: ["whisper-large-v3"] if mtype == "audio" else [])
seen = {}
def fake_words(loop, path, model, language):
seen["model"] = model
return [{"word": "hello", "start": 0.0, "end": 0.9},
{"word": "world", "start": 0.9, "end": 1.0}]
monkeypatch.setattr(compose_mod, "_stt_words", fake_words)
spec = {"canvas": {"width": 160, "height": 284, "fps": 24},
"captions": {"enabled": True, "timing": "stt", "stt_model": "whisper-large-v3",
"font": "DejaVu Sans", "max_words_per_line": 2},
"outputs": ["video", "srt"],
"scenes": [{"text": "hello world", "audio": _uri(_tone(tmp_path / "s.wav", 1.0)),
"padding": 0.0, "visuals": [{"type": "color"}]}]}
doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
assert doc["status"] == "done", doc
assert seen["model"] == "whisper-large-v3"
srt = (files_dir / doc["result"]["srt"]["path"].split("/")[-1]).read_text()
# "world" starts at 0.9 per the model, not at the halfway point of the clip.
assert "00:00:00,000 --> 00:00:01,0" in srt and "hello world" in srt
# ================================================================= uploads
def test_upload_is_content_addressed_and_dedupes(tmp_path, files_dir):
client, _ = _client(files_dir)
png = Path(_png(tmp_path / "u.png")).read_bytes()
r = client.post("/v1/files/upload", files={"file": ("u.png", png, "image/png")})
assert r.status_code == 200, r.text
doc = r.json()
assert doc["id"].startswith("sha256:") and len(doc["id"]) == 71
assert doc["mime"] == "image/png" and doc["kind"] == "image"
assert doc["bytes"] == len(png) and doc["existed"] is False
assert doc["path"].startswith("/v1/files/up-")
assert (files_dir / doc["path"].split("/")[-1]).read_bytes() == png
again = client.post("/v1/files/upload",
files={"file": ("other-name.png", png, "application/octet-stream")})
assert again.json()["existed"] is True and again.json()["id"] == doc["id"]
# The existence check lets a client skip the upload entirely.
h = doc["id"].split(":")[1]
assert client.get(f"/v1/files/blob/{h}").json()["exists"] is True
assert client.get(f"/v1/files/blob/sha256:{h}").status_code == 200
assert client.get("/v1/files/blob/" + "0" * 64).status_code == 404
assert client.get("/v1/files/blob/not-a-hash").status_code == 404
def test_upload_accepts_json_and_raw_bodies_and_refuses_junk(tmp_path, files_dir):
import base64
client, _ = _client(files_dir)
wav = Path(_tone(tmp_path / "u.wav", 0.2)).read_bytes()
r = client.post("/v1/files/upload",
json={"file": base64.b64encode(wav).decode(), "filename": "u.wav"})
assert r.status_code == 200 and r.json()["kind"] == "audio"
mp4 = Path(_clip(tmp_path / "u.mp4", 0.3)).read_bytes()
r = client.post("/v1/files/upload", content=mp4,
headers={"content-type": "video/mp4"})
assert r.status_code == 200 and r.json()["mime"] == "video/mp4"
r = client.post("/v1/files/upload", content=b"#!/bin/sh\nrm -rf /\n",
headers={"content-type": "application/x-sh"})
assert r.status_code == 415 and "unsupported media type" in r.json()["detail"]
assert client.post("/v1/files/upload", content=b"",
headers={"content-type": "video/mp4"}).status_code == 400
def test_an_uploaded_file_can_be_used_as_a_visual(tmp_path, files_dir):
client, _ = _client(files_dir)
clip = Path(_clip(tmp_path / "stock.mp4", 1.0, 320, 180, "blue")).read_bytes()
up = client.post("/v1/files/upload",
files={"file": ("stock.mp4", clip, "video/mp4")}).json()
spec = {"canvas": {"width": 160, "height": 284, "fps": 24},
"captions": {"enabled": False}, "outputs": ["video"],
"scenes": [{"text": "with my own footage",
"audio": _uri(_tone(tmp_path / "own.wav", 0.8)),
"visuals": [{"type": "video", "src": up["path"]}]}]}
doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
assert doc["status"] == "done", doc
assert doc["warnings"] == [] # resolved locally, nothing downloaded
def test_compose_and_uploads_are_never_forwarded_to_a_remote():
"""A pod has ffmpeg but not this install's models or files: composition and
uploads must stay here even when a video capability is remoted."""
from codai.api.remote_gateway import capability_for, resolve_target
assert capability_for("/v1/video/compose") == "video" # by prefix…
for path in ("/v1/video/compose", "/v1/video/compose/cmp_1",
"/v1/video/compose/cmp_1/cancel", "/v1/files/upload",
"/v1/files/blob/" + "a" * 64):
assert resolve_target(path, "POST", "", b"", "application/json") is None, path
def test_a_cloned_voice_profile_is_used_for_narration(tmp_path, files_dir, monkeypatch):
"""engine=clone goes through /v1/audio/clone, whose answer is a url when the
install has an output directory and inline audio when it does not."""
client, compose_mod = _client(files_dir)
monkeypatch.setattr(compose_mod, "_model_ids", lambda mtype: [])
monkeypatch.setattr(compose_mod, "_voice_available", lambda req: True)
wav = Path(_tone(tmp_path / "cloned.wav", 0.6)).read_bytes()
(files_dir / "cloned-out.wav").write_bytes(wav)
seen = []
async def fake_clone(req, http_request=None):
seen.append({"text": req.text, "voice_name": req.voice_name, "speed": req.speed})
return {"created": 0, "data": [{"url": "/v1/files/cloned-out.wav"}], "engine": "f5"}
import codai.api.voice_clone as vc
monkeypatch.setattr(vc, "clone_voice", fake_clone)
spec = {"canvas": {"width": 160, "height": 284, "fps": 24},
"voice": {"engine": "clone", "voice_name": "nextime", "speed": 0.9},
"captions": {"enabled": False}, "outputs": ["video", "narration"],
"scenes": [{"text": "in my own voice", "visuals": [{"type": "color"}]}]}
doc = _wait(client, client.post("/v1/video/compose", json=spec).json()["id"])
assert doc["status"] == "done", doc
assert seen == [{"text": "in my own voice", "voice_name": "nextime", "speed": 0.9}]
# The scene is as long as the cloned clip (0.6s) plus the default padding.
assert doc["result"]["duration"] == pytest.approx(0.75, abs=0.1)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment