from __future__ import annotations import html import json import os import random import shutil import subprocess import time from datetime import datetime from pathlib import Path from typing import Optional from urllib.parse import quote import gradio as gr from .settings import * def get_aspect_ratio_choices_for_task(task: str) -> list[tuple[str, str]]: """Get Aspect Ratio choices with default/recommended marker for the given task.""" internal_task = normalize_task(task) default_ratio = DEFAULT_IMAGE_ASPECT_RATIO if internal_task in IMAGE_TASKS else DEFAULT_VIDEO_ASPECT_RATIO return [ (f"{ratio}" if ratio == default_ratio else ratio, ratio) for ratio in ASPECT_RATIO_CHOICES ] def get_video_duration_choices() -> list[tuple[str, int]]: return [(f"{seconds}s", seconds) for seconds in range(1, 11)] def display_path(path: Path) -> str: path_text = path.as_posix() if path.is_absolute(): try: path_text = path.relative_to(Path.cwd()).as_posix() except ValueError: return path_text if path_text == "." or path_text.startswith("./"): return path_text return f"./{path_text}" def get_model_base_dir() -> Path: """Return the local model directory only. Local-only mode never selects remote storage and never downloads model assets from remote repositories. Override with LANCE_MODEL_BASE_DIR when your local weights live somewhere else. """ configured = os.getenv("LANCE_MODEL_BASE_DIR") return Path(configured).expanduser() if configured else LOCAL_MODEL_BASE_DIR def normalize_model_variant(model_variant: Optional[str] = None) -> str: variant = (model_variant or os.getenv("LANCE_MODEL_VARIANT", DEFAULT_MODEL_VARIANT)).strip().lower() if variant in {"image", "t2i", "i2t"}: return MODEL_VARIANT_IMAGE return MODEL_VARIANT_VIDEO def get_model_path(model_variant: Optional[str] = None) -> Path: variant = normalize_model_variant(model_variant) variant_env_name = "LANCE_IMAGE_MODEL_PATH" if variant == MODEL_VARIANT_IMAGE else "LANCE_VIDEO_MODEL_PATH" variant_configured = os.getenv(variant_env_name) if variant_configured: return Path(variant_configured).expanduser() configured = os.getenv("LANCE_MODEL_PATH") if configured: return Path(configured).expanduser() model_dir_name = MODEL_VARIANT_TO_DIR[variant] return get_model_base_dir() / model_dir_name def get_required_model_asset_paths(model_base_dir: Path, model_path: Path) -> list[Path]: return [ model_path / "llm_config.json", model_path / "model.safetensors", model_base_dir / "Qwen2.5-VL-ViT" / "vit.safetensors", model_base_dir / "Wan2.2_VAE.pth", ] def ensure_model_assets(model_variant: Optional[str] = None) -> Path: """Verify that all required model assets exist locally. Expected layout by default: downloads/ Lance_3B_Video/ Lance_3B/ Qwen2.5-VL-ViT/ Wan2.2_VAE.pth Set LANCE_MODEL_BASE_DIR, LANCE_MODEL_PATH, LANCE_VIDEO_MODEL_PATH or LANCE_IMAGE_MODEL_PATH to point at local files. No remote download is attempted. """ model_base_dir = get_model_base_dir() os.environ["LANCE_MODEL_BASE_DIR"] = display_path(model_base_dir) model_path = get_model_path(model_variant) required_paths = get_required_model_asset_paths(model_base_dir, model_path) if all(path.exists() for path in required_paths): return model_path missing = "\n".join(f"- {display_path(path)}" for path in required_paths if not path.exists()) raise FileNotFoundError( "Local Lance model assets are missing. This local-only build does not " "download from remote repositories. Set LANCE_MODEL_BASE_DIR " "or the model path environment variables to your local weights.\n" f"Missing files:\n{missing}" ) def ensure_dirs() -> None: TMP_INPUT_DIR.mkdir(parents=True, exist_ok=True) RESULTS_ROOT.mkdir(parents=True, exist_ok=True) def save_generation_record(record: dict, save_dir: Path) -> None: ensure_dirs() run_record_path = save_dir / RUN_RECORD_FILENAME with run_record_path.open("w", encoding="utf-8") as f: json.dump(record, f, ensure_ascii=False, indent=2) with RECORD_WRITE_LOCK: with GLOBAL_RECORDS_FILE.open("a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") def normalize_seed(seed: int) -> int: return random.randint(0, 2**31 - 1) if seed == -1 else seed def video_seconds_to_num_frames(seconds: int) -> int: seconds = max(1, min(10, int(seconds))) return 12 * seconds + 1 def get_default_video_duration_seconds(task: str) -> int: internal_task = normalize_task(task) if internal_task == TASK_I2V: return DEFAULT_I2V_DURATION_SECONDS return DEFAULT_T2V_DURATION_SECONDS def normalize_task(task: str) -> str: task_key = (task or TASK_LABEL_VIDEO_GENERATION).strip() task = TASK_LABEL_TO_INTERNAL.get(task_key, TASK_LABEL_TO_INTERNAL.get(task_key.lower(), "")) if task not in GENERATION_TASKS | UNDERSTANDING_TASKS: raise ValueError(f"Unsupported task type: {task}") return task def normalize_resolution_choice_value(resolution: str, task: str) -> str: resolution_text = str(resolution or "").strip() for choice in get_resolution_choices_for_task(task): if isinstance(choice, tuple): label, value = choice if resolution_text in {str(label), str(value)}: return str(value) elif resolution_text == str(choice): return str(choice) return resolution_text def get_resolution_choice_values_for_task(task: str) -> list[str]: choices = get_resolution_choices_for_task(task) values = [] for choice in choices: values.append(choice[1] if isinstance(choice, tuple) else choice) return values def get_resolution_choices_for_task(task: str) -> list[str | tuple[str, str]]: internal_task = normalize_task(task) if internal_task in IMAGE_TASKS: return IMAGE_RESOLUTION_CHOICES if internal_task in {TASK_T2V, TASK_I2V}: return VIDEO_RESOLUTION_DISPLAY_CHOICES if internal_task == TASK_VIDEO_EDIT: return VIDEO_EDIT_RESOLUTION_CHOICES if internal_task in VIDEO_TASKS: return VIDEO_EDIT_RESOLUTION_CHOICES return VIDEO_RESOLUTION_CHOICES def get_default_resolution_for_task(task: str) -> str: internal_task = normalize_task(task) if internal_task in IMAGE_TASKS: return DEFAULT_IMAGE_RESOLUTION # Text-to-Video and Image-to-Video default to the lightweight/recommended 360p profile. if internal_task in {TASK_T2V, TASK_I2V}: return DEFAULT_RESOLUTION if internal_task == TASK_VIDEO_EDIT: return DEFAULT_VIDEO_EDIT_RESOLUTION if internal_task in VIDEO_TASKS: return DEFAULT_VIDEO_EDIT_RESOLUTION return DEFAULT_RESOLUTION def normalize_resolution_for_backend(resolution: str, task: str) -> str: internal_task = normalize_task(task) normalized_resolution = normalize_resolution_choice_value(resolution, internal_task) choices = get_resolution_choice_values_for_task(internal_task) if normalized_resolution in choices: return normalized_resolution return get_default_resolution_for_task(internal_task) def get_default_aspect_ratio(task: str) -> str: internal_task = normalize_task(task) return DEFAULT_IMAGE_ASPECT_RATIO if internal_task in IMAGE_TASKS else DEFAULT_VIDEO_ASPECT_RATIO def normalize_aspect_ratio_for_task(task: str, aspect_ratio: Optional[str]) -> str: internal_task = normalize_task(task) if internal_task == TASK_I2V: return get_default_aspect_ratio(internal_task) return aspect_ratio if aspect_ratio in ASPECT_RATIO_CHOICES else get_default_aspect_ratio(internal_task) def normalize_video_resolution(resolution: Optional[str], task: Optional[str] = None) -> str: if task is None: return resolution if resolution in VIDEO_RESOLUTION_CHOICES else DEFAULT_RESOLUTION normalized_resolution = normalize_resolution_choice_value(resolution, task) choices = get_resolution_choice_values_for_task(task) return normalized_resolution if normalized_resolution in choices else get_default_resolution_for_task(task) def get_size_for_aspect_ratio(task: str, aspect_ratio: str, video_resolution: Optional[str] = None) -> tuple[int, int]: internal_task = normalize_task(task) aspect_ratio = normalize_aspect_ratio_for_task(internal_task, aspect_ratio) if internal_task in IMAGE_TASKS: size_map = IMAGE_ASPECT_RATIO_TO_SIZE else: size_map = VIDEO_RESOLUTION_TO_SIZE_MAP[normalize_video_resolution(video_resolution, internal_task)] return size_map[aspect_ratio] def format_size_markdown(task: str, width: int, height: int) -> str: internal_task = normalize_task(task) if internal_task in UNDERSTANDING_TASKS: return "" #return f"**Output Resolution:** `{width} x {height}`" return f"{width} x {height}" def get_size_map_for_task(task: str, video_resolution: Optional[str] = None) -> dict[str, tuple[int, int]]: internal_task = normalize_task(task) if internal_task in IMAGE_TASKS: return IMAGE_ASPECT_RATIO_TO_SIZE return VIDEO_RESOLUTION_TO_SIZE_MAP[normalize_video_resolution(video_resolution, internal_task)] def get_output_resolution_choices_for_task(task: str, video_resolution: Optional[str] = None) -> list[tuple[str, str]]: """Get Output Resolution choices with a one-to-one mapping to aspect ratios.""" internal_task = normalize_task(task) default_ratio = get_default_aspect_ratio(internal_task) size_map = get_size_map_for_task(internal_task, video_resolution) choices = [] for ratio in ASPECT_RATIO_CHOICES: width, height = size_map[ratio] resolution_text = format_size_markdown(internal_task, width, height) label = f"{resolution_text}" if ratio == default_ratio else resolution_text choices.append((label, resolution_text)) return choices def build_lance_label_html(text: str, *extra_classes: str) -> str: class_names = " ".join(["lance-section-label", *extra_classes]).strip() return f'
{html.escape(text)}
' def build_lance_icon_label_html(text: str, icon: str, *extra_classes: str) -> str: icon_map = { "video": """ """, "image": """ """, "text": """ """, "logs": """ """, } icon_html = icon_map.get(icon, "") class_names = " ".join(["lance-section-label", "lance-icon-label", *extra_classes]).strip() return f'
{icon_html}{html.escape(text)}
' def update_size_from_aspect_ratio(task: str, aspect_ratio: str, video_resolution: Optional[str] = None): aspect_ratio = normalize_aspect_ratio_for_task(task, aspect_ratio) width, height = get_size_for_aspect_ratio(task, aspect_ratio, video_resolution) return height, width, gr.update( choices=get_output_resolution_choices_for_task(task, video_resolution), value=format_size_markdown(task, width, height), ) def update_output_resolution_from_video_profile(task: str, aspect_ratio: str, video_resolution: str): aspect_ratio = normalize_aspect_ratio_for_task(task, aspect_ratio) width, height = get_size_for_aspect_ratio(task, aspect_ratio, video_resolution) return ( gr.update( choices=get_output_resolution_choices_for_task(task, video_resolution), value=format_size_markdown(task, width, height), ), height, width, ) def reset_generation_defaults_for_task(task: str): internal_task = normalize_task(task) aspect_ratio = get_default_aspect_ratio(internal_task) resolution = get_default_resolution_for_task(internal_task) width, height = get_size_for_aspect_ratio(internal_task, aspect_ratio, resolution) num_frames = get_default_video_duration_seconds(internal_task) return aspect_ratio, height, width, num_frames, resolution, gr.update( choices=get_output_resolution_choices_for_task(internal_task, resolution), value=format_size_markdown(internal_task, width, height), ) def make_prompt_example_click_handler(prompt_text: str): """Create a click handler for custom text-to-visual prompt-example rows.""" def _handler(task: str): defaults = reset_generation_defaults_for_task(task) return (prompt_text, "", *defaults) return _handler def make_media_prompt_example_click_handler( prompt_text: str, input_video_path: Optional[str] = None, input_image_path: Optional[str] = None, ): """Create a click handler for edit/understanding example rows.""" def _handler(task: str): internal_task = normalize_task(task) defaults = reset_generation_defaults_for_task(internal_task) system_prompt = normalize_understanding_system_prompt(internal_task, None) if internal_task in UNDERSTANDING_TASKS else "" return (prompt_text, input_video_path, input_image_path, system_prompt, *defaults) return _handler def get_understanding_system_prompt_choices(task: str) -> list[str]: internal_task = normalize_task(task) if internal_task == TASK_X2T_IMAGE: return [I2T_QA_SYSTEM_PROMPT] return [V2T_QA_SYSTEM_PROMPT] def normalize_understanding_system_prompt(task: str, system_prompt: Optional[str]) -> str: return get_understanding_system_prompt_choices(task)[0] def create_request_json( task: str, prompt: str, input_video: Optional[str], input_image: Optional[str], system_prompt: Optional[str] = None, ) -> Path: ensure_dirs() timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") prompt_file = TMP_INPUT_DIR / f"{task}_{timestamp}.json" if task == TASK_T2V: payload = {"000000.mp4": prompt} elif task == TASK_T2I: payload = {"000000.png": prompt} elif task == TASK_I2V: if not input_image: raise ValueError("The image-to-video task requires an input image.") payload = { "000000": { "interleave_array": [prompt, input_image], "element_dtype_array": ["text", "image"], "istarget_in_interleave": [0, 0], } } elif task == TASK_VIDEO_EDIT: if not input_video: raise ValueError("The video edit task requires an input video.") payload = { "000000": { "interleave_array": [prompt, input_video, input_video], "element_dtype_array": ["text", "video", "video"], "istarget_in_interleave": [0, 0, 1], } } elif task == TASK_IMAGE_EDIT: if not input_image: raise ValueError("The image edit task requires an input image.") payload = { "000000": { "interleave_array": [prompt, input_image, input_image], "element_dtype_array": ["text", "image", "image"], "istarget_in_interleave": [0, 0, 1], } } elif task == TASK_X2T_VIDEO: if not input_video: raise ValueError("The video understanding task requires an input video.") system_prompt = normalize_understanding_system_prompt(task, system_prompt) payload = { "000000": { "interleave_array": [input_video, [system_prompt, prompt, ""]], "element_dtype_array": ["video", "text"], "istarget_in_interleave": [0, 1], } } elif task == TASK_X2T_IMAGE: if not input_image: raise ValueError("The image understanding task requires an input image.") system_prompt = normalize_understanding_system_prompt(task, system_prompt) payload = { "000000": { "interleave_array": [input_image, [system_prompt, prompt, ""]], "element_dtype_array": ["image", "text"], "istarget_in_interleave": [0, 1], } } else: raise ValueError(f"Unsupported task type: {task}") with prompt_file.open("w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2) return prompt_file def resolve_example_path(path: str) -> str: candidate = Path(path) if candidate.is_absolute(): return str(candidate) repo_candidate = (REPO_ROOT / candidate) if repo_candidate.exists(): return str(repo_candidate.resolve()) if candidate.exists(): return str(candidate.resolve()) return path def resolve_video_example_paths(path: str) -> tuple[str, str]: """Return (browser_preview_path, model_input_path). Model input keeps the original sample path. Browser preview uses a H.264/yuv420p copy when the source codec is not reliably playable. """ original_path = resolve_example_path(path) return prepare_browser_preview_video(original_path), original_path def _probe_video_stream(video_path: Path) -> dict[str, str]: if not shutil.which("ffprobe"): return {} try: result = subprocess.run( [ "ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_name,pix_fmt", "-of", "json", str(video_path), ], check=True, capture_output=True, text=True, ) data = json.loads(result.stdout or "{}") streams = data.get("streams") or [] return streams[0] if streams else {} except Exception: return {} def _is_browser_playable_mp4(video_path: Path) -> bool: stream = _probe_video_stream(video_path) return stream.get("codec_name") == "h264" and stream.get("pix_fmt") == "yuv420p" def prepare_browser_preview_video(video_path: str) -> str: source = _resolve_existing_media_path(video_path) if source is None: return video_path if _is_browser_playable_mp4(source): return str(source) if not shutil.which("ffmpeg"): return str(source) PREVIEW_VIDEO_DIR.mkdir(parents=True, exist_ok=True) preview_path = PREVIEW_VIDEO_DIR / f"{source.stem}_h264.mp4" if preview_path.exists() and preview_path.stat().st_mtime >= source.stat().st_mtime: return str(preview_path) try: subprocess.run( [ "ffmpeg", "-y", "-i", str(source), "-an", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-movflags", "+faststart", str(preview_path), ], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) return str(preview_path) except Exception: return str(source) def _resolve_existing_media_path(media_path: Optional[str]) -> Optional[Path]: if not media_path: return None candidate = Path(str(media_path)) candidates = [candidate] if candidate.is_absolute() else [REPO_ROOT / candidate, candidate] for item in candidates: try: resolved = item.expanduser().resolve() except Exception: continue if resolved.exists(): return resolved return None def build_gradio_media_url(media_path: Optional[str]) -> str: """Build a Gradio file-serving URL for local recommended-case media.""" existing = _resolve_existing_media_path(media_path) source = str(existing if existing else media_path or "") if not source: return "" try: from gradio.route_utils import API_PREFIX except Exception: API_PREFIX = "" return f"{API_PREFIX or ''}/file={quote(source, safe='/:')}" def build_example_media_html(media_path: Optional[str], media_type: str, fallback_media_path: Optional[str] = None) -> str: """Build a lightweight complete-fit media preview for recommended cases.""" if media_type == "video": sources = [] for candidate in (media_path, fallback_media_path): url = build_gradio_media_url(candidate) if url and url not in sources: sources.append(url) if not sources: return '
Video file not found
' source_tags = "".join( f'' for url in sources ) return ( '' ) url = build_gradio_media_url(media_path) if not url: return '
Image file not found
' alt_text = html.escape(Path(str(media_path)).name or "example image", quote=True) return f'{alt_text}' def load_json_examples(relative_path: str) -> dict: path = REPO_ROOT / relative_path with path.open("r", encoding="utf-8") as f: return json.load(f) T2V_EXAMPLE_SUMMARIES = { "000000.mp4": "Red panda surfing on a bright seaside wave.", "000002.mp4": "Panda cub skateboarding in a creative loft.", "000004.mp4": "Young woman shaping clay in a sunlit pottery workshop.", "000005.mp4": "Panda boxing a robot in a luxurious palace ring.", "000008.mp4": "Fantasy pastel horse stepping through a glowing cloud valley.", } def make_generation_examples( task_label: str, relative_path: str, limit: int, image_task: bool, selected_keys: Optional[list[str]] = None, summaries: Optional[dict[str, str]] = None, ) -> list[list]: data = load_json_examples(relative_path) items = [(key, data[key]) for key in selected_keys if key in data] if selected_keys else list(data.items())[:limit] return [[prompt] for _output_name, prompt in items] def make_edit_examples(task_label: str, relative_path: str, limit: int, media_type: str) -> list[list]: data = load_json_examples(relative_path) examples = [] for sample in list(data.values())[:limit]: interleave = sample["interleave_array"] prompt = interleave[0] if media_type == "video": preview_video_path, input_video_path = resolve_video_example_paths(interleave[1]) examples.append([prompt, preview_video_path, input_video_path, None, None]) else: image_path = resolve_example_path(interleave[1]) examples.append([prompt, None, None, image_path, image_path]) return examples def make_i2v_examples(relative_path: str, limit: int) -> list[list]: data = load_json_examples(relative_path) examples = [] for sample in list(data.values())[:limit]: interleave = sample["interleave_array"] prompt = interleave[0] image_path = resolve_example_path(interleave[1]) examples.append([prompt, None, None, image_path, image_path]) return examples def make_understanding_examples(task_label: str, relative_path: str, limit: int, media_type: str) -> list[list]: data = load_json_examples(relative_path) examples = [] for sample in list(data.values())[:limit]: interleave = sample["interleave_array"] text_payload = interleave[1] question = text_payload[1] if isinstance(text_payload, list) and len(text_payload) > 1 else "" if media_type == "video": preview_video_path, input_video_path = resolve_video_example_paths(interleave[0]) examples.append([question, preview_video_path, input_video_path, None, None]) else: image_path = resolve_example_path(interleave[0]) examples.append([question, None, None, image_path, image_path]) return examples VIDEO_GENERATION_EXAMPLES = make_generation_examples( TASK_LABEL_VIDEO_GENERATION, "config/examples/t2v_example.json", limit=7, image_task=False, #selected_keys=["000000.mp4", "000002.mp4", "000005.mp4", "000004.mp4", "000008.mp4"], selected_keys=["000004.mp4", "000002.mp4", "000000.mp4", "000005.mp4", "000008.mp4", "000007.mp4", "000001.mp4"], summaries=T2V_EXAMPLE_SUMMARIES, ) VIDEO_EDIT_EXAMPLES = make_edit_examples( TASK_LABEL_VIDEO_EDIT, "config/examples/video_edit_example.json", limit=3, media_type="video", ) IMAGE_TO_VIDEO_EXAMPLES = make_i2v_examples( "config/examples/i2v_example.json", limit=6, ) VIDEO_UNDERSTANDING_EXAMPLES = make_understanding_examples( TASK_LABEL_VIDEO_UNDERSTANDING, "config/examples/x2t_video_example.json", limit=3, media_type="video", ) IMAGE_GENERATION_EXAMPLES = make_generation_examples( TASK_LABEL_IMAGE_GENERATION, "config/examples/t2i_example.json", limit=9, image_task=True, selected_keys=["000000.png", "000003.png", "000002.png", "000005.png", "000006.png", "000007.png", "000008.png", "000009.png", "000010.png"], ) IMAGE_EDIT_EXAMPLES = make_edit_examples( TASK_LABEL_IMAGE_EDIT, "config/examples/image_edit_example.json", limit=5, media_type="image", ) IMAGE_UNDERSTANDING_EXAMPLES = make_understanding_examples( TASK_LABEL_IMAGE_UNDERSTANDING, "config/examples/x2t_image_example.json", limit=6, media_type="image", ) def build_save_dir(task: str) -> Path: ensure_dirs() timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") return RESULTS_ROOT / f"{task}_{timestamp}_{int(time.time() * 1000) % 1000:03d}" def find_generated_video(save_dir: Path) -> Optional[Path]: videos = sorted(save_dir.glob("*.mp4"), key=lambda p: p.stat().st_mtime, reverse=True) return videos[0] if videos else None def find_generated_image(save_dir: Path) -> Optional[Path]: images = sorted(save_dir.glob("*.png"), key=lambda p: p.stat().st_mtime, reverse=True) return images[0] if images else None def extract_text_result(save_dir: Path) -> str: prompt_result_path = save_dir / PROMPT_JSON_FILENAME if not prompt_result_path.exists(): return "" with prompt_result_path.open("r", encoding="utf-8") as f: data = json.load(f) if not data: return "" first_value = next(iter(data.values())) return first_value if isinstance(first_value, str) else json.dumps(first_value, ensure_ascii=False)