Skip to content

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

normalise a language before transcribing, stop inflating weights that are... 

normalise a language before transcribing, stop inflating weights that are already quantized, and never answer 200 with nothing (v0.5.9)

Five bugs and five features from a client report. The bugs share a shape: the
server held a belief about a model that nothing checked against the model.

A1. A TTS voice is chosen for an accent, so the language a compose render
    carries is "en-us" / "pt-br". Passed to transformers' Whisper -- which
    accepts only the English NAME -- it raised "Unsupported language: en-us"
    and 500'd every captioned scene of every render. Reproduced verbatim
    against the running server before the fix. New codai/api/lang_codes.py
    normalises once, per backend: the NAME for the transformers pipeline, the
    ISO code for whisper.cpp, Canary and faster-whisper, and None (auto-detect)
    for a tag that maps to nothing -- refusing to transcribe because the caller
    named a dialect is never the better outcome. Language VALIDATION now
    compares by meaning, so a model declaring "english" no longer refuses
    "en-gb", and compose's model picker stops disqualifying the only model that
    can time words.

A3. FLUX.2 (dev/klein/pro) is a unified generate-and-edit model whose name
    carries neither "edit" nor "kontext", so it fell through to the generation
    branch and advertised plain img2img -- telling clients to show a `strength`
    slider for a pipeline that has no such argument. Confirmed live:
    ['image_generation','image_to_image','inpainting']. diffusers 0.37 agrees
    it is an instruction edit (Flux2Pipeline takes `image`, not `strength`),
    which is what the runtime already decided on; only the advertised list
    disagreed.

A4. The estimator normalises a disk size to the load precision. A GGUF was
    special-cased to 1.0 because its bytes ARE its footprint; an HF checkpoint
    published pre-quantized is the same case by another route and was not.
    Z-Image-Turbo-unsloth-bnb-4bit: 16.3 GB of 4-bit weights, 1.70x factor,
    estimated 30.03 GB against a 24 GB card. Nothing can free 30 of 24, so
    every still ran partly on the CPU -- ~72 s for one image from a few-step
    turbo model.

