""" OmniLottie Gradio Demo - Hugging Face Compatible Version This version uses decoder_hf.py with from_pretrained() to load models, supporting automatic downloading from Hugging Face Hub. Environment Variables: MODEL_PATH: Model path or HF Hub ID (default: "OmniLottie/OmniLottie") PROCESSOR_PATH: Processor path (default: "Qwen/Qwen2.5-VL-3B-Instruct") Usage: # Using HF Hub model (automatic download) python app_hf.py # Using local model MODEL_PATH=/path/to/local/model python app_hf.py # Custom configuration MODEL_PATH=your-org/your-model PROCESSOR_PATH=/path/to/processor python app_hf.py """ import gradio as gr import json import torch import os import numpy as np import random import re import tempfile import base64 import threading import time from PIL import Image as PILImage from decord import VideoReader, cpu from decoder_hf import LottieDecoder # Use HF decoder with from_pretrained from transformers import AutoProcessor from qwen_vl_utils import process_vision_info from lottie.objects.lottie_tokenize import LottieTensor from lottie.objects.lottie_param import ( from_sequence, ShapeLayer, NullLayer, PreCompLayer, TextLayer, SolidColorLayer, Font, Chars, shape_layer_to_json, null_layer_to_json, precomp_layer_to_json, text_layer_to_json, solid_layer_to_json, font_to_json, char_to_json ) SYSTEM_PROMPT = "You are a Lottie animation expert." VIDEO_PROMPT = "Turn this video into Lottie code." LOTTIE_BOS = 192398 LOTTIE_EOS = 192399 PAD_TOKEN = 151643 model = None processor = None device = None generation_lock = threading.Lock() def load_model_once(): """ Load OmniLottie model using from_pretrained() for HuggingFace compatibility Supports automatic downloading from HF Hub or loading from local path """ global model, processor, device if model is not None: return model, processor, device # Model path - can be HF Hub ID or local path # Examples: "OmniLottie/OmniLottie" or "/path/to/local/model" model_path = os.environ.get("MODEL_PATH", "OmniLottie/OmniLottie") device = torch.device("cuda:0" if torch.cuda.is_available() else "xpu:0" if torch.xpu.is_available() else "cpu") print(f"Loading model from {model_path}...") # Load model using from_pretrained (supports HF Hub and local paths) model = LottieDecoder.from_pretrained( model_path, torch_dtype=torch.bfloat16, trust_remote_code=True ) model = model.to(device).eval() # Load processor processor_path = os.environ.get("PROCESSOR_PATH", "Qwen/Qwen2.5-VL-3B-Instruct") processor = AutoProcessor.from_pretrained( processor_path, padding_side="left", trust_remote_code=True ) print(f"✅ Model loaded on {device}") return model, processor, device def simplify_to_animation_description(text): if not text or not isinstance(text, str): return text prefixes = [ r'^The video features?\s+', r'^The scene shows?\s+', r'^An animation of\s+', r'^There is\s+', r'^It shows?\s+' ] for pattern in prefixes: text = re.sub(pattern, '', text, flags=re.IGNORECASE) if text: text = text[0].upper() + text[1:] return text.strip() def add_random_background(img): if img.mode != 'RGBA': return img.convert('RGB') light_colors = [(255, 255, 255), (245, 245, 245), (250, 250, 250)] bg_color = random.choice(light_colors) background = PILImage.new('RGB', img.size, bg_color) background.paste(img, (0, 0), img) return background def load_frames_from_video(video_path, num_frames=8, max_size=336): import os ext = os.path.splitext(video_path)[1].lower() frames = [] if ext in ('.gif', '.webp'): try: img = PILImage.open(video_path) total_frames = getattr(img, 'n_frames', 1) if total_frames < 1: raise ValueError(f"No frames in {ext.upper()}: {video_path}") indices = np.linspace(0, total_frames - 1, min(num_frames, total_frames)).astype(int) for idx in indices: img.seek(idx) frame = img.convert('RGB') if max(frame.size) > max_size: frame.thumbnail((max_size, max_size), PILImage.LANCZOS) frames.append(frame) img.close() except Exception as e: raise ValueError(f"Failed to load {ext.upper()}: {str(e)}") else: try: vr = VideoReader(video_path, ctx=cpu(0)) total_frames = len(vr) if total_frames < 1: raise ValueError(f"Video has no frames: {video_path}") indices = np.linspace(0, total_frames - 1, num_frames).astype(int) frames_np = vr.get_batch(indices).asnumpy() for f in frames_np: img = PILImage.fromarray(f) if max(img.size) > max_size: img.thumbnail((max_size, max_size), PILImage.LANCZOS) frames.append(img) except Exception as e: raise ValueError(f"Failed to load video: {str(e)}") while len(frames) < num_frames: frames.append(frames[-1].copy()) return frames def build_messages(task_type, text_prompt=None, image=None, video_frames=None): messages = [{"role": "system", "content": SYSTEM_PROMPT}] if task_type == "text": text = simplify_to_animation_description(text_prompt) messages.append({ "role": "user", "content": [{"type": "text", "text": f"Generate Lottie code: {text}"}] }) elif task_type == "image": text = simplify_to_animation_description(text_prompt) messages.append({ "role": "user", "content": [ {"type": "image", "image": image}, {"type": "text", "text": f"Animate this image: {text}"} ] }) elif task_type == "video": messages.append({ "role": "user", "content": [ {"type": "video", "video": video_frames, "fps": 8.0}, {"type": "text", "text": VIDEO_PROMPT} ] }) return messages def prepare_inference_input(processor, messages, device): text_input = processor.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) image_inputs, video_inputs = process_vision_info(messages) inputs = processor( text=[text_input], images=image_inputs if image_inputs else None, videos=video_inputs if video_inputs else None, padding=False, return_tensors="pt" ) input_ids = inputs['input_ids'] attention_mask = inputs['attention_mask'] target_len = 1500 if input_ids.shape[1] < target_len: pad_len = target_len - input_ids.shape[1] input_ids = torch.cat([ torch.full((1, pad_len), PAD_TOKEN, dtype=torch.long), input_ids ], dim=1) attention_mask = torch.cat([ torch.zeros((1, pad_len), dtype=torch.long), attention_mask ], dim=1) result = { 'input_ids': input_ids.to(device), 'attention_mask': attention_mask.to(device), 'pixel_values': inputs.get('pixel_values').to(device) if inputs.get('pixel_values') is not None else None, 'image_grid_thw': inputs.get('image_grid_thw').to(device) if inputs.get('image_grid_thw') is not None else None, 'pixel_values_videos': inputs.get('pixel_values_videos').to(device) if inputs.get('pixel_values_videos') is not None else None, 'video_grid_thw': inputs.get('video_grid_thw').to(device) if inputs.get('video_grid_thw') is not None else None, } return result def generate_lottie(model, inputs, max_tokens, device, use_sampling=False, temperature=0.95, top_p=0.25, top_k=5): """生成 Lottie tokens""" model.transformer.rope_deltas = None position_ids, _ = model.transformer.get_rope_index( input_ids=inputs['input_ids'], attention_mask=inputs['attention_mask'], image_grid_thw=inputs.get('image_grid_thw'), video_grid_thw=inputs.get('video_grid_thw'), ) position_ids = position_ids * inputs['attention_mask'][None, ] kwargs = { 'input_ids': inputs['input_ids'], 'attention_mask': inputs['attention_mask'], 'pixel_values': inputs.get('pixel_values'), 'image_grid_thw': inputs.get('image_grid_thw'), 'pixel_values_videos': inputs.get('pixel_values_videos'), 'video_grid_thw': inputs.get('video_grid_thw'), 'position_ids': position_ids, 'max_new_tokens': max_tokens, 'eos_token_id': LOTTIE_EOS, 'pad_token_id': PAD_TOKEN, 'use_cache': True, } if use_sampling: kwargs.update({'do_sample': True, 'temperature': temperature, 'top_p': top_p, 'top_k': top_k}) else: kwargs.update({'do_sample': False, 'num_beams': 1}) with torch.no_grad(): outputs = model.transformer.generate(**kwargs) input_len = inputs['input_ids'].shape[1] generated_ids = outputs[0][input_len:].tolist() del outputs, kwargs, position_ids if generated_ids and generated_ids[0] == LOTTIE_BOS: generated_ids = generated_ids[1:] if LOTTIE_EOS in generated_ids: generated_ids = generated_ids[:generated_ids.index(LOTTIE_EOS)] return generated_ids def fix_lottie_json(anim): anim_ip = int(round(anim.get("ip", 0))) anim_op = int(round(anim.get("op", 16))) anim["ip"] = anim_ip anim["op"] = anim_op anim["fr"] = int(round(anim.get("fr", 8))) anim["ddd"] = int(anim.get("ddd", 0)) def fix_t_recursive(obj): if isinstance(obj, dict): if obj.get("a") == 1 and isinstance(obj.get("k"), list): for kf in obj["k"]: if isinstance(kf, dict) and "t" in kf: kf["t"] = int(round(kf["t"])) for v in obj.values(): fix_t_recursive(v) elif isinstance(obj, list): for item in obj: fix_t_recursive(item) fix_t_recursive(anim) max_x = float(anim.get("w", 512)) max_y = float(anim.get("h", 512)) def collect_pos(layer): nonlocal max_x, max_y p = layer.get("ks", {}).get("p", {}) if isinstance(p, dict): if p.get("a", 0) == 0: pv = p.get("k", [0, 0]) if isinstance(pv, list) and len(pv) >= 2: max_x = max(max_x, float(pv[0])) max_y = max(max_y, float(pv[1])) else: for kf in p.get("k", []): if isinstance(kf, dict): for sv in (kf.get("s", []), kf.get("e", [])): if isinstance(sv, list) and len(sv) >= 2: max_x = max(max_x, float(sv[0])) max_y = max(max_y, float(sv[1])) for sub in layer.get("layers", []): collect_pos(sub) for layer in anim.get("layers", []): collect_pos(layer) anim["w"] = max(512, int((max_x * 1.1 + 15) // 16 * 16)) anim["h"] = max(512, int((max_y * 1.1 + 15) // 16 * 16)) valid_inds = set() for layer in anim.get("layers", []): if "ind" in layer: valid_inds.add(int(layer["ind"])) def clean_shapes(shapes): if not isinstance(shapes, list): return shapes cleaned = [] for sh in shapes: if not isinstance(sh, dict): continue if sh.get("ty") == "gr": sh["it"] = clean_shapes(sh.get("it", [])) if not sh["it"]: continue has_tr = any(item.get("ty") == "tr" for item in sh["it"] if isinstance(item, dict)) if not has_tr: sh["it"].append({ "ty": "tr", "nm": "", "a": {"a": 0, "k": [0, 0], "ix": 1}, "p": {"a": 0, "k": [0, 0], "ix": 2}, "s": {"a": 0, "k": [100, 100], "ix": 3}, "r": {"a": 0, "k": 0, "ix": 6}, "o": {"a": 0, "k": 100, "ix": 7}, "sk": {"a": 0, "k": 0, "ix": 4}, "sa": {"a": 0, "k": 0, "ix": 5}, "hd": False }) cleaned.append(sh) return cleaned def fix_layer(layer): ip = int(round(layer.get("ip", anim_ip))) op = int(round(layer.get("op", anim_op))) layer["ip"] = max(anim_ip, ip) layer["op"] = min(anim_op, max(layer["ip"] + 1, op)) layer["st"] = int(round(layer.get("st", anim_ip))) if "ind" in layer: layer["ind"] = int(layer["ind"]) if "parent" in layer: p = int(layer["parent"]) if p in valid_inds: layer["parent"] = p else: del layer["parent"] layer.pop("ct", None) if "shapes" in layer: layer["shapes"] = clean_shapes(layer["shapes"]) for sub in layer.get("layers", []): fix_layer(sub) return layer fixed_layers = [] for l in anim.get("layers", []): fix_layer(l) shapes = l.get("shapes", []) if l.get("ty") == 4 and not shapes: continue fixed_layers.append(l) anim["layers"] = fixed_layers for asset in anim.get("assets", []): if "layers" in asset: fixed = [] for l in asset["layers"]: fix_layer(l) if l.get("ty") == 4 and not l.get("shapes"): continue fixed.append(l) asset["layers"] = fixed return anim def tokens_to_lottie_json(generated_ids): reconstructed_tensor = LottieTensor.from_list(generated_ids) reconstructed_sequence = reconstructed_tensor.to_sequence() reconstructed = from_sequence(reconstructed_sequence) json_animation = { "v": reconstructed.get("v", "5.5.2"), "fr": reconstructed.get("fr", 8), "ip": reconstructed.get("ip", 0), "op": reconstructed.get("op", 16), "w": reconstructed.get("w", 512), "h": reconstructed.get("h", 512), "nm": reconstructed.get("nm", "Animation"), "ddd": reconstructed.get("ddd", 0), "assets": [], "layers": [], } if "fonts" in reconstructed and reconstructed["fonts"]: fonts_data = reconstructed["fonts"] if isinstance(fonts_data, dict) and "list" in fonts_data: fonts_json = {"list": []} for font in fonts_data["list"]: if isinstance(font, Font): fonts_json["list"].append(font_to_json(font)) else: fonts_json["list"].append(font) json_animation["fonts"] = fonts_json if "chars" in reconstructed and reconstructed["chars"]: chars_json = [] for char in reconstructed["chars"]: if isinstance(char, Chars): chars_json.append(char_to_json(char)) else: chars_json.append(char) json_animation["chars"] = chars_json for asset in reconstructed.get("assets", []): asset_json = dict(asset) if "layers" in asset: asset_json["layers"] = [] for layer in asset["layers"]: if isinstance(layer, ShapeLayer): asset_json["layers"].append(shape_layer_to_json(layer)) elif isinstance(layer, NullLayer): asset_json["layers"].append(null_layer_to_json(layer)) elif isinstance(layer, PreCompLayer): asset_json["layers"].append(precomp_layer_to_json(layer)) elif isinstance(layer, TextLayer): asset_json["layers"].append(text_layer_to_json(layer)) elif isinstance(layer, SolidColorLayer): asset_json["layers"].append(solid_layer_to_json(layer)) else: asset_json["layers"].append(layer) json_animation["assets"].append(asset_json) for layer in reconstructed.get("layers", []): if isinstance(layer, ShapeLayer): json_animation["layers"].append(shape_layer_to_json(layer)) elif isinstance(layer, NullLayer): json_animation["layers"].append(null_layer_to_json(layer)) elif isinstance(layer, PreCompLayer): json_animation["layers"].append(precomp_layer_to_json(layer)) elif isinstance(layer, TextLayer): json_animation["layers"].append(text_layer_to_json(layer)) elif isinstance(layer, SolidColorLayer): json_animation["layers"].append(solid_layer_to_json(layer)) else: json_animation["layers"].append(layer) json_animation = fix_lottie_json(json_animation) return json_animation _lottie_js_cache = None def get_lottie_js(): global _lottie_js_cache if _lottie_js_cache is not None: return _lottie_js_cache local_path = "lottie.min.js" if os.path.exists(local_path): with open(local_path, 'r', encoding='utf-8') as f: _lottie_js_cache = f.read() print(f"✅ Loaded local lottie.min.js ({len(_lottie_js_cache)} bytes)") else: _lottie_js_cache = '' print("⚠️ Using CDN lottie.min.js (local file not found)") return _lottie_js_cache def create_lottie_html(animation_data, height=600): bg_style = """ background-image: linear-gradient(45deg, #666666 25%, transparent 25%), linear-gradient(-45deg, #666666 25%, transparent 25%), linear-gradient(45deg, transparent 75%, #666666 75%), linear-gradient(-45deg, transparent 75%, #666666 75%); background-size: 16px 16px; background-position: 0 0, 0 8px, 8px -8px, -8px 0px; background-color: #444444; """ lottie_js = get_lottie_js() if lottie_js.startswith('" animation_json_escaped = json.dumps(animation_data).replace('\\', '\\\\').replace("'", "\\'") anim_width = animation_data.get('w', 512) anim_height = animation_data.get('h', 512) inner_html = f"""
{lottie_script}