Skip to content

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

Add character/environment profiles, 3D conversion, archive, multi-character dialog 

- Character profiles: extract from images/video or generate from text prompt;
  up to 6 profiles per image/video generation via IP-Adapter conditioning
- Environment profiles: same multi-slot system for scene/background conditioning
- Multi-character dialog for video: per-character TTS synthesis with voice profile
  support, sequential or manual timing, ffmpeg amix, Wav2Lip/SadTalker lip sync
- 2D↔3D conversion: image/video to stereo/anaglyph/depth/mesh and back;
  text/image to 3D model (GLB); turntable video rendering
- Generation archive: auto-save all outputs with configurable retention and
  hourly cleanup; browse/delete from web UI and API
- Voice profile enhancements: extract from video, PATCH update, GET by name;
  TTS endpoint now accepts voice_profile for F5-TTS cloning
- Image generation progress tracking via GET /v1/images/progress
- Web Studio: Profiles tab (Characters, Environments, Voices) with full managers;
  multi-slot character/environment selectors in all panels; character dialog
  section in all video panels; archive settings in Settings page
- Fix: Profiles tab sub-tab switching now correctly updates the content panel
Co-Authored-By: 's avatarClaude Sonnet 4.6 <noreply@anthropic.com>
parent 19e28d3d
Show whitespace changes
Inline Side-by-side
Showing with 5447 additions and 102 deletions
README.md
View file @ cb542996
......@@ -42,6 +42,9 @@ An OpenAI-compatible API server to run models on your local GPU with web adminis
- **Face Swap**: InsightFace INSwapper — swap faces in images and videos
- **Depth Estimation**: Monocular depth maps
- **Segmentation**: SAM-based object segmentation
- **Character Profiles**: Apply up to 6 named character profiles (IP-Adapter) to anchor appearance across generations
- **Environment Profiles**: Apply up to 6 named environment profiles to condition scene/background style
- **Generation Progress**: Real-time per-step progress tracking via `GET /v1/images/progress`
### Video Generation
- **Text-to-Video**: Generate video from text prompts
......@@ -52,6 +55,9 @@ An OpenAI-compatible API server to run models on your local GPU with web adminis
- **Upscaling**: Real-ESRGAN video upscaling
- **Subtitles**: Whisper transcription + optional translation + burn-in
- **Dubbing**: Transcribe → translate → TTS → replace audio track
- **Character Profiles**: Apply up to 6 saved character profiles for visual consistency
- **Environment Profiles**: Apply up to 6 saved environment profiles to condition the scene
- **Multi-Character Dialog**: Assign spoken lines to individual characters — each line picks a character profile, voice profile or TTS voice ID, and text; lines are TTS-synthesised, mixed with correct timing (sequential or manual), and lip-synced to the video via Wav2Lip or SadTalker
### Audio
- **Text-to-Speech**: Kokoro TTS with voice selection and speed control
......@@ -59,7 +65,35 @@ An OpenAI-compatible API server to run models on your local GPU with web adminis
- **Music/SFX Generation**: MusicGen, AudioGen, AudioLDM2
- **Voice Cloning**: F5-TTS zero-shot voice cloning from a reference audio clip
- **Voice Conversion (SVC)**: Seed-VC — converts timbre while preserving pitch, melody and expression; **singing mode** for music
- **Voice Profiles**: Save named voice profiles (reference audio + transcript) for reuse
- **Voice Profiles**: Save named voice profiles (reference audio + transcript) for reuse; voice profiles can now be extracted directly from video files and updated via PATCH
### Character Profiles
Named collections of reference images used to condition character appearance via IP-Adapter across any image or video generation. Up to 6 profiles can be selected per generation.
- **Extract from images/video**: Automatic face detection and cropping to build a reference set
- **Generate from text prompt**: Create reference images from a visual description (no source images needed) using any registered image model
- **Reuse across generations**: Select saved profiles by name in the API or web UI
- **CRUD API**: Create, list, view, patch (add/remove images), and delete profiles
### Environment Profiles
Named collections of reference images that condition scene background and environment style via IP-Adapter. Same multi-slot selection (up to 6) as character profiles.
- **Extract from images/video**: Selects the sharpest frames/images (no face crop — full image used)
- **Generate from text prompt**: Create reference images from a scene description using any registered image model
- **Reuse across generations**: Select saved profiles by name in image and video generation
### 2D ↔ 3D Conversion
- **Image → 3D**: Convert any image to stereo pair, anaglyph, depth map, or mesh (GLB/OBJ) via depth estimation + optional 3D reconstruction
- **3D → Image**: Render a GLB/OBJ model to a 2D image from a specified viewpoint
- **Video → 3D**: Apply frame-by-frame depth processing to produce a 3D video
- **3D → Video**: Render a 3D model as a turntable rotation video
- **Text/Image → 3D Model**: Generate a 3D GLB model from a text prompt or reference image (requires a compatible 3D generation model)
### Generation Archive
- **Auto-save**: Every generated file (image, video, audio) is optionally saved to a configurable local directory
- **Configurable retention**: `1h`, `1d`, `2d`, `1w`, `1m`, `3m`, `6m`, `1y`, or `never` — automatic hourly cleanup
- **Browse & delete**: View and delete archived files from the web UI Archive tab or via the API
- **Settings page**: Archive directory and retention are configurable from the web UI Settings page without a restart
### Pipelines
Built-in multi-step pipelines callable from the API or web UI:
......@@ -221,10 +255,18 @@ Config files live in `~/.coderai/` (or `--config` path):
"backend": { "type": "auto" },
"models": { "default_load_mode": "ondemand" },
"offload": { "load_in_4bit": false, "flash_attention": false },
"vulkan": { "n_gpu_layers": -1, "n_ctx": 2048, "device_id": 0 }
"vulkan": { "n_gpu_layers": -1, "n_ctx": 2048, "device_id": 0 },
"archive": {
"enabled": true,
"directory": "",
"retention": "1w"
}
}
```
`archive.directory` — absolute path, or empty to use `<config_dir>/archive`.
`archive.retention` — one of: `1h`, `1d`, `2d`, `1w`, `1m`, `3m`, `6m`, `1y`, `never`.
### models.json
```json
......@@ -264,30 +306,75 @@ Config files live in `~/.coderai/` (or `--config` path):
| `POST /v1/images/faceswap` | Face swap (image or video) |
| `POST /v1/images/depth` | Depth estimation |
| `POST /v1/images/segment` | Object segmentation |
| `POST /v1/images/to3d` | Image → stereo / anaglyph / depth map / mesh |
| `POST /v1/images/from3d` | 3D model → rendered 2D image |
| `GET /v1/images/progress` | Current generation progress (step/total) |
### Video
| Endpoint | Description |
|---|---|
| `POST /v1/video/generations` | Generate video (t2v/i2v/v2v/ti2v/interp) |
| `POST /v1/video/generations` | Generate video (t2v/i2v/v2v/ti2v/interp) — supports `character_profiles`, `environment_profiles`, `dialogs` |
| `POST /v1/video/upscale` | Upscale video |
| `POST /v1/video/subtitle` | Generate/burn subtitles |
| `POST /v1/video/interpolate` | Frame interpolation |
| `POST /v1/video/dub` | Dub video to another language |
| `POST /v1/video/to3d` | Video → 3D video (frame-by-frame depth) |
| `POST /v1/video/from3d` | 3D model → turntable video |
### Audio
| Endpoint | Description |
|---|---|
| `POST /v1/audio/speech` | Text-to-speech |
| `POST /v1/audio/speech` | Text-to-speech (supports `voice_profile` for F5-TTS cloning) |
| `POST /v1/audio/transcriptions` | Speech-to-text (Whisper) |
| `POST /v1/audio/generate` | Music/SFX generation |
| `POST /v1/audio/clone` | Voice cloning TTS (F5-TTS) |
| `POST /v1/audio/convert` | Voice conversion / SVC (Seed-VC) |
| `GET /v1/audio/voices` | List saved voice profiles |
| `POST /v1/audio/voices` | Save a voice profile |
| `GET /v1/audio/voices/{name}` | Get a specific voice profile |
| `PATCH /v1/audio/voices/{name}` | Update a voice profile |
| `POST /v1/audio/voices/extract` | Extract voice profile from audio/video |
| `DELETE /v1/audio/voices/{name}` | Delete a voice profile |
### Character Profiles
| Endpoint | Description |
|---|---|
| `GET /v1/characters` | List all character profiles |
| `POST /v1/characters` | Save a character profile from images/videos |
| `POST /v1/characters/extract` | Extract a profile from source images/videos (face-crop + sharpness ranking) |
| `POST /v1/characters/generate` | Generate reference images from a text prompt and save as profile |
| `GET /v1/characters/{name}` | Get a character profile with images |
| `PATCH /v1/characters/{name}` | Update description or add/remove reference images |
| `DELETE /v1/characters/{name}` | Delete a character profile |
### Environment Profiles
| Endpoint | Description |
|---|---|
| `GET /v1/environments` | List all environment profiles |
| `POST /v1/environments` | Save an environment profile from images/videos |
| `POST /v1/environments/extract` | Extract a profile from source images/videos (sharpness ranking, full image) |
| `POST /v1/environments/generate` | Generate reference images from a scene description and save as profile |
| `GET /v1/environments/{name}` | Get an environment profile with images |
| `PATCH /v1/environments/{name}` | Update description or add/remove reference images |
| `DELETE /v1/environments/{name}` | Delete an environment profile |
### 3D Generation
| Endpoint | Description |
|---|---|
| `POST /v1/3d/generate` | Text or image → 3D model (GLB) |
### Archive
| Endpoint | Description |
|---|---|
| `GET /v1/archive` | List all archived generated files |
| `DELETE /v1/archive/{filename}` | Delete an archived file |
### Pipelines
| Endpoint | Description |
......@@ -304,6 +391,51 @@ Config files live in `~/.coderai/` (or `--config` path):
| `POST /v1/pipelines/run` | Run an inline pipeline definition |
| `GET /v1/pipelines/step-types` | List available step types |
### Character & Environment Profile Usage
Profiles are selected by name in any image or video generation request. Up to 6 of each type can be combined:
```json
{
"model": "wan-model",
"prompt": "Alice and Bob walking in a forest",
"character_profiles": ["Alice", "Bob"],
"character_strength": 0.8,
"environment_profiles": ["forest-summer"],
"environment_strength": 0.6
}
```
### Multi-Character Dialog
Add spoken dialog to a video generation. Each line specifies a character, a voice, and text. Lines are TTS-synthesised, timed, mixed, and lip-synced:
```json
{
"model": "wan-model",
"prompt": "Two characters having a conversation",
"character_profiles": ["Alice", "Bob"],
"lip_sync_method": "wav2lip",
"dialogs": [
{
"character": "Alice",
"voice": "alice-voice-profile",
"text": "Hello, how are you today?",
"lip_sync": true
},
{
"character": "Bob",
"voice": "en-US-GuyNeural",
"text": "I'm doing great, thanks for asking!",
"lip_sync": true,
"start_time": 3.5
}
]
}
```
Fields per dialog line: `character` (profile name for lip-sync face selection), `voice` (saved voice profile name or TTS voice ID), `text`, `start_time` (seconds; omit for sequential auto-timing), `lip_sync` (bool), `lang`, `speed`.
### Custom Pipeline Definition
```json
......@@ -339,7 +471,7 @@ Config files live in `~/.coderai/` (or `--config` path):
Template variables: `{{input}}`, `{{stepN.output}}`, `{{stepN.url}}`.
Available step types: `text_gen`, `image_gen`, `image_edit`, `image_inpaint`, `image_upscale`, `image_deblur`, `image_unpix`, `image_outfit`, `image_faceswap`, `video_gen`, `video_upscale`, `video_sub`, `video_interp`, `video_dub`, `tts`, `audio_gen`, `voice_clone`, `voice_convert`.
Available step types: `text_gen`, `image_gen`, `image_edit`, `image_inpaint`, `image_upscale`, `image_deblur`, `image_unpix`, `image_outfit`, `image_faceswap`, `image_to3d`, `video_gen`, `video_upscale`, `video_sub`, `video_interp`, `video_dub`, `video_to3d`, `tts`, `audio_gen`, `voice_clone`, `voice_convert`.
---
......@@ -456,3 +588,5 @@ Merge requests welcome.
- [F5-TTS](https://github.com/SWivid/F5-TTS) — voice cloning
- [Seed-VC](https://github.com/Plachta/Seed-VC) — singing voice conversion
- [Real-ESRGAN](https://github.com/xinntao/Real-ESRGAN) — image/video upscaling
- [Wav2Lip](https://github.com/Rudrabha/Wav2Lip) — audio-driven lip sync
- [SadTalker](https://github.com/OpenTalker/SadTalker) — talking head / lip sync generation
codai/admin/routes.py
View file @ cb542996
......@@ -1556,6 +1556,11 @@ async def api_get_settings(username: str = Depends(require_admin)):
"device_id": c.vulkan.device_id,
"single_gpu": c.vulkan.single_gpu,
},
"archive": {
"enabled": c.archive.enabled,
"directory": c.archive.directory,
"retention": c.archive.retention,
},
"system_prompt": c.system_prompt,
"tools_closer_prompt": c.tools_closer_prompt,
"grammar_guided": c.grammar_guided,
......@@ -1632,8 +1637,95 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
if "parser" in data:
c.parser = data["parser"]
if "archive" in data:
import os as _os
from codai.api.archive import archive_manager, RETENTION_OPTIONS
arc = data["archive"]
c.archive.enabled = bool(arc.get("enabled", c.archive.enabled))
raw_dir = (arc.get("directory") or "").strip()
c.archive.directory = raw_dir # store as-is (empty = default)
ret = arc.get("retention", c.archive.retention)
c.archive.retention = ret if ret in RETENTION_OPTIONS else "never"
# Resolve for live reconfiguration
cfg_dir = str(config_manager.config_dir)
resolved = raw_dir if raw_dir and _os.path.isabs(raw_dir) else _os.path.join(cfg_dir, raw_dir or "archive")
archive_manager.configure(c.archive.enabled, resolved, c.archive.retention)
config_manager.save_config()
return {"success": True}
# =============================================================================
# Archive management
# =============================================================================
@router.get("/admin/archive", response_class=HTMLResponse)
async def archive_page(request: Request, username: str = Depends(require_admin)):
return templates.TemplateResponse(request, "archive.html", {"username": username, "is_admin": True})
@router.get("/admin/api/archive")
async def api_archive_list(
limit: int = 50,
offset: int = 0,
username: str = Depends(require_admin),
):
from codai.api.archive import archive_manager
entries = archive_manager.list_entries(limit=limit, offset=offset)
total = archive_manager.count_entries()
return {"entries": entries, "total": total}
@router.get("/admin/api/archive/{gen_id}")
async def api_archive_get(gen_id: str, username: str = Depends(require_admin)):
from codai.api.archive import archive_manager
entry = archive_manager.get_entry(gen_id)
if entry is None:
raise HTTPException(status_code=404, detail="Entry not found")
return entry
@router.delete("/admin/api/archive/{gen_id}")
async def api_archive_delete(gen_id: str, username: str = Depends(require_admin)):
from codai.api.archive import archive_manager
if not archive_manager.delete_entry(gen_id):
raise HTTPException(status_code=404, detail="Entry not found")
return {"success": True}
@router.get("/admin/api/archive/{gen_id}/files/{filename}")
async def api_archive_file(
gen_id: str,
filename: str,
username: str = Depends(require_auth),
):
from fastapi.responses import FileResponse
from codai.api.archive import archive_manager
path = archive_manager.get_file_path(gen_id, filename)
if path is None:
raise HTTPException(status_code=404, detail="File not found")
import mimetypes
media_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
return FileResponse(path, media_type=media_type)
@router.get("/admin/api/archive-settings")
async def api_archive_settings_get(username: str = Depends(require_admin)):
if config_manager is None or config_manager.config is None:
raise HTTPException(status_code=503, detail="Config not ready")
import os as _os
c = config_manager.config.archive
from codai.api.archive import RETENTION_OPTIONS
default_dir = _os.path.join(str(config_manager.config_dir), "archive")
return {
"enabled": c.enabled,
"directory": c.directory,
"default_directory": default_dir,
"retention": c.retention,
"retention_options": RETENTION_OPTIONS,
}
# --- HuggingFace model search proxy ---
import re as _re
......@@ -1794,6 +1886,110 @@ async def api_hf_model_files(model_id: str, username: str = Depends(require_admi
return files
# =============================================================================
# Character profile management proxy (admin UI)
# =============================================================================
@router.get("/admin/api/characters")
async def api_list_characters(username: str = Depends(require_auth)):
from codai.api.characters import _list_characters
return {"characters": _list_characters()}
@router.get("/admin/api/characters/{name}")
async def api_get_character(name: str, username: str = Depends(require_auth)):
from codai.api.characters import _load_character_meta, _load_character_images
meta = _load_character_meta(name)
if not meta:
raise HTTPException(status_code=404, detail=f"Character '{name}' not found")
images = _load_character_images(name)
return {
"name": meta['name'],
"description": meta.get('description', ''),
"image_count": meta['image_count'],
"created_at": meta['created_at'],
"images": [img.model_dump() for img in images],
}
@router.delete("/admin/api/characters/{name}")
async def api_delete_character(name: str, username: str = Depends(require_auth)):
import os as _os, shutil
from codai.api.characters import _char_dir
cdir = _char_dir(name)
if not _os.path.isdir(cdir):
raise HTTPException(status_code=404, detail=f"Character '{name}' not found")
shutil.rmtree(cdir)
return {"ok": True, "name": name}
# =============================================================================
# Environment profile management proxy (admin UI)
# =============================================================================
@router.get("/admin/api/environments")
async def api_list_environments(username: str = Depends(require_auth)):
from codai.api.environments import _list_environments
return {"environments": _list_environments()}
@router.get("/admin/api/environments/{name}")
async def api_get_environment(name: str, username: str = Depends(require_auth)):
from codai.api.environments import _load_environment_meta, _load_environment_images
meta = _load_environment_meta(name)
if not meta:
raise HTTPException(status_code=404, detail=f"Environment '{name}' not found")
images = _load_environment_images(name)
return {
"name": meta['name'],
"description": meta.get('description', ''),
"image_count": meta['image_count'],
"created_at": meta['created_at'],
"images": [img.model_dump() for img in images],
}
@router.delete("/admin/api/environments/{name}")
async def api_delete_environment(name: str, username: str = Depends(require_auth)):
import os as _os, shutil
from codai.api.environments import _env_dir
edir = _env_dir(name)
if not _os.path.isdir(edir):
raise HTTPException(status_code=404, detail=f"Environment '{name}' not found")
shutil.rmtree(edir)
return {"ok": True, "name": name}
# =============================================================================
# Voice profile management proxy (admin UI)
# =============================================================================
@router.get("/admin/api/voices")
async def api_list_voices(username: str = Depends(require_auth)):
from codai.api.voice_clone import _list_voices
return {"voices": _list_voices()}
@router.get("/admin/api/voices/{name}")
async def api_get_voice(name: str, username: str = Depends(require_auth)):
from codai.api.voice_clone import _load_voice
meta = _load_voice(name)
if not meta:
raise HTTPException(status_code=404, detail=f"Voice '{name}' not found")
return {"voice": meta}
@router.delete("/admin/api/voices/{name}")
async def api_delete_voice(name: str, username: str = Depends(require_auth)):
import os as _os, shutil
from codai.api.voice_clone import _voice_path
vdir = _voice_path(name)
if not _os.path.exists(vdir):
raise HTTPException(status_code=404, detail=f"Voice '{name}' not found")
shutil.rmtree(vdir)
return {"deleted": True, "name": name}
@router.get("/admin/api/hf-model-info")
async def api_hf_model_info(model_id: str, username: str = Depends(require_admin)):
"""Full metadata for a single HuggingFace model repo."""
......
codai/admin/templates/archive.html 0 → 100644
View file @ cb542996
{% extends "base.html" %}
{% block title %}Archive — CoderAI{% endblock %}
{% block head %}
<style>
.archive-toolbar{display:flex;align-items:center;gap:.75rem;flex-wrap:wrap;margin-bottom:1.25rem}
.archive-filters{display:flex;align-items:center;gap:.5rem;flex:1;flex-wrap:wrap}
.filter-chip{padding:.25rem .75rem;border-radius:999px;border:1.5px solid var(--border);background:transparent;color:var(--text-muted);font-size:12px;font-weight:500;cursor:pointer;transition:.15s}
.filter-chip.active,.filter-chip:hover{border-color:var(--accent);color:var(--accent);background:rgba(99,102,241,.08)}
.archive-grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(200px,1fr));gap:1rem}
.archive-card{border-radius:10px;border:1.5px solid var(--border);background:var(--surface);overflow:hidden;cursor:pointer;transition:.15s;position:relative}
.archive-card:hover{border-color:var(--accent);transform:translateY(-2px);box-shadow:0 6px 24px rgba(0,0,0,.18)}
.archive-card:hover .card-overlay{opacity:1}
.card-thumb{width:100%;aspect-ratio:1/1;object-fit:cover;background:var(--surface-2);display:block}
.card-thumb-audio{width:100%;aspect-ratio:1/1;display:flex;align-items:center;justify-content:center;background:var(--surface-2);font-size:2.5rem}
.card-overlay{position:absolute;inset:0;background:rgba(0,0,0,.55);display:flex;align-items:center;justify-content:center;gap:.5rem;opacity:0;transition:.15s}
.card-meta{padding:.6rem .75rem}
.card-type{font-size:10px;font-weight:700;letter-spacing:.05em;text-transform:uppercase;color:var(--text-muted);margin-bottom:.2rem}
.card-prompt{font-size:12px;color:var(--text);line-height:1.4;overflow:hidden;display:-webkit-box;-webkit-line-clamp:2;-webkit-box-orient:vertical;word-break:break-word}
.card-date{font-size:11px;color:var(--text-muted);margin-top:.3rem}
.archive-empty{text-align:center;padding:4rem 1rem;color:var(--text-muted)}
.archive-empty-icon{font-size:3rem;opacity:.3;margin-bottom:.75rem}
/* Detail modal */
.detail-grid{display:grid;grid-template-columns:1fr 1fr;gap:1.5rem;align-items:start}
@media(max-width:640px){.detail-grid{grid-template-columns:1fr}}
.detail-preview{border-radius:8px;overflow:hidden;background:var(--surface-2);display:flex;align-items:center;justify-content:center;min-height:220px}
.detail-preview img,.detail-preview video{max-width:100%;max-height:400px;object-fit:contain;display:block}
.detail-info .kv{display:flex;gap:.5rem;margin-bottom:.5rem;font-size:13px;flex-wrap:wrap}
.detail-info .kv .k{font-weight:600;color:var(--text-muted);min-width:90px}
.detail-info .kv .v{color:var(--text);word-break:break-all}
.detail-info .prompt-box{background:var(--surface-2);border-radius:6px;padding:.75rem;font-size:13px;line-height:1.55;word-break:break-word;margin-bottom:.75rem}
.param-list{display:flex;flex-wrap:wrap;gap:.4rem;margin-top:.5rem}
.param-chip{font-size:11px;padding:.18rem .55rem;border-radius:4px;background:var(--surface-2);color:var(--text-muted);border:1px solid var(--border)}
.pagination{display:flex;align-items:center;gap:.4rem;justify-content:center;margin-top:1.5rem}
.page-btn{padding:.3rem .75rem;border-radius:6px;border:1.5px solid var(--border);background:transparent;color:var(--text-muted);font-size:13px;cursor:pointer;transition:.15s}
.page-btn:hover:not([disabled]){border-color:var(--accent);color:var(--accent)}
.page-btn[disabled]{opacity:.35;cursor:default}
.page-btn.active{background:var(--accent);border-color:var(--accent);color:#fff}
/* Settings panel */
.archive-settings-panel{background:var(--surface);border:1.5px solid var(--border);border-radius:10px;padding:1.25rem;margin-bottom:1.25rem}
</style>
{% endblock %}
{% block content %}
<div class="page-header">
<div>
<h1>Archive</h1>
<p>All generated images, videos, and audio</p>
</div>
<div class="header-actions">
<button class="btn" onclick="toggleSettings()" id="settings-btn">Settings</button>
</div>
</div>
<div id="settings-panel" class="archive-settings-panel" style="display:none">
<div class="card-title" style="margin-bottom:1rem">Archive Settings</div>
<div style="display:grid;grid-template-columns:1fr 1fr 1fr;gap:1rem;align-items:end">
<div class="form-row" style="margin:0">
<label class="form-label">Archive directory <span class="muted">(leave blank for default)</span></label>
<input type="text" id="arc-dir" class="form-input" placeholder="Loading…">
<span class="form-hint" id="arc-dir-hint"></span>
</div>
<div class="form-row" style="margin:0">
<label class="form-label">Auto-delete after</label>
<select id="arc-retention" class="form-input">
<option value="never">Never (keep forever)</option>
<option value="1h">1 hour</option>
<option value="1d">1 day</option>
<option value="2d">2 days</option>
<option value="1w">1 week</option>
<option value="1m">1 month</option>
<option value="3m">3 months</option>
<option value="6m">6 months</option>
<option value="1y">1 year</option>
</select>
</div>
<div class="form-row" style="margin:0">
<label class="form-label">Enabled</label>
<label style="display:flex;align-items:center;gap:.5rem;cursor:pointer;padding:.5rem 0">
<input type="checkbox" id="arc-enabled">
<span style="font-size:13px">Save generations to archive</span>
</label>
</div>
</div>
<div style="display:flex;gap:.75rem;margin-top:1rem;align-items:center">
<button class="btn btn-primary" onclick="saveArchiveSettings()">Save settings</button>
<span id="arc-save-status" class="muted small"></span>
</div>
</div>
<div class="archive-toolbar">
<div class="archive-filters" id="type-filters">
<button class="filter-chip active" onclick="setFilter('all',this)">All</button>
<button class="filter-chip" onclick="setFilter('image',this)">Images</button>
<button class="filter-chip" onclick="setFilter('video',this)">Video</button>
<button class="filter-chip" onclick="setFilter('audio',this)">Audio</button>
<button class="filter-chip" onclick="setFilter('tts',this)">TTS</button>
</div>
<span id="arc-count" class="muted small"></span>
</div>
<div id="archive-grid" class="archive-grid"></div>
<div id="archive-empty" class="archive-empty" style="display:none">
<div class="archive-empty-icon">&#128190;</div>
<p>No generations archived yet.</p>
<p class="small">Images, videos, and audio will appear here after generation.</p>
</div>
<div id="pagination" class="pagination" style="display:none"></div>
<!-- Detail modal -->
<div class="modal" id="detailModal" onclick="if(event.target===this)closeDetail()">
<div class="modal-box" style="max-width:780px;width:95%">
<div class="modal-head">
<span class="modal-title" id="detail-title">Generation Detail</span>
<button class="modal-close" onclick="closeDetail()">&times;</button>
</div>
<div class="modal-body" id="detail-body" style="padding:1.25rem"></div>
<div style="display:flex;gap:.75rem;padding:.75rem 1.25rem 1.25rem;border-top:1px solid var(--border)">
<a id="detail-download-btn" class="btn btn-primary" download>Download</a>
<button class="btn btn-danger" onclick="deleteEntry()">Delete</button>
<button class="btn" onclick="closeDetail()">Close</button>
</div>
</div>
</div>
{% endblock %}
{% block scripts %}
<script>
const PAGE_SIZE = 48;
let currentPage = 0;
let currentFilter = 'all';
let currentEntry = null;
let totalEntries = 0;
const TYPE_ICONS = {image:'🖼️', video:'🎬', audio:'🎵', tts:'🗣️'};
const TYPE_LABELS = {image:'Image', video:'Video', audio:'Audio', tts:'TTS'};
function setFilter(type, btn) {
currentFilter = type;
currentPage = 0;
document.querySelectorAll('.filter-chip').forEach(b => b.classList.remove('active'));
btn.classList.add('active');
loadArchive();
}
function toggleSettings() {
const p = document.getElementById('settings-panel');
p.style.display = p.style.display === 'none' ? '' : 'none';
}
function formatDate(ts) {
if (!ts) return '';
const d = new Date(ts * 1000);
return d.toLocaleString();
}
function shortPrompt(p, len=60) {
if (!p) return '(no prompt)';
return p.length > len ? p.slice(0, len) + '…' : p;
}
function previewUrl(entry) {
if (!entry.preview) return null;
return `/admin/api/archive/${entry.id}/files/${entry.preview}`;
}
function buildCard(entry) {
const div = document.createElement('div');
div.className = 'archive-card';
div.onclick = () => openDetail(entry);
const purl = previewUrl(entry);
let thumbHtml;
if (purl && (entry.type === 'image' || entry.type === 'video')) {
thumbHtml = `<img class="card-thumb" src="${purl}" loading="lazy" alt="">`;
} else if (purl && entry.type === 'video' && entry.preview?.endsWith('.webp')) {
thumbHtml = `<img class="card-thumb" src="${purl}" loading="lazy" alt="">`;
} else {
const icon = TYPE_ICONS[entry.type] || '📄';
thumbHtml = `<div class="card-thumb-audio">${icon}</div>`;
}
div.innerHTML = `
${thumbHtml}
<div class="card-overlay">
<span style="color:#fff;font-size:1.5rem">&#128269;</span>
</div>
<div class="card-meta">
<div class="card-type">${TYPE_LABELS[entry.type] || entry.type}</div>
<div class="card-prompt">${shortPrompt(entry.prompt)}</div>
<div class="card-date">${formatDate(entry.created_at)}</div>
</div>`;
return div;
}
async function loadArchive() {
const grid = document.getElementById('archive-grid');
const empty = document.getElementById('archive-empty');
const countEl = document.getElementById('arc-count');
grid.innerHTML = '<span class="muted small">Loading…</span>';
try {
const resp = await fetch(`/admin/api/archive?limit=${PAGE_SIZE}&offset=${currentPage * PAGE_SIZE}`);
const data = await resp.json();
totalEntries = data.total || 0;
let entries = data.entries || [];
if (currentFilter !== 'all') {
entries = entries.filter(e => e.type === currentFilter);
}
grid.innerHTML = '';
if (entries.length === 0) {
empty.style.display = '';
countEl.textContent = '';
} else {
empty.style.display = 'none';
countEl.textContent = `${totalEntries} total`;
entries.forEach(e => grid.appendChild(buildCard(e)));
}
renderPagination(totalEntries);
} catch(err) {
grid.innerHTML = `<span class="muted small">Failed to load: ${err.message}</span>`;
}
}
function renderPagination(total) {
const pages = Math.ceil(total / PAGE_SIZE);
const el = document.getElementById('pagination');
if (pages <= 1) { el.style.display = 'none'; return; }
el.style.display = 'flex';
el.innerHTML = '';
const prev = document.createElement('button');
prev.className = 'page-btn';
prev.textContent = '←';
prev.disabled = currentPage === 0;
prev.onclick = () => { currentPage--; loadArchive(); };
el.appendChild(prev);
for (let i = 0; i < pages; i++) {
const btn = document.createElement('button');
btn.className = 'page-btn' + (i === currentPage ? ' active' : '');
btn.textContent = i + 1;
btn.onclick = () => { currentPage = i; loadArchive(); };
el.appendChild(btn);
}
const next = document.createElement('button');
next.className = 'page-btn';
next.textContent = '→';
next.disabled = currentPage >= pages - 1;
next.onclick = () => { currentPage++; loadArchive(); };
el.appendChild(next);
}
function openDetail(entry) {
currentEntry = entry;
document.getElementById('detail-title').textContent =
`${TYPE_LABELS[entry.type] || entry.type} — ${formatDate(entry.created_at)}`;
const purl = previewUrl(entry);
let previewHtml = '';
if (purl) {
if (entry.type === 'video') {
const firstMp4 = (entry.files || []).find(f => f.endsWith('.mp4'));
const src = firstMp4 ? `/admin/api/archive/${entry.id}/files/${firstMp4}` : purl;
previewHtml = `<video controls style="max-width:100%;max-height:360px"><source src="${src}"></video>`;
} else if (entry.type === 'audio' || entry.type === 'tts') {
const firstAudio = (entry.files || []).find(f => /\.(wav|mp3|ogg|flac)$/.test(f));
if (firstAudio) {
previewHtml = `<audio controls style="width:100%"><source src="/admin/api/archive/${entry.id}/files/${firstAudio}"></audio>`;
} else {
previewHtml = `<div style="font-size:3rem;padding:2rem;text-align:center">${TYPE_ICONS[entry.type]}</div>`;
}
} else {
previewHtml = `<img src="${purl}" style="max-width:100%;max-height:360px;object-fit:contain" alt="">`;
}
} else {
previewHtml = `<div style="font-size:3rem;padding:2rem;text-align:center">${TYPE_ICONS[entry.type] || '📄'}</div>`;
}
// Build params
const params = entry.parameters || {};
const paramChips = Object.entries(params)
.filter(([k,v]) => v !== null && v !== undefined && v !== '')
.map(([k,v]) => `<span class="param-chip">${k}: ${typeof v === 'object' ? JSON.stringify(v) : v}</span>`)
.join('');
document.getElementById('detail-body').innerHTML = `
<div class="detail-grid">
<div class="detail-preview">${previewHtml}</div>
<div class="detail-info">
<div class="kv"><span class="k">Type</span><span class="v">${TYPE_LABELS[entry.type] || entry.type}</span></div>
<div class="kv"><span class="k">Model</span><span class="v">${entry.model || '—'}</span></div>
<div class="kv"><span class="k">Created</span><span class="v">${formatDate(entry.created_at)}</span></div>
<div class="kv"><span class="k">ID</span><span class="v" style="font-family:monospace;font-size:11px">${entry.id}</span></div>
${entry.prompt ? `<div style="margin:.75rem 0 .25rem;font-size:12px;font-weight:600;color:var(--text-muted)">Prompt</div>
<div class="prompt-box">${entry.prompt}</div>` : ''}
${paramChips ? `<div style="margin:.5rem 0 .25rem;font-size:12px;font-weight:600;color:var(--text-muted)">Parameters</div>
<div class="param-list">${paramChips}</div>` : ''}
${entry.files?.length > 1 ? `<div style="margin:.75rem 0 .25rem;font-size:12px;font-weight:600;color:var(--text-muted)">Files</div>
<div style="display:flex;flex-direction:column;gap:.3rem">${entry.files.map(f =>
`<a href="/admin/api/archive/${entry.id}/files/${f}" target="_blank" style="font-size:12px;color:var(--accent)">${f}</a>`
).join('')}</div>` : ''}
</div>
</div>`;
// Set download button
const dlBtn = document.getElementById('detail-download-btn');
const dlFile = entry.files?.[0];
if (dlFile) {
dlBtn.href = `/admin/api/archive/${entry.id}/files/${dlFile}`;
dlBtn.download = dlFile;
dlBtn.style.display = '';
} else {
dlBtn.style.display = 'none';
}
document.getElementById('detailModal').classList.add('show');
}
function closeDetail() {
document.getElementById('detailModal').classList.remove('show');
currentEntry = null;
}
async function deleteEntry() {
if (!currentEntry) return;
if (!confirm('Delete this generation from archive?')) return;
try {
const r = await fetch(`/admin/api/archive/${currentEntry.id}`, {method:'DELETE'});
if (!r.ok) throw new Error(await r.text());
closeDetail();
loadArchive();
} catch(e) {
alert('Delete failed: ' + e.message);
}
}
// Settings
async function loadArchiveSettings() {
try {
const d = await fetch('/admin/api/archive-settings').then(r=>r.json());
document.getElementById('arc-dir').value = d.directory || '';
document.getElementById('arc-dir').placeholder = d.default_directory || '<config_dir>/archive';
document.getElementById('arc-dir-hint').textContent = `Default: ${d.default_directory || '<config_dir>/archive'}`;
document.getElementById('arc-retention').value = d.retention || 'never';
document.getElementById('arc-enabled').checked = !!d.enabled;
} catch(e) {}
}
async function saveArchiveSettings() {
const status = document.getElementById('arc-save-status');
try {
const payload = {
archive: {
enabled: document.getElementById('arc-enabled').checked,
directory: document.getElementById('arc-dir').value.trim(),
retention: document.getElementById('arc-retention').value,
}
};
const r = await fetch('/admin/api/settings', {
method: 'POST',
headers: {'Content-Type':'application/json'},
body: JSON.stringify(payload),
});
if (!r.ok) throw new Error(await r.text());
status.textContent = 'Saved.';
setTimeout(() => status.textContent = '', 3000);
} catch(e) {
status.textContent = 'Error: ' + e.message;
}
}
loadArchive();
loadArchiveSettings();
</script>
{% endblock %}
codai/admin/templates/base.html
View file @ cb542996
......@@ -27,6 +27,7 @@
<a href="/admin/models" class="nav-link {% if '/models' in request.url.path %}active{% endif %}">Models</a>
<a href="/admin/tokens" class="nav-link {% if '/tokens' in request.url.path %}active{% endif %}">Tokens</a>
<a href="/admin/users" class="nav-link {% if '/users' in request.url.path %}active{% endif %}">Users</a>
<a href="/admin/archive" class="nav-link {% if '/archive' in request.url.path %}active{% endif %}">Archive</a>
<a href="/admin/settings" class="nav-link {% if '/settings' in request.url.path %}active{% endif %}">Settings</a>
{% endif %}
<button class="nav-link nav-donate-btn" onclick="document.getElementById('donateModal').classList.add('show')">Donate &#9829;</button>
......
codai/admin/templates/chat.html
View file @ cb542996
This source diff could not be displayed because it is too large. You can view the blob instead.
codai/admin/templates/settings.html
View file @ cb542996
......@@ -69,6 +69,38 @@
<span class="form-hint">Models will inherit this as default when configured</span>
</div>
</div>
<!-- Archive -->
<div class="card mb-0" style="margin-top:1rem">
<div class="card-title">Generation Archive</div>
<div class="form-row">
<label style="display:flex;align-items:center;gap:.5rem;cursor:pointer">
<input type="checkbox" id="s-arc-enabled">
<span style="font-size:13px;font-weight:500">Enable generation archive</span>
</label>
<span class="form-hint">When enabled, every generation (image, video, audio, text) is logged with its prompt, model, and output files.</span>
</div>
<div class="form-row">
<label class="form-label">Archive directory <span class="muted" id="s-arc-dir-hint"></span></label>
<input type="text" id="s-arc-dir" class="form-input" placeholder="(default)">
<span class="form-hint">Absolute path or relative to the config directory. Leave blank to use the default.</span>
</div>
<div class="form-row" style="margin:0">
<label class="form-label">Retention period</label>
<select id="s-arc-retention" class="form-input" style="max-width:200px">
<option value="1h">1 hour</option>
<option value="1d">1 day</option>
<option value="2d">2 days</option>
<option value="1w">1 week</option>
<option value="1m">1 month</option>
<option value="3m">3 months</option>
<option value="6m">6 months</option>
<option value="1y">1 year</option>
<option value="never">Never (keep forever)</option>
</select>
<span class="form-hint">Archived entries older than this are automatically deleted. Takes effect immediately on save.</span>
</div>
</div>
{% endblock %}
{% block scripts %}
......@@ -99,6 +131,18 @@ async function loadSettings(){
document.getElementById('s-gguf-cache').value = d.models?.gguf_cache_dir ?? '';
document.getElementById('s-offload-dir').value = d.offload?.directory ?? './offload';
toggleHttps();
// Archive
const arc = d.archive || {};
document.getElementById('s-arc-enabled').checked = !!arc.enabled;
document.getElementById('s-arc-dir').value = arc.directory ?? '';
document.getElementById('s-arc-retention').value = arc.retention ?? 'never';
// Show effective default dir as placeholder hint
try {
const as = await fetch('/admin/api/archive-settings').then(r=>r.json());
const hint = document.getElementById('s-arc-dir-hint');
if (hint && as.default_directory) hint.textContent = `(default: ${as.default_directory})`;
document.getElementById('s-arc-dir').placeholder = as.default_directory || '(default)';
} catch(_){}
}catch(e){ showAlert('error','Failed to load settings: '+e.message); }
}
......@@ -119,14 +163,19 @@ async function saveSettings(){
},
offload:{
directory: document.getElementById('s-offload-dir').value.trim() || './offload',
}
},
archive:{
enabled: document.getElementById('s-arc-enabled').checked,
directory: document.getElementById('s-arc-dir').value.trim(),
retention: document.getElementById('s-arc-retention').value,
},
};
try{
const r = await fetch('/admin/api/settings',{
method:'POST', headers:{'Content-Type':'application/json'},
body: JSON.stringify(data)
});
if(r.ok) showAlert('info','Settings saved. Restart CoderAI to apply.');
if(r.ok) showAlert('info','Settings saved. Archive changes take effect immediately; restart CoderAI for other changes.');
else{ const e=await r.json(); showAlert('error', e.detail||'Save failed'); }
}catch(e){ showAlert('error','Error: '+e.message); }
}
......
codai/api/app.py
View file @ cb542996
......@@ -66,8 +66,33 @@ def set_global_file_path_wrapper(path: str):
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Lifespan context manager for startup/shutdown."""
# Startup
import asyncio as _asyncio
async def _archive_cleanup_loop():
from codai.api.archive import archive_manager
while True:
await _asyncio.sleep(3600)
try:
archive_manager.run_cleanup()
except Exception:
pass
cleanup_task = _asyncio.create_task(_archive_cleanup_loop())
# Run initial cleanup on startup
try:
from codai.api.archive import archive_manager
archive_manager.run_cleanup()
except Exception:
pass
yield
cleanup_task.cancel()
try:
await cleanup_task
except _asyncio.CancelledError:
pass
# Shutdown
multi_model_manager.cleanup()
model_manager.cleanup()
......@@ -101,6 +126,8 @@ from codai.api.voice_clone import router as voice_clone_router
from codai.api.voice_convert import router as voice_convert_router
from codai.api.faceswap import router as faceswap_router
from codai.api.characters import router as characters_router
from codai.api.spatial import router as spatial_router
from codai.api.environments import router as environments_router
from codai.admin.routes import router as admin_router
# Import and add middleware
......@@ -133,6 +160,8 @@ app.include_router(voice_clone_router)
app.include_router(voice_convert_router)
app.include_router(faceswap_router)
app.include_router(characters_router)
app.include_router(environments_router)
app.include_router(spatial_router)
app.include_router(admin_router)
......@@ -167,3 +196,58 @@ async def get_file(filename: str):
if not os.path.isfile(candidate):
raise HTTPException(status_code=404, detail="File not found")
return FileResponse(candidate)
_IMAGE_EXTS = {'.png', '.jpg', '.jpeg', '.webp', '.gif'}
_VIDEO_EXTS = {'.mp4', '.webm', '.avi', '.mov'}
_AUDIO_EXTS = {'.wav', '.mp3', '.ogg', '.flac', '.aac', '.m4a'}
@app.get("/v1/archive")
async def list_archive():
"""List all generated files in the output directory."""
if not global_file_path or not os.path.isdir(global_file_path):
return {"files": []}
files = []
try:
names = os.listdir(global_file_path)
except OSError:
return {"files": []}
for fname in names:
fpath = os.path.join(global_file_path, fname)
if not os.path.isfile(fpath):
continue
ext = os.path.splitext(fname)[1].lower()
if ext in _IMAGE_EXTS:
ftype = 'image'
elif ext in _VIDEO_EXTS:
ftype = 'video'
elif ext in _AUDIO_EXTS:
ftype = 'audio'
else:
continue
stat = os.stat(fpath)
files.append({
'filename': fname,
'type': ftype,
'size': stat.st_size,
'created': stat.st_mtime,
'url': f'/v1/files/{fname}',
})
files.sort(key=lambda f: f['created'], reverse=True)
return {"files": files}
@app.delete("/v1/archive/{filename}")
async def delete_archive_file(filename: str):
"""Delete a generated file from the output directory."""
if not global_file_path:
raise HTTPException(status_code=404, detail="File not found")
safe_base = os.path.realpath(global_file_path)
candidate = os.path.realpath(os.path.join(global_file_path, filename))
if not (candidate == safe_base or candidate.startswith(safe_base + os.sep)):
raise HTTPException(status_code=403, detail="Access denied")
if not os.path.isfile(candidate):
raise HTTPException(status_code=404, detail="File not found")
os.remove(candidate)
return {"deleted": filename}
\ No newline at end of file
codai/api/archive.py 0 → 100644
View file @ cb542996
# 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/>.
"""Generation archive — persists all generated artifacts with metadata on disk."""
import io
import json
import logging
import os
import shutil
import threading
import time
import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
_log = logging.getLogger(__name__)
RETENTION_OPTIONS = ["1h", "1d", "2d", "1w", "1m", "3m", "6m", "1y", "never"]
_RETENTION_SECONDS: Dict[str, Optional[int]] = {
"1h": 3_600,
"1d": 86_400,
"2d": 172_800,
"1w": 604_800,
"1m": 2_592_000, # 30 days
"3m": 7_776_000, # 90 days
"6m": 15_552_000, # 180 days
"1y": 31_536_000, # 365 days
"never": None,
}
def _safe_id(s: str) -> bool:
return bool(s) and "/" not in s and ".." not in s and "\\" not in s
def _safe_filename(s: str) -> bool:
return bool(s) and "/" not in s and ".." not in s and "\\" not in s
class ArchiveManager:
"""Manages the on-disk generation archive."""
def __init__(self):
self.enabled: bool = True
self.directory: str = os.path.abspath("./archive")
self.retention: str = "never"
self._lock = threading.Lock()
def configure(self, enabled: bool, directory: str, retention: str):
self.enabled = enabled
self.directory = os.path.abspath(directory or "./archive")
self.retention = retention if retention in _RETENTION_SECONDS else "never"
def _gen_dir(self, gen_id: str) -> str:
return os.path.join(self.directory, gen_id)
# ------------------------------------------------------------------
# Save
# ------------------------------------------------------------------
def save_generation(
self,
gen_type: str,
endpoint: str,
model: Optional[str],
prompt: Optional[str],
params: Dict[str, Any],
artifacts: List[Tuple[bytes, str]], # [(raw_bytes, file_ext), ...]
) -> Optional[str]:
"""Save a generation to archive. Returns gen_id on success, None otherwise."""
if not self.enabled:
return None
try:
gen_id = f"{int(time.time())}_{uuid.uuid4().hex[:8]}"
gen_dir = self._gen_dir(gen_id)
os.makedirs(gen_dir, exist_ok=True)
saved_files: List[str] = []
preview_file: Optional[str] = None
for i, (data, ext) in enumerate(artifacts):
if not data:
continue
fname = f"output_{i}.{ext}"
with open(os.path.join(gen_dir, fname), "wb") as fh:
fh.write(data)
saved_files.append(fname)
if i == 0 and preview_file is None:
try:
preview_file = self._make_preview(data, ext, gen_dir)
except Exception as e:
_log.debug("Archive preview failed: %s", e)
meta: Dict[str, Any] = {
"id": gen_id,
"type": gen_type,
"endpoint": endpoint,
"model": model or "",
"prompt": prompt or "",
"parameters": params,
"files": saved_files,
"preview": preview_file or (saved_files[0] if saved_files else None),
"created_at": int(time.time()),
"created_at_iso": datetime.utcnow().isoformat() + "Z",
}
with open(os.path.join(gen_dir, "metadata.json"), "w") as fh:
json.dump(meta, fh, indent=2)
return gen_id
except Exception as e:
_log.warning("Archive save failed: %s", e)
return None
# ------------------------------------------------------------------
# Query
# ------------------------------------------------------------------
def list_entries(self, limit: int = 50, offset: int = 0) -> List[Dict]:
"""Return metadata dicts, newest first."""
if not os.path.isdir(self.directory):
return []
entries: List[Dict] = []
try:
for name in os.listdir(self.directory):
meta_path = os.path.join(self.directory, name, "metadata.json")
if not os.path.isfile(meta_path):
continue
try:
with open(meta_path) as fh:
entries.append(json.load(fh))
except Exception:
pass
except Exception:
pass
entries.sort(key=lambda x: x.get("created_at", 0), reverse=True)
return entries[offset: offset + limit]
def count_entries(self) -> int:
if not os.path.isdir(self.directory):
return 0
count = 0
try:
for name in os.listdir(self.directory):
if os.path.isfile(os.path.join(self.directory, name, "metadata.json")):
count += 1
except Exception:
pass
return count
def get_entry(self, gen_id: str) -> Optional[Dict]:
if not _safe_id(gen_id):
return None
meta_path = os.path.join(self._gen_dir(gen_id), "metadata.json")
if not os.path.isfile(meta_path):
return None
try:
with open(meta_path) as fh:
return json.load(fh)
except Exception:
return None
def delete_entry(self, gen_id: str) -> bool:
if not _safe_id(gen_id):
return False
gen_dir = self._gen_dir(gen_id)
if not os.path.isdir(gen_dir):
return False
try:
shutil.rmtree(gen_dir)
return True
except Exception:
return False
def get_file_path(self, gen_id: str, filename: str) -> Optional[str]:
if not _safe_id(gen_id) or not _safe_filename(filename):
return None
path = os.path.join(self._gen_dir(gen_id), filename)
return path if os.path.isfile(path) else None
# ------------------------------------------------------------------
# Cleanup
# ------------------------------------------------------------------
def run_cleanup(self):
"""Remove entries older than the configured retention window."""
max_age = _RETENTION_SECONDS.get(self.retention)
if max_age is None:
return
if not os.path.isdir(self.directory):
return
cutoff = time.time() - max_age
deleted = 0
try:
for name in list(os.listdir(self.directory)):
meta_path = os.path.join(self.directory, name, "metadata.json")
try:
with open(meta_path) as fh:
meta = json.load(fh)
if meta.get("created_at", time.time()) < cutoff:
shutil.rmtree(os.path.join(self.directory, name), ignore_errors=True)
deleted += 1
except Exception:
pass
except Exception:
pass
if deleted:
_log.info("Archive cleanup: removed %d entry/entries", deleted)
# ------------------------------------------------------------------
# Preview generation
# ------------------------------------------------------------------
def _make_preview(self, data: bytes, ext: str, gen_dir: str) -> Optional[str]:
ext = ext.lower()
if ext in ("png", "jpg", "jpeg", "webp", "bmp", "gif"):
from PIL import Image
img = Image.open(io.BytesIO(data)).convert("RGB")
img.thumbnail((512, 512))
dst = os.path.join(gen_dir, "preview.webp")
img.save(dst, "WEBP", quality=75)
return "preview.webp"
if ext in ("mp4", "webm", "avi", "mov"):
import subprocess
import tempfile
with tempfile.NamedTemporaryFile(suffix=f".{ext}", delete=False) as tmp:
tmp.write(data)
tmp_path = tmp.name
frame_path = os.path.join(gen_dir, "_frame_tmp.jpg")
try:
r = subprocess.run(
["ffmpeg", "-y", "-i", tmp_path, "-vframes", "1", "-q:v", "3", frame_path],
capture_output=True, timeout=15,
)
if r.returncode == 0 and os.path.exists(frame_path):
from PIL import Image
img = Image.open(frame_path).convert("RGB")
img.thumbnail((512, 512))
dst = os.path.join(gen_dir, "preview.webp")
img.save(dst, "WEBP", quality=75)
return "preview.webp"
except Exception:
pass
finally:
try:
os.unlink(tmp_path)
except Exception:
pass
try:
os.unlink(frame_path)
except Exception:
pass
# Audio or unrecognised — no visual preview
return None
# Module-level singleton
archive_manager = ArchiveManager()
codai/api/audio_gen.py
View file @ cb542996
......@@ -199,4 +199,24 @@ async def audio_generate(request: AudioGenerationRequest, http_request: Request
raise HTTPException(status_code=500, detail=f"Audio generation failed: {e}")
result = _save_audio_response(audio_bytes, ext, http_request)
try:
from codai.api.archive import archive_manager
asyncio.get_event_loop().create_task(asyncio.to_thread(
archive_manager.save_generation,
"audio", "/v1/audio/generate",
model_name,
request.prompt,
{
"duration": request.duration,
"top_k": request.top_k,
"top_p": request.top_p,
"temperature": request.temperature,
"cfg_coef": request.cfg_coef,
},
[(audio_bytes, ext)],
))
except Exception:
pass
return AudioGenerationResponse(created=int(time.time()), data=[result])
\ No newline at end of file
codai/api/characters.py
View file @ cb542996
......@@ -18,21 +18,26 @@
Character profile endpoints.
Saved character profiles are named collections of reference images used to
maintain visual consistency of a character across multiple video generations.
maintain visual consistency of a character across image/video generations.
POST /v1/characters – save / update a character profile
POST /v1/characters/extract – extract character from images and/or videos
GET /v1/characters – list all saved profiles (no images)
GET /v1/characters/{name} – get a profile including base64 images
PATCH /v1/characters/{name} – update description / add / remove images
DELETE /v1/characters/{name} – delete a profile
"""
import base64
import io
import json
import os
import subprocess
import tempfile
import time
from typing import List, Optional
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, ConfigDict
router = APIRouter()
......@@ -79,6 +84,34 @@ class CharacterSaveRequest(BaseModel):
model_config = ConfigDict(extra="allow")
class CharacterExtractRequest(BaseModel):
name: str
description: Optional[str] = ""
images: Optional[List[str]] = None # base64 source images
videos: Optional[List[str]] = None # base64/URL source videos
max_images: Optional[int] = 5 # max reference images to extract
model_config = ConfigDict(extra="allow")
class CharacterGenerateRequest(BaseModel):
name: str
description: Optional[str] = ""
prompt: str # visual description prompt
model: Optional[str] = None # image model to use
n: Optional[int] = 3 # number of reference images to generate
steps: Optional[int] = None # inference steps (None = model default)
width: Optional[int] = 512
height: Optional[int] = 512
model_config = ConfigDict(extra="allow")
class CharacterPatchRequest(BaseModel):
description: Optional[str] = None
add_images: Optional[List[CharacterImage]] = None
remove_indices: Optional[List[int]] = None # 0-based indices to remove
model_config = ConfigDict(extra="allow")
class CharacterProfile(BaseModel):
name: str
description: Optional[str] = ""
......@@ -163,6 +196,141 @@ def _list_characters() -> list:
return sorted(profiles, key=lambda p: p.get('created_at', 0))
def _decode_source(data: str) -> bytes:
"""Decode base64 or URL source into raw bytes."""
if data.startswith('data:'):
_, b64 = data.split(',', 1)
return base64.b64decode(b64)
if data.startswith('http://') or data.startswith('https://'):
import urllib.request
with urllib.request.urlopen(data, timeout=60) as r:
return r.read()
return base64.b64decode(data)
def _detect_faces_cv2(img_bytes: bytes):
"""Return list of (x,y,w,h) face rects using Haar cascade, or [] if cv2 unavailable."""
try:
import cv2
import numpy as np
arr = np.frombuffer(img_bytes, dtype=np.uint8)
img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
if img is None:
return []
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
cascade_path = cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
cascade = cv2.CascadeClassifier(cascade_path)
faces = cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=5, minSize=(40, 40))
if len(faces) == 0:
return []
return [(int(x), int(y), int(w), int(h)) for x, y, w, h in faces]
except Exception:
return []
def _crop_face(img_bytes: bytes, rect) -> Optional[bytes]:
"""Crop a face rect (with padding) from an image, return PNG bytes."""
try:
import cv2
import numpy as np
x, y, w, h = rect
arr = np.frombuffer(img_bytes, dtype=np.uint8)
img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
if img is None:
return None
ih, iw = img.shape[:2]
pad = int(max(w, h) * 0.4)
x1 = max(0, x - pad)
y1 = max(0, y - pad)
x2 = min(iw, x + w + pad)
y2 = min(ih, y + h + pad)
crop = img[y1:y2, x1:x2]
ok, buf = cv2.imencode('.png', crop)
return bytes(buf) if ok else None
except Exception:
return None
def _image_sharpness(img_bytes: bytes) -> float:
"""Return a sharpness score (Laplacian variance) for frame selection."""
try:
import cv2
import numpy as np
arr = np.frombuffer(img_bytes, dtype=np.uint8)
img = cv2.imdecode(arr, cv2.IMREAD_GRAYSCALE)
if img is None:
return 0.0
return float(cv2.Laplacian(img, cv2.CV_64F).var())
except Exception:
return 0.0
def _extract_from_image(img_bytes: bytes) -> List[bytes]:
"""Return list of PNG crops from a single source image (one per detected face, or whole image)."""
faces = _detect_faces_cv2(img_bytes)
if faces:
crops = [c for f in faces for c in [_crop_face(img_bytes, f)] if c]
if crops:
return crops
# No face detected — use whole image as reference
try:
from PIL import Image as PILImage
img = PILImage.open(io.BytesIO(img_bytes)).convert('RGB')
buf = io.BytesIO()
img.save(buf, 'PNG')
return [buf.getvalue()]
except Exception:
return [img_bytes]
def _extract_frames_from_video(video_bytes: bytes, max_frames: int = 10) -> List[bytes]:
"""Extract diverse frames from a video using ffmpeg, return PNG byte lists."""
with tempfile.TemporaryDirectory() as tmpdir:
in_path = os.path.join(tmpdir, 'input.mp4')
with open(in_path, 'wb') as f:
f.write(video_bytes)
# Get duration
probe = subprocess.run(
['ffprobe', '-v', 'error', '-show_entries', 'format=duration',
'-of', 'default=nw=1:nk=1', in_path],
capture_output=True, text=True, timeout=30,
)
try:
duration = float(probe.stdout.strip())
except (ValueError, AttributeError):
duration = 10.0
# Extract evenly-spaced frames
frames_dir = os.path.join(tmpdir, 'frames')
os.makedirs(frames_dir, exist_ok=True)
fps_target = max(1, max_frames) / max(1.0, duration)
subprocess.run(
['ffmpeg', '-y', '-i', in_path, '-vf', f'fps={fps_target:.4f}',
'-frames:v', str(max_frames * 3), os.path.join(frames_dir, '%04d.png')],
capture_output=True, timeout=120,
)
frame_files = sorted(os.listdir(frames_dir))
frames = []
for fname in frame_files:
fpath = os.path.join(frames_dir, fname)
with open(fpath, 'rb') as f:
frames.append(f.read())
return frames
def _select_best_frames(frames: List[bytes], max_count: int) -> List[bytes]:
"""Pick the sharpest frames, spaced across the video."""
if not frames:
return []
scored = sorted(enumerate(frames), key=lambda t: _image_sharpness(t[1]), reverse=True)
top = scored[:max(max_count * 2, 6)]
top.sort(key=lambda t: t[0]) # restore temporal order
step = max(1, len(top) // max_count)
return [top[i][1] for i in range(0, len(top), step)][:max_count]
def resolve_character_profiles(profile_names: List[str]) -> List[str]:
"""Resolve saved profile names → flat list of base64 image strings."""
out = []
......@@ -216,3 +384,199 @@ async def delete_character(name: str):
import shutil
shutil.rmtree(cdir)
return {"ok": True, "name": name}
@router.patch("/v1/characters/{name}")
async def patch_character(name: str, req: CharacterPatchRequest):
"""Update a character profile: description, add images, or remove images by index."""
meta = _load_character_meta(name)
if not meta:
raise HTTPException(status_code=404, detail=f"Character '{name}' not found")
cdir = _char_dir(name)
img_files = list(meta.get('images', []))
# Remove by index (highest first so indices stay valid)
if req.remove_indices:
for idx in sorted(set(req.remove_indices), reverse=True):
if 0 <= idx < len(img_files):
try:
os.unlink(os.path.join(cdir, img_files[idx]['file']))
except Exception:
pass
img_files.pop(idx)
# Add new images
if req.add_images:
next_i = len(img_files)
for img in req.add_images:
raw = img.data
if raw.startswith('data:'):
_, b64 = raw.split(',', 1)
else:
b64 = raw
img_bytes = base64.b64decode(b64)
ext = '.png' if img_bytes[:4] == b'\x89PNG' else '.jpg'
fname = f"ref{next_i:02d}{ext}"
fpath = os.path.join(cdir, fname)
with open(fpath, 'wb') as f:
f.write(img_bytes)
img_files.append({'file': fname, 'label': img.label or f'ref{next_i}'})
next_i += 1
if req.description is not None:
meta['description'] = req.description
meta['images'] = img_files
meta['image_count'] = len(img_files)
with open(os.path.join(cdir, 'meta.json'), 'w') as f:
json.dump(meta, f)
return {"ok": True, "name": name, "image_count": meta['image_count']}
@router.post("/v1/characters/generate")
async def generate_character(req: CharacterGenerateRequest, request: Request):
"""
Generate a character profile from a text prompt.
Calls the local image generation pipeline to produce `n` reference images,
then saves them as a named character profile.
"""
if not req.name or '/' in req.name or '..' in req.name:
raise HTTPException(status_code=400, detail="Invalid character name")
if not req.prompt or not req.prompt.strip():
raise HTTPException(status_code=400, detail="A visual description prompt is required")
n = max(1, min(8, req.n or 3))
payload: dict = {
"prompt": req.prompt,
"n": n,
"response_format": "b64_json",
"size": f"{req.width or 512}x{req.height or 512}",
}
if req.model:
payload["model"] = req.model
if req.steps:
payload["steps"] = req.steps
# Forward the caller's auth token so rate-limit / auth middleware passes
auth_header = request.headers.get("authorization", "")
headers = {"Content-Type": "application/json"}
if auth_header:
headers["Authorization"] = auth_header
try:
from httpx import AsyncClient, ASGITransport
async with AsyncClient(
transport=ASGITransport(app=request.app),
base_url="http://internal",
timeout=300,
) as client:
r = await client.post("/v1/images/generations", json=payload, headers=headers)
if not r.is_success:
try:
detail = r.json().get("detail", r.text)
except Exception:
detail = r.text
raise HTTPException(status_code=r.status_code, detail=f"Image generation failed: {detail}")
images_data = r.json().get("data", [])
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Image generation error: {e}")
if not images_data:
raise HTTPException(status_code=422, detail="No images were generated")
char_images: List[CharacterImage] = []
for i, item in enumerate(images_data):
b64 = item.get("b64_json") or item.get("data", "")
if not b64:
continue
if b64.startswith("data:"):
header, b64_pure = b64.split(",", 1)
mime = header.split(";")[0].split(":")[1] if ":" in header else "image/png"
else:
mime, b64_pure = "image/png", b64
char_images.append(CharacterImage(
label=f"generated_{i:02d}",
data=f"data:{mime};base64,{b64_pure}",
))
if not char_images:
raise HTTPException(status_code=422, detail="Generated images could not be decoded")
meta = _save_character(req.name, req.description or req.prompt, char_images)
return {"ok": True, "name": meta["name"], "image_count": meta["image_count"]}
@router.post("/v1/characters/extract")
async def extract_character(req: CharacterExtractRequest):
"""
Extract a character profile from source images and/or videos.
Images and videos are analysed for faces; the best crops are saved as
reference images for the named character profile.
"""
import asyncio
if not req.name or '/' in req.name or '..' in req.name:
raise HTTPException(status_code=400, detail="Invalid character name")
if not req.images and not req.videos:
raise HTTPException(status_code=400, detail="Provide at least one image or video")
max_per_source = max(1, (req.max_images or 5))
collected: List[bytes] = []
# --- images ---
if req.images:
for src in req.images:
try:
img_bytes = await asyncio.get_event_loop().run_in_executor(
None, _decode_source, src)
crops = await asyncio.get_event_loop().run_in_executor(
None, _extract_from_image, img_bytes)
collected.extend(crops[:max_per_source])
except Exception as e:
raise HTTPException(status_code=400, detail=f"Failed to process image: {e}")
# --- videos ---
if req.videos:
for src in req.videos:
try:
vid_bytes = await asyncio.get_event_loop().run_in_executor(
None, _decode_source, src)
frames = await asyncio.get_event_loop().run_in_executor(
None, _extract_frames_from_video, vid_bytes, max_per_source * 4)
best = await asyncio.get_event_loop().run_in_executor(
None, _select_best_frames, frames, max_per_source * 2)
face_crops: List[bytes] = []
for frame in best:
crops = await asyncio.get_event_loop().run_in_executor(
None, _extract_from_image, frame)
face_crops.extend(crops)
collected.extend(face_crops[:max_per_source])
except Exception as e:
raise HTTPException(status_code=400, detail=f"Failed to process video: {e}")
if not collected:
raise HTTPException(status_code=422, detail="No usable frames could be extracted")
# Deduplicate and trim
collected = collected[: req.max_images or 5]
# Build CharacterImage list from raw PNG bytes
char_images = []
for i, img_bytes in enumerate(collected):
b64 = base64.b64encode(img_bytes).decode()
mime = 'image/png' if img_bytes[:4] == b'\x89PNG' else 'image/jpeg'
char_images.append(CharacterImage(
label=f'extracted_{i:02d}',
data=f'data:{mime};base64,{b64}',
))
meta = _save_character(req.name, req.description or '', char_images)
return {"ok": True, "name": meta['name'], "image_count": meta['image_count']}
codai/api/environments.py 0 → 100644
View file @ cb542996
# 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/>.
"""
Environment profile endpoints.
Saved environment profiles are named collections of reference images that
describe a setting, background, or location used to maintain visual
consistency of an environment across image/video generations.
POST /v1/environments – save / update an environment profile
POST /v1/environments/extract – extract from images and/or videos
POST /v1/environments/generate – generate from a text prompt
GET /v1/environments – list all saved profiles (no images)
GET /v1/environments/{name} – get a profile including base64 images
PATCH /v1/environments/{name} – update description / add / remove images
DELETE /v1/environments/{name} – delete a profile
"""
import base64
import io
import json
import os
import subprocess
import tempfile
import time
from typing import List, Optional
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, ConfigDict
router = APIRouter()
_ENVS_DIR: Optional[str] = None
def set_global_args(args):
global _ENVS_DIR
base = getattr(args, 'file_path', None) or os.path.expanduser('~/.coderai')
root = base if os.path.isdir(base) else (os.path.dirname(base) if base else os.path.expanduser('~/.coderai'))
_ENVS_DIR = os.path.join(root, 'environments')
os.makedirs(_ENVS_DIR, exist_ok=True)
def set_global_file_path(path: str):
pass # not needed for environments
def _envs_dir() -> str:
if _ENVS_DIR:
return _ENVS_DIR
d = os.path.expanduser('~/.coderai/environments')
os.makedirs(d, exist_ok=True)
return d
def _env_dir(name: str) -> str:
return os.path.join(_envs_dir(), name)
# ── Pydantic models ───────────────────────────────────────────────────────────
class EnvironmentImage(BaseModel):
label: Optional[str] = None
data: str # base64 image (with or without data: prefix)
model_config = ConfigDict(extra="allow")
class EnvironmentSaveRequest(BaseModel):
name: str
description: Optional[str] = ""
images: List[EnvironmentImage]
model_config = ConfigDict(extra="allow")
class EnvironmentExtractRequest(BaseModel):
name: str
description: Optional[str] = ""
images: Optional[List[str]] = None # base64 source images
videos: Optional[List[str]] = None # base64/URL source videos
max_images: Optional[int] = 5
model_config = ConfigDict(extra="allow")
class EnvironmentGenerateRequest(BaseModel):
name: str
description: Optional[str] = ""
prompt: str
model: Optional[str] = None
n: Optional[int] = 3
steps: Optional[int] = None
width: Optional[int] = 512
height: Optional[int] = 512
model_config = ConfigDict(extra="allow")
class EnvironmentPatchRequest(BaseModel):
description: Optional[str] = None
add_images: Optional[List[EnvironmentImage]] = None
remove_indices: Optional[List[int]] = None
model_config = ConfigDict(extra="allow")
# ── Helpers ───────────────────────────────────────────────────────────────────
def _save_environment(name: str, description: str, images: List[EnvironmentImage]) -> dict:
edir = _env_dir(name)
os.makedirs(edir, exist_ok=True)
img_files = []
for i, img in enumerate(images):
raw = img.data
if raw.startswith('data:'):
_, b64 = raw.split(',', 1)
else:
b64 = raw
img_bytes = base64.b64decode(b64)
ext = '.png' if img_bytes[:4] == b'\x89PNG' else '.jpg'
fname = f"ref{i:02d}{ext}"
fpath = os.path.join(edir, fname)
with open(fpath, 'wb') as f:
f.write(img_bytes)
img_files.append({'file': fname, 'label': img.label or f'ref{i}'})
meta = {
'name': name,
'description': description,
'images': img_files,
'image_count': len(img_files),
'created_at': int(time.time()),
}
with open(os.path.join(edir, 'meta.json'), 'w') as f:
json.dump(meta, f)
return meta
def _load_environment_meta(name: str) -> Optional[dict]:
meta_path = os.path.join(_env_dir(name), 'meta.json')
if not os.path.exists(meta_path):
return None
with open(meta_path) as f:
return json.load(f)
def _load_environment_images(name: str) -> List[EnvironmentImage]:
meta = _load_environment_meta(name)
if not meta:
return []
edir = _env_dir(name)
result = []
for img_info in meta.get('images', []):
fpath = os.path.join(edir, img_info['file'])
if not os.path.exists(fpath):
continue
with open(fpath, 'rb') as f:
raw = f.read()
ext = img_info['file'].rsplit('.', 1)[-1]
mime = 'image/png' if ext == 'png' else 'image/jpeg'
b64 = base64.b64encode(raw).decode()
result.append(EnvironmentImage(
label=img_info.get('label'),
data=f"data:{mime};base64,{b64}",
))
return result
def _list_environments() -> list:
d = _envs_dir()
profiles = []
for entry in os.scandir(d):
if entry.is_dir():
meta = _load_environment_meta(entry.name)
if meta:
profiles.append({k: v for k, v in meta.items() if k != 'images'})
return sorted(profiles, key=lambda p: p.get('created_at', 0))
def _decode_source(data: str) -> bytes:
if data.startswith('data:'):
_, b64 = data.split(',', 1)
return base64.b64decode(b64)
if data.startswith('http://') or data.startswith('https://'):
import urllib.request
with urllib.request.urlopen(data, timeout=60) as r:
return r.read()
return base64.b64decode(data)
def _image_sharpness(img_bytes: bytes) -> float:
try:
import cv2
import numpy as np
arr = np.frombuffer(img_bytes, dtype=np.uint8)
img = cv2.imdecode(arr, cv2.IMREAD_GRAYSCALE)
if img is None:
return 0.0
return float(cv2.Laplacian(img, cv2.CV_64F).var())
except Exception:
return 0.0
def _normalise_image(img_bytes: bytes) -> bytes:
"""Convert any source image to a consistent PNG, resized to max 1024px on the long edge."""
try:
from PIL import Image as PILImage
img = PILImage.open(io.BytesIO(img_bytes)).convert('RGB')
img.thumbnail((1024, 1024), PILImage.LANCZOS)
buf = io.BytesIO()
img.save(buf, 'PNG')
return buf.getvalue()
except Exception:
return img_bytes
def _extract_frames_from_video(video_bytes: bytes, max_frames: int = 10) -> List[bytes]:
with tempfile.TemporaryDirectory() as tmpdir:
in_path = os.path.join(tmpdir, 'input.mp4')
with open(in_path, 'wb') as f:
f.write(video_bytes)
probe = subprocess.run(
['ffprobe', '-v', 'error', '-show_entries', 'format=duration',
'-of', 'default=nw=1:nk=1', in_path],
capture_output=True, text=True, timeout=30,
)
try:
duration = float(probe.stdout.strip())
except (ValueError, AttributeError):
duration = 10.0
frames_dir = os.path.join(tmpdir, 'frames')
os.makedirs(frames_dir, exist_ok=True)
fps_target = max(1, max_frames) / max(1.0, duration)
subprocess.run(
['ffmpeg', '-y', '-i', in_path, '-vf', f'fps={fps_target:.4f}',
'-frames:v', str(max_frames * 3), os.path.join(frames_dir, '%04d.png')],
capture_output=True, timeout=120,
)
frame_files = sorted(os.listdir(frames_dir))
frames = []
for fname in frame_files:
fpath = os.path.join(frames_dir, fname)
with open(fpath, 'rb') as f:
frames.append(f.read())
return frames
def _select_best_frames(frames: List[bytes], max_count: int) -> List[bytes]:
if not frames:
return []
scored = sorted(enumerate(frames), key=lambda t: _image_sharpness(t[1]), reverse=True)
top = scored[:max(max_count * 2, 6)]
top.sort(key=lambda t: t[0])
step = max(1, len(top) // max_count)
return [top[i][1] for i in range(0, len(top), step)][:max_count]
def resolve_environment_profiles(profile_names: List[str]) -> List[str]:
"""Resolve saved profile names → flat list of base64 image strings."""
out = []
for name in profile_names:
for img in _load_environment_images(name):
out.append(img.data)
return out
# ── Endpoints ─────────────────────────────────────────────────────────────────
@router.post("/v1/environments")
async def save_environment(req: EnvironmentSaveRequest):
"""Save or update a named environment profile."""
if not req.name or '/' in req.name or '..' in req.name:
raise HTTPException(status_code=400, detail="Invalid environment name")
if not req.images:
raise HTTPException(status_code=400, detail="At least one reference image required")
meta = _save_environment(req.name, req.description or '', req.images)
return {"ok": True, "name": meta['name'], "image_count": meta['image_count']}
@router.get("/v1/environments")
async def list_environments():
"""List all saved environment profiles (metadata only)."""
return {"environments": _list_environments()}
@router.get("/v1/environments/{name}")
async def get_environment(name: str):
"""Get an environment profile including its reference images as base64."""
meta = _load_environment_meta(name)
if not meta:
raise HTTPException(status_code=404, detail=f"Environment '{name}' not found")
images = _load_environment_images(name)
return {
"name": meta['name'],
"description": meta.get('description', ''),
"image_count": meta['image_count'],
"created_at": meta['created_at'],
"images": [img.model_dump() for img in images],
}
@router.delete("/v1/environments/{name}")
async def delete_environment(name: str):
"""Delete an environment profile."""
edir = _env_dir(name)
if not os.path.isdir(edir):
raise HTTPException(status_code=404, detail=f"Environment '{name}' not found")
import shutil
shutil.rmtree(edir)
return {"ok": True, "name": name}
@router.patch("/v1/environments/{name}")
async def patch_environment(name: str, req: EnvironmentPatchRequest):
"""Update an environment profile: description, add images, or remove images by index."""
meta = _load_environment_meta(name)
if not meta:
raise HTTPException(status_code=404, detail=f"Environment '{name}' not found")
edir = _env_dir(name)
img_files = list(meta.get('images', []))
if req.remove_indices:
for idx in sorted(set(req.remove_indices), reverse=True):
if 0 <= idx < len(img_files):
try:
os.unlink(os.path.join(edir, img_files[idx]['file']))
except Exception:
pass
img_files.pop(idx)
if req.add_images:
next_i = len(img_files)
for img in req.add_images:
raw = img.data
if raw.startswith('data:'):
_, b64 = raw.split(',', 1)
else:
b64 = raw
img_bytes = base64.b64decode(b64)
ext = '.png' if img_bytes[:4] == b'\x89PNG' else '.jpg'
fname = f"ref{next_i:02d}{ext}"
fpath = os.path.join(edir, fname)
with open(fpath, 'wb') as f:
f.write(img_bytes)
img_files.append({'file': fname, 'label': img.label or f'ref{next_i}'})
next_i += 1
if req.description is not None:
meta['description'] = req.description
meta['images'] = img_files
meta['image_count'] = len(img_files)
with open(os.path.join(edir, 'meta.json'), 'w') as f:
json.dump(meta, f)
return {"ok": True, "name": name, "image_count": meta['image_count']}
@router.post("/v1/environments/generate")
async def generate_environment(req: EnvironmentGenerateRequest, request: Request):
"""
Generate an environment profile from a text prompt.
Calls the local image generation pipeline to produce `n` reference images,
then saves them as a named environment profile.
"""
if not req.name or '/' in req.name or '..' in req.name:
raise HTTPException(status_code=400, detail="Invalid environment name")
if not req.prompt or not req.prompt.strip():
raise HTTPException(status_code=400, detail="A visual description prompt is required")
n = max(1, min(8, req.n or 3))
payload: dict = {
"prompt": req.prompt,
"n": n,
"response_format": "b64_json",
"size": f"{req.width or 512}x{req.height or 512}",
}
if req.model:
payload["model"] = req.model
if req.steps:
payload["steps"] = req.steps
auth_header = request.headers.get("authorization", "")
headers = {"Content-Type": "application/json"}
if auth_header:
headers["Authorization"] = auth_header
try:
from httpx import AsyncClient, ASGITransport
async with AsyncClient(
transport=ASGITransport(app=request.app),
base_url="http://internal",
timeout=300,
) as client:
r = await client.post("/v1/images/generations", json=payload, headers=headers)
if not r.is_success:
try:
detail = r.json().get("detail", r.text)
except Exception:
detail = r.text
raise HTTPException(status_code=r.status_code, detail=f"Image generation failed: {detail}")
images_data = r.json().get("data", [])
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Image generation error: {e}")
if not images_data:
raise HTTPException(status_code=422, detail="No images were generated")
env_images: List[EnvironmentImage] = []
for i, item in enumerate(images_data):
b64 = item.get("b64_json") or item.get("data", "")
if not b64:
continue
if b64.startswith("data:"):
header, b64_pure = b64.split(",", 1)
mime = header.split(";")[0].split(":")[1] if ":" in header else "image/png"
else:
mime, b64_pure = "image/png", b64
env_images.append(EnvironmentImage(
label=f"generated_{i:02d}",
data=f"data:{mime};base64,{b64_pure}",
))
if not env_images:
raise HTTPException(status_code=422, detail="Generated images could not be decoded")
meta = _save_environment(req.name, req.description or req.prompt, env_images)
return {"ok": True, "name": meta["name"], "image_count": meta["image_count"]}
@router.post("/v1/environments/extract")
async def extract_environment(req: EnvironmentExtractRequest):
"""
Extract an environment profile from source images and/or videos.
Whole frames are used as-is (no face cropping); the sharpest and most
diverse frames are selected as reference images for the profile.
"""
import asyncio
if not req.name or '/' in req.name or '..' in req.name:
raise HTTPException(status_code=400, detail="Invalid environment name")
if not req.images and not req.videos:
raise HTTPException(status_code=400, detail="Provide at least one image or video")
max_per_source = max(1, req.max_images or 5)
collected: List[bytes] = []
if req.images:
for src in req.images:
try:
img_bytes = await asyncio.get_event_loop().run_in_executor(
None, _decode_source, src)
norm = await asyncio.get_event_loop().run_in_executor(
None, _normalise_image, img_bytes)
collected.append(norm)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Failed to process image: {e}")
if req.videos:
for src in req.videos:
try:
vid_bytes = await asyncio.get_event_loop().run_in_executor(
None, _decode_source, src)
frames = await asyncio.get_event_loop().run_in_executor(
None, _extract_frames_from_video, vid_bytes, max_per_source * 4)
best = await asyncio.get_event_loop().run_in_executor(
None, _select_best_frames, frames, max_per_source)
normed = []
for frame in best:
n = await asyncio.get_event_loop().run_in_executor(
None, _normalise_image, frame)
normed.append(n)
collected.extend(normed)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Failed to process video: {e}")
if not collected:
raise HTTPException(status_code=422, detail="No usable frames could be extracted")
collected = collected[: req.max_images or 5]
env_images = []
for i, img_bytes in enumerate(collected):
b64 = base64.b64encode(img_bytes).decode()
mime = 'image/png' if img_bytes[:4] == b'\x89PNG' else 'image/jpeg'
env_images.append(EnvironmentImage(
label=f'extracted_{i:02d}',
data=f'data:{mime};base64,{b64}',
))
meta = _save_environment(req.name, req.description or '', env_images)
return {"ok": True, "name": meta['name'], "image_count": meta['image_count']}
codai/api/images.py
View file @ cb542996
......@@ -115,6 +115,23 @@ global_file_path = None
model_semaphores = {}
queue_flags = {}
# =============================================================================
# Generation progress tracking
# =============================================================================
_gen_progress: dict = {"current": 0, "total": 0, "active": False}
def _progress_reset(total: int):
_gen_progress["current"] = 0
_gen_progress["total"] = total
_gen_progress["active"] = True
def _progress_done():
_gen_progress["current"] = _gen_progress["total"]
_gen_progress["active"] = False
def _progress_step(step: int):
_gen_progress["current"] = step
# =============================================================================
# Helper Functions
......@@ -451,7 +468,7 @@ def _load_diffusers_pipeline(model_name: str, global_args):
return pipeline
def _generate_with_diffusers(pipeline, request, global_args, http_request=None):
async def _generate_with_diffusers(pipeline, request, global_args, http_request=None):
"""Generate images using a diffusers pipeline (with prompt-embedding cache)."""
import torch
import numpy as np
......@@ -498,6 +515,8 @@ def _generate_with_diffusers(pipeline, request, global_args, http_request=None):
getattr(global_args, 'image_cfg_scale', 7.5) if quality == "standard" else 9.0
)
_progress_reset(num_steps)
# ------------------------------------------------------------------
# Prompt embedding cache
# Try to encode the prompt once and reuse the embeddings.
......@@ -549,6 +568,22 @@ def _generate_with_diffusers(pipeline, request, global_args, http_request=None):
print(f"Warning: prompt encode/cache failed ({e}), using plain text prompt")
embed_kwargs = {}
def _step_cb(pipe, step_index, timestep, callback_kwargs):
_progress_step(step_index + 1)
return callback_kwargs
# Resolve character references (saved profiles + inline images)
char_images = []
try:
profiles = getattr(request, 'character_profiles', None) or []
if profiles:
from codai.api.characters import resolve_character_profiles
char_images += resolve_character_profiles(profiles)
inline = getattr(request, 'character_references', None) or []
char_images += list(inline)
except Exception:
pass
# Build call kwargs
if embed_kwargs:
call_kwargs = dict(
......@@ -558,6 +593,7 @@ def _generate_with_diffusers(pipeline, request, global_args, http_request=None):
generator=generator,
guidance_scale=cfg_scale,
num_inference_steps=num_steps,
callback_on_step_end=_step_cb,
**embed_kwargs,
)
else:
......@@ -570,9 +606,35 @@ def _generate_with_diffusers(pipeline, request, global_args, http_request=None):
generator=generator,
guidance_scale=cfg_scale,
num_inference_steps=num_steps,
callback_on_step_end=_step_cb,
)
result = pipeline(**call_kwargs)
# Inject IP-Adapter images if character references provided
if char_images and hasattr(pipeline, 'set_ip_adapter_scale'):
try:
strength = getattr(request, 'character_strength', 0.6) or 0.6
ref_imgs = []
for ref in char_images:
from PIL import Image as PILImage
if ref.startswith('data:'):
_, b64 = ref.split(',', 1)
raw = base64.b64decode(b64)
else:
raw = base64.b64decode(ref)
ref_imgs.append(PILImage.open(io.BytesIO(raw)).convert('RGB'))
pipeline.set_ip_adapter_scale(strength)
call_kwargs['ip_adapter_image'] = ref_imgs[0] if len(ref_imgs) == 1 else ref_imgs
except Exception as _ip_err:
print(f"Warning: IP-Adapter injection failed ({_ip_err}), continuing without character refs")
try:
result = await asyncio.to_thread(pipeline, **call_kwargs)
except TypeError:
# Older pipeline that doesn't support callback_on_step_end
call_kwargs.pop('callback_on_step_end', None)
result = await asyncio.to_thread(pipeline, **call_kwargs)
finally:
_progress_done()
# Extract images
images = []
......@@ -583,12 +645,44 @@ def _generate_with_diffusers(pipeline, request, global_args, http_request=None):
if result_images is None:
raise Exception(f"Could not extract images from diffusers result: {img_err}")
_archive_artifacts = []
for img in result_images:
if isinstance(img, np.ndarray):
img = np.nan_to_num(img, nan=0.0, posinf=1.0, neginf=0.0)
img = np.clip(img, 0.0, 1.0)
img_data = save_image_response(img, request.response_format, http_request)
images.append(img_data)
try:
_buf = io.BytesIO()
if isinstance(img, Image.Image):
img.convert("RGB").save(_buf, "PNG")
else:
Image.fromarray(img).convert("RGB").save(_buf, "PNG")
_archive_artifacts.append((_buf.getvalue(), "png"))
except Exception:
pass
if _archive_artifacts:
try:
from codai.api.archive import archive_manager
asyncio.create_task(asyncio.to_thread(
archive_manager.save_generation,
"image", "/v1/images/generations",
getattr(request, 'model', None) or model_id,
request.prompt,
{
"size": request.size,
"n": request.n,
"steps": num_steps,
"guidance_scale": cfg_scale,
"seed": seed,
"quality": quality,
"negative_prompt": neg_prompt or None,
},
_archive_artifacts,
))
except Exception:
pass
return {
"created": timestamp,
......@@ -613,11 +707,30 @@ async def _generate_with_sdcpp(sd_model, request, global_args, http_request=None
pass
# Use default steps for fast generation
steps = 4
steps = request.steps if request.steps else 4
_progress_reset(steps)
def _sdcpp_progress(step: int, total: int, elapsed: float):
_progress_step(step)
# Use request seed if provided, otherwise use CLI default seed
seed = request.seed if request.seed is not None else getattr(global_args, 'image_seed', None)
try:
result = await asyncio.to_thread(
sd_model.generate_image,
prompt=request.prompt,
negative_prompt='',
width=width,
height=height,
cfg_scale=get_cfg_scale(),
sample_steps=steps,
seed=seed if seed is not None else 42,
batch_count=request.n if request.n else 1,
progress_callback=_sdcpp_progress,
)
except TypeError:
result = await asyncio.to_thread(
sd_model.generate_image,
prompt=request.prompt,
......@@ -629,15 +742,47 @@ async def _generate_with_sdcpp(sd_model, request, global_args, http_request=None
seed=seed if seed is not None else 42,
batch_count=request.n if request.n else 1,
)
finally:
_progress_done()
# Small delay to let Vulkan driver settle after generation
time.sleep(0.1)
# Convert results to response format
images = []
_archive_artifacts = []
for img in result:
img_data = save_image_response(img, http_request=http_request)
images.append(img_data)
try:
_buf = io.BytesIO()
if isinstance(img, Image.Image):
img.convert("RGB").save(_buf, "PNG")
else:
Image.fromarray(img).convert("RGB").save(_buf, "PNG")
_archive_artifacts.append((_buf.getvalue(), "png"))
except Exception:
pass
if _archive_artifacts:
try:
import asyncio as _asyncio
from codai.api.archive import archive_manager
_asyncio.create_task(_asyncio.to_thread(
archive_manager.save_generation,
"image", "/v1/images/generations",
getattr(request, 'model', None),
request.prompt,
{
"size": request.size,
"n": request.n,
"steps": steps,
"seed": seed,
},
_archive_artifacts,
))
except Exception:
pass
return {
"created": int(time.time()),
......@@ -712,9 +857,15 @@ def _load_sdcpp_model(model_path: str, global_args, model_config: dict = None):
# Check if sd.cpp failed to identify the model architecture.
# In this case new_sd_ctx returns a non-null but broken context that
# will segfault on generate_image — reject it early.
# will segfault on free_sd_ctx — null out the pointer before raising so
# the destructor skips the free call and doesn't kill the server process.
failed_version = any('get sd version from file failed' in l for l in log_lines)
if failed_version:
try:
if hasattr(sd_model, '_model') and hasattr(sd_model._model, 'model'):
sd_model._model.model = None
except Exception:
pass
raise ValueError(
f"sd.cpp could not identify the model architecture in '{model_path}'. "
"This model may require a newer version of stable-diffusion-cpp-python, "
......@@ -731,6 +882,18 @@ def _load_sdcpp_model(model_path: str, global_args, model_config: dict = None):
router = APIRouter()
@router.get("/v1/images/progress")
async def get_image_progress():
"""Return current image generation step progress."""
return {
"current": _gen_progress["current"],
"total": _gen_progress["total"],
"active": _gen_progress["active"],
"pct": int(_gen_progress["current"] / _gen_progress["total"] * 100)
if _gen_progress["total"] > 0 else 0,
}
@router.post("/v1/images/generations")
async def create_image_generation(request: ImageGenerationRequest, http_request: Request = None):
"""
......@@ -832,7 +995,7 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
else:
# Assume it's a diffusers pipeline
print(f"Using cached diffusers pipeline for generation")
return _generate_with_diffusers(pipeline, request, global_args, http_request)
return await _generate_with_diffusers(pipeline, request, global_args, http_request)
# =====================================================================
# Step 4: Model not loaded - try to load it
......@@ -853,7 +1016,7 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
multi_model_manager.current_model_key = model_key
print(f"Loaded diffusers model: {model_name}")
return _generate_with_diffusers(pipeline, request, global_args, http_request)
return await _generate_with_diffusers(pipeline, request, global_args, http_request)
except ImportError as e:
diffusers_error = str(e)
......
codai/api/ratelimit.py
View file @ cb542996
......@@ -34,9 +34,11 @@ from starlette.middleware.base import BaseHTTPMiddleware
class BearerAuthMiddleware(BaseHTTPMiddleware):
"""Reject /v1/ API requests that lack a valid Bearer token or active web session."""
_EXEMPT_PATHS = {"/v1/images/progress"}
async def dispatch(self, request: Request, call_next):
path = request.url.path
if not path.startswith("/v1/"):
if not path.startswith("/v1/") or path in self._EXEMPT_PATHS:
return await call_next(request)
from codai.admin import routes as _admin_routes
......@@ -112,14 +114,20 @@ class RateLimitMiddleware(BaseHTTPMiddleware):
return prefix
return ""
# Lightweight polling endpoints that must never be rate-limited
_EXEMPT_PATHS = {"/v1/images/progress"}
async def dispatch(self, request: Request, call_next):
if not RATE_LIMITING_ENABLED:
return await call_next(request)
path = request.url.path
# Queue-size enforcement for authenticated API requests
if any(path.startswith(p) for p in _QUEUED_PREFIXES):
if path in self._EXEMPT_PATHS:
return await call_next(request)
# Queue-size enforcement for authenticated API requests (not for status polls)
if path not in self._EXEMPT_PATHS and any(path.startswith(p) for p in _QUEUED_PREFIXES):
from codai.queue.manager import queue_manager
if await queue_manager.is_full():
return JSONResponse(
......
codai/api/spatial.py 0 → 100644
View file @ cb542996
# 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/>.
"""
2D ↔ 3D conversion and 3D model generation endpoints.
Endpoints:
POST /v1/images/to3d – 2D image → stereo/anaglyph/depth-map/mesh (GLB)
POST /v1/images/from3d – 3D model (GLB/OBJ) → rendered 2D image
POST /v1/video/to3d – 2D video → 3D video (frame-by-frame)
POST /v1/video/from3d – 3D model → turntable video
POST /v1/3d/generate – text / image → 3D model (GLB)
"""
import asyncio
import base64
import io
import os
import subprocess
import tempfile
import time
import uuid
from typing import Optional
import numpy as np
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, ConfigDict
router = APIRouter()
global_args = None
global_file_path = None
def set_global_args(args):
global global_args
global_args = args
def set_global_file_path(path):
global global_file_path
global_file_path = path
# =============================================================================
# Shared helpers
# =============================================================================
def _decode_b64(data: str) -> bytes:
if data.startswith("data:"):
_, enc = data.split(",", 1)
return base64.b64decode(enc)
return base64.b64decode(data)
def _build_url(filename: str, http_request) -> str:
url_setting = getattr(global_args, 'url', 'auto') if global_args else 'auto'
if url_setting == 'auto':
host = (http_request.headers.get('host', '127.0.0.1') if http_request else '127.0.0.1')
if ':' in host:
parts = host.split(':')
if len(parts) == 2 and parts[1].isdigit():
host = parts[0]
use_https = getattr(global_args, 'https', False) or getattr(global_args, 'pubkey', None)
proto = 'https' if use_https else 'http'
port = getattr(global_args, 'port', 8000) if global_args else 8000
return f"{proto}://{host}:{port}/v1/files/{filename}"
return f"{url_setting.rstrip('/')}/v1/files/{filename}"
def _save_output(data: bytes, ext: str, http_request) -> dict:
filename = f"{uuid.uuid4().hex}.{ext}"
if global_file_path:
os.makedirs(global_file_path, exist_ok=True)
out_path = os.path.join(global_file_path, filename)
with open(out_path, 'wb') as f:
f.write(data)
return {"url": _build_url(filename, http_request)}
return {f"b64_{ext}": base64.b64encode(data).decode()}
def _derive_device() -> str:
if global_args:
for attr in ('image_vulkan_device', 'vulkan_device'):
d = getattr(global_args, attr, None)
if d is not None:
return f"cuda:{d}"
return "cuda:0"
def _img_to_bytes(img_pil, fmt: str = "PNG") -> bytes:
buf = io.BytesIO()
img_pil.save(buf, format=fmt)
return buf.getvalue()
# =============================================================================
# Depth estimation
# =============================================================================
def _estimate_depth(img_pil) -> np.ndarray:
"""Return depth array normalised to 0–1 (float32). Higher = closer."""
device = _derive_device()
# Prefer transformers depth-estimation pipeline
try:
from transformers import pipeline as hf_pipeline
pipe = hf_pipeline("depth-estimation", device=device)
result = pipe(img_pil)
depth = np.array(result['depth'], dtype=np.float32)
d_min, d_max = depth.min(), depth.max()
if d_max > d_min:
return (depth - d_min) / (d_max - d_min)
return np.zeros_like(depth)
except Exception:
pass
# MiDaS small fallback
try:
import torch
model = torch.hub.load("intel-isl/MiDaS", "MiDaS_small")
model.eval().to(device)
transforms = torch.hub.load("intel-isl/MiDaS", "transforms").small_transform
inp = transforms(np.array(img_pil)).to(device)
with torch.no_grad():
depth = model(inp).squeeze().cpu().numpy()
d_min, d_max = depth.min(), depth.max()
if d_max > d_min:
return (depth - d_min) / (d_max - d_min)
return np.zeros_like(depth)
except Exception as e:
raise RuntimeError(f"Depth estimation failed (no backend available): {e}")
# =============================================================================
# Stereo helpers
# =============================================================================
def _depth_to_stereo(img_pil, depth: np.ndarray, max_shift: int = 20):
"""Return (left_pil, right_pil) with horizontal parallax shift."""
import cv2
img_arr = np.array(img_pil)
h, w = img_arr.shape[:2]
depth_r = cv2.resize(depth, (w, h), interpolation=cv2.INTER_LINEAR)
shift_map = (depth_r * max_shift).astype(np.float32)
x_coords = np.arange(w, dtype=np.float32)
y_coords = np.arange(h, dtype=np.float32)
xx, yy = np.meshgrid(x_coords, y_coords)
map_x_l = (xx + shift_map).clip(0, w - 1)
map_x_r = (xx - shift_map).clip(0, w - 1)
left_arr = cv2.remap(img_arr, map_x_l, yy, cv2.INTER_LINEAR)
right_arr = cv2.remap(img_arr, map_x_r, yy, cv2.INTER_LINEAR)
from PIL import Image as PILImage
return PILImage.fromarray(left_arr), PILImage.fromarray(right_arr)
def _make_anaglyph(left_pil, right_pil):
"""Red-cyan anaglyph: red channel from left eye, green+blue from right eye."""
from PIL import Image as PILImage
left = np.array(left_pil.convert("RGB"), dtype=np.uint8)
right = np.array(right_pil.convert("RGB"), dtype=np.uint8)
ana = np.zeros_like(left)
ana[:, :, 0] = left[:, :, 0]
ana[:, :, 1] = right[:, :, 1]
ana[:, :, 2] = right[:, :, 2]
return PILImage.fromarray(ana)
# =============================================================================
# Depth → GLB mesh
# =============================================================================
def _depth_to_mesh_glb(img_pil, depth: np.ndarray, depth_scale: float = 0.3) -> bytes:
"""Build a textured grid mesh from depth + colour, export as GLB."""
import trimesh
import cv2
from PIL import Image as PILImage
img_arr = np.array(img_pil.convert("RGB"))
h, w = img_arr.shape[:2]
max_dim = 256
scale = min(max_dim / w, max_dim / h, 1.0)
nw, nh = max(int(w * scale), 2), max(int(h * scale), 2)
img_s = cv2.resize(img_arr, (nw, nh), interpolation=cv2.INTER_AREA)
depth_s = cv2.resize(depth, (nw, nh), interpolation=cv2.INTER_AREA)
ys, xs = np.mgrid[0:nh, 0:nw]
x_norm = (xs / (nw - 1)) * 2 - 1
y_norm = (ys / (nh - 1)) * 2 - 1
z = depth_s * depth_scale
vertices = np.column_stack([x_norm.ravel(), -y_norm.ravel(), z.ravel()])
faces = []
for row in range(nh - 1):
for col in range(nw - 1):
tl = row * nw + col
tr = tl + 1
bl = tl + nw
br = bl + 1
faces.append([tl, bl, tr])
faces.append([tr, bl, br])
faces = np.array(faces)
uvs = np.column_stack([xs.ravel() / (nw - 1), ys.ravel() / (nh - 1)])
texture_pil = PILImage.fromarray(img_s)
material = trimesh.visual.texture.SimpleMaterial(image=texture_pil)
visual = trimesh.visual.TextureVisuals(uv=uvs, material=material)
mesh = trimesh.Trimesh(vertices=vertices, faces=faces, visual=visual, process=False)
glb_buf = io.BytesIO()
mesh.export(glb_buf, file_type='glb')
return glb_buf.getvalue()
# =============================================================================
# 3D → 2D rendering
# =============================================================================
def _build_camera_pose(eye: np.ndarray) -> np.ndarray:
target = np.zeros(3)
up = np.array([0.0, 1.0, 0.0])
z_axis = (eye - target)
z_axis /= np.linalg.norm(z_axis)
x_axis = np.cross(up, z_axis)
x_axis_norm = np.linalg.norm(x_axis)
if x_axis_norm < 1e-6:
# Eye is directly above/below — use a different up
up = np.array([0.0, 0.0, 1.0])
x_axis = np.cross(up, z_axis)
x_axis_norm = np.linalg.norm(x_axis)
x_axis /= x_axis_norm
y_axis = np.cross(z_axis, x_axis)
pose = np.eye(4, dtype=np.float32)
pose[:3, 0] = x_axis
pose[:3, 1] = y_axis
pose[:3, 2] = z_axis
pose[:3, 3] = eye
return pose
def _render_mesh_to_png(mesh_bytes: bytes, mesh_fmt: str,
distance: float, elevation: float, azimuth: float,
width: int, height: int) -> bytes:
import trimesh
import pyrender
import math
mesh = trimesh.load(io.BytesIO(mesh_bytes), file_type=mesh_fmt)
if isinstance(mesh, trimesh.Scene):
mesh = mesh.dump(concatenate=True)
mesh.apply_translation(-mesh.centroid)
max_ext = max(mesh.extents) if max(mesh.extents) > 0 else 1.0
mesh.apply_scale(1.0 / max_ext)
scene = pyrender.Scene(bg_color=[0.12, 0.12, 0.14, 1.0])
scene.add(pyrender.Mesh.from_trimesh(mesh))
el = math.radians(elevation)
az = math.radians(azimuth)
eye = np.array([
distance * math.cos(el) * math.sin(az),
distance * math.sin(el),
distance * math.cos(el) * math.cos(az),
], dtype=np.float32)
pose = _build_camera_pose(eye)
camera = pyrender.PerspectiveCamera(yfov=np.pi / 3.0)
scene.add(camera, pose=pose)
scene.add(pyrender.DirectionalLight(color=[1, 1, 1], intensity=3.0), pose=pose)
renderer = pyrender.OffscreenRenderer(width, height)
color, _ = renderer.render(scene)
renderer.delete()
from PIL import Image as PILImage
return _img_to_bytes(PILImage.fromarray(color))
def _render_turntable_mp4(mesh_bytes: bytes, mesh_fmt: str,
n_frames: int, fps: int,
elevation: float, distance: float,
width: int, height: int) -> bytes:
import trimesh
import pyrender
import math
import imageio
mesh = trimesh.load(io.BytesIO(mesh_bytes), file_type=mesh_fmt)
if isinstance(mesh, trimesh.Scene):
mesh = mesh.dump(concatenate=True)
mesh.apply_translation(-mesh.centroid)
max_ext = max(mesh.extents) if max(mesh.extents) > 0 else 1.0
mesh.apply_scale(1.0 / max_ext)
renderer = pyrender.OffscreenRenderer(width, height)
scene = pyrender.Scene(bg_color=[0.12, 0.12, 0.14, 1.0])
scene.add(pyrender.Mesh.from_trimesh(mesh))
camera = pyrender.PerspectiveCamera(yfov=np.pi / 3.0)
light = pyrender.DirectionalLight(color=[1, 1, 1], intensity=3.0)
el = math.radians(elevation)
frames_list = []
for i in range(n_frames):
az = 2 * math.pi * i / n_frames
eye = np.array([
distance * math.cos(el) * math.sin(az),
distance * math.sin(el),
distance * math.cos(el) * math.cos(az),
], dtype=np.float32)
pose = _build_camera_pose(eye)
cam_node = scene.add(camera, pose=pose)
light_node = scene.add(light, pose=pose)
color, _ = renderer.render(scene)
scene.remove_node(cam_node)
scene.remove_node(light_node)
frames_list.append(color)
renderer.delete()
out_path = tempfile.mktemp(suffix='_3d_turntable.mp4')
imageio.mimsave(out_path, frames_list, fps=fps, codec='libx264', quality=8)
with open(out_path, 'rb') as f:
data = f.read()
os.unlink(out_path)
return data
# =============================================================================
# 3D model generation backends
# =============================================================================
def _generate_3d_triposr(image_bytes: bytes) -> bytes:
import torch
from PIL import Image as PILImage
from tsr.system import TSR
device = 'cuda' if torch.cuda.is_available() else 'cpu'
model = TSR.from_pretrained(
"stabilityai/TripoSR",
config_name="config.yaml",
weight_name="model.ckpt",
)
model = model.to(device)
img = PILImage.open(io.BytesIO(image_bytes)).convert("RGB")
with torch.no_grad():
scene_codes = model([img], device=device)
meshes = model.extract_mesh(scene_codes, resolution=256)
glb_buf = io.BytesIO()
meshes[0].export(glb_buf, file_type='glb')
return glb_buf.getvalue()
def _generate_3d_shape_e(prompt: Optional[str], image_bytes: Optional[bytes],
steps: int) -> bytes:
import torch
from shap_e.diffusion.sample import sample_latents
from shap_e.diffusion.gaussian_diffusion import diffusion_from_config
from shap_e.models.download import load_model, load_config
from shap_e.util.notebooks import decode_latent_mesh
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
xm = load_model('transmitter', device=device)
diffusion = diffusion_from_config(load_config('diffusion'))
if image_bytes:
from PIL import Image as PILImage
cond_img = PILImage.open(io.BytesIO(image_bytes)).convert("RGB")
model = load_model('image300M', device=device)
latents = sample_latents(
batch_size=1, model=model, diffusion=diffusion,
guidance_scale=3.0,
model_kwargs=dict(images=[cond_img]),
progress=True, clip_denoised=True, use_fp16=True,
use_karras=True, karras_steps=steps,
sigma_min=1e-3, sigma_max=160, s_churn=0,
)
else:
model = load_model('text300M', device=device)
latents = sample_latents(
batch_size=1, model=model, diffusion=diffusion,
guidance_scale=15.0,
model_kwargs=dict(texts=[prompt or "a 3D object"]),
progress=True, clip_denoised=True, use_fp16=True,
use_karras=True, karras_steps=steps,
sigma_min=1e-3, sigma_max=160, s_churn=0,
)
mesh = decode_latent_mesh(xm, latents[0]).tri_mesh()
glb_buf = io.BytesIO()
mesh.write_glb(glb_buf)
return glb_buf.getvalue()
def _generate_3d_from_depth(image_bytes: bytes) -> bytes:
"""Fallback: depth-based mesh from an image when no dedicated 3D model is available."""
from PIL import Image as PILImage
img = PILImage.open(io.BytesIO(image_bytes)).convert("RGB")
depth = _estimate_depth(img)
return _depth_to_mesh_glb(img, depth)
# =============================================================================
# Video 2D→3D frame processing
# =============================================================================
def _process_video_to_3d(video_bytes: bytes, method: str, max_shift: int) -> bytes:
import shutil
from PIL import Image as PILImage
temps = []
try:
in_path = tempfile.mktemp(suffix='.mp4')
temps.append(in_path)
with open(in_path, 'wb') as f:
f.write(video_bytes)
frames_dir = tempfile.mkdtemp()
temps.append(frames_dir)
subprocess.run(['ffmpeg', '-y', '-i', in_path, f'{frames_dir}/%08d.png'],
capture_output=True, check=True)
probe = subprocess.run(
['ffprobe', '-v', 'error', '-select_streams', 'v:0',
'-show_entries', 'stream=r_frame_rate', '-of', 'default=nw=1:nk=1', in_path],
capture_output=True, text=True)
fps_str = probe.stdout.strip() or '25/1'
num, den = fps_str.split('/')
fps = float(num) / float(den)
out_dir = tempfile.mkdtemp()
temps.append(out_dir)
for fname in sorted(os.listdir(frames_dir)):
fpath = os.path.join(frames_dir, fname)
img = PILImage.open(fpath).convert("RGB")
depth = _estimate_depth(img)
if method == "depth":
result_img = PILImage.fromarray((depth * 255).astype(np.uint8))
elif method == "anaglyph":
left, right = _depth_to_stereo(img, depth, max_shift)
result_img = _make_anaglyph(left, right)
else: # stereo
left, right = _depth_to_stereo(img, depth, max_shift)
w, h = left.size
result_img = PILImage.new("RGB", (w * 2, h))
result_img.paste(left, (0, 0))
result_img.paste(right, (w, 0))
result_img.save(os.path.join(out_dir, fname))
out_path = tempfile.mktemp(suffix='_3d.mp4')
temps.append(out_path)
subprocess.run(
['ffmpeg', '-y', '-framerate', str(fps), '-i', f'{out_dir}/%08d.png',
'-i', in_path, '-map', '0:v', '-map', '1:a?',
'-c:v', 'libx264', '-c:a', 'copy', '-shortest', out_path],
capture_output=True, check=True)
with open(out_path, 'rb') as f:
return f.read()
finally:
for t in temps:
try:
if os.path.isdir(t):
shutil.rmtree(t)
else:
os.unlink(t)
except Exception:
pass
# =============================================================================
# Endpoint: Image 2D → 3D
# =============================================================================
class ImageTo3DRequest(BaseModel):
image: str
method: Optional[str] = "stereo" # stereo | anaglyph | depth | mesh
max_shift: Optional[int] = 20
response_format: Optional[str] = "url"
model_config = ConfigDict(extra="allow")
@router.post("/v1/images/to3d")
async def image_to_3d(request: ImageTo3DRequest, http_request: Request = None):
"""Convert a 2D image to a 3D representation.
Methods:
stereo – side-by-side left/right pair (PNG)
anaglyph – red-cyan 3D image (PNG)
depth – normalised depth map (PNG)
mesh – textured 3D mesh (GLB) built from depth
"""
raw = _decode_b64(request.image)
from PIL import Image as PILImage
img = PILImage.open(io.BytesIO(raw)).convert("RGB")
method = request.method or "stereo"
try:
depth = await asyncio.get_event_loop().run_in_executor(None, _estimate_depth, img)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Depth estimation failed: {e}")
try:
if method == "depth":
out_img = PILImage.fromarray((depth * 255).astype(np.uint8))
result = _save_output(_img_to_bytes(out_img), "png", http_request)
elif method == "anaglyph":
left, right = await asyncio.get_event_loop().run_in_executor(
None, _depth_to_stereo, img, depth, request.max_shift or 20)
ana = await asyncio.get_event_loop().run_in_executor(
None, _make_anaglyph, left, right)
result = _save_output(_img_to_bytes(ana), "png", http_request)
elif method == "mesh":
glb = await asyncio.get_event_loop().run_in_executor(
None, _depth_to_mesh_glb, img, depth)
result = _save_output(glb, "glb", http_request)
else: # stereo (default)
left, right = await asyncio.get_event_loop().run_in_executor(
None, _depth_to_stereo, img, depth, request.max_shift or 20)
w, h = left.size
composite = PILImage.new("RGB", (w * 2, h))
composite.paste(left, (0, 0))
composite.paste(right, (w, 0))
result = _save_output(_img_to_bytes(composite), "png", http_request)
except Exception as e:
raise HTTPException(status_code=500, detail=f"2D→3D conversion failed: {e}")
return {"created": int(time.time()), "data": [result], "method": method}
# =============================================================================
# Endpoint: 3D model → 2D image
# =============================================================================
class ImageFrom3DRequest(BaseModel):
model_data: str # base64 GLB / OBJ
format: Optional[str] = "glb"
camera_distance: Optional[float] = 2.0
camera_elevation: Optional[float] = 30.0
camera_azimuth: Optional[float] = 45.0
width: Optional[int] = 512
height: Optional[int] = 512
response_format: Optional[str] = "url"
model_config = ConfigDict(extra="allow")
@router.post("/v1/images/from3d")
async def image_from_3d(request: ImageFrom3DRequest, http_request: Request = None):
"""Render a 3D model (GLB/OBJ) to a 2D PNG image from a specified camera angle."""
raw = _decode_b64(request.model_data)
try:
png_bytes = await asyncio.get_event_loop().run_in_executor(
None, _render_mesh_to_png, raw,
request.format or "glb",
request.camera_distance or 2.0,
request.camera_elevation or 30.0,
request.camera_azimuth or 45.0,
request.width or 512,
request.height or 512,
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"3D render failed: {e}")
result = _save_output(png_bytes, "png", http_request)
return {"created": int(time.time()), "data": [result]}
# =============================================================================
# Endpoint: Video 2D → 3D
# =============================================================================
class VideoTo3DRequest(BaseModel):
video: str
method: Optional[str] = "anaglyph" # stereo | anaglyph | depth
max_shift: Optional[int] = 15
response_format: Optional[str] = "url"
model_config = ConfigDict(extra="allow")
@router.post("/v1/video/to3d")
async def video_to_3d(request: VideoTo3DRequest, http_request: Request = None):
"""Convert a 2D video to a 3D video frame-by-frame.
Methods:
stereo – side-by-side left/right video
anaglyph – red-cyan 3D video
depth – greyscale depth video
"""
raw = _decode_b64(request.video)
method = request.method or "anaglyph"
try:
out_bytes = await asyncio.get_event_loop().run_in_executor(
None, _process_video_to_3d, raw, method, request.max_shift or 15)
except subprocess.CalledProcessError as e:
raise HTTPException(status_code=500, detail=f"ffmpeg error: {e.stderr.decode()[:200]}")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Video 2D→3D failed: {e}")
result = _save_output(out_bytes, "mp4", http_request)
return {"created": int(time.time()), "data": [result], "method": method}
# =============================================================================
# Endpoint: 3D model → turntable video
# =============================================================================
class VideoFrom3DRequest(BaseModel):
model_data: str # base64 GLB
format: Optional[str] = "glb"
frames: Optional[int] = 36
fps: Optional[int] = 12
camera_elevation: Optional[float] = 20.0
camera_distance: Optional[float] = 2.5
width: Optional[int] = 512
height: Optional[int] = 512
response_format: Optional[str] = "url"
model_config = ConfigDict(extra="allow")
@router.post("/v1/video/from3d")
async def video_from_3d(request: VideoFrom3DRequest, http_request: Request = None):
"""Render a 3D model as a 360° turntable video."""
raw = _decode_b64(request.model_data)
try:
mp4_bytes = await asyncio.get_event_loop().run_in_executor(
None, _render_turntable_mp4,
raw, request.format or "glb",
request.frames or 36, request.fps or 12,
request.camera_elevation or 20.0, request.camera_distance or 2.5,
request.width or 512, request.height or 512,
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"3D→video render failed: {e}")
result = _save_output(mp4_bytes, "mp4", http_request)
return {"created": int(time.time()), "data": [result]}
# =============================================================================
# Endpoint: Generate 3D model
# =============================================================================
class Generate3DRequest(BaseModel):
prompt: Optional[str] = None
image: Optional[str] = None
model: Optional[str] = None
steps: Optional[int] = 64
seed: Optional[int] = None
response_format: Optional[str] = "url"
model_config = ConfigDict(extra="allow")
@router.post("/v1/3d/generate")
async def generate_3d(request: Generate3DRequest, http_request: Request = None):
"""Generate a 3D model (GLB) from a text prompt and/or an image.
Backend priority:
1. TripoSR – image-to-3D mesh (best quality, requires image)
2. Shap-E – text or image to 3D latent mesh
3. Depth mesh – depth-map fallback (requires image, no ML 3D backend needed)
"""
if not request.prompt and not request.image:
raise HTTPException(status_code=400, detail="Either prompt or image is required")
image_bytes = _decode_b64(request.image) if request.image else None
# 1. TripoSR (image only)
if image_bytes:
try:
glb = await asyncio.get_event_loop().run_in_executor(
None, _generate_3d_triposr, image_bytes)
result = _save_output(glb, "glb", http_request)
return {"created": int(time.time()), "data": [result], "backend": "triposr"}
except (ImportError, ModuleNotFoundError):
pass
except Exception:
pass # fall through to Shap-E
# 2. Shap-E
try:
glb = await asyncio.get_event_loop().run_in_executor(
None, _generate_3d_shape_e, request.prompt, image_bytes, request.steps or 64)
result = _save_output(glb, "glb", http_request)
return {"created": int(time.time()), "data": [result], "backend": "shap-e"}
except (ImportError, ModuleNotFoundError):
pass
except Exception:
pass
# 3. Depth-based mesh fallback (image required)
if image_bytes:
try:
glb = await asyncio.get_event_loop().run_in_executor(
None, _generate_3d_from_depth, image_bytes)
result = _save_output(glb, "glb", http_request)
return {"created": int(time.time()), "data": [result], "backend": "depth-mesh"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"3D generation failed: {e}")
raise HTTPException(
status_code=500,
detail=(
"No 3D generation backend available. "
"Install one of: pip install tsr OR pip install shap-e. "
"For image-only mesh (no ML), provide an image — depth-based fallback will be used."
)
)
codai/api/tts.py
View file @ cb542996
......@@ -18,10 +18,12 @@
Text-to-speech endpoints for the codai API.
"""
import asyncio
import base64
import os
from typing import Optional
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, ConfigDict
# Import from codai modules
......@@ -51,6 +53,7 @@ class TTSRequest(BaseModel):
voice: str = "af_sarah"
response_format: str = "mp3"
speed: float = 1.0
voice_profile: Optional[str] = None # saved voice profile name (uses F5-TTS cloning)
model_config = ConfigDict(extra="allow")
......@@ -62,13 +65,39 @@ class TTSResponse(BaseModel):
@router.post("/v1/audio/speech")
async def create_speech(request: TTSRequest):
async def create_speech(request: TTSRequest, http_request: Request = None):
"""
Text-to-speech endpoint (OpenAI-compatible).
Supports:
- Kokoro TTS models (when --tts-model is specified)
"""
# If a voice profile is requested, delegate to voice cloning (F5-TTS)
if request.voice_profile:
from codai.api.voice_clone import _load_voice, _f5tts_clone
meta = _load_voice(request.voice_profile)
if not meta:
raise HTTPException(status_code=404,
detail=f"Voice profile '{request.voice_profile}' not found")
ref_audio_path = meta['audio_file']
ref_text = meta.get('transcript', '')
if not ref_text:
raise HTTPException(status_code=400,
detail="Voice profile has no transcript; update it with PATCH /v1/audio/voices/{name}")
try:
audio_bytes = await asyncio.get_event_loop().run_in_executor(
None, _f5tts_clone,
ref_audio_path, ref_text, request.input,
request.speed or 1.0, None,
)
except ImportError:
raise HTTPException(status_code=501,
detail="f5-tts not installed. Run: pip install f5-tts")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Voice cloning failed: {e}")
audio_base64 = base64.b64encode(audio_bytes).decode('utf-8')
return {"audio": audio_base64}
# Use the manager to resolve the model and manage VRAM
model_info = multi_model_manager.request_model(
requested_model=request.model,
......@@ -120,6 +149,19 @@ async def create_speech(request: TTSRequest):
audio_bytes = kokoro_model.generate(request.input, voice=voice, speed=speed)
try:
from codai.api.archive import archive_manager
asyncio.get_event_loop().create_task(asyncio.to_thread(
archive_manager.save_generation,
"tts", "/v1/audio/speech",
model_name,
request.input,
{"voice": voice, "speed": speed, "response_format": request.response_format},
[(audio_bytes, request.response_format or "mp3")],
))
except Exception:
pass
# Convert to base64
audio_base64 = base64.b64encode(audio_bytes).decode('utf-8')
......
codai/api/video.py
View file @ cb542996
......@@ -42,6 +42,7 @@ from codai.pydantic.videorequest import (
VideoGenerationRequest, VideoGenerationResponse,
VideoUpscaleRequest, VideoSubtitleRequest,
VideoInterpolateRequest, VideoDubRequest,
CharacterDialogLine,
)
from codai.api.images import _disable_safety_checker
......@@ -363,7 +364,7 @@ def _generate_video(pipe, request: VideoGenerationRequest):
def _postprocess_video(mp4_bytes: bytes, request: VideoGenerationRequest,
http_request, temp_paths: list) -> bytes:
"""Apply upscale / interpolation / audio steps to a raw mp4 blob."""
"""Apply upscale / interpolation / audio / dialog steps to a raw mp4 blob."""
path = _tmp_write(mp4_bytes, '.mp4')
temp_paths.append(path)
......@@ -376,6 +377,10 @@ def _postprocess_video(mp4_bytes: bytes, request: VideoGenerationRequest,
if request.add_audio:
path = _add_audio_to_video(path, request, temp_paths)
if request.dialogs:
path = _process_dialogs(path, request.dialogs,
request.lip_sync_method or 'wav2lip', temp_paths)
if request.generate_subtitles or request.burn_subtitles:
path = _add_subtitles(path, request, temp_paths)
......@@ -491,6 +496,220 @@ def _generate_tts(text: str, voice: Optional[str], speed: float,
return None
def _get_audio_duration(path: str) -> float:
"""Return audio/video duration in seconds via ffprobe."""
try:
r = subprocess.run(
['ffprobe', '-v', 'quiet', '-show_entries', 'format=duration',
'-of', 'default=noprint_wrappers=1:nokey=1', path],
capture_output=True, text=True)
return float(r.stdout.strip())
except Exception:
return 0.0
def _generate_tts_for_line(line: CharacterDialogLine, temps: list) -> Optional[str]:
"""Generate TTS for a single dialog line, using the voice profile's reference audio if available."""
voice = line.voice
text = line.text
speed = line.speed or 1.0
lang = line.lang
# Try to load voice profile reference audio first (for kokoro/RVC cloning)
ref_audio = None
if voice:
try:
from codai.api.voice_clone import _load_voice, _voice_path
meta = _load_voice(voice)
audio_file = meta.get('audio_file') or meta.get('audio_path')
if audio_file and os.path.isfile(audio_file):
ref_audio = audio_file
except Exception:
pass
# edge_tts with voice id
try:
import edge_tts, asyncio as _aio
voice_id = voice if (voice and not ref_audio) else (
f"{lang.split('-')[0]}-" if lang else 'en-'
) + 'US-JennyNeural'
if not voice or ref_audio:
voice_id = 'en-US-JennyNeural'
out = tempfile.mktemp(suffix='.mp3')
temps.append(out)
tts = edge_tts.Communicate(text, voice_id, rate=f"+{int((speed - 1) * 100)}%")
_aio.get_event_loop().run_until_complete(tts.save(out))
if os.path.exists(out) and os.path.getsize(out) > 0:
return out
except Exception:
pass
# kokoro with optional reference voice
try:
from kokoro import KPipeline
import soundfile as sf, numpy as np
lang_code = 'a'
if lang:
lang_code = lang.split('-')[0][:1].lower()
pipe = KPipeline(lang_code=lang_code)
kokoro_voice = voice if (voice and not ref_audio) else 'af_sky'
audio, sr = pipe(text, voice=kokoro_voice, speed=speed)
out = tempfile.mktemp(suffix='.wav')
temps.append(out)
sf.write(out, np.concatenate(audio), sr)
return out
except Exception:
pass
return None
def _mix_dialog_audio(clips: list, temps: list) -> Optional[str]:
"""
Mix a list of (start_time_sec, audio_path) clips into one audio file.
Uses ffmpeg adelay + amix. Returns path to mixed audio or None.
"""
if not clips:
return None
if len(clips) == 1:
return clips[0][1]
# Build complex filter: delay each stream, then amix
filter_parts = []
inputs = []
for i, (start_sec, apath) in enumerate(clips):
inputs += ['-i', apath]
delay_ms = int(start_sec * 1000)
filter_parts.append(f'[{i}]adelay={delay_ms}|{delay_ms}[a{i}]')
mix_inputs = ''.join(f'[a{i}]' for i in range(len(clips)))
filter_parts.append(f'{mix_inputs}amix=inputs={len(clips)}:duration=longest:dropout_transition=0[out]')
filter_str = ';'.join(filter_parts)
out = tempfile.mktemp(suffix='_mixed.wav')
temps.append(out)
cmd = ['ffmpeg', '-y'] + inputs + ['-filter_complex', filter_str, '-map', '[out]', out]
r = subprocess.run(cmd, capture_output=True)
if r.returncode == 0 and os.path.exists(out):
return out
# Fallback: simple concatenation (ignore timing)
import logging
logging.getLogger(__name__).warning(
"ffmpeg amix failed (rc=%d), falling back to concat: %s",
r.returncode, r.stderr.decode(errors='replace'))
list_path = tempfile.mktemp(suffix='.txt')
temps.append(list_path)
with open(list_path, 'w') as f:
for _, apath in clips:
f.write(f"file '{apath}'\n")
out2 = tempfile.mktemp(suffix='_cat.wav')
temps.append(out2)
r2 = subprocess.run(['ffmpeg', '-y', '-f', 'concat', '-safe', '0',
'-i', list_path, out2], capture_output=True)
return out2 if r2.returncode == 0 else None
def _apply_lipsync(video_path: str, audio_path: str, method: str, temps: list) -> str:
"""Apply lip sync to video using wav2lip or sadtalker. Returns new video path."""
import logging, shutil
_log = logging.getLogger(__name__)
out = tempfile.mktemp(suffix='_lipsync.mp4')
temps.append(out)
if method == 'wav2lip':
wav2lip_bin = shutil.which('wav2lip') or shutil.which('Wav2Lip')
if wav2lip_bin:
cmd = [wav2lip_bin, '--face', video_path, '--audio', audio_path, '--outfile', out]
r = subprocess.run(cmd, capture_output=True)
if r.returncode == 0 and os.path.exists(out):
return out
_log.warning("wav2lip failed (rc=%d): %s", r.returncode, r.stderr.decode(errors='replace'))
else:
# Try wav2lip Python API
try:
import inference as wav2lip_inference # noqa
wav2lip_inference.main(face=video_path, audio=audio_path, outfile=out)
if os.path.exists(out):
return out
except Exception as e:
_log.warning("wav2lip Python API failed: %s", e)
elif method == 'sadtalker':
sadtalker_bin = shutil.which('sadtalker')
if sadtalker_bin:
cmd = [sadtalker_bin, '--driven_audio', audio_path, '--source_video', video_path,
'--result_dir', os.path.dirname(out)]
r = subprocess.run(cmd, capture_output=True)
if r.returncode == 0:
# Find output file
out_dir = os.path.dirname(out)
for f in sorted(os.listdir(out_dir)):
if f.endswith('.mp4'):
return os.path.join(out_dir, f)
_log.warning("sadtalker failed (rc=%d): %s", r.returncode, r.stderr.decode(errors='replace'))
# Fallback: just mux audio onto video without lip sync
_log.warning("Lip sync unavailable (%s not found/working), merging audio only", method)
out_fallback = tempfile.mktemp(suffix='_nosync.mp4')
temps.append(out_fallback)
cmd = ['ffmpeg', '-y', '-i', video_path, '-i', audio_path,
'-c:v', 'copy', '-c:a', 'aac', '-shortest', out_fallback]
r = subprocess.run(cmd, capture_output=True)
return out_fallback if r.returncode == 0 else video_path
def _process_dialogs(path: str, dialogs: list, lip_sync_method: str, temps: list) -> str:
"""
Generate TTS for each dialog line, mix with correct timing, apply lip sync to full video.
Returns new video path.
"""
import logging
_log = logging.getLogger(__name__)
# First pass: generate TTS audio for each line and calculate timing
clips = [] # (start_sec, audio_path)
cursor = 0.0
for line in dialogs:
if not line.text.strip():
continue
audio_path = _generate_tts_for_line(line, temps)
if not audio_path or not os.path.exists(audio_path):
_log.warning("TTS generation failed for dialog line: %r", line.text[:40])
continue
if line.start_time is not None:
start = float(line.start_time)
else:
start = cursor
duration = _get_audio_duration(audio_path)
clips.append((start, audio_path))
cursor = start + duration + 0.1 # small gap between sequential lines
if not clips:
return path
# Mix all clips into one audio track
mixed_audio = _mix_dialog_audio(clips, temps)
if not mixed_audio or not os.path.exists(mixed_audio):
return path
# Determine if any line wants lip sync
wants_lip_sync = any(getattr(line, 'lip_sync', True) for line in dialogs if line.text.strip())
if wants_lip_sync and lip_sync_method:
return _apply_lipsync(path, mixed_audio, lip_sync_method, temps)
# No lip sync — just mux audio
out = tempfile.mktemp(suffix='_dialog.mp4')
temps.append(out)
cmd = ['ffmpeg', '-y', '-i', path, '-i', mixed_audio,
'-c:v', 'copy', '-c:a', 'aac', '-shortest', out]
r = subprocess.run(cmd, capture_output=True)
return out if r.returncode == 0 else path
def _add_subtitles(path: str, request: VideoGenerationRequest, temps: list) -> str:
"""Transcribe video audio → subtitles, optionally burn them in."""
try:
......@@ -635,6 +854,7 @@ async def video_generations(request: VideoGenerationRequest,
request.upscale_output,
request.interpolate_output,
request.add_audio,
bool(request.dialogs),
request.generate_subtitles,
request.burn_subtitles,
])
......@@ -653,6 +873,29 @@ async def video_generations(request: VideoGenerationRequest,
pass
result = _save_file(mp4_bytes, 'mp4', http_request)
try:
from codai.api.archive import archive_manager
asyncio.get_event_loop().create_task(asyncio.to_thread(
archive_manager.save_generation,
"video", "/v1/video/generations",
request.model,
request.prompt or "",
{
"mode": request.mode,
"num_frames": request.num_frames,
"fps": request.fps,
"width": request.width,
"height": request.height,
"num_inference_steps": request.num_inference_steps,
"guidance_scale": request.guidance_scale,
"seed": request.seed,
},
[(mp4_bytes, "mp4")],
))
except Exception:
pass
return VideoGenerationResponse(created=int(time.time()), data=[result])
......
codai/api/voice_clone.py
View file @ cb542996
......@@ -4,6 +4,8 @@ Voice cloning endpoints.
POST /v1/audio/clone — synthesize speech in a cloned voice
GET /v1/audio/voices — list saved voice profiles
POST /v1/audio/voices — save a named voice profile (ref audio + transcript)
POST /v1/audio/voices/extract — extract voice profile from audio or video file
PATCH /v1/audio/voices/{name} — update description / replace reference audio
DELETE /v1/audio/voices/{name} — delete a voice profile
"""
......@@ -12,6 +14,7 @@ import base64
import io
import json
import os
import subprocess
import tempfile
import time
from typing import Optional
......@@ -103,6 +106,17 @@ def _decode_audio(data: str) -> tuple[bytes, str]:
return base64.b64decode(data), '.wav'
def _decode_b64_or_url(data: str) -> bytes:
if data.startswith('data:'):
_, b64 = data.split(',', 1)
return base64.b64decode(b64)
if data.startswith('http://') or data.startswith('https://'):
import urllib.request
with urllib.request.urlopen(data, timeout=60) as r:
return r.read()
return base64.b64decode(data)
def _f5tts_clone(ref_audio_path: str, ref_text: str, gen_text: str,
speed: float = 1.0, seed: Optional[int] = None) -> bytes:
"""Run F5-TTS voice cloning, return WAV bytes."""
......@@ -152,6 +166,26 @@ def _save_audio_response(audio_bytes: bytes, http_request: Request) -> dict:
return {"b64_wav": base64.b64encode(audio_bytes).decode()}
# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------
class VoiceExtractRequest(BaseModel):
name: str
description: str = ''
audio: Optional[str] = None # base64/URL audio file
video: Optional[str] = None # base64/URL video (audio track extracted)
transcript: Optional[str] = '' # optional; auto-transcribed if omitted
model_config = ConfigDict(extra="allow")
class VoicePatchRequest(BaseModel):
description: Optional[str] = None
transcript: Optional[str] = None
audio: Optional[str] = None # replace reference audio (base64)
model_config = ConfigDict(extra="allow")
# ---------------------------------------------------------------------------
# Voice profile management
# ---------------------------------------------------------------------------
......@@ -198,6 +232,122 @@ async def delete_voice(name: str):
return {"deleted": True, "name": name}
@router.patch("/v1/audio/voices/{name}")
async def patch_voice(name: str, req: VoicePatchRequest):
"""Update description, transcript, or reference audio of a saved voice profile."""
meta = _load_voice(name)
if not meta:
raise HTTPException(status_code=404, detail=f"Voice '{name}' not found")
vdir = _voice_path(name)
if req.description is not None:
meta['description'] = req.description
if req.transcript is not None:
meta['transcript'] = req.transcript
if req.audio:
audio_bytes, ext = _decode_audio(req.audio)
audio_file = os.path.join(vdir, f'ref{ext}')
# Remove old audio file(s)
for f in os.listdir(vdir):
if f.startswith('ref.') or f.startswith('ref'):
try:
os.unlink(os.path.join(vdir, f))
except Exception:
pass
with open(audio_file, 'wb') as f:
f.write(audio_bytes)
meta['audio_file'] = audio_file
meta['audio_ext'] = ext
with open(os.path.join(vdir, 'meta.json'), 'w') as f:
json.dump(meta, f)
return {"updated": True, "voice": meta}
@router.get("/v1/audio/voices/{name}")
async def get_voice(name: str):
"""Get a single voice profile metadata."""
meta = _load_voice(name)
if not meta:
raise HTTPException(status_code=404, detail=f"Voice '{name}' not found")
return {"voice": meta}
@router.post("/v1/audio/voices/extract")
async def extract_voice(req: VoiceExtractRequest):
"""
Extract a voice profile from a source audio or video file.
- audio: base64/URL audio file (wav, mp3, flac, …)
- video: base64/URL video file — audio track is extracted automatically
- transcript: optional; auto-transcribed with Whisper when omitted
"""
if not req.name.replace('-', '').replace('_', '').isalnum():
raise HTTPException(status_code=400, detail="Voice name must be alphanumeric (hyphens/underscores allowed)")
if not req.audio and not req.video:
raise HTTPException(status_code=400, detail="Provide audio or video")
temps = []
try:
audio_bytes: Optional[bytes] = None
audio_ext = '.wav'
if req.video:
# Extract audio track from video with ffmpeg
raw = _decode_b64_or_url(req.video)
in_tmp = tempfile.NamedTemporaryFile(suffix='.mp4', delete=False)
in_tmp.write(raw)
in_tmp.close()
temps.append(in_tmp.name)
out_tmp = tempfile.NamedTemporaryFile(suffix='.wav', delete=False)
out_tmp.close()
temps.append(out_tmp.name)
r = subprocess.run(
['ffmpeg', '-y', '-i', in_tmp.name, '-vn', '-acodec', 'pcm_s16le',
'-ar', '22050', '-ac', '1', out_tmp.name],
capture_output=True, timeout=120,
)
if r.returncode != 0:
raise HTTPException(status_code=422,
detail=f"Audio extraction from video failed: {r.stderr.decode(errors='replace')[:200]}")
with open(out_tmp.name, 'rb') as f:
audio_bytes = f.read()
else:
audio_bytes, audio_ext = _decode_audio(req.audio)
# Auto-transcribe if transcript not provided
transcript = req.transcript or ''
if not transcript:
try:
import whisper
in_tmp2 = tempfile.NamedTemporaryFile(suffix=audio_ext, delete=False)
in_tmp2.write(audio_bytes)
in_tmp2.close()
temps.append(in_tmp2.name)
model = whisper.load_model('base')
result = model.transcribe(in_tmp2.name)
transcript = result.get('text', '').strip()
except Exception:
pass
# Validate audio
try:
import soundfile as sf
sf.info(io.BytesIO(audio_bytes))
except Exception as e:
raise HTTPException(status_code=400, detail=f"Invalid audio: {e}")
meta = _save_voice(req.name, audio_bytes, audio_ext, transcript, req.description)
return {"created": True, "voice": meta}
finally:
for t in temps:
try:
os.unlink(t)
except Exception:
pass
# ---------------------------------------------------------------------------
# Voice cloning TTS
# ---------------------------------------------------------------------------
......
codai/config.py
View file @ cb542996
......@@ -98,6 +98,14 @@ class WhisperConfig:
server_port: int = 8744
@dataclass
class ArchiveConfig:
"""Generation archive configuration."""
enabled: bool = True
directory: str = "" # empty = <config_dir>/archive; relative paths resolve from config_dir
retention: str = "never" # one of: 1h 1d 2d 1w 1m 3m 6m 1y never
@dataclass
class Config:
"""Main configuration class."""
......@@ -109,6 +117,7 @@ class Config:
vulkan: VulkanConfig = field(default_factory=VulkanConfig)
image: ImageConfig = field(default_factory=ImageConfig)
whisper: WhisperConfig = field(default_factory=WhisperConfig)
archive: ArchiveConfig = field(default_factory=ArchiveConfig)
system_prompt: Optional[str] = None
tools_closer_prompt: bool = False
grammar_guided: bool = False
......@@ -244,6 +253,7 @@ class ConfigManager:
vulkan=VulkanConfig(**config_data.get("vulkan", {})),
image=ImageConfig(**config_data.get("image", {})),
whisper=WhisperConfig(**config_data.get("whisper", {})),
archive=ArchiveConfig(**config_data.get("archive", {})),
system_prompt=config_data.get("system_prompt"),
tools_closer_prompt=config_data.get("tools_closer_prompt", False),
grammar_guided=config_data.get("grammar_guided", False),
......@@ -347,6 +357,11 @@ class ConfigManager:
"vae_tiling": self.config.image.vae_tiling,
"clip_on_cpu": self.config.image.clip_on_cpu
},
"archive": {
"enabled": self.config.archive.enabled,
"directory": self.config.archive.directory,
"retention": self.config.archive.retention,
},
"system_prompt": self.config.system_prompt,
"tools_closer_prompt": self.config.tools_closer_prompt,
"grammar_guided": self.config.grammar_guided,
......
codai/main.py
View file @ cb542996
......@@ -22,6 +22,7 @@ import os
from codai.cli import parse_args
from codai.config import ConfigManager
from codai.admin.routes import init_session_manager, set_config_manager
from codai.api.archive import archive_manager
def main():
......@@ -55,6 +56,18 @@ def main():
if config.models.gguf_cache_dir:
os.environ['CODERAI_CACHE_DIR'] = config.models.gguf_cache_dir
# Configure generation archive
_arc_dir = config.archive.directory
if not _arc_dir:
_arc_dir = os.path.join(config_dir, "archive")
elif not os.path.isabs(_arc_dir):
_arc_dir = os.path.join(config_dir, _arc_dir)
archive_manager.configure(
enabled=config.archive.enabled,
directory=_arc_dir,
retention=config.archive.retention,
)
# Initialize admin session manager and expose config to admin routes
from pathlib import Path
init_session_manager(Path(config_dir))
......@@ -364,6 +377,8 @@ def main():
multi_model_manager.model_backend_types[mid] = (
m.get("backend", "auto") if isinstance(m, dict) else "auto"
)
if isinstance(m, dict) and m.get("alias"):
multi_model_manager.set_model_alias(m["alias"], mid)
# Audio models
audio_models = models_config.get("audio_models", [])
......@@ -400,6 +415,8 @@ def main():
mid = _model_id(m)
if mid:
multi_model_manager.set_image_model(mid, config=_model_cfg(m, "image"))
if isinstance(m, dict) and m.get("alias"):
multi_model_manager.set_model_alias(m["alias"], mid)
# Vision models
vision_models = models_config.get("vision_models", [])
......@@ -407,6 +424,8 @@ def main():
mid = _model_id(m)
if mid:
multi_model_manager.set_vision_model(mid, config=_model_cfg(m, "vision"))
if isinstance(m, dict) and m.get("alias"):
multi_model_manager.set_model_alias(m["alias"], mid)
# TTS models
tts_models = models_config.get("tts_models", [])
......@@ -414,6 +433,8 @@ def main():
mid = _model_id(m)
if mid:
multi_model_manager.set_tts_model(mid, config=_model_cfg(m, "tts") if isinstance(m, dict) else {})
if isinstance(m, dict) and m.get("alias"):
multi_model_manager.set_model_alias(m["alias"], mid)
# Video generation models
video_models = models_config.get("video_models", [])
......@@ -421,6 +442,8 @@ def main():
mid = _model_id(m)
if mid:
multi_model_manager.set_video_model(mid, config=_model_cfg(m, "video") if isinstance(m, dict) else {})
if isinstance(m, dict) and m.get("alias"):
multi_model_manager.set_model_alias(m["alias"], mid)
# Audio generation models (MusicGen, AudioLDM2, …)
audio_gen_models = models_config.get("audio_gen_models", [])
......@@ -428,6 +451,8 @@ def main():
mid = _model_id(m)
if mid:
multi_model_manager.set_audio_gen_model(mid, config=_model_cfg(m, "audio_gen") if isinstance(m, dict) else {})
if isinstance(m, dict) and m.get("alias"):
multi_model_manager.set_model_alias(m["alias"], mid)
# Embedding models
embedding_models = models_config.get("embedding_models", [])
......@@ -435,6 +460,8 @@ def main():
mid = _model_id(m)
if mid:
multi_model_manager.set_embedding_model(mid, config=_model_cfg(m, "embedding") if isinstance(m, dict) else {})
if isinstance(m, dict) and m.get("alias"):
multi_model_manager.set_model_alias(m["alias"], mid)
# Register aliases
aliases = models_config.get("aliases", {})
......@@ -623,10 +650,20 @@ def main():
if global_file_path:
set_fs_file_path(global_file_path)
# Set spatial (3D) module global args
from codai.api.spatial import set_global_args as set_spatial_global_args, set_global_file_path as set_spatial_file_path
set_spatial_global_args(global_args)
if global_file_path:
set_spatial_file_path(global_file_path)
# Set character profiles module global args
from codai.api.characters import set_global_args as set_chars_global_args
set_chars_global_args(global_args)
# Set environment profiles module global args
from codai.api.environments import set_global_args as set_envs_global_args
set_envs_global_args(global_args)
# Set embeddings module global args
from codai.api.embeddings import set_global_args as set_embed_global_args
set_embed_global_args(global_args)
......
codai/models/capabilities.py
View file @ cb542996
......@@ -62,6 +62,12 @@ class ModelCapabilities:
lip_sync: bool = False # Wav2Lip, SadTalker
video_dubbing: bool = False # Translation + TTS + lip sync
# 3D generation and conversion
image_to_3d: bool = False # 2D image → stereo / anaglyph / depth / mesh
video_to_3d: bool = False # 2D video → stereo / anaglyph / depth video
model_3d_generation: bool = False # text / image → 3D model (GLB)
model_3d_to_image: bool = False # 3D model → rendered 2D image / video
def to_list(self) -> List[str]:
out = []
for name, val in self.__dataclass_fields__.items():
......@@ -85,6 +91,15 @@ def detect_model_capabilities(model_name: str) -> ModelCapabilities:
n = model_name.lower()
# ── 3D generation ────────────────────────────────────────────────────────
if any(x in n for x in ['triposr', 'tsr', 'shap-e', 'shape-e', 'point-e', 'pointe',
'zero123', 'wonder3d', 'instant3d', 'one-2-3-45',
'mvdiffusion', 'syncdreamer', 'dreamgaussian']):
caps.model_3d_generation = True
caps.image_to_3d = True
caps.model_3d_to_image = True
return caps
# ── Video generation ─────────────────────────────────────────────────────
if any(x in n for x in ['cogvideox', 'cogvideo', 'ltx-video', 'ltxvideo',
'hunyuan-video', 'mochi-1', 'dynamicrafter',
......
codai/models/manager.py
View file @ cb542996
......@@ -1632,6 +1632,23 @@ class MultiModelManager:
'error': f"Model '{resolved_name}' is not available. Use one of: {', '.join(allowed_ids)}",
}
# Step 1c: Resolve short-name to canonical registered name (e.g. "Foo" → "org/Foo")
_type_registry = {
"image": self.image_models,
"audio": self.audio_models,
"tts": [self.tts_model] if self.tts_model else [],
"vision": self.vision_models,
"video": self.video_models,
"audio_gen": self.audio_gen_models,
"embedding": self.embedding_models,
}
_candidates = _type_registry.get(model_type, []) if model_type else []
if resolved_name not in _candidates:
for _m in _candidates:
if "/" in _m and _m.split("/")[-1] == resolved_name:
resolved_name = _m
break
# Step 2: Build the model key (prefixed with type)
if model_type and model_type != "text":
model_key = f"{model_type}:{resolved_name}"
......@@ -1641,6 +1658,19 @@ class MultiModelManager:
# Step 3: Check if already loaded in self.models
existing_model = self.models.get(model_key)
# Fuzzy fallback: if not found by exact key, check for org-prefix
# variations (e.g. "image:Model" vs "image:org/Model").
if existing_model is None:
type_prefix = f"{model_type}:" if model_type and model_type != "text" else ""
short = resolved_name.split("/")[-1]
for k, v in self.models.items():
k_name = k[len(type_prefix):] if type_prefix and k.startswith(type_prefix) else k
if k_name == resolved_name or k_name.split("/")[-1] == short:
model_key = k
resolved_name = k_name
existing_model = v
break
# =====================================================================
# PER-MODEL LOAD MODE: Check per-model config first.
# Per-model "load" = pre-loaded (treat as loadall for this model).
......
codai/pydantic/audiogenrequest.py
View file @ cb542996
......@@ -31,6 +31,8 @@ class AudioGenerationRequest(BaseModel):
seed: Optional[int] = None
# Reference audio for melody conditioning (MusicGen Melody)
melody: Optional[str] = None # base64/URL
# Voice profile for singing/speech conditioning
voice_profile: Optional[str] = None # saved voice profile name
# Output
response_format: Optional[str] = "url" # url | b64_wav | b64_mp3
user: Optional[str] = None
......
codai/pydantic/imagerequest.py
View file @ cb542996
......@@ -34,6 +34,12 @@ class ImageGenerationRequest(BaseModel):
seed: Optional[int] = None
user: Optional[str] = None
disable_safety_checker: Optional[bool] = False
negative_prompt: Optional[str] = None
# Character consistency
character_profiles: Optional[List[str]] = None # saved profile names
character_references: Optional[List[str]] = None # inline base64 images
character_strength: Optional[float] = 0.6 # IP-Adapter scale
model_config = ConfigDict(extra="allow")
......
codai/pydantic/videorequest.py
View file @ cb542996
......@@ -20,6 +20,18 @@ from typing import Dict, List, Optional
from pydantic import BaseModel, ConfigDict
class CharacterDialogLine(BaseModel):
"""One spoken line in a multi-character dialog sequence."""
character: Optional[str] = None # character profile name (used for lip-sync face)
voice: Optional[str] = None # voice profile name OR TTS voice id
text: str # dialog text to synthesise
start_time: Optional[float] = None # explicit offset in seconds (None = sequential)
lip_sync: Optional[bool] = True # apply lip sync for this line
lang: Optional[str] = None # language hint for TTS
speed: Optional[float] = 1.0
model_config = ConfigDict(extra="allow")
class VideoGenerationRequest(BaseModel):
model: str
prompt: str = ""
......@@ -78,6 +90,9 @@ class VideoGenerationRequest(BaseModel):
lip_sync: Optional[bool] = False # warp mouth to match audio
lip_sync_method: Optional[str] = "wav2lip" # wav2lip | sadtalker
# ── Multi-character dialog ────────────────────────────────────────────
dialogs: Optional[List[CharacterDialogLine]] = None
# ── Subtitles ────────────────────────────────────────────────────────
generate_subtitles: Optional[bool] = False
burn_subtitles: Optional[bool] = False
......
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