A5. Qwen3.5-9B answered content "" with HTTP 200 for four weeks. All three
    rungs of the generation ladder raised ("Pointer argument (at 0) cannot be
    accessed from Triton (cpu tensor?)") and the last one assigned
    generated_text = "" and returned normally. Now raises. The Triton error is
    a quantized model whose weights were CPU-offloaded: such a load succeeds
    and then fails every request, so it is refused at load time and a quantized
    model tries full-GPU whatever the estimate says. That also breaks a loop --
    measured_vram_gb = GPU + offloaded + reserve means one offloaded load writes
    a number bigger than the card, and the next load reads it and offloads again.

B1. A scene can carry its clip's own sound: visual.keep_audio, scene.audio_mix
    (voice|clip|both) and clip_volume. "clip" makes the scene as long as the
    clip and skips the TTS call; "both" ducks the clip under the voice. Folded
    into the narration track, so the music bed ducks under it like a voice.

B2. Two characters talking: scene.speakers, each line in its own voice, each
    character lip-synced in their own column (or the frame cutting to whoever
    speaks). One sync run per CHARACTER, not per line -- each gets a track
    holding only their own lines, so the mouth moves on their words and stays
    closed while the other talks. Captions never mix two speakers and carry
    the attribution into the .srt/.vtt.

B3. characters: ["alice","bob"] on images and video. The gap was never the
    field: IP-Adapter conditions on the MEAN of its references, so a second
    character's pictures produced one person who looked like neither. Reference
    sets are kept apart and handed to a model that takes several (FLUX.2,
    Qwen-Image-Edit); a model that cannot do it says so instead of blending.

B4. POST /v1/moderation -- nsfw, faces, age, face_match over images or sampled
    video frames. It fails CLOSED: a missing model is a 503 naming it, not a
    result with the check quietly absent, because a caller cannot tell a
    missing check from a passed one. minor_detected is nullable (explicit
    content with no face cannot be cleared on age) and keys on the LOW end of
    an age interval, not a point estimate from a model with a multi-year error.

C5. Per-model defaults (steps, guidance, sampler) that fill in only what the
    caller omitted, published in /v1/models alongside modes and licence -- so a
    client stops keeping a table keyed on the model NAME and an 8-step model
    stops being run for 30. A sampler switcher that refuses to cross the
    flow-matching boundary (a FLUX-family pipeline handed a DDIM produces noise,
    not a worse picture). Stable Audio gets its own pipeline: it was mapped to
    AudioLDM2, so a configured entry could only fail at load.

Two of my own regressions, caught by the suite's guard tests and fixed here:
/v1/moderation was unmapped in the remote gateway, and the moderation
availability probe LOADED each model -- which turned the /v1/models listing
into a 350 MB download on the event loop.

Suite: 15 failed, 5146 passed -- the same 15 as the baseline.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01S8LAP3ueNzQJtwzhgVbq66
parent 055270bb
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 4106 additions and 67 deletions
codai/__init__.py
View file @ 3f9b2680
......@@ -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.5.8"
__version__ = "0.5.9"
__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 @ 3f9b2680
......@@ -271,6 +271,7 @@ from codai.api.loras import router as loras_router
from codai.api.spatial import router as spatial_router
from codai.api.environments import router as environments_router
from codai.api.ocr import router as ocr_router
from codai.api.moderation import router as moderation_router
from codai.api.rerank import router as rerank_router
from codai.admin.routes import router as admin_router
......@@ -856,6 +857,7 @@ app.include_router(chunking_router, tags=["Chunking"])
from codai.ocr.gateway import router as ocr_gateway_router # noqa: E402
app.include_router(ocr_gateway_router, include_in_schema=False)
app.include_router(rerank_router, tags=["Rerank"])
app.include_router(moderation_router, tags=["Moderation"])
from codai.api.models_transfer import router as models_transfer_router
app.include_router(models_transfer_router, tags=["Models"])
from codai.api.model_test import router as model_test_router
......
codai/api/audio_gen.py
View file @ 3f9b2680
......@@ -251,13 +251,43 @@ def _load_audioldm(model_name: str, device: str, model_config: dict = None):
def _detect_audio_gen_type(model_name: str) -> str:
n = model_name.lower()
if 'audioldm' in n or 'stable-audio' in n:
# Stable Audio is NOT AudioLDM. It was mapped here to 'audioldm', which
# loads AudioLDM2Pipeline -- a different architecture with a different
# projection model and a different sample rate -- so a configured
# stable-audio-open entry could only ever fail at load. It has its own
# pipeline in diffusers and its own branch below.
if 'stable-audio' in n or 'stable_audio' in n:
return 'stable_audio'
if 'audioldm' in n:
return 'audioldm'
if 'audiogen' in n:
return 'audiogen'
if 'ace-step' in n or 'acestep' in n:
return 'ace_step'
return 'musicgen'
def _load_stable_audio(model_name: str, device: str, model_config: dict = None):
"""Stable Audio Open: text (and a negative prompt) to music or an effect.
44.1 kHz stereo, up to about 47 seconds, which is what makes it the right
model for a loop or a sound effect rather than a whole track.
"""
import torch
from diffusers import StableAudioPipeline
from codai.models.hf_loading import resolve_dtype
dtype = resolve_dtype(model_config, default='f16')
pipe = StableAudioPipeline.from_pretrained(model_name, torch_dtype=dtype)
_off = (model_config or {}).get('offload_strategy')
if _off in ('cpu', 'sequential', 'model', 'disk') and hasattr(
pipe, 'enable_model_cpu_offload'):
pipe.enable_model_cpu_offload()
else:
pipe = pipe.to(device)
return pipe
def _generate_audio(pipe, model_name: str, request: AudioGenerationRequest, task_id=None):
"""Run generation and return (audio_bytes, ext)."""
import numpy as np, io as _io
......@@ -284,6 +314,36 @@ def _generate_audio(pipe, model_name: str, request: AudioGenerationRequest, task
audio_np = wav[0, 0].cpu().numpy()
sr = pipe.sample_rate
elif model_type == 'stable_audio':
num_steps = int(getattr(request, 'steps', None) or 100)
_aud_progress_reset(num_steps, unit="it")
def _sa_step_cb(pipe, step_index, timestep, callback_kwargs):
task_registry.raise_if_cancelled(task_id)
task_registry.wait_if_paused(task_id)
task_registry.step(task_id, step_index + 1)
_aud_progress_step(step_index + 1)
return callback_kwargs
import torch
_gen = None
if getattr(request, 'seed', None) is not None:
_gen = torch.Generator(pipe.device).manual_seed(int(request.seed))
result = pipe(
request.prompt,
negative_prompt=getattr(request, 'negative_prompt', None),
num_inference_steps=num_steps,
audio_end_in_s=float(request.duration or 10.0),
num_waveforms_per_prompt=1,
generator=_gen,
callback_on_step_end=_sa_step_cb,
)
# (channels, samples) float32 at 44.1 kHz. Kept STEREO: this model's
# point is that it produces a usable stereo bed, and folding it to mono
# here would throw that away.
audio_np = result.audios[0].T.float().cpu().numpy()
sr = int(pipe.vae.config.sampling_rate)
elif model_type == 'audioldm':
num_steps = 50
_aud_progress_reset(num_steps, unit="it")
......@@ -307,7 +367,13 @@ def _generate_audio(pipe, model_name: str, request: AudioGenerationRequest, task
# Write to wav
import scipy.io.wavfile as wavfile
buf = _io.BytesIO()
audio_int16 = (audio_np * 32767).astype(np.int16)
# scipy writes (samples,) as mono and (samples, channels) as multichannel.
# Clipped before the cast: a float sample above 1.0 wraps round to a loud
# negative one on int16 overflow, which is heard as a click.
audio_np = np.asarray(audio_np)
if audio_np.ndim > 1 and audio_np.shape[0] < audio_np.shape[-1]:
audio_np = audio_np.T # (channels, samples) -> (samples, channels)
audio_int16 = (np.clip(audio_np, -1.0, 1.0) * 32767).astype(np.int16)
wavfile.write(buf, sr, audio_int16)
return buf.getvalue(), 'wav'
......@@ -369,6 +435,20 @@ async def audio_generate(request: AudioGenerationRequest, http_request: Request
if model_type in ('musicgen', 'audiogen'):
pipe = await asyncio.get_event_loop().run_in_executor(
None, _load_musicgen, model_name, device, _ag_cfg)
elif model_type == 'stable_audio':
pipe = await asyncio.get_event_loop().run_in_executor(
None, _load_stable_audio, model_name, device, _ag_cfg)
elif model_type == 'ace_step':
# ACE-Step ships its own package, which pins a stack this
# venv cannot hold. It belongs in an isolated-venv worker
# like CrisperWhisper and Canary; until that exists, say
# so plainly rather than loading it as a MusicGen and
# failing with a shape error three minutes later.
raise RuntimeError(
f"{model_name} is an ACE-Step model, which needs the "
"`acestep` package in its own venv; coderai has no "
"ACE-Step worker yet. Use a Stable Audio or MusicGen "
"model for now.")
else:
pipe = await asyncio.get_event_loop().run_in_executor(
None, _load_audioldm, model_name, device, _ag_cfg)
......
codai/api/characters.py
View file @ 3f9b2680
......@@ -417,6 +417,73 @@ def resolve_character_profiles(profile_names: List[str]) -> List[str]:
return out
def character_groups(request) -> List[dict]:
"""Every character this request names, each with its OWN references.
Returns ``[{"name": str|None, "images": [b64, ...]}, ...]`` in the order
the caller gave them, reading whichever of the four fields they used:
* ``characters`` — a list of profile NAMES (``["alice", "bob"]``) or of
slots (``[{"name": ..., "images": [...]}, ...]``). Both spellings, because
a caller who has saved profiles should not have to load and resend the
images to name two of them.
* ``character_profiles`` — saved profile names (the long-standing field).
* ``character_references`` + ``character_names`` — inline images, which
cannot be split per character unless names align with them, so they
become one group.
Grouping matters, and it is the whole point of this function. The flat list
that :func:`resolve_character_profiles` returns is what IP-Adapter wants,
and IP-Adapter AVERAGES its references: hand it four pictures of Alice and
four of Bob and it conditions on one person who looks like neither. Two
characters in one frame therefore cannot be done by appending to that list
-- they need a model that takes several references and keeps them distinct
(FLUX.2 takes up to ten; Qwen-Image-Edit-2511 takes several), one
reference per character. This returns the structure that makes that
possible; the caller decides what the pipeline can do with it.
"""
groups: List[dict] = []
seen = set()
def _add(name, images):
imgs = [i for i in (images or []) if i]
if not imgs:
return
key = (name or "").strip().lower() or f"_inline{len(groups)}"
if key in seen:
return
seen.add(key)
groups.append({"name": name, "images": imgs})
for entry in (getattr(request, "characters", None) or []):
if isinstance(entry, str):
_add(entry, resolve_character_profiles([entry]))
elif isinstance(entry, dict):
nm = entry.get("name")
imgs = entry.get("images") or []
if not imgs and nm:
imgs = resolve_character_profiles([nm])
_add(nm, imgs)
for nm in (getattr(request, "character_profiles", None) or []):
_add(nm, resolve_character_profiles([nm]))
inline = list(getattr(request, "character_references", None) or [])
if inline:
nms = list(getattr(request, "character_names", None) or [])
if len(nms) == len(inline) and len(set(nms)) > 1:
# One name per image: they can be split properly.
per: dict = {}
for nm, img in zip(nms, inline):
per.setdefault(nm, []).append(img)
for nm, imgs in per.items():
_add(nm, imgs)
else:
_add(nms[0] if nms else None, inline)
return groups
# ── Endpoints ─────────────────────────────────────────────────────────────────
@router.post("/v1/characters", summary="Create or replace a character profile")
......
codai/api/compose.py
View file @ 3f9b2680
This diff is collapsed.
codai/api/images.py
View file @ 3f9b2680
This diff is collapsed.
codai/api/lang_codes.py 0 → 100644
View file @ 3f9b2680
# 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.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""One language tag, four spellings: normalisation for the speech backends.
A language reaches the server as whatever the caller naturally has. A voice is
chosen for an accent, so a TTS language is a REGIONAL tag -- ``en-us``,
``en-gb``, ``pt-br`` -- and that is the tag a compose request carries from the
voice to the captions. The speech backends then each want something else:
* HF Whisper (``whisper_hf``, ``crisperwhisper``) wants the English NAME:
``english``. Handed ``en-us`` it raises
``Unsupported language: en-us. Language should be one of: ['english', ...]``,
which the worker returns as a 500 -- measured on every scene of every render
that asked for word-timed captions (2026-10-27).
* whisper.cpp (``whisper-server``) and NeMo/Canary want the ISO-639-1 CODE:
``en``. whisper.cpp also accepts the literal ``auto``.
* wav2vec2 is per-language (the code is advisory) and vosk picks the language
by model directory, so for those the code is the sane pass-through.
So the tag is normalised ONCE, here, for the backend that will receive it, and
an unmappable tag becomes ``None`` -- which every backend reads as
auto-detect. Refusing a transcription because the caller named a dialect is
never the better outcome: the words are the same words.
Nothing here validates. :func:`codai.api.transcriptions._validate_language`
decides what a model accepts, and uses :func:`aliases` so a model that declares
``english`` also answers to ``en`` and ``en-gb``.
"""
from typing import Optional
# ISO-639-1 -> the English name Whisper uses. This is Whisper's own language
# table (the 99 it was trained on); the names are what transformers lists in
# the error above, so they are the spelling to send.
WHISPER_NAMES = {
"en": "english", "zh": "chinese", "de": "german", "es": "spanish",
"ru": "russian", "ko": "korean", "fr": "french", "ja": "japanese",
"pt": "portuguese", "tr": "turkish", "pl": "polish", "ca": "catalan",
"nl": "dutch", "ar": "arabic", "sv": "swedish", "it": "italian",
"id": "indonesian", "hi": "hindi", "fi": "finnish", "vi": "vietnamese",
"he": "hebrew", "uk": "ukrainian", "el": "greek", "ms": "malay",
"cs": "czech", "ro": "romanian", "da": "danish", "hu": "hungarian",
"ta": "tamil", "no": "norwegian", "th": "thai", "ur": "urdu",
"hr": "croatian", "bg": "bulgarian", "lt": "lithuanian", "la": "latin",
"mi": "maori", "ml": "malayalam", "cy": "welsh", "sk": "slovak",
"te": "telugu", "fa": "persian", "lv": "latvian", "bn": "bengali",
"sr": "serbian", "az": "azerbaijani", "sl": "slovenian", "kn": "kannada",
"et": "estonian", "mk": "macedonian", "br": "breton", "eu": "basque",
"is": "icelandic", "hy": "armenian", "ne": "nepali", "mn": "mongolian",
"bs": "bosnian", "kk": "kazakh", "sq": "albanian", "sw": "swahili",
"gl": "galician", "mr": "marathi", "pa": "punjabi", "si": "sinhala",
"km": "khmer", "sn": "shona", "yo": "yoruba", "so": "somali",
"af": "afrikaans", "oc": "occitan", "ka": "georgian", "be": "belarusian",
"tg": "tajik", "sd": "sindhi", "gu": "gujarati", "am": "amharic",
"yi": "yiddish", "lo": "lao", "uz": "uzbek", "fo": "faroese",
"ht": "haitian creole", "ps": "pashto", "tk": "turkmen", "nn": "nynorsk",
"mt": "maltese", "sa": "sanskrit", "lb": "luxembourgish", "my": "myanmar",
"bo": "tibetan", "tl": "tagalog", "mg": "malagasy", "as": "assamese",
"tt": "tatar", "haw": "hawaiian", "ln": "lingala", "ha": "hausa",
"ba": "bashkir", "jw": "javanese", "su": "sundanese", "yue": "cantonese",
}
# The reverse direction, plus the spellings people actually type. Whisper's own
# table carries several of these aliases and they are cheap to honour.
_NAME_TO_CODE = {v: k for k, v in WHISPER_NAMES.items()}
_NAME_TO_CODE.update({
"mandarin": "zh", "chinese (simplified)": "zh", "chinese (traditional)": "zh",
"brazilian portuguese": "pt", "castilian": "es", "flemish": "nl",
"burmese": "my", "valencian": "ca", "moldavian": "ro", "moldovan": "ro",
"sinhalese": "si", "panjabi": "pa", "pushto": "ps", "farsi": "fa",
"nepali (individual language)": "ne", "greek (modern)": "el",
"norwegian nynorsk": "nn", "bokmal": "no", "norwegian bokmal": "no",
"letzeburgesch": "lb", "javanese (jw)": "jw", "haitian": "ht",
})
# Three-letter codes worth understanding, because ``eng``/``ita`` turn up in
# corpora and in some TTS catalogues. Only where unambiguous.
_ISO3 = {
"eng": "en", "ita": "it", "deu": "de", "ger": "de", "fra": "fr",
"fre": "fr", "spa": "es", "por": "pt", "rus": "ru", "nld": "nl",
"dut": "nl", "pol": "pl", "ces": "cs", "cze": "cs", "jpn": "ja",
"kor": "ko", "zho": "zh", "chi": "zh", "cmn": "zh", "ara": "ar",
"heb": "he", "hin": "hi", "tur": "tr", "swe": "sv", "dan": "da",
"nor": "no", "fin": "fi", "ell": "el", "gre": "el", "ukr": "uk",
"ron": "ro", "rum": "ro", "hun": "hu", "bul": "bg", "hrv": "hr",
"srp": "sr", "slk": "sk", "slo": "sk", "slv": "sl", "cat": "ca",
"eus": "eu", "baq": "eu", "glg": "gl", "vie": "vi", "tha": "th",
"ind": "id", "msa": "ms", "may": "ms", "fas": "fa", "per": "fa",
"urd": "ur", "ben": "bn", "tam": "ta", "tel": "te", "mar": "mr",
"guj": "gu", "kan": "kn", "mal": "ml", "pan": "pa", "swa": "sw",
"afr": "af", "isl": "is", "ice": "is", "est": "et", "lav": "lv",
"lit": "lt", "kat": "ka", "geo": "ka", "hye": "hy", "arm": "hy",
"aze": "az", "kaz": "kk", "uzb": "uz", "mkd": "mk", "mac": "mk",
"sqi": "sq", "alb": "sq", "bel": "be", "mlt": "mt", "cym": "cy",
"wel": "cy", "lat": "la", "yue": "yue", "haw": "haw",
}
# Families whose ``language`` argument is an English NAME rather than a code.
# Only the transformers pipeline is in here. faster-whisper and whisper.cpp are
# both built on Whisper too but take the CODE, so they are deliberately absent:
# the distinction is the API, not the model.
_NAME_FAMILIES = {"whisper_hf", "crisperwhisper", "whisper-hf",
"whisper-transformers"}
def primary(code) -> str:
"""The primary language subtag, lowercased: ``en-GB`` -> ``en``.
Accepts the separators that turn up in the wild (``-``, ``_``) and the
extensions that follow a tag (``zh-Hans-CN``). Returns ``""`` for nothing.
"""
s = str(code or "").strip().lower().replace("_", "-")
if not s:
return ""
return s.split("-", 1)[0].strip()
def to_code(code) -> Optional[str]:
"""Any spelling of a language -> its ISO-639-1 code, or ``None``.
Understands a regional tag (``pt-br``), a name (``Portuguese``), a
three-letter code (``por``) and a code that is already right (``pt``).
"""
s = str(code or "").strip().lower().replace("_", "-")
if not s or s in ("auto", "automatic", "detect", "none"):
return None
# A full name may itself contain a space or a hyphen ("haitian creole"),
# so names are tried on the whole string BEFORE it is split on "-".
if s in _NAME_TO_CODE:
return _NAME_TO_CODE[s]
head = primary(s)
if head in WHISPER_NAMES:
return head
if head in _ISO3:
return _ISO3[head]
if head in _NAME_TO_CODE:
return _NAME_TO_CODE[head]
return None
def to_name(code) -> Optional[str]:
"""Any spelling -> the English name Whisper expects, or ``None``."""
c = to_code(code)
return WHISPER_NAMES.get(c) if c else None
def normalize(code, family=None) -> Optional[str]:
"""The spelling ``family`` wants, or ``None`` for auto-detect.
``None`` is a deliberate answer, not a failure: every backend treats a
missing language as "detect it", which is a far better outcome for the
caller than a 500 because they named a dialect this model has never heard
of. The caller should omit the argument entirely when this returns
``None`` -- passing ``None`` through is equivalent for all four backends.
"""
fam = str(family or "").strip().lower()
if fam in _NAME_FAMILIES:
return to_name(code)
return to_code(code)
def aliases(code) -> set:
"""Every spelling that means this language, for membership tests.
``aliases("en-gb")`` -> ``{"en-gb", "en", "english"}``. Used by language
validation so a model declaring ``["english"]`` is not made to refuse
``en`` -- and vice versa. The original string is always included, so an
unmappable tag still matches a config that declares exactly it.
"""
s = str(code or "").strip().lower().replace("_", "-")
out = {s} if s else set()
c = to_code(code)
if c:
out.add(c)
name = WHISPER_NAMES.get(c)
if name:
out.add(name)
return out
codai/api/moderation.py 0 → 100644
View file @ 3f9b2680
This diff is collapsed.
codai/api/remote_gateway.py
View file @ 3f9b2680
......@@ -72,6 +72,14 @@ _CAPABILITY_PREFIXES = (
("/v1/video", "video"),
("/v1/3d", "spatial"),
("/v1/embeddings", "embeddings"),
# Moderation runs where the models are. It rides the IMAGES capability
# rather than inventing one of its own: its models are an image
# classifier and a face detector, the images pod already carries that
# kind of weight, and a `moderation` capability would be one nothing
# advertises -- so every request would fall back to local anyway. Keeping
# it with images also means a pod that generated the picture can screen it
# without the bytes crossing the wire twice.
("/v1/moderation", "images"),
# Retrieval chunking rides the EMBEDDINGS capability rather than inventing
# one of its own. Two reasons: chunks are on their way to be embedded, so
# the machine that will embed them is the natural place to cut them; and the
......
codai/api/stt_backends.py
View file @ 3f9b2680
......@@ -44,6 +44,8 @@ import tempfile
import wave
from typing import Optional
from codai.api.lang_codes import normalize as _norm_lang
# --------------------------------------------------------------------------- #
# family detection
......@@ -172,12 +174,19 @@ class _Wav2Vec2Backend:
gen_kwargs = {}
if self._is_whisper:
# Whisper seq2seq: choose transcribe vs translate, pass source language.
# Whisper's `language` is an English NAME, so a regional tag from a
# TTS voice ("en-us") has to be mapped first -- handed through as is
# it raises "Unsupported language: en-us" and the whole request 500s.
# An unmappable tag becomes None, i.e. auto-detect.
gen_kwargs["task"] = "translate" if target_language else "transcribe"
if language:
gen_kwargs["language"] = language
lang = _norm_lang(language, "whisper_hf")
if lang:
gen_kwargs["language"] = lang
elif language and self.config.get("language_aware"):
# MMS / Seamless accept a target language; plain wav2vec2 ignores it.
gen_kwargs["language"] = language
lang = _norm_lang(language, "wav2vec2")
if lang:
gen_kwargs["language"] = lang
# return_timestamps: CTC uses "chunk" / "word". Whisper uses "word" or
# True (segment) — but some checkpoints (e.g. CrisperWhisper) ship a custom
# generation config whose timestamp postprocessing breaks on transformers
......@@ -303,7 +312,11 @@ class _VoskBackend:
wf.close()
return {"text": " ".join(t for t in texts if t).strip(),
"segments": segments, "words": words,
"language": language or self.config.get("language")}
# Vosk chooses the language by model directory and ignores
# the argument; report the normalised code so the response
# doesn't echo a dialect tag Vosk never honoured.
"language": (_norm_lang(language, "vosk")
or self.config.get("language"))}
finally:
try:
os.unlink(wav_path)
......@@ -350,10 +363,16 @@ class _RemoteNemoBackend:
word_timestamps: bool = False) -> dict:
import requests
params = {}
if language:
params["language"] = language
# Canary names its languages by ISO-639-1 code, so a regional tag from a
# TTS voice ("pt-br") is reduced to "pt" first. Unmappable -> omitted,
# which the worker reads as auto-detect.
lang = _norm_lang(language, "nemo")
if lang:
params["language"] = lang
if target_language:
params["target_language"] = target_language
tgt = _norm_lang(target_language, "nemo")
if tgt:
params["target_language"] = tgt
if word_timestamps:
params["word_timestamps"] = "1"
with open(audio_path, "rb") as f:
......@@ -411,8 +430,12 @@ class _RemoteCrisperWhisperBackend:
word_timestamps: bool = False) -> dict:
import requests
params = {"task": "translate" if target_language else "transcribe"}
if language:
params["language"] = language
# The worker hands `language` to transformers' Whisper generate(), which
# accepts only the English NAME: "en-us" raised "Unsupported language"
# and surfaced as a 500 on every captioned scene (2026-10-27).
lang = _norm_lang(language, "crisperwhisper")
if lang:
params["language"] = lang
with open(audio_path, "rb") as f:
audio = f.read()
......
codai/api/transcriptions.py
View file @ 3f9b2680
......@@ -53,6 +53,7 @@ def _upload_name(file) -> str:
return getattr(file, "filename", None) or ""
# Import from codai modules
from codai.api import lang_codes
from codai.models.manager import multi_model_manager
......@@ -170,8 +171,16 @@ def _validate_language(model, config: dict, language, target_language):
no languages accept anything (pass-through). ``target_language`` additionally
requires the model to advertise ``supports_translation``."""
langs = config.get("languages") or []
norm = {str(x).strip().lower() for x in langs}
if language and norm and str(language).strip().lower() not in norm:
# Compared by MEANING, not by spelling. A config declares its languages in
# whatever form its backend uses -- "english" for a whisper model, "en" for
# Canary -- while the caller sends what they have, which for a compose
# render is the voice's regional tag ("en-gb"). Matching those literally
# refused every request that named a dialect, so both sides are expanded to
# their alias sets and the test is whether the sets intersect.
norm = set()
for x in langs:
norm |= lang_codes.aliases(x)
if language and norm and not (lang_codes.aliases(language) & norm):
raise HTTPException(
status_code=400,
detail=f"Language '{language}' not supported by '{model}'. "
......@@ -182,7 +191,7 @@ def _validate_language(model, config: dict, language, target_language):
status_code=400,
detail=f"Model '{model}' does not support translation "
"(target_language). Use a NeMo/Canary model that does.")
if norm and str(target_language).strip().lower() not in norm:
if norm and not (lang_codes.aliases(target_language) & norm):
raise HTTPException(
status_code=400,
detail=f"Target language '{target_language}' not supported by "
......@@ -534,9 +543,11 @@ async def _run_transcription_inner(
multi_model_manager.start_whisper_server, whisper_model_id or model)
if not whisper_server.is_running():
raise HTTPException(status_code=500, detail="whisper-server failed to start")
# whisper.cpp takes an ISO-639-1 code (or "auto"); a regional tag is
# rejected outright, so normalise and let an unmappable tag auto-detect.
result = whisper_server.transcribe(
file_content,
language=language,
language=lang_codes.normalize(language, "whisper-server"),
prompt=prompt
)
if "error" in result:
......@@ -664,7 +675,7 @@ async def _run_transcription_inner(
raw_segments, _ = whisper_model.transcribe(
tmp_path,
language=language,
language=lang_codes.normalize(language, "faster_whisper"),
initial_prompt=prompt,
temperature=temperature,
word_timestamps=word_timestamps,
......
codai/api/video.py
View file @ 3f9b2680
......@@ -1954,6 +1954,10 @@ def _load_rife(model_name: str, device: str):
# =============================================================================
def _build_call_kwargs(request: VideoGenerationRequest) -> dict:
# NOTE: this populates steps/guidance ONLY when the REQUEST set them. The
# caller's accel and model-card defaults are layered on afterwards with
# setdefault, and both of those depend on an unset key meaning "the caller
# did not ask".
kw = {}
if request.prompt:
kw['prompt'] = request.prompt
......@@ -2032,10 +2036,29 @@ def _resolve_character_inputs(request) -> tuple[List[str], List[str]]:
except Exception:
pass
# 2. Named character slots [{name, images:[...]}, ...]
# 2. Named character slots [{name, images:[...]}, ...] -- or bare profile
# names (["alice", "bob"]), which are resolved from the saved profiles.
# A caller who has saved a character should not have to load its
# reference images and send them back to name it.
if request.characters:
for slot in request.characters:
if isinstance(slot, str):
try:
from codai.api.characters import resolve_character_profiles
images += resolve_character_profiles([slot])
names.append(slot)
except Exception:
pass
continue
if not isinstance(slot, dict):
continue
slot_imgs = slot.get('images') or []
if not slot_imgs and slot.get('name'):
try:
from codai.api.characters import resolve_character_profiles
slot_imgs = resolve_character_profiles([slot['name']])
except Exception:
slot_imgs = []
images += slot_imgs
if slot.get('name'):
names.append(slot['name'])
......@@ -2611,6 +2634,34 @@ def _generate_video(pipe, request: VideoGenerationRequest):
print(" [video][accel] distill LoRA not fused — ignoring the preset's "
"low step count and using safe defaults (25 steps) to avoid a "
"collapsed/solid-colour result")
# The model card's own recommendation, below a fused distill preset and
# above the generic floor. Without it a client needs a table of per-model
# step counts keyed on the model NAME -- which breaks the moment a second
# quantization of the same model is added -- and a client without one runs
# an 8-step model for 25 steps: three times the cost for a worse clip.
try:
from codai.api.images import model_defaults
from codai.models.manager import multi_model_manager
_mcfg = multi_model_manager._config_for_model(
getattr(request, 'model', None) or '') or {}
# The manager keeps runtime kwargs flat and the original models.json
# entry under `_raw_cfg`; `defaults` lives in the entry.
if isinstance(_mcfg.get('_raw_cfg'), dict):
_mcfg = {**_mcfg['_raw_cfg'], **{k: v for k, v in _mcfg.items()
if k != '_raw_cfg'}}
_card = model_defaults(_mcfg)
except Exception:
_card = {}
_c_steps = _card.get('steps') or _card.get('num_inference_steps')
if _c_steps:
kw.setdefault('num_inference_steps', int(_c_steps))
_c_cfg = _card.get('guidance_scale')
if _c_cfg is None:
_c_cfg = _card.get('cfg_scale')
if _c_cfg is not None:
kw.setdefault('guidance_scale', float(_c_cfg))
if _card.get('num_frames'):
kw.setdefault('num_frames', int(_card['num_frames']))
kw.setdefault('num_inference_steps', 25)
kw.setdefault('guidance_scale', 7.5)
kw.setdefault('num_frames', 16)
......@@ -4132,8 +4183,27 @@ async def _generate_h3(request: VideoGenerationRequest, model_name: str,
print(f" [h3] character profile resolution failed: {e}")
for img in (request.character_references or []):
refs.append({"type": "image", "data": _as_b64(img)})
# `characters` holds either slots or bare profile names; a string has no
# .get(), so both shapes are handled here rather than crashing on one.
for entry in (request.characters or []):
for img in (entry.get("images") or []):
if isinstance(entry, str):
try:
from codai.api.characters import resolve_character_profiles
for img in resolve_character_profiles([entry]):
refs.append({"type": "image", "data": _as_b64(img)})
except Exception as e:
print(f" [h3] character {entry!r} could not be resolved: {e}")
continue
if not isinstance(entry, dict):
continue
_imgs = entry.get("images") or []
if not _imgs and entry.get("name"):
try:
from codai.api.characters import resolve_character_profiles
_imgs = resolve_character_profiles([entry["name"]])
except Exception:
_imgs = []
for img in _imgs:
refs.append({"type": "image", "data": _as_b64(img)})
if refs:
# The checkpoint takes at most 9 image references; keep the first ones,
......@@ -4561,6 +4631,14 @@ async def _maybe_build_identity_keyframe(request: VideoGenerationRequest,
payload["character_profiles"] = list(request.character_profiles)
if request.character_references:
payload["character_references"] = list(request.character_references)
# Forward `characters` as well, or a clip asking for two of the caller's
# characters would render its identity keyframe from neither. It also means
# the keyframe is built by the image path's multi-reference routing, so two
# characters arrive as two people in the frame the video is then driven
# from -- which is the only point in this pipeline where that can be
# decided.
if getattr(request, 'characters', None):
payload["characters"] = list(request.characters)
if getattr(request, 'environment_profiles', None):
payload["environment_profiles"] = list(request.environment_profiles)
if getattr(request, 'keyframe_model', None):
......
codai/backends/cuda.py
View file @ 3f9b2680
This diff is collapsed.
codai/compose/captions.py
View file @ 3f9b2680
......@@ -188,6 +188,25 @@ def break_lines(words: List[dict], max_words_per_line: int = 4,
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
# A caption must never put two characters' words in one event: in a
# dialogue that would render a question and its answer as a single line
# attributed to nobody. So the word list is first split wherever the
# speaker changes, and events are grouped inside each run.
runs: List[List[dict]] = []
for w in words:
spk = w.get("speaker")
if runs and runs[-1][0].get("speaker") == spk:
runs[-1].append(w)
else:
runs.append([w])
out = []
for run in runs:
out += _events_from_run(run, per_line, per_event)
return out
def _events_from_run(words: List[dict], per_line: int,
per_event: int) -> List[dict]:
out = []
for i in range(0, len(words), per_event):
chunk = words[i:i + per_event]
......@@ -206,6 +225,10 @@ def break_lines(words: List[dict], max_words_per_line: int = 4,
mv = max((int(w.get("margin_v") or 0) for w in chunk), default=0)
if mv:
ev["margin_v"] = mv
# Who said it, carried through so a style can distinguish speakers and
# a caller reading the .srt/.vtt sidecar knows the attribution.
if chunk[0].get("speaker"):
ev["speaker"] = chunk[0]["speaker"]
out.append(ev)
return out
......
codai/compose/presenter.py
View file @ 3f9b2680
......@@ -325,7 +325,7 @@ def _data_uri(path: str) -> str:
# --------------------------------------------------------------- the scene
def build_scene_presenter(spec: dict, *, narration: Optional[str], duration: float,
width: int, height: int, fps: int, workdir: str,
index: int = 0, loop=None, cancel=None, log=None,
index=0, loop=None, cancel=None, log=None, # int, or any str tag (a dialogue needs scene+column)
gpu_lease=None) -> dict:
"""The presenter clip for one scene.
......
codai/compose/render.py
View file @ 3f9b2680
......@@ -490,6 +490,187 @@ def build_music_bed(src: str, out_path: str, *, total: float, volume: float = 0.
return out_path
def place_at(placements: List[tuple], out_path: str, *, duration: float,
cancel=None, log=None) -> str:
"""One track holding each clip at its own offset, silence in between.
``placements`` is ``[(path, start_seconds), ...]``. Built for one speaker in
a dialogue: their lines where they fall in the scene, nothing where
somebody else is talking.
That shape is what lets a multi-speaker scene cost one lip-sync run per
SPEAKER rather than one per line. A sync engine drives the mouth from the
audio it is given, so a track that is silent while the other character
speaks produces a closed mouth there -- which is what a person listening
looks like, and what a per-line approach has to stitch together by hand.
"""
if not placements:
return silence(out_path, duration, cancel=cancel, log=log)
inputs: List[str] = []
graph: List[str] = []
labels: List[str] = []
for i, (path, start) in enumerate(placements):
inputs += ["-i", path]
ms = max(0, int(round(float(start) * 1000)))
graph.append(f"[{i}:a]aresample=48000,adelay={ms}|{ms}[d{i}]")
labels.append(f"[d{i}]")
if len(labels) == 1:
graph.append(f"{labels[0]}apad,atrim=0:{max(0.02, float(duration)):.3f},"
f"asetpts=N/SR/TB[aout]")
else:
graph.append("".join(labels)
+ f"amix=inputs={len(labels)}:duration=longest:"
f"dropout_transition=0:normalize=0,"
f"apad,atrim=0:{max(0.02, float(duration)):.3f},"
f"asetpts=N/SR/TB[aout]")
run(inputs + ["-filter_complex", ";".join(graph), "-map", "[aout]",
"-ac", "2", "-ar", "48000", "-c:a", "pcm_s16le", out_path],
cancel=cancel, log=log)
return out_path
def cut_segment(src: str, out_path: str, *, start: float, duration: float,
fps: int, width: Optional[int] = None,
height: Optional[int] = None, cancel=None, log=None) -> str:
"""``duration`` seconds of ``src`` from ``start``, at the canvas size.
Re-encoded rather than stream-copied: the cut has to land on the exact
frame, and a copy would snap to the nearest keyframe and drift the dialogue
against its own audio.
"""
chain = [f"fps={fps}"]
if width and height:
chain.append(fit_filter(int(width), int(height), "cover"))
chain += [f"tpad=stop_mode=clone:stop_duration={max(0.05, duration):.3f}",
f"trim=0:{max(0.08, duration):.3f}", "setpts=PTS-STARTPTS",
"format=yuv420p"]
run(["-ss", f"{max(0.0, float(start)):.3f}", "-i", src,
"-vf", ",".join(chain), "-an", "-r", str(fps)]
+ _INTER_ARGS + [out_path], cancel=cancel, log=log)
return out_path
def composite_row(clips: List[str], out_path: str, *, width: int, height: int,
fps: int, cancel=None, log=None) -> str:
"""Several speakers side by side, each filling its own column.
Each column is cropped to fill rather than letterboxed, because these are
portraits: a face with black bars beside it reads as a mistake, and a face
cropped at the shoulders reads as framing.
"""
n = max(1, len(clips))
if n == 1:
return hold_to(clips[0], out_path, duration=0.0, fps=fps,
width=width, height=height, cancel=cancel, log=log)
col = int(width / n / 2) * 2
fill = (f"scale={col}:{height}:force_original_aspect_ratio=increase,"
f"crop={col}:{height},setsar=1")
inputs: List[str] = []
graph: List[str] = []
labels: List[str] = []
for i, c in enumerate(clips):
inputs += ["-i", c]
graph.append(f"[{i}:v]{fill}[c{i}]")
labels.append(f"[c{i}]")
graph.append("".join(labels) + f"hstack=inputs={n},"
f"pad={width}:{height}:0:0,format=yuv420p,fps={fps}[vout]")
run(inputs + ["-filter_complex", ";".join(graph), "-map", "[vout]", "-an"]
+ _INTER_ARGS + [out_path], cancel=cancel, log=log)
return out_path
def clip_audio(src: str, out_path: str, *, duration: float, trim_start: float = 0.0,
volume: float = 1.0, loop: bool = False, cancel=None,
log=None) -> str:
"""One visual's OWN sound, cut to its slot in the scene.
The video track of a visual is already trimmed, looped and padded to the
scene's length by :func:`build_visual`; this does the same to its audio so
the two stay in step. ``loop`` must match what the video did, or the sound
drifts out of sync with the picture it belongs to.
Always 48 kHz stereo PCM, like every other audio part, so the scene parts
can be concatenated without a resample surprise in the middle.
"""
pre = ["-stream_loop", "-1"] if loop else []
if trim_start > 0:
pre += ["-ss", f"{float(trim_start):.3f}"]
af = ["aresample=48000", "apad",
f"atrim=0:{max(0.02, float(duration)):.3f}", "asetpts=N/SR/TB"]
v = max(0.0, float(volume))
if abs(v - 1.0) > 1e-6:
af.append(f"volume={v:.4f}")
run(pre + ["-i", src, "-vn", "-af", ",".join(af),
"-ac", "2", "-ar", "48000", "-c:a", "pcm_s16le", out_path],
cancel=cancel, log=log)
return out_path
def has_audio_stream(src: str) -> bool:
"""Whether a file carries an audio stream at all.
Asked before extracting, because ``keep_audio`` on silent footage would
otherwise fail the render over something the caller cannot see from the
file name. A probe failure answers "no": the extraction would fail anyway.
"""
try:
out = subprocess.run(
[media._ffprobe_bin(), "-v", "error", "-select_streams", "a",
"-show_entries", "stream=index", "-of", "csv=p=0", src],
capture_output=True, text=True, timeout=60)
except Exception:
return False
return bool((out.stdout or "").strip())
def mix_voice_clip(voice: Optional[str], clip: Optional[str], out_path: str, *,
duration: float, duck: bool = True, cancel=None,
log=None) -> str:
"""One scene's narration and its clip's sound, as a single track.
Mixed HERE, per scene, rather than as a third input to the final mix, for
two reasons: the clip's sound belongs to one scene and must not bleed into
the next, and folding it into the narration bed means the music ducking
downstream keys on it too -- which is what "treat clip sound like
narration" asks for, with no second sidechain.
With ``duck`` the clip is pushed down while the voice is speaking and comes
back up between lines, so narration stays intelligible over a noisy shot.
Without it the two are simply summed at their given levels.
"""
if not clip:
if voice:
return pad_to(voice, out_path, duration, cancel=cancel, log=log)
return silence(out_path, duration, cancel=cancel, log=log)
if not voice:
run(["-i", clip, "-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
# The scene's exact length is held INSIDE the graph. It cannot be an `-af`
# on the output: ffmpeg refuses simple and complex filtering on the same
# stream ("Filtergraph ... was specified for a stream fed from a complex
# filtergraph"), and the mix would never be written at all.
fit = (f"apad,atrim=0:{max(0.02, float(duration)):.3f},asetpts=N/SR/TB")
if duck:
graph = ("[1:a]asplit=2[c1][c2];"
"[0:a]asplit=2[v1][v2];"
# v2 is only the sidechain key; c1 is what we hear.
"[c1][v2]sidechaincompress=threshold=0.04:ratio=6:attack=20:"
"release=350:makeup=1[cduck];"
"[v1][cduck]amix=inputs=2:duration=longest:dropout_transition=0,"
f"alimiter=limit=0.97,{fit}[aout];"
"[c2]anullsink")
else:
graph = ("[0:a][1:a]amix=inputs=2:duration=longest:dropout_transition=0,"
f"alimiter=limit=0.97,{fit}[aout]")
run(["-i", voice, "-i", clip, "-filter_complex", graph, "-map", "[aout]",
"-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."""
......
codai/models/capabilities.py
View file @ 3f9b2680
......@@ -54,6 +54,15 @@ class ModelCapabilities:
object_detection: bool = False # YOLO, DETR
style_transfer: bool = False # Neural style transfer
# Content moderation (served at /v1/moderation). Split by what each model
# actually decides, because a caller's obligations differ: an explicitness
# score is a policy question, an age estimate is not.
moderation: bool = False # any of the checks below
nsfw_detection: bool = False # explicitness classifier
age_estimation: bool = False # age per detected face
face_detection: bool = False # where the faces are
face_embedding: bool = False # identity vector, for performer match
# Video generation & editing
video_generation: bool = False # Text-to-video (CogVideoX, LTX, …)
image_to_video: bool = False # Image-to-video (SVD, I2VGen, …)
......@@ -227,8 +236,19 @@ def detect_model_capabilities(model_name: str) -> ModelCapabilities:
# ── Image: instruction editing ───────────────────────────────────────────
# Before the inpaint/generation branches: these ids carry "edit", and
# falling through to img2img would advertise a `strength` they do not have.
# FLUX.2 (dev / klein / pro) is a UNIFIED model: one checkpoint that makes a
# picture from text and edits one from an instruction plus up to ten
# reference images. Its name carries neither "edit" nor "kontext", so it
# fell through to the generation branch below and was advertised as plain
# img2img -- which told clients to show a `strength` slider for a pipeline
# that has no such argument. Confirmed against diffusers 0.37:
# Flux2Pipeline and Flux2KleinPipeline both take `image` and neither takes
# `strength`, which is exactly what images._is_instruction_edit decides on,
# so the runtime already treated it as an instruction edit and only the
# capability list disagreed.
if (any(x in n for x in ['qwen-image-edit', 'qwen_image_edit',
'flux.1-kontext', 'flux-kontext', 'flux_kontext',
'flux.2', 'flux-2', 'flux2',
'image-edit-rapid', 'firered-image-edit'])
or re.search(r'\bstep1x-edit', n)):
caps.instruction_edit = True
......
codai/models/manager.py
View file @ 3f9b2680
......@@ -3945,6 +3945,8 @@ class MultiModelManager:
'I16': 2, 'I8': 1, 'U8': 1, 'BOOL': 1,
}
_storage_dtype_cache: Dict[str, float] = {}
# model key -> whether its weights are ALREADY quantized on disk.
_prequantized_cache: Dict[str, bool] = {}
@classmethod
def _safetensors_numel_bytes(cls, path: str):
......@@ -4128,6 +4130,121 @@ class MultiModelManager:
self._component_shares_cache[ck] = shares
return shares
# Repo-name markers for a checkpoint published pre-quantized. Only used
# when config.json can't be read: the name is a hint, the file is proof.
_PREQUANT_NAME_MARKERS = (
"bnb-4bit", "bnb_4bit", "bnb-8bit", "bnb_8bit",
"-4bit", "_4bit", "-8bit", "_8bit",
"-awq", "_awq", "-gptq", "_gptq", "-nf4", "_nf4",
"-int4", "_int4", "-int8", "_int8", "-fp8", "_fp8",
)
def _is_prequantized_checkpoint(self, model_key: str, resolved_name: str,
cfg: dict) -> bool:
"""Whether this checkpoint's weights are ALREADY quantized on disk.
The same situation as a GGUF, reached by a different route: a repo
published pre-quantized (``unsloth/...-bnb-4bit``, an AWQ or GPTQ
build, an fp8 release) stores a ``quantization_config`` in its
``config.json``, and transformers/diffusers then load those weights AS
THEY ARE. Nothing dequantizes them.
That matters because the estimate normalises storage precision to load
precision, and ``_storage_bytes_per_elem`` correctly measures such a
checkpoint at ~1 byte/elem. Dividing the default 2-byte load precision
by that gives a factor of ~1.7 -- i.e. the estimate assumed the model
would expand to bf16 on the way in.
Measured on zeiss 2026-10-27: ``Z-Image-Turbo-unsloth-bnb-4bit``, a
16.3 GB 4-bit checkpoint, estimated **30.03 GB** against a 24 GB card.
Nothing could ever free that much, so every still was generated with
the pipeline partly on the CPU -- ~72 s for one 768x1344 image from a
few-step turbo model that should take seconds. The operator had no
``load_in_4bit`` set, and correctly so: there is nothing to ask for
when the weights arrive quantized.
Decided from ``config.json`` where it can be read (that is the field
transformers itself dispatches on), falling back to the repo name.
Cached per model; never raises.
"""
ck = resolved_name or model_key or ""
cached = self._prequantized_cache.get(ck)
if cached is not None:
return cached
import json as _json
import os
result = False
candidates = [v for v in (resolved_name, cfg.get("path"),
cfg.get("model_path"), cfg.get("model"),
model_key)
if v and isinstance(v, str)]
# 1) config.json -- the authoritative answer. A diffusers pipeline keeps
# the quantized component one level down (transformer/, text_encoder/),
# so those are checked too.
conf_paths = []
for c in candidates:
base = c.split(":", 1)[1] if ":" in c and not os.path.exists(c) else c
if os.path.isdir(base):
conf_paths.append(os.path.join(base, "config.json"))
conf_paths.append(os.path.join(base, "model_index.json"))
for sub in ("transformer", "text_encoder", "unet"):
conf_paths.append(os.path.join(base, sub, "config.json"))
if not conf_paths:
# HuggingFace hub cache: find the newest snapshot directory.
try:
from huggingface_hub import scan_cache_dir
from codai.models.cache import (get_all_cache_dirs,
is_huggingface_model_id)
hf_dir = get_all_cache_dirs().get("huggingface")
for c in candidates:
repo_id = c.split(":", 1)[1] if ":" in c else c
if not (hf_dir and is_huggingface_model_id(repo_id)):
continue
info = scan_cache_dir(hf_dir)
for repo in info.repos:
if repo.repo_id != repo_id:
continue
revs = sorted(repo.revisions,
key=lambda r: r.last_modified, reverse=True)
for fobj in (revs[0].files if revs else []):
fp = str(fobj.file_path)
if os.path.basename(fp) == "config.json":
conf_paths.append(fp)
if conf_paths:
break
except Exception:
pass
for cp in conf_paths[:24]:
try:
with open(cp) as fh:
blob = _json.load(fh)
except Exception:
continue
if not isinstance(blob, dict):
continue
# Top level, or nested under a component entry.
if blob.get("quantization_config"):
result = True
break
for v in blob.values():
if isinstance(v, dict) and v.get("quantization_config"):
result = True
break
if result:
break
# 2) The repo name, when no config could be read.
if not result and not conf_paths:
blob = " ".join(candidates).lower()
result = any(m in blob for m in self._PREQUANT_NAME_MARKERS)
self._prequantized_cache[ck] = result
return result
@staticmethod
def _load_bytes_per_elem(cfg: dict) -> float:
"""Bytes/elem for the configured LOAD precision (default fp16 = 2)."""
......@@ -4477,6 +4594,22 @@ class MultiModelManager:
_gguf_path = str(cfg.get("path") or resolved_name or model_key or "")
_is_gguf = (_gguf_path.endswith(".gguf") or "gguf" in _gguf_path.lower()
or cfg.get("model_type") == "gguf_models")
# An HF checkpoint published pre-quantized is the same case as a GGUF:
# the bytes on disk ARE the loaded footprint, so neither the fp16-
# relative quant divisor nor the storage->load precision normalization
# applies. Without this a 4-bit release was inflated by ~1.7x and the
# planner concluded it could not fit a card it fits comfortably.
_is_prequant = False
if not _is_gguf:
try:
_is_prequant = self._is_prequantized_checkpoint(
model_key, resolved_name, cfg)
except Exception:
_is_prequant = False
if _is_prequant:
quant_mult = 1.0
prec_factor = 1.0
if _is_gguf:
quant_mult = 1.0
prec_factor = 1.0
......@@ -4502,6 +4635,7 @@ class MultiModelManager:
f"measured_cfg={cfg.get('measured_vram_gb')} "
f"component_quantization={cfg.get('component_quantization')} "
f"load_in_4bit={load_in_4bit} load_in_8bit={load_in_8bit} "
f"prequantized={_is_prequant} "
f"quant_mult={quant_mult:.3f} "
f"prec(load{_load_bpe:.0f}/disk{_storage_bpe:.0f})={prec_factor:.2f}")
except Exception:
......@@ -6036,6 +6170,7 @@ class MultiModelManager:
"audio_gen": "audio_generation",
"embedding": "embeddings",
"spatial": "depth_estimation",
"moderation": "moderation",
}
def _add(model_id: str, model_type: str = None, meta: Dict[str, Any] = None):
......@@ -6103,6 +6238,15 @@ class MultiModelManager:
supports_translation=meta.get("supports_translation"),
video=video_geom,
max_model_len=context_window_for(self, model_id, meta),
defaults=(meta.get("defaults") or None),
# `modes` may live in `defaults` (where a card's settings
# naturally go) or at the top level; both are read, because an
# operator should not have to remember which.
modes=((meta.get("defaults") or {}).get("modes")
or meta.get("modes") or None),
license=((meta.get("defaults") or {}).get("license")
or meta.get("license") or meta.get("licence") or None),
available=meta.get("available"),
))
# --- Models from config (the authoritative source) ---
......@@ -6185,6 +6329,24 @@ class MultiModelManager:
except Exception:
pass
# --- Moderation (served at /v1/moderation) — like OCR and diarization,
# built-in endpoints rather than model-manager models, so their models
# are surfaced here. Published even when a model is NOT loadable, with
# `available` saying so: a caller screening adult renders needs to know
# a check is unavailable BEFORE sending content for it, and an absent
# entry reads as "this server does no moderation" rather than "this
# check would fail". ---
try:
from codai.api.moderation import available_models as _mod_models
for _m in _mod_models():
_add(_m["id"], "moderation",
{"capabilities": _m["capabilities"],
"backend": "moderation",
"role": _m["role"],
"available": _m["available"]})
except Exception:
pass
# --- Speaker diarization (pyannote) — a built-in endpoint, not a
# model-manager model, so surface its configured model here so /v1/models
# advertises it (served at /v1/audio/diarization). ---
......
codai/pydantic/imagerequest.py
View file @ 3f9b2680
......@@ -16,7 +16,7 @@
"""Pydantic models for image generation API."""
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, ConfigDict, Field
......@@ -77,10 +77,22 @@ class ImageGenerationRequest(BaseModel):
"in as many windows as it needs, so nothing is truncated at 77 tokens."))
# Per-request component overrides
sampler: Optional[str] = Field(None, description=(
"Scheduler for this request (euler, euler_a, dpmpp_2m, ddim, unipc, "
"lcm, flow_euler, ...). Defaults to the model's own recommendation "
"where its card carries one, else the pipeline's. A sampler from the "
"wrong family for the model is a 400, not a silently worse picture."))
vae_model: Optional[str] = Field(None, description="Override the VAE for this request.")
loras: Optional[List[LoraConfig]] = Field(None, description="Additional LoRA adapters to apply for this request.")
# Character consistency
characters: Optional[List[Any]] = Field(None, description=(
'The characters in the picture: profile names (["alice", "bob"]) or '
'slots ([{"name": "alice", "images": ["b64...", ...]}]). Several are '
'kept as SEPARATE reference sets, so two named characters appear as two '
'people on a model that takes multiple references (FLUX.2, '
'Qwen-Image-Edit); IP-Adapter averages its references and would blend '
'them into one.'))
character_profiles: Optional[List[str]] = Field(None, description="Saved character profile names to apply (IP-Adapter).")
character_references: Optional[List[str]] = Field(None, description="Inline base64 reference images for character consistency.")
character_strength: Optional[float] = Field(0.6, description="IP-Adapter scale for character references.")
......
codai/pydantic/textrequest.py
View file @ 3f9b2680
......@@ -183,6 +183,28 @@ class ModelInfo(BaseModel):
# be, how many scenes fill a minute) instead of hardcoding a table per family.
# See codai/models/video_geometry.py.
video: Optional[Dict] = None
# What this model recommends for itself, from its card: steps, guidance
# scale, sampler. Published because a client cannot otherwise know: a
# distilled model wants 8 steps at CFG 1 and its non-distilled sibling
# wants 30 at CFG 4, and a client with no table of its own gets 30 from
# the 8-step model -- four times the cost for a worse picture. The server
# applies these when a request omits them; this field is so a UI can SHOW
# them.
defaults: Optional[Dict] = None
# Which directions this model generates in: "t2v", "i2v", "v2v" for video,
# "t2i", "i2i" for images. A caller otherwise discovers that a model has no
# image-to-video path by sending one and getting a 400.
modes: Optional[List[str]] = None
# The licence the weights are under, as the operator recorded it. Some
# models are free only below a revenue threshold or require a content
# filter, and that is a fact about the model which belongs beside it
# rather than in a document somewhere.
license: Optional[str] = None
# Whether this model is usable RIGHT NOW. Used by the built-in endpoints
# whose models are not model-manager models (moderation): an unavailable
# check has to be visible before content is sent for it, and an omitted
# entry would read as "this server does not do that at all".
available: Optional[bool] = None
class ModelList(BaseModel):
......
codai/pydantic/videorequest.py
View file @ 3f9b2680
......@@ -16,7 +16,7 @@
"""Pydantic models for video generation API."""
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, ConfigDict, Field
......@@ -126,7 +126,11 @@ class VideoGenerationRequest(BaseModel):
camera_motion: Optional[str] = None # zoom-in | zoom-out | pan-left | pan-right | tilt-up | tilt-down | rotate
# ── Character consistency ─────────────────────────────────────────────
characters: Optional[List[dict]] = Field(None, description='Per-character references, each {"name": "Alice", "images": ["b64...", ...]}.')
characters: Optional[List[Any]] = Field(None, description=(
'The characters in the clip: profile names (["alice", "bob"]) or slots '
'({"name": "Alice", "images": ["b64...", ...]}). A bare name is '
'resolved from the saved profile, so naming two characters does not '
'mean loading and resending their reference images.'))
character_references: Optional[List[str]] = Field(None, description="Legacy flat list of base64/URL reference images.")
character_strength: Optional[float] = Field(0.8, description="IP-Adapter scale for character references.")
character_names: Optional[List[str]] = Field(None, description="Optional names aligned with character_references.")
......
docs/README.md
View file @ 3f9b2680
......@@ -27,6 +27,7 @@ reference; these are the deep dives.
| [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 |
| [chunking.md](chunking.md) | `POST /v1/chunk`: splitting a document for a retrieval index, with the structure decided per request by a template |
| [moderation.md](moderation.md) | `POST /v1/moderation`: screening generated images and video before release (explicitness, faces, age, performer match) — and why it fails closed |
| [expressive-tts.md](expressive-tts.md) | Expressive text-to-speech |
| [image-prompts.md](image-prompts.md) | Image prompts: the `(tag:1.3)` / `[tag]` / `BREAK` grammar, prompts past CLIP's 77 tokens, `clip_skip`, negative prompts, textual-inversion embeddings, size validation |
| [zimage-lora-training.md](zimage-lora-training.md) | LoRA training for image models |
......
docs/moderation.md 0 → 100644
View file @ 3f9b2680
# Screening generated media — `POST /v1/moderation`
Four checks over an image, or over frames sampled from a video: how explicit it
is, where the faces are, how old they look, and whether they are the performer
who consented.
## The rule this endpoint is built around
**It fails closed.** Every reassuring value it can return —
`minor_detected: false`, `flagged: false`, `unmatched_faces: 0` — is a
statement that a model actually examined the content and found nothing. When a
model a requested check needs is missing or unloadable, the request is a
**503** naming the model, not a result with that check quietly absent.
This is the whole point. A caller cannot tell a missing check from a passed
one, so an endpoint that degraded silently would be worse than no endpoint at
all: it would be a green light generated by a model that was never there.
Two consequences, both deliberate:
* **An explicit image with no detectable face cannot be cleared on age.** There
is nothing to estimate an age from, so `minor_detected` comes back `null`
(not `false`), `review_required` is true and the item is flagged. A body
without a face is precisely the case you most need told about.
* **`minor_detected` keys on the low end of the age interval, not the point
estimate.** A face estimated at 24 with an ±8-year band has a credible chance
of being under 18 and is flagged. An age estimator's error is several years
wide; a minor/adult decision made on one number is a decision made on noise.
A face too small to assess (under 48 px) is reported as a face with a `null`
age and `age_assessed: false`, and counts toward `faces_not_assessed` — it is
not an adult face.
None of this is legal advice or a substitute for your own obligations. It is a
screen, and the response says what was measured and by which model so you can
tell the difference.
## Ask what is available first
```http
GET /v1/moderation/models
```
```json
{"models": [{"id": "Falconsai/nsfw_image_detection", "role": "nsfw",
"type": "moderation",
"capabilities": ["moderation", "nsfw_detection"],
"available": true},
{"id": "insightface:genderage", "role": "age", "available": true},
{"id": "buffalo_l", "role": "face", "available": true}],
"unavailable": []}
```
Worth asking **before** sending content: an unavailable check is a 503 at
moderation time. The same entries appear in `/v1/models` as `type:
"moderation"`, with `available` saying whether each can be loaded right now.
## A request
```jsonc
POST /v1/moderation
{
"input": [
{"type": "image", "src": "/v1/files/out-1.png"},
{"type": "video", "src": "/v1/files/out-2.mp4", "sample_fps": 1}
],
"checks": ["nsfw", "age", "faces", "face_match"],
"nsfw_threshold": 0.7,
"face_match": {
"reference": ["/v1/files/performer-selfie.jpg"],
"character_profiles": ["performer-a"], // or a saved profile
"threshold": 0.6
}
}
```
`src` takes a `/v1/files/…` reference, a URL, a local path or base64. An
unknown check name is a 400, never a skipped check.
## The response
```jsonc
{
"results": [{
"src": "/v1/files/out-1.png",
"frames_examined": 1,
"nsfw": {"score": 0.97, "labels": {"nsfw": 0.97, "normal": 0.03}},
"faces": [{"frame": 0, "box": [x, y, w, h], "det_score": 0.93,
"age_estimate": 27.0, "age_low": 19.0, "age_high": 35.0,
"age_assessed": true, "minor_probability": 0.0,
"match": {"reference": 0, "similarity": 0.83, "matched": true}}],
"min_age_estimate": 19.0,
"minor_detected": false,
"unmatched_faces": 0,
"review_required": false,
"flagged": false,
"reasons": []
}],
"flagged": false,
"checks": ["nsfw", "age", "faces", "face_match"],
"models": {"nsfw": "…", "age": "…", "face": "buffalo_l"}
}
```
`reasons` is the list of plain-language findings behind `flagged`; it is empty
when nothing was found. For a video, the fields roll up the **worst** frame,
not an average: one bad frame in a clean clip is a bad clip.
### Video sampling
`sample_fps` defaults to 1 — plenty for screening, since a frame that appears
for under a second still appears. Frames are sampled uniformly across the
**whole** duration, and when the per-video frame cap binds the rate is reduced
rather than the tail dropped: screening the first four minutes of a ten-minute
video is not screening it.
## Configuring the models
Three roles, each configurable through a `moderation_models` entry:
```jsonc
"moderation_models": [
{"role": "nsfw", "id": "Falconsai/nsfw_image_detection"},
{"role": "age", "id": "iitolstykh/mivolo_imdb"}, // else insightface's own head
{"role": "face", "id": "buffalo_l"}
]
```
Defaults, with no configuration at all:
* **faces and face_match** — insightface `buffalo_l`, which the image already
ships for face swapping. Detection plus a 512-d identity embedding.
* **age** — `buffalo_l`'s own genderage head, so age needs no extra download.
It reports a single number, which this endpoint widens into an interval
(see above). A configured MiVOLO is the better estimator where you have one.
* **nsfw** — `Falconsai/nsfw_image_detection` (ViT, Apache-2.0). Any
image-classification model works; labels containing *nsfw*, *porn*, *hentai*,
*explicit*, *nudity*, *nude*, *sexy* or *suggestive* count toward the score,
and every label is reported so your policy can differ from the default.
`CODERAI_NSFW_MODEL` overrides the nsfw default without touching the config.
## Notes
* A `face_match` reference with no detectable face is a **400**. Skipping it
quietly would make every real face read as "matches no consenting performer"
for the wrong reason.
* Similarity is cosine between embeddings; `0.6` separates identities on
`buffalo_l`'s recognition head. Raise it for a stricter match.
* Faces are reported per sampled frame, so one person across 60 frames is 60
entries. `min_age_estimate`, `minor_detected` and `unmatched_faces` are the
rolled-up answers.
docs/video-compose.md
View file @ 3f9b2680
......@@ -39,8 +39,13 @@ and every artefact is served from `/v1/files/`.
"audio": null, // pre-rendered narration instead; overrides TTS
"min_duration": 2.0, // floor, seconds
"padding": 0.15, // silence after the narration
"audio_mix": "voice", // voice | clip | both (see "The clip's own sound")
"clip_volume": 0.35, // the clip's level under the voice, for "both"
"speakers": [], // two characters talking (see "A dialogue")
"visuals": [ // split evenly across the scene, in order
{"type": "video", "src": "https://…/clip.mp4", "trim_start": 0.0},
{"type": "video", "src": "https://…/clip.mp4", "trim_start": 0.0,
"keep_audio": false, // keep THIS clip's own sound (see below)
"volume": null}, // its level under the voice; else clip_volume
{"type": "image", "src": "/v1/files/abc.png",
"ken_burns": {"zoom_from": 1.0, "zoom_to": 1.2, "pan": "auto"}},
{"type": "gradient", "colors": ["#6D28D9", "#DB2777"]},
......@@ -68,8 +73,88 @@ and every artefact is served from `/v1/files/`.
}
```
A scene needs either narration (`text` and/or `audio`) or an explicit
`"duration"`, and at least one visual. Anything else has a default.
A scene needs either narration (`text`, `audio` or `speakers`) or an explicit
`"duration"`, and at least one visual. Anything else has a default. A dialogue
scene needs neither `text` nor visuals: its lines are the narration and its
characters are the picture.
## The clip's own sound
Every visual is rendered silent, which is right for a stock shot under a
voice-over and wrong for footage whose sound **is** the content — a crowd, a
product demo, a piece to camera that was filmed with its audio. Two fields:
```jsonc
{
"audio_mix": "both", // voice | clip | both
"clip_volume": 0.35,
"visuals": [{"type": "video", "src": "/v1/files/demo.mp4", "keep_audio": true}]
}
```
* `"voice"` (the default) — the narration only. Unchanged from before; a
request written without these fields renders exactly as it did.
* `"clip"` — the clip's sound **instead of** narration. No TTS call is made for
this scene, and the scene lasts as long as the clip: cutting it at a
narration's length would truncate the content itself. An explicit `duration`
still wins.
* `"both"` — the narration over the clip, with the clip ducked while the voice
is speaking and brought back up between lines.
The clip's sound is folded into the scene's narration track, so the music bed
ducks under it exactly as it ducks under a voice. With one video visual in the
scene, `keep_audio` is optional — asking for the clip's sound and not saying
which clip can only mean the one clip. With two, mark the one you mean.
Silent footage with `keep_audio: true` is a **warning, not a failure**: the
scene renders with its narration (or in silence for `"clip"`) and the job says
so. Only the first marked clip in a scene contributes; a second clip's sound
belongs in its own scene, where it also gets its own duration.
## A dialogue — two characters in one scene
`presenter` is one character talking to camera, which is a monologue. For an
exchange, give the scene `speakers`:
```jsonc
{
"speakers": [
{"character": "alice", "text": "Did you see the new build?",
"voice": {"model": "kokoro", "voice": "af_sarah", "language": "en-gb"},
"layout": "left", "gap": 0.25},
{"character": "bob", "text": "It shipped this morning.",
"voice": {"model": "kokoro", "voice": "am_adam"}, "layout": "right"}
]
}
```
The lines are spoken in order, each in its own voice, and each character is
lip-synced in their own region of the frame. A character may speak several
times; their lines group into one region in first-speaking order.
* `layout: "left" | "right"` (the default) — every character stays on screen in
their own column for the whole scene. A listener's reaction is half of a
dialogue, so this is the default.
* `layout: "full"` — the frame cuts to whoever is speaking.
All the speakers in one scene must use columns or all must use `full`; mixing
them is a 400, because a character cannot both hold a column and be off screen.
Each character needs a `character` profile or an `image` (or a request-level
`presenter` to inherit a portrait from). A profile name that does not exist is a
400 **before** the render starts, so a typo does not cost you the narration.
Captions follow the speaker: a caption never mixes two characters' words, and
each event carries a `speaker` so the `.srt`/`.vtt` sidecars keep the
attribution.
One lip-sync run happens per **character**, not per line: each character is
given a track holding only their own lines, so their mouth moves on their words
and stays closed while the other one talks.
A dialogue replaces `text`, `voice` and `presenter` for that scene. Keep an
exchange to a beat — a longer conversation is several scenes, which is also how
it gets its own visuals.
## What the server decides
......
tests/test_caption_timings_and_status_polls.py
View file @ 3f9b2680
......@@ -410,14 +410,29 @@ def test_a_ctc_pipeline_is_not_handed_a_none_generate_kwargs(monkeypatch, tmp_pa
def test_a_whisper_checkpoint_still_gets_its_generate_kwargs(monkeypatch, tmp_path):
"""The seq2seq path needs language/task; only the EMPTY dict is dropped."""
"""The seq2seq path needs language/task; only the EMPTY dict is dropped.
The language arrives as the English NAME, which is the only spelling
transformers' Whisper accepts. This test used to assert the code was
passed through verbatim; that pass-through is what made every captioned
render fail with ``Unsupported language: en-us``, so the expectation here
is now the normalised value (see codai.api.lang_codes). What this test is
really guarding is unchanged: the seq2seq path still BUILDS generate_kwargs
rather than dropping them.
"""
import codai.api.stt_backends as stt
be, pipe, audio = _ctc_backend(monkeypatch, tmp_path)
be._is_whisper = True
out = be.transcribe(audio, language="it", word_timestamps=True)
assert out["words"]
assert pipe.calls[0]["generate_kwargs"] == {"task": "transcribe",
"language": "it"}
"language": "italian"}
# And the regional tag a TTS voice carries maps to the same place.
be2, pipe2, audio2 = _ctc_backend(monkeypatch, tmp_path)
be2._is_whisper = True
be2.transcribe(audio2, language="en-gb", word_timestamps=True)
assert pipe2.calls[0]["generate_kwargs"] == {"task": "transcribe",
"language": "english"}
def test_a_whisper_server_model_is_recognised_from_models_json(monkeypatch):
......
tests/test_compose_clip_audio.py 0 → 100644
View file @ 3f9b2680
"""B1: a scene can carry its clip's own sound, not only the narration.
Every visual was rendered with ffmpeg ``-an``, so footage whose sound IS the
content -- a crowd, a demo, a piece to camera -- arrived silent and the only
audio a scene could have was a synthesised voice over it.
"""
import os
import subprocess
import tempfile
import pytest
from codai.api.compose import Scene, Visual, _scene_clip_visual, _scene_mix
# ------------------------------------------------------------ the schema bits
def test_defaults_are_the_old_behaviour():
sc = Scene(text="hi", visuals=[Visual(type="video", src="a.mp4")])
assert _scene_mix(sc) == "voice"
assert sc.clip_volume == 0.35
assert sc.visuals[0].keep_audio is False
# Nothing is picked up without being asked for: a request written before
# this feature existed must render byte-for-byte as it did.
assert _scene_clip_visual(sc) is None
@pytest.mark.parametrize("given,expect", [
("voice", "voice"), ("clip", "clip"), ("both", "both"),
("CLIP", "clip"), (" both ", "both"),
("nonsense", "voice"), (None, "voice"), ("", "voice"),
])
def test_the_mix_is_validated_down_to_three(given, expect):
sc = Scene(text="hi") if given is None else Scene(text="hi", audio_mix=given)
assert _scene_mix(sc) == expect
def test_keep_audio_picks_that_visual():
sc = Scene(text="hi", audio_mix="both", visuals=[
Visual(type="image", src="a.png"),
Visual(type="video", src="b.mp4"),
Visual(type="video", src="c.mp4", keep_audio=True),
])
v = _scene_clip_visual(sc)
assert v is not None and v.src == "c.mp4"
def test_the_first_marked_clip_wins_not_all_of_them():
# Summing several soundtracks under one narration is not what anyone asked
# for; a second clip's sound belongs in its own scene.
sc = Scene(text="hi", audio_mix="both", visuals=[
Visual(type="video", src="b.mp4", keep_audio=True),
Visual(type="video", src="c.mp4", keep_audio=True),
])
assert _scene_clip_visual(sc).src == "b.mp4"
def test_a_single_clip_needs_no_keep_audio_flag():
# Asking for the clip's sound and not saying which clip can only mean the
# one clip in the scene.
sc = Scene(audio_mix="clip", visuals=[Visual(type="video", src="b.mp4")])
assert _scene_clip_visual(sc).src == "b.mp4"
# But with two, it is ambiguous and must be marked.
sc2 = Scene(audio_mix="clip", visuals=[Visual(type="video", src="b.mp4"),
Visual(type="video", src="c.mp4")])
assert _scene_clip_visual(sc2) is None
def test_images_are_never_offered_as_an_audio_source():
sc = Scene(audio_mix="clip", visuals=[Visual(type="image", src="a.png",
keep_audio=True)])
assert _scene_clip_visual(sc) is None
# ------------------------------------------------- the render helpers, for real
def _ffmpeg():
from codai.compose import media
return media.ffmpeg_bin()
def _have_ffmpeg():
try:
subprocess.run([_ffmpeg(), "-version"], capture_output=True, timeout=30)
return True
except Exception:
return False
needs_ffmpeg = pytest.mark.skipif(not _have_ffmpeg(), reason="ffmpeg not available")
def _make_clip(path, *, seconds=3.0, with_audio=True, freq=440):
args = [_ffmpeg(), "-y", "-f", "lavfi", "-i",
f"color=c=red:s=320x240:r=24:d={seconds}"]
if with_audio:
args += ["-f", "lavfi", "-i", f"sine=frequency={freq}:duration={seconds}"]
args += ["-t", str(seconds), "-pix_fmt", "yuv420p"]
if with_audio:
args += ["-shortest"]
args += [path]
subprocess.run(args, capture_output=True, check=True, timeout=120)
return path
@needs_ffmpeg
def test_has_audio_stream_tells_the_two_apart():
from codai.compose import render as rnd
with tempfile.TemporaryDirectory() as d:
loud = _make_clip(os.path.join(d, "loud.mp4"))
mute = _make_clip(os.path.join(d, "mute.mp4"), with_audio=False)
assert rnd.has_audio_stream(loud) is True
assert rnd.has_audio_stream(mute) is False
# A file that is not there at all answers "no" rather than raising:
# the extraction would fail anyway and the render says so.
assert rnd.has_audio_stream(os.path.join(d, "nope.mp4")) is False
@needs_ffmpeg
def test_clip_audio_is_cut_to_exactly_its_slot():
from codai.compose import media
from codai.compose import render as rnd
with tempfile.TemporaryDirectory() as d:
src = _make_clip(os.path.join(d, "s.mp4"), seconds=3.0)
# Shorter than the clip: trimmed.
a = rnd.clip_audio(src, os.path.join(d, "a.wav"), duration=1.5)
assert abs(media.duration_of(a) - 1.5) < 0.08, media.duration_of(a)
# Longer than the clip: padded with silence, never short.
b = rnd.clip_audio(src, os.path.join(d, "b.wav"), duration=5.0)
assert abs(media.duration_of(b) - 5.0) < 0.08, media.duration_of(b)
# Looped, to match a video track that build_visual looped.
c = rnd.clip_audio(src, os.path.join(d, "c.wav"), duration=7.0, loop=True)
assert abs(media.duration_of(c) - 7.0) < 0.08, media.duration_of(c)
@needs_ffmpeg
def test_trim_start_skips_the_same_audio_the_picture_skips():
from codai.compose import media
from codai.compose import render as rnd
with tempfile.TemporaryDirectory() as d:
src = _make_clip(os.path.join(d, "s.mp4"), seconds=4.0)
a = rnd.clip_audio(src, os.path.join(d, "a.wav"), duration=4.0,
trim_start=2.0)
# Four seconds of slot, two seconds skipped: two of sound, two of pad.
assert abs(media.duration_of(a) - 4.0) < 0.08
@needs_ffmpeg
def test_volume_is_applied():
from codai.compose import render as rnd
with tempfile.TemporaryDirectory() as d:
src = _make_clip(os.path.join(d, "s.mp4"), seconds=2.0)
loud = rnd.clip_audio(src, os.path.join(d, "l.wav"), duration=2.0,
volume=1.0)
quiet = rnd.clip_audio(src, os.path.join(d, "q.wav"), duration=2.0,
volume=0.1)
def _peak(p):
out = subprocess.run(
[_ffmpeg(), "-i", p, "-af", "volumedetect", "-f", "null", "-"],
capture_output=True, text=True, timeout=120)
for line in (out.stderr or "").splitlines():
if "max_volume" in line:
return float(line.split("max_volume:")[1].replace("dB", "").strip())
return None
pl, pq = _peak(loud), _peak(quiet)
assert pl is not None and pq is not None
assert pq < pl - 15, (pl, pq) # ~20 dB down for 0.1
@needs_ffmpeg
def test_mix_voice_clip_covers_every_combination():
from codai.compose import media
from codai.compose import render as rnd
with tempfile.TemporaryDirectory() as d:
src = _make_clip(os.path.join(d, "s.mp4"), seconds=3.0)
clip = rnd.clip_audio(src, os.path.join(d, "c.wav"), duration=3.0)
voice = rnd.silence(os.path.join(d, "v.wav"), 2.0)
both = rnd.mix_voice_clip(voice, clip, os.path.join(d, "both.wav"),
duration=3.0, duck=True)
assert abs(media.duration_of(both) - 3.0) < 0.1
only_clip = rnd.mix_voice_clip(None, clip, os.path.join(d, "oc.wav"),
duration=3.0)
assert abs(media.duration_of(only_clip) - 3.0) < 0.1
only_voice = rnd.mix_voice_clip(voice, None, os.path.join(d, "ov.wav"),
duration=3.0)
assert abs(media.duration_of(only_voice) - 3.0) < 0.1
neither = rnd.mix_voice_clip(None, None, os.path.join(d, "n.wav"),
duration=3.0)
assert abs(media.duration_of(neither) - 3.0) < 0.1
tests/test_compose_dialogue.py 0 → 100644
View file @ 3f9b2680
"""B2: two characters talking in one scene.
A scene had one voice and one presenter, which is a monologue. A dialogue
needs an order, a voice per turn, and somewhere for each character to be while
the other one talks.
"""
import os
import subprocess
import tempfile
import pytest
from fastapi import HTTPException
from codai.api.compose import (ComposeRequest, Scene, SpeakerLine, Visual,
_speaker_key, _speaker_layout, _speaker_order,
_validate)
def _sp(**kw):
return SpeakerLine(**kw)
# ----------------------------------------------------------------- identity
def test_a_character_is_identified_by_profile_then_portrait_then_position():
assert _speaker_key(_sp(character="Alice"), 0) == "alice"
assert _speaker_key(_sp(image="a.png"), 0) == "a.png"
assert _speaker_key(_sp(), 3) == "_speaker3"
def test_the_same_character_speaking_twice_is_one_column():
sc = Scene(speakers=[_sp(character="alice", text="hi"),
_sp(character="bob", text="hello"),
_sp(character="Alice", text="how are you")])
order = _speaker_order(sc)
assert [k for k, _ in order] == ["alice", "bob"]
def test_the_column_order_is_first_speaking_order():
sc = Scene(speakers=[_sp(character="bob", text="x"),
_sp(character="alice", text="y")])
assert [k for k, _ in _speaker_order(sc)] == ["bob", "alice"]
# ------------------------------------------------------------------ layout
def test_columns_are_the_default():
sc = Scene(speakers=[_sp(character="a", text="x"), _sp(character="b", text="y")])
assert _speaker_layout(sc) == "row"
@pytest.mark.parametrize("layouts,expect", [
(["left", "right"], "row"),
(["full", "full"], "cut"),
([None, None], "row"),
(["left", None], "row"),
])
def test_the_layout_is_derived_from_what_the_speakers_ask_for(layouts, expect):
sc = Scene(speakers=[_sp(character=f"c{i}", text="x", layout=l)
for i, l in enumerate(layouts)])
assert _speaker_layout(sc) == expect
def test_mixing_a_column_and_full_is_refused():
# A character cannot both hold a column and be off screen while the other
# one speaks; rendering it would have to silently pick one.
sc = Scene(speakers=[_sp(character="a", text="x", layout="left"),
_sp(character="b", text="y", layout="full")])
with pytest.raises(ValueError) as e:
_speaker_layout(sc)
assert "column" in str(e.value)
# -------------------------------------------------------------- validation
def _req(scene):
return ComposeRequest(scenes=[scene])
def test_a_dialogue_scene_needs_neither_text_nor_visuals():
# Its lines ARE the narration and the presenter. Refusing it for having
# neither is what the old validation did. Portraits given inline, so the
# test does not depend on what is saved on this install.
_validate(_req(Scene(speakers=[
_sp(image="data:image/png;base64,AAAA", text="hi"),
_sp(image="data:image/png;base64,BBBB", text="hello")])))
def test_a_line_with_nothing_to_say_is_refused():
with pytest.raises(HTTPException) as e:
_validate(_req(Scene(speakers=[_sp(image="data:image/png;base64,AAAA")])))
assert e.value.status_code == 400
assert "text" in str(e.value.detail)
def test_a_speaker_with_no_face_is_refused_up_front():
# It cannot be lip-synced, and finding that out mid-render costs the caller
# the narration they already paid to synthesise.
with pytest.raises(HTTPException) as e:
_validate(_req(Scene(speakers=[_sp(text="hi")])))
assert e.value.status_code == 400
assert "character" in str(e.value.detail)
def test_a_speaker_may_inherit_the_request_presenter_portrait():
from codai.api.compose import PresenterSpec
req = ComposeRequest(
scenes=[Scene(speakers=[_sp(text="hi")])],
presenter=PresenterSpec(image="data:image/png;base64,AAAA"))
_validate(req)
def test_a_misspelt_character_profile_is_caught_before_the_render(monkeypatch):
import codai.api.characters as C
monkeypatch.setattr(C, "_load_character_meta",
lambda n: {"name": n} if n == "alice" else None)
_validate(_req(Scene(speakers=[_sp(character="alice", text="hi")])))
with pytest.raises(HTTPException) as e:
_validate(_req(Scene(speakers=[_sp(character="alise", text="hi")])))
assert e.value.status_code == 400
assert "alise" in str(e.value.detail)
def test_a_bad_layout_mix_is_a_400_not_a_500():
with pytest.raises(HTTPException) as e:
_validate(_req(Scene(speakers=[
_sp(image="data:image/png;base64,AAAA", text="x", layout="left"),
_sp(image="data:image/png;base64,BBBB", text="y", layout="full")])))
assert e.value.status_code == 400
def test_too_many_lines_in_one_scene_is_refused():
from codai.api.compose import _MAX_SPEAKERS_PER_SCENE
sc = Scene(speakers=[_sp(image="data:image/png;base64,AAAA", text="x")
for _ in range(_MAX_SPEAKERS_PER_SCENE + 1)])
with pytest.raises(HTTPException) as e:
_validate(_req(sc))
assert "at most" in str(e.value.detail)
# --------------------------------------------------- captions, per speaker
def test_a_caption_never_mixes_two_speakers():
from codai.compose.captions import break_lines
words = ([{"word": f"a{i}", "start": i, "end": i + 1, "speaker": "alice"}
for i in range(3)]
+ [{"word": f"b{i}", "start": 3 + i, "end": 4 + i, "speaker": "bob"}
for i in range(3)])
evs = break_lines(words, max_words_per_line=4, max_lines=2)
# Without the speaker split these six words are one event of a question
# and its answer, attributed to nobody.
assert len(evs) == 2
assert evs[0]["speaker"] == "alice"
assert evs[1]["speaker"] == "bob"
assert "b0" not in evs[0]["text"]
def test_a_long_run_by_one_speaker_still_splits_by_length():
from codai.compose.captions import break_lines
words = [{"word": f"w{i}", "start": i, "end": i + 1, "speaker": "alice"}
for i in range(10)]
evs = break_lines(words, max_words_per_line=2, max_lines=1)
assert len(evs) == 5
assert all(e["speaker"] == "alice" for e in evs)
def test_captions_without_speakers_are_unchanged():
from codai.compose.captions import break_lines
words = [{"word": f"w{i}", "start": i, "end": i + 1} for i in range(4)]
evs = break_lines(words, max_words_per_line=4, max_lines=1)
assert len(evs) == 1
assert "speaker" not in evs[0]
# ----------------------------------------------- the render helpers, for real
def _ffmpeg():
from codai.compose import media
return media.ffmpeg_bin()
def _have_ffmpeg():
try:
subprocess.run([_ffmpeg(), "-version"], capture_output=True, timeout=30)
return True
except Exception:
return False
needs_ffmpeg = pytest.mark.skipif(not _have_ffmpeg(), reason="ffmpeg not available")
@needs_ffmpeg
def test_place_at_puts_each_line_where_it_falls():
from codai.compose import media
from codai.compose import render as rnd
with tempfile.TemporaryDirectory() as d:
a = rnd.silence(os.path.join(d, "a.wav"), 1.0)
b = rnd.silence(os.path.join(d, "b.wav"), 1.0)
out = rnd.place_at([(a, 0.0), (b, 2.0)], os.path.join(d, "t.wav"),
duration=4.0)
assert abs(media.duration_of(out) - 4.0) < 0.1
@needs_ffmpeg
def test_place_at_with_nothing_is_silence_not_an_error():
# A character who says nothing in a scene still needs a track of the
# scene's length, or their lip-sync run has nothing to drive.
from codai.compose import media
from codai.compose import render as rnd
with tempfile.TemporaryDirectory() as d:
out = rnd.place_at([], os.path.join(d, "t.wav"), duration=2.5)
assert abs(media.duration_of(out) - 2.5) < 0.1
@needs_ffmpeg
def test_composite_row_puts_two_speakers_side_by_side():
from codai.compose import media
from codai.compose import render as rnd
with tempfile.TemporaryDirectory() as d:
clips = []
for i, colour in enumerate(("red", "blue")):
p = os.path.join(d, f"c{i}.mp4")
subprocess.run([_ffmpeg(), "-y", "-f", "lavfi", "-i",
f"color=c={colour}:s=240x320:r=24:d=2",
"-pix_fmt", "yuv420p", p],
capture_output=True, check=True, timeout=120)
clips.append(p)
out = rnd.composite_row(clips, os.path.join(d, "row.mp4"),
width=480, height=320, fps=24)
info = media.probe(out)
assert info["width"] == 480 and info["height"] == 320
@needs_ffmpeg
def test_cut_segment_lands_on_the_asked_for_window():
from codai.compose import media
from codai.compose import render as rnd
with tempfile.TemporaryDirectory() as d:
src = os.path.join(d, "s.mp4")
subprocess.run([_ffmpeg(), "-y", "-f", "lavfi", "-i",
"color=c=green:s=320x240:r=24:d=6",
"-pix_fmt", "yuv420p", src],
capture_output=True, check=True, timeout=120)
out = rnd.cut_segment(src, os.path.join(d, "cut.mp4"), start=2.0,
duration=1.5, fps=24, width=320, height=240)
assert abs(media.duration_of(out) - 1.5) < 0.12
tests/test_model_defaults_and_samplers.py 0 → 100644
View file @ 3f9b2680
"""C5: a model's own recommended settings, and the sampler machinery.
Without these a client keeps a table of per-model step counts keyed on the
model NAME -- which breaks the moment a second quantization of the same model
is added -- and a client without one runs an 8-step distilled model for 30
steps: four times the cost for a worse picture.
"""
import pytest
from codai.api.images import SAMPLERS, _apply_sampler, model_defaults
# ------------------------------------------------------------------ defaults
def test_defaults_are_read_from_the_model_entry():
d = model_defaults({"defaults": {"steps": 8, "guidance_scale": 1.0,
"sampler": "euler"}})
assert d["steps"] == 8 and d["guidance_scale"] == 1.0
@pytest.mark.parametrize("cfg", [None, {}, {"defaults": None},
{"defaults": "not a dict"}])
def test_a_model_without_defaults_is_not_an_error(cfg):
assert model_defaults(cfg) == {}
def test_defaults_are_not_call_kwargs():
# `call_kwargs` applies on EVERY call because the card says the model does
# not work without it; `defaults` fill in only what the caller omitted.
# Confused for one another, a caller's explicit steps would be overridden.
import inspect
from codai.api import images
src = inspect.getsource(images.model_defaults)
assert "never override" in src
def test_the_request_beats_the_card_and_the_card_beats_the_floor():
import inspect
from codai.api import images
src = inspect.getsource(images._generate_with_diffusers)
i = src.index("_dflt = model_defaults(model_config)")
window = src[i:i + 900]
# request.steps first, then the fused-accel preset, then the card, then 30.
assert "request.steps if request.steps" in window
assert "_accel_steps if _accel_steps" in window
assert "int(_d_steps) if _d_steps" in window
def test_video_layers_the_card_under_the_accel_preset_and_over_the_floor():
import inspect
from codai.api import video
src = inspect.getsource(video._generate_video)
i_accel = src.index("accel_call_defaults")
i_card = src.index("model_defaults(")
i_floor = src.index("kw.setdefault('num_inference_steps', 25)")
assert i_accel < i_card < i_floor, "the precedence order is wrong"
def test_build_call_kwargs_still_sets_only_what_the_request_asked():
# The accel and card layers both use setdefault, so an unset key has to
# keep meaning "the caller did not ask". Breaking this makes the card
# silently outrank a fused distill preset.
import types
from codai.api.video import _build_call_kwargs
r = types.SimpleNamespace(prompt="x", negative_prompt=None,
num_inference_steps=None, guidance_scale=None,
num_frames=None, width=None, height=None,
seed=None)
kw = _build_call_kwargs(r)
assert "num_inference_steps" not in kw
assert "guidance_scale" not in kw
# ------------------------------------------------------------------ samplers
class _Pipe:
def __init__(self, sched):
self.scheduler = sched
def _sched(name):
import diffusers
return getattr(diffusers, name)()
def test_a_known_sampler_swaps_the_scheduler():
p = _Pipe(_sched("EulerDiscreteScheduler"))
assert _apply_sampler(p, "euler_a") == "EulerAncestralDiscreteScheduler"
assert type(p.scheduler).__name__ == "EulerAncestralDiscreteScheduler"
def test_the_spelling_is_forgiving():
for spelling in ("DPMPP-2M", "dpmpp_2m", " dpmpp 2m "):
p = _Pipe(_sched("EulerDiscreteScheduler"))
assert _apply_sampler(p, spelling) == "DPMSolverMultistepScheduler"
def test_an_unknown_sampler_names_the_valid_ones():
p = _Pipe(_sched("EulerDiscreteScheduler"))
with pytest.raises(ValueError) as e:
_apply_sampler(p, "nonsense")
assert "unknown sampler" in str(e.value)
assert "euler" in str(e.value)
def test_crossing_the_flow_matching_boundary_is_refused():
"""Not a worse picture -- noise.
Flow-matching and DDPM-family schedulers parameterise the trajectory
differently, so substituting one for the other does not degrade the
result, it destroys it. A silent swap would look like the model being
broken.
"""
flow = _Pipe(_sched("FlowMatchEulerDiscreteScheduler"))
with pytest.raises(ValueError) as e:
_apply_sampler(flow, "ddim")
assert "not compatible" in str(e.value)
classic = _Pipe(_sched("EulerDiscreteScheduler"))
with pytest.raises(ValueError):
_apply_sampler(classic, "flow_euler")
# Within a family it is fine.
assert _apply_sampler(flow, "flow_heun") == "FlowMatchHeunDiscreteScheduler"
def test_asking_for_the_scheduler_already_in_place_is_a_no_op():
p = _Pipe(_sched("EulerDiscreteScheduler"))
before = p.scheduler
assert _apply_sampler(p, "euler") == "EulerDiscreteScheduler"
assert p.scheduler is before
def test_an_empty_sampler_does_nothing():
p = _Pipe(_sched("EulerDiscreteScheduler"))
before = p.scheduler
assert _apply_sampler(p, "") == ""
assert _apply_sampler(p, None) == ""
assert p.scheduler is before
def test_every_mapped_sampler_exists_in_this_diffusers():
import diffusers
missing = sorted({c for c in SAMPLERS.values()
if not hasattr(diffusers, c)})
assert not missing, missing
# ------------------------------------------------- what /v1/models publishes
def test_the_listing_can_carry_defaults_modes_and_licence():
from codai.pydantic.textrequest import ModelInfo
m = ModelInfo(id="x", defaults={"steps": 8}, modes=["t2v", "i2v"],
license="Apache-2.0", available=True)
d = m.model_dump()
assert d["defaults"] == {"steps": 8}
assert d["modes"] == ["t2v", "i2v"]
assert d["license"] == "Apache-2.0"
assert d["available"] is True
def test_modes_and_licence_are_read_from_either_place():
# An operator should not have to remember whether these go at the top
# level of the entry or inside `defaults`.
import inspect
from codai.models import manager
src = inspect.getsource(manager)
i = src.index('defaults=(meta.get("defaults") or None)')
window = src[i:i + 700]
assert 'meta.get("modes")' in window
assert 'meta.get("license")' in window
assert '"licence"' in window
tests/test_moderation_fails_closed.py 0 → 100644
View file @ 3f9b2680
This diff is collapsed.
tests/test_multi_character.py 0 → 100644
View file @ 3f9b2680
"""B3: two of the caller's characters in one frame.
`character_profiles` already took a list, but every reference went into one
IP-Adapter set -- and IP-Adapter conditions on the MEAN of its references, so
two characters produced one person who looked like neither. Keeping the
reference sets apart, and handing them to a model that takes several, is what
makes two people two people.
"""
import types
import pytest
class _Req:
def __init__(self, **kw):
self.characters = kw.get("characters")
self.character_profiles = kw.get("character_profiles")
self.character_references = kw.get("character_references")
self.character_names = kw.get("character_names")
@pytest.fixture
def saved(monkeypatch):
"""Two saved profiles, with two reference images each."""
store = {"alice": ["A1", "A2"], "bob": ["B1", "B2"]}
import codai.api.characters as C
monkeypatch.setattr(
C, "resolve_character_profiles",
lambda names: [i for n in names for i in store.get(str(n).lower(), [])])
return store
# ------------------------------------------------------------------ grouping
def test_bare_names_become_one_group_each(saved):
from codai.api.characters import character_groups
g = character_groups(_Req(characters=["alice", "bob"]))
assert [x["name"] for x in g] == ["alice", "bob"]
assert g[0]["images"] == ["A1", "A2"]
assert g[1]["images"] == ["B1", "B2"]
def test_slots_work_and_a_named_slot_without_images_is_resolved(saved):
from codai.api.characters import character_groups
g = character_groups(_Req(characters=[
{"name": "carol", "images": ["C1"]},
{"name": "alice"},
]))
assert g[0]["images"] == ["C1"]
assert g[1]["images"] == ["A1", "A2"]
def test_the_long_standing_field_still_groups_per_character(saved):
from codai.api.characters import character_groups
g = character_groups(_Req(character_profiles=["alice", "bob"]))
assert len(g) == 2
# This is the case that silently blended: two profiles, one flat list.
assert g[0]["images"] != g[1]["images"]
def test_a_character_named_twice_is_one_character(saved):
from codai.api.characters import character_groups
g = character_groups(_Req(characters=["alice"], character_profiles=["Alice"]))
assert len(g) == 1
def test_one_character_yields_one_group(saved):
from codai.api.characters import character_groups
assert len(character_groups(_Req(characters=["alice"]))) == 1
assert character_groups(_Req()) == []
def test_inline_references_split_only_when_names_align(saved):
from codai.api.characters import character_groups
g = character_groups(_Req(character_references=["X1", "X2"],
character_names=["alice", "bob"]))
assert len(g) == 2
# Without aligned names there is no way to know which image is whose, so
# they stay one group rather than being guessed at.
g2 = character_groups(_Req(character_references=["X1", "X2"]))
assert len(g2) == 1 and g2[0]["images"] == ["X1", "X2"]
def test_an_empty_profile_is_not_a_character(saved):
from codai.api.characters import character_groups
assert character_groups(_Req(characters=["nobody"])) == []
# --------------------------------------------------- the video request shapes
def test_video_resolves_bare_names_slots_and_legacy(saved):
from codai.api.video import _resolve_character_inputs
r = types.SimpleNamespace(
character_profiles=None, characters=["alice", {"name": "bob"}],
character_references=None, character_names=None)
images, names = _resolve_character_inputs(r)
assert names == ["alice", "bob"]
assert images == ["A1", "A2", "B1", "B2"]
def test_a_bare_name_does_not_crash_the_minimax_path(saved):
# The widened type is only safe if every consumer stopped calling .get()
# on it. This is the one that would have raised AttributeError.
import inspect
from codai.api import video
src = inspect.getsource(video)
i = src.index('for entry in (request.characters or []):')
window = src[i:i + 900]
assert "isinstance(entry, str)" in window
assert "isinstance(entry, dict)" in window
def test_the_keyframe_bridge_forwards_characters():
import inspect
from codai.api import video
src = inspect.getsource(video)
assert 'payload["characters"]' in src, (
"a two-character clip would render its identity keyframe from neither")
# ------------------------------------------------- the image routing decision
def test_multi_character_needs_a_multi_reference_pipeline():
# The decision rests on _is_instruction_edit, which is (image yes,
# strength no) -- true for FLUX.2 and Qwen-Image-Edit, false for SDXL.
from codai.api.images import _is_instruction_edit
class _Edit:
def __call__(self, image=None, prompt=None):
pass
class _Img2Img:
def __call__(self, image=None, strength=None, prompt=None):
pass
assert _is_instruction_edit(_Edit()) is True
assert _is_instruction_edit(_Img2Img()) is False
def test_the_routing_is_present_and_clears_the_blended_set():
import inspect
from codai.api import images
src = inspect.getsource(images._generate_with_diffusers)
assert "_multi = [g for g in char_groups" in src
i = src.index("_multi = [g for g in char_groups")
window = src[i:i + 1600]
# Passed as a list of references …
assert 'call_kwargs["image"] = _refs' in window
# … and NOT also blended through IP-Adapter, which would undo it.
assert "char_images = []" in window
# A model that cannot do it says so rather than quietly blending.
assert "_log.warning" in window
tests/test_prequant_vram_and_empty_completion.py 0 → 100644
View file @ 3f9b2680
"""A3/A4/A5: an advertised capability, an inflated estimate, a silent failure.
The three share a shape: the server held a belief about a model that nothing
checked against the model itself, and the client paid for it.
"""
import json
import os
import types
import pytest
# ============================================================ A3: FLUX.2 edit
def test_flux2_reports_instruction_edit():
from codai.models.capabilities import detect_model_capabilities
for mid in ("FLUX.2-klein-4B", "black-forest-labs/FLUX.2-klein-4B",
"FLUX.2-dev", "flux2-klein", "FLUX-2-pro"):
caps = detect_model_capabilities(mid).to_list()
assert "instruction_edit" in caps, (mid, caps)
assert "image_generation" in caps, (mid, caps)
# It is not a denoising img2img model: advertising `strength` would make
# clients show a slider for an argument the pipeline has not got.
assert "inpainting" not in caps, (mid, caps)
def test_flux1_is_not_swept_up():
from codai.models.capabilities import detect_model_capabilities
caps = detect_model_capabilities("FLUX.1-dev").to_list()
assert "instruction_edit" not in caps, caps
# Kontext is the FLUX.1 that does edit by instruction, and still does.
assert "instruction_edit" in detect_model_capabilities(
"FLUX.1-Kontext-dev").to_list()
def test_the_pipeline_signature_agrees_with_the_advertised_capability():
# The claim is only worth making if the runtime discriminator agrees:
# images._is_instruction_edit decides on (image yes, strength no).
diffusers = pytest.importorskip("diffusers")
from codai.api.images import _is_instruction_edit
for name in ("Flux2Pipeline", "Flux2KleinPipeline"):
cls = getattr(diffusers, name, None)
if cls is None:
pytest.skip(f"{name} not in this diffusers")
assert _is_instruction_edit(cls) is True, name
# ================================================= A4: pre-quantized estimate
def _mgr():
from codai.models.manager import MultiModelManager
m = object.__new__(MultiModelManager)
m._prequantized_cache = {}
m._storage_dtype_cache = {}
return m
def test_a_bnb_checkpoint_is_recognised_from_its_config(tmp_path):
d = tmp_path / "zimage"
d.mkdir()
(d / "config.json").write_text(json.dumps({
"_class_name": "ZImagePipeline",
"quantization_config": {"quant_method": "bitsandbytes_4bit"},
}))
m = _mgr()
assert m._is_prequantized_checkpoint("image:x", str(d), {"path": str(d)}) is True
def test_a_quantized_pipeline_component_counts(tmp_path):
d = tmp_path / "pipe"
(d / "transformer").mkdir(parents=True)
(d / "config.json").write_text(json.dumps({"_class_name": "P"}))
(d / "transformer" / "config.json").write_text(json.dumps({
"quantization_config": {"quant_method": "gptq"}}))
m = _mgr()
assert m._is_prequantized_checkpoint("image:x", str(d), {"path": str(d)}) is True
def test_a_plain_checkpoint_is_not_flagged(tmp_path):
d = tmp_path / "plain"
d.mkdir()
(d / "config.json").write_text(json.dumps({"_class_name": "P",
"torch_dtype": "bfloat16"}))
m = _mgr()
assert m._is_prequantized_checkpoint("image:x", str(d), {"path": str(d)}) is False
@pytest.mark.parametrize("repo,expect", [
("unsloth/Z-Image-Turbo-unsloth-bnb-4bit", True),
("TheBloke/Model-AWQ", True),
("org/Model-GPTQ", True),
("org/Model-int8", True),
("org/Model-fp8", True),
("Qwen/Qwen3-8B", False),
("black-forest-labs/FLUX.2-klein-4B", False), # "4B" is a size, not 4-bit
])
def test_the_name_is_the_fallback_when_no_config_is_reachable(repo, expect):
m = _mgr()
# A repo id that is not in any local cache: no config.json is reachable, so
# the name decides.
got = m._is_prequantized_checkpoint("text:" + repo, repo, {"path": repo})
assert got is expect, (repo, got)
def test_a_prequantized_checkpoint_is_not_inflated_to_load_precision():
"""The A4 regression itself, at the arithmetic.
16.3 GB of 4-bit weights (~1.17 bytes/elem on disk) were normalised to a
2-byte load precision -- a 1.70x factor -- and estimated at 30.03 GB
against a 24 GB card. Nothing can free 30 GB of 24, so every still was
generated with the pipeline partly on the CPU: ~72 s for one image from a
few-step turbo model.
"""
from codai.models.manager import MultiModelManager
m = object.__new__(MultiModelManager)
m._prequantized_cache = {}
m._storage_dtype_cache = {}
m._measured_vram_gb = {}
m.config = {}
key = "image:unsloth/Z-Image-Turbo-unsloth-bnb-4bit"
cfg = {"path": "unsloth/Z-Image-Turbo-unsloth-bnb-4bit",
"used_vram_gb": 16.316, "model_type": "image_models"}
m._config_for_model = types.MethodType(lambda self, k: cfg, m)
m._resolve_config_key = types.MethodType(lambda self, k: key, m)
m._storage_bytes_per_elem = types.MethodType(lambda self, *a: 1.1736, m)
m._quantized_cache_gb = types.MethodType(lambda self, *a: 0.0, m)
m._config_lora_vram_gb = types.MethodType(lambda self, c: 0.0, m)
m._runtime_reserve_gb = types.MethodType(lambda self, c, k, g: 2.0, m)
m._longcat_resident_gb = types.MethodType(lambda self, *a: 0.0, m)
m._effective_quant_multiplier = types.MethodType(lambda self, *a: 1.0, m)
est = m._get_model_used_vram_gb(key)
# 16.316 + 2.0 reserve, NOT 16.316 * 1.70 + reserve = 30.03.
assert 17.0 <= est <= 20.0, est
assert est < 24.0, f"still does not fit a 24 GB card: {est}"
# ============================================== A5: the silent empty answer
def test_a_failed_generation_raises_instead_of_returning_empty():
from codai.backends.cuda import GenerationFailed, _generation_failed
exc = _generation_failed(
RuntimeError("Pointer argument (at 0) cannot be accessed from Triton "
"(cpu tensor?)"))
assert isinstance(exc, GenerationFailed)
assert isinstance(exc, RuntimeError) # so existing handlers still see it
msg = str(exc)
assert "Triton" in msg
# The hint names the cause, not the symptom: an operator reading "Pointer
# argument" alone has nowhere to go.
assert "resident on the CPU" in msg, msg
assert "used_vram_gb" in msg, msg
def test_the_first_error_leads_and_duplicates_collapse():
from codai.backends.cuda import _generation_failed
msg = str(_generation_failed(RuntimeError("first"), RuntimeError("second"),
RuntimeError("second")))
assert msg.index("first") < msg.index("second")
assert msg.count("second") == 1
def test_an_oom_gets_its_own_hint():
from codai.backends.cuda import _generation_failed
assert "ran out of memory" in str(
_generation_failed(RuntimeError("CUDA out of memory")))
def test_the_empty_answer_is_no_longer_reachable_from_the_ladder():
# The three-rung ladder is still there -- it is real recovery -- but the
# bottom rung no longer assigns "" and returns. Asserted on the source so
# the guarantee survives a refactor of the ladder itself.
import inspect
from codai.backends import cuda
src = inspect.getsource(cuda.NvidiaBackend.generate_chat)
assert "no-cache fallback failed" in src
tail = src[src.index("no-cache fallback failed"):]
assert "_generation_failed" in tail
assert 'generated_text = ""' not in tail
class _FakeQuantModel:
def __init__(self, device_map):
self.hf_device_map = device_map
self.config = types.SimpleNamespace(quantization_config={"q": 1})
def test_a_quantized_model_offloaded_to_cpu_is_refused():
from codai.backends.cuda import NvidiaBackend
b = object.__new__(NvidiaBackend)
model = _FakeQuantModel({"layers.0": 0, "layers.1": "cpu", "layers.2": "disk"})
with pytest.raises(RuntimeError) as e:
b._reject_offloaded_quantized(model, "Qwen/Qwen3.5-9B",
{"quantization_config": object()})
msg = str(e.value)
assert "quantized" in msg
assert "2 of its 3 modules" in msg, msg
assert "fail on every request" in msg
def test_a_quantized_model_fully_on_the_gpu_is_accepted():
from codai.backends.cuda import NvidiaBackend
b = object.__new__(NvidiaBackend)
model = _FakeQuantModel({"layers.0": 0, "layers.1": 0})
b._reject_offloaded_quantized(model, "m", {"quantization_config": object()})
def test_an_unquantized_model_may_still_offload():
# Offload is a legitimate, merely slower path for dense weights; only
# quantized kernels cannot read them.
from codai.backends.cuda import NvidiaBackend
b = object.__new__(NvidiaBackend)
model = types.SimpleNamespace(
hf_device_map={"layers.0": 0, "layers.1": "cpu"},
config=types.SimpleNamespace(quantization_config=None))
b._reject_offloaded_quantized(model, "m", {})
tests/test_stt_language_normalization.py 0 → 100644
View file @ 3f9b2680
"""A TTS voice's regional tag must not break speech-to-text (A1).
A compose render picks a voice for its accent, so the language it carries is
``en-us`` / ``en-gb`` / ``pt-br``. Every speech backend wants something else,
and transformers' Whisper RAISES on a tag it doesn't know -- which arrived at
the client as ``500: CrisperWhisper worker error 500: Unsupported language:
en-us`` on every captioned scene.
"""
import pytest
from codai.api import lang_codes as L
# ---------------------------------------------------------------- the mapping
@pytest.mark.parametrize("tag,code,name", [
("en-us", "en", "english"),
("en-GB", "en", "english"),
("en_gb", "en", "english"),
("pt-br", "pt", "portuguese"),
("it", "it", "italian"),
("italian", "it", "italian"),
("Italian", "it", "italian"),
("ita", "it", "italian"),
("zh-Hans-CN", "zh", "chinese"),
("mandarin", "zh", "chinese"),
("fr-ca", "fr", "french"),
("es-419", "es", "spanish"),
])
def test_every_spelling_reaches_the_same_language(tag, code, name):
assert L.to_code(tag) == code
assert L.to_name(tag) == name
@pytest.mark.parametrize("family,expected", [
("crisperwhisper", "english"), # transformers: the NAME
("whisper_hf", "english"),
("whisper-transformers", "english"),
("nemo", "en"), # Canary: the CODE
("whisper-server", "en"), # whisper.cpp: the CODE
("faster_whisper", "en"), # faster-whisper: the CODE, not the name
("wav2vec2", "en"),
("vosk", "en"),
(None, "en"),
])
def test_each_backend_gets_the_spelling_its_api_wants(family, expected):
assert L.normalize("en-us", family) == expected
@pytest.mark.parametrize("tag", ["klingon", "xx-YY", "auto", "", None, " "])
def test_an_unmappable_tag_becomes_autodetect_not_an_error(tag):
# None means "omit the argument", i.e. let the model detect. Never a raise:
# refusing to transcribe because the caller named a dialect is strictly
# worse than transcribing with detection.
assert L.normalize(tag, "crisperwhisper") is None
assert L.normalize(tag, "nemo") is None
def test_names_with_spaces_survive_the_subtag_split():
# "haitian creole" must not be truncated at the first word, and a name
# containing a hyphen must not be split either.
assert L.to_code("haitian creole") == "ht"
assert L.to_name("haitian creole") == "haitian creole"
# ------------------------------------------------------------- the validation
def test_declared_english_accepts_a_regional_tag():
from codai.api.transcriptions import _validate_language
# No raise: this is the exact combination that 400'd before.
_validate_language("crisperwhisper", {"languages": ["english", "italian"]},
"en-gb", None)
_validate_language("canary", {"languages": ["en", "it"]}, "english", None)
_validate_language("canary", {"languages": ["en"]}, "en-us", None)
def test_a_language_the_model_really_lacks_is_still_refused():
from fastapi import HTTPException
from codai.api.transcriptions import _validate_language
with pytest.raises(HTTPException) as e:
_validate_language("wav2vec2-it", {"languages": ["italian"]}, "en-us", None)
assert e.value.status_code == 400
# And a model that declares nothing still accepts anything.
_validate_language("whatever", {}, "en-us", None)
def test_aliases_are_symmetric():
assert L.aliases("en-gb") == {"en-gb", "en", "english"}
assert "en" in L.aliases("english")
# An unmappable tag still matches a config that declares exactly it.
assert L.aliases("klingon") == {"klingon"}
# --------------------------------------------------- the backends, not the map
def test_crisperwhisper_backend_sends_the_name(monkeypatch):
import codai.api.stt_backends as sb
sent = {}
class _Resp:
status_code = 200
def json(self):
return {"text": "hi", "words": [], "segments": []}
class _Req:
exceptions = __import__("requests").exceptions
@staticmethod
def post(url, params=None, data=None, timeout=None, headers=None):
sent.update(params or {})
return _Resp()
monkeypatch.setitem(__import__("sys").modules, "requests", _Req)
b = sb._RemoteCrisperWhisperBackend("crisperwhisper", {}, "http://x")
import tempfile, os
fd, path = tempfile.mkstemp()
os.close(fd)
try:
b.transcribe(path, language="en-us")
finally:
os.unlink(path)
assert sent.get("language") == "english", sent
def test_nemo_backend_sends_the_code(monkeypatch):
import codai.api.stt_backends as sb
sent = {}
class _Resp:
status_code = 200
def json(self):
return {"text": "hi"}
class _Req:
exceptions = __import__("requests").exceptions
@staticmethod
def post(url, params=None, data=None, timeout=None, headers=None):
sent.update(params or {})
return _Resp()
monkeypatch.setitem(__import__("sys").modules, "requests", _Req)
b = sb._RemoteNemoBackend("canary", {}, "http://x")
import tempfile, os
fd, path = tempfile.mkstemp()
os.close(fd)
try:
b.transcribe(path, language="pt-br", target_language="en-us")
finally:
os.unlink(path)
assert sent.get("language") == "pt", sent
assert sent.get("target_language") == "en", sent
def test_compose_model_picker_accepts_a_regional_tag(monkeypatch):
import codai.api.compose as C
monkeypatch.setattr(C, "_stt_entry", lambda m: {"languages": ["english"]})
# Before: "en-gb" not in {"english"} -> the only word-timing model was
# disqualified and captions silently fell back to estimated timings.
assert C._stt_speaks("crisperwhisper", ["en-gb"]) is True
monkeypatch.setattr(C, "_stt_entry", lambda m: {"languages": ["italian"]})
assert C._stt_speaks("wav2vec2-it", ["en-gb"]) is False
tools/crisperwhisper_service.py
View file @ 3f9b2680
......@@ -17,6 +17,7 @@ import json
import os
import subprocess
import tempfile
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
......@@ -92,11 +93,37 @@ class _Engine:
"end": (ts[1] or 0.0) + offset})
return text.strip(), words
def _language_ok(self, language) -> bool:
"""Whether this checkpoint's tokenizer knows ``language``.
Whisper carries its table as TO_LANGUAGE_CODE (name -> code) and
LANGUAGES (code -> name); a value in either, in either direction, is
accepted by generate(). When the table can't be read, assume yes and
let the model decide."""
try:
from transformers.models.whisper.tokenization_whisper import (
LANGUAGES, TO_LANGUAGE_CODE)
except Exception:
return True
want = str(language).strip().lower()
return want in TO_LANGUAGE_CODE or want in LANGUAGES
def transcribe(self, wav_path: str, language: str = None, task: str = None) -> dict:
import soundfile as sf
gen = {"task": task or "transcribe"}
if language:
gen["language"] = language
# Belt and braces: the caller (codai.api.lang_codes) now sends the
# English name transformers wants, but this worker is reachable on its
# own and transformers RAISES on an unknown language rather than
# ignoring it -- which turned one odd tag into a 500 for the whole
# request. Probe the model's own table and drop the argument if it
# would be rejected, so the worst case is auto-detect.
if gen.get("language") and not self._language_ok(gen["language"]):
sys.stderr.write(
"crisperwhisper: language %r not recognised; auto-detecting\n"
% gen["language"])
gen.pop("language")
data, sr = sf.read(wav_path, dtype="float32")
if getattr(data, "ndim", 1) > 1:
data = data.mean(axis=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