Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| # ========================================================== | |
| # FILE: ghostpack.py | |
| # ========================================================== | |
| import os, sys, time, json, argparse, importlib.util, subprocess, traceback | |
| import torch, einops, numpy as np, gradio as gr | |
| from PIL import Image | |
| from diffusers import AutoencoderKLHunyuanVideo | |
| from transformers import ( | |
| LlamaModel, CLIPTextModel, LlamaTokenizerFast, CLIPTokenizer, | |
| SiglipImageProcessor, SiglipVisionModel | |
| ) | |
| try: | |
| from diffusers_helper.hf_login import login | |
| from diffusers_helper.hunyuan import ( | |
| encode_prompt_conds, vae_decode, vae_encode, vae_decode_fake | |
| ) | |
| from diffusers_helper.utils import ( | |
| save_bcthw_as_mp4, crop_or_pad_yield_mask, soft_append_bcthw, | |
| resize_and_center_crop, generate_timestamp | |
| ) | |
| from diffusers_helper.models.hunyuan_video_packed import HunyuanVideoTransformer3DModelPacked | |
| from diffusers_helper.pipelines.k_diffusion_hunyuan import sample_hunyuan | |
| from diffusers_helper.memory import ( | |
| gpu, get_cuda_free_memory_gb, move_model_to_device_with_memory_preservation, | |
| offload_model_from_device_for_memory_preservation, fake_diffusers_current_device, | |
| DynamicSwapInstaller, unload_complete_models, load_model_as_complete | |
| ) | |
| from diffusers_helper.thread_utils import AsyncStream, async_run | |
| from diffusers_helper.gradio.progress_bar import make_progress_bar_css, make_progress_bar_html | |
| from diffusers_helper.clip_vision import hf_clip_vision_encode | |
| from diffusers_helper.bucket_tools import find_nearest_bucket | |
| except ImportError as e: | |
| with open(os.path.join(os.path.abspath(os.path.dirname(__file__)), 'outputs', 'install_logs.txt'), 'a') as f: | |
| f.write(f"[Dependency Error] {str(e)}\n") | |
| print(f"Dependency error: {str(e)}. Check outputs/install_logs.txt.") | |
| sys.exit(1) | |
| try: | |
| from huggingface_hub import hf_hub_download | |
| from safetensors.torch import load_file | |
| except ImportError as e: | |
| with open(os.path.join(os.path.abspath(os.path.dirname(__file__)), 'outputs', 'install_logs.txt'), 'a') as f: | |
| f.write(f"[Dependency Error] {str(e)}\n") | |
| print(f"Dependency error: {str(e)}. Install huggingface_hub and safetensors: pip install huggingface_hub safetensors") | |
| sys.exit(1) | |
| # ------------------------- CLI ---------------------------- | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument('--share', action='store_true') | |
| parser.add_argument('--server', type=str, default='0.0.0.0') | |
| parser.add_argument('--port', type=int) | |
| parser.add_argument('--inbrowser', action='store_true') | |
| parser.add_argument('--cli', action='store_true') | |
| args = parser.parse_args() | |
| BASE = os.path.abspath(os.path.dirname(__file__)) | |
| os.environ['HF_HOME'] = os.path.join(BASE, 'hf_download') | |
| LORA_CACHE = os.path.join(BASE, 'dlora') | |
| os.makedirs(LORA_CACHE, exist_ok=True) | |
| # Set HF token from environment variable | |
| HF_TOKEN = os.getenv('HF_TOKEN', 'XXXXXXXXXXXXXXXXXXXXXXXX') | |
| if args.cli: | |
| print("👻 GhostPack F1 Pro CLI\n") | |
| print("python ghostpack.py # launch UI") | |
| print("python ghostpack.py --cli # show help\n") | |
| sys.exit(0) | |
| # ---------------------- Paths ----------------------------- | |
| OUT_BASE = os.path.join(BASE, 'outputs') | |
| OUT_IMG = os.path.join(OUT_BASE, 'img') | |
| OUT_TMP = os.path.join(OUT_BASE, 'tmp_vid') | |
| OUT_VID = os.path.join(OUT_BASE, 'vid') | |
| PROMPT_LOG = os.path.join(OUT_BASE, 'prompts.txt') | |
| SAVED_PROMPTS = os.path.join(OUT_BASE, 'saved_prompts.json') | |
| INSTALL_LOG = os.path.join(OUT_BASE, 'install_logs.txt') | |
| for d in (OUT_BASE, OUT_IMG, OUT_TMP, OUT_VID): | |
| os.makedirs(d, exist_ok=True) | |
| if not os.path.exists(SAVED_PROMPTS): | |
| json.dump([], open(SAVED_PROMPTS,'w')) | |
| if not os.path.exists(INSTALL_LOG): | |
| open(INSTALL_LOG,'w').close() | |
| # ---------------- Auto-Downloader ------------------------ | |
| def auto_download_fastvideo_lora(): | |
| repo_id = "Kijai/HunyuanVideo_comfy" | |
| filename = "hyvideo_FastVideo_LoRA-fp8.safetensors" | |
| try: | |
| msg, lora_path = download_lora(repo_id, filename, HF_TOKEN) | |
| return msg | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[Auto-Download Error] {repo_id}/{filename}: {str(e)}\n") | |
| return f"❌ Auto-download failed: {str(e)}" | |
| # Run auto-downloader at startup | |
| auto_download_status = auto_download_fastvideo_lora() | |
| # ---------------- Prompt utils --------------------------- | |
| def get_last_prompts(): | |
| return json.load(open(SAVED_PROMPTS))[-5:][::-1] | |
| def save_prompt_fn(p, n): | |
| if not p: | |
| return "❌ No prompt" | |
| data = json.load(open(SAVED_PROMPTS)) | |
| entry = {'prompt': p, 'negative': n} | |
| if entry not in data: | |
| data.append(entry) | |
| json.dump(data, open(SAVED_PROMPTS,'w')) | |
| return "✅ Saved" | |
| def load_prompt_fn(idx): | |
| lst = get_last_prompts() | |
| return lst[idx]['prompt'] if idx < len(lst) else "" | |
| # ---------------- Cleanup utils -------------------------- | |
| def clear_temp_videos(): | |
| try: | |
| [os.remove(os.path.join(OUT_TMP,f)) for f in os.listdir(OUT_TMP)] | |
| return "✅ Temp cleared" | |
| except Exception as e: | |
| return f"❌ Failed to clear temp: {str(e)}" | |
| def clear_old_files(): | |
| try: | |
| cutoff = time.time() - 7*24*3600 | |
| c = 0 | |
| for d in (OUT_TMP, OUT_IMG): | |
| for f in os.listdir(d): | |
| p = os.path.join(d, f) | |
| if os.path.isfile(p) and os.path.getmtime(p) < cutoff: | |
| os.remove(p) | |
| c += 1 | |
| return f"✅ {c} old files removed" | |
| except Exception as e: | |
| return f"❌ Failed to clear old files: {str(e)}" | |
| def clear_images(): | |
| try: | |
| [os.remove(os.path.join(OUT_IMG,f)) for f in os.listdir(OUT_IMG)] | |
| return "✅ Images cleared" | |
| except Exception as e: | |
| return f"❌ Failed to clear images: {str(e)}" | |
| def clear_videos(): | |
| try: | |
| [os.remove(os.path.join(OUT_VID,f)) for f in os.listdir(OUT_VID)] | |
| return "✅ Videos cleared" | |
| except Exception as e: | |
| return f"❌ Failed to clear videos: {str(e)}" | |
| # ---------------- Gallery helpers ------------------------ | |
| def list_images(): | |
| try: | |
| return sorted( | |
| [os.path.join(OUT_IMG,f) for f in os.listdir(OUT_IMG) if f.lower().endswith(('.png','.jpg'))], | |
| key=os.path.getmtime | |
| ) | |
| except Exception: | |
| return [] | |
| def list_videos(): | |
| try: | |
| return sorted( | |
| [os.path.join(OUT_VID,f) for f in os.listdir(OUT_VID) if f.lower().endswith('.mp4')], | |
| key=os.path.getmtime | |
| ) | |
| except Exception: | |
| return [] | |
| def list_loras(): | |
| try: | |
| return sorted( | |
| [os.path.join(LORA_CACHE,f) for f in os.listdir(LORA_CACHE) if f.lower().endswith('.safetensors')], | |
| key=os.path.getmtime | |
| ) | |
| except Exception: | |
| return [] | |
| def load_image(sel): | |
| try: | |
| imgs = list_images() | |
| if sel in [os.path.basename(p) for p in imgs]: | |
| pth = imgs[[os.path.basename(p) for p in imgs].index(sel)] | |
| return gr.update(value=pth), gr.update(value=os.path.basename(pth)) | |
| return gr.update(), gr.update() | |
| except Exception as e: | |
| return gr.update(), gr.update(value=f"❌ Error: {str(e)}") | |
| def load_video(sel): | |
| try: | |
| vids = list_videos() | |
| if sel in [os.path.basename(p) for p in vids]: | |
| pth = vids[[os.path.basename(p) for p in vids].index(sel)] | |
| return gr.update(value=pth), gr.update(value=os.path.basename(pth)) | |
| return gr.update(), gr.update() | |
| except Exception as e: | |
| return gr.update(), gr.update(value=f"❌ Error: {str(e)}") | |
| def load_lora_select(sel): | |
| try: | |
| loras = list_loras() | |
| if sel in [os.path.basename(p) for p in loras]: | |
| pth = loras[[os.path.basename(p) for p in loras].index(sel)] | |
| return gr.update(value=pth), gr.update(value=os.path.basename(pth)) | |
| return gr.update(), gr.update() | |
| except Exception as e: | |
| return gr.update(), gr.update(value=f"❌ Error: {str(e)}") | |
| def next_image_and_load(sel): | |
| try: | |
| imgs = list_images() | |
| if not imgs: | |
| return gr.update(), gr.update() | |
| names = [os.path.basename(i) for i in imgs] | |
| idx = (names.index(sel)+1) % len(names) if sel in names else 0 | |
| pth = imgs[idx] | |
| return gr.update(value=pth), gr.update(value=os.path.basename(pth)) | |
| except Exception: | |
| return gr.update(), gr.update() | |
| def next_video_and_load(sel): | |
| try: | |
| vids = list_videos() | |
| if not vids: | |
| return gr.update(), gr.update() | |
| names = [os.path.basename(v) for v in vids] | |
| idx = (names.index(sel)+1) % len(names) if sel in names else 0 | |
| pth = vids[idx] | |
| return gr.update(value=pth), gr.update(value=os.path.basename(pth)) | |
| except Exception: | |
| return gr.update(), gr.update() | |
| def next_lora_and_load(sel): | |
| try: | |
| loras = list_loras() | |
| if not loras: | |
| return gr.update(), gr.update() | |
| names = [os.path.basename(l) for l in loras] | |
| idx = (names.index(sel)+1) % len(names) if sel in names else 0 | |
| pth = loras[idx] | |
| return gr.update(value=pth), gr.update(value=os.path.basename(pth)) | |
| except Exception: | |
| return gr.update(), gr.update() | |
| def gallery_image_select(evt: gr.SelectData): | |
| try: | |
| imgs = list_images() | |
| if evt.index is not None and evt.index < len(imgs): | |
| pth = imgs[evt.index] | |
| return gr.update(value=pth), gr.update(value=os.path.basename(pth)) | |
| return gr.update(), gr.update() | |
| except Exception: | |
| return gr.update(), gr.update() | |
| def gallery_video_select(evt: gr.SelectData): | |
| try: | |
| vids = list_videos() | |
| if evt.index is not None and evt.index < len(vids): | |
| pth = vids[evt.index] | |
| return gr.update(value=pth), gr.update(value=os.path.basename(pth)) | |
| return gr.update(), gr.update() | |
| except Exception: | |
| return gr.update(), gr.update() | |
| def gallery_lora_select(evt: gr.SelectData): | |
| try: | |
| loras = list_loras() | |
| if evt.index is not None and evt.index < len(loras): | |
| pth = loras[evt.index] | |
| return gr.update(value=pth), gr.update(value=os.path.basename(pth)) | |
| return gr.update(), gr.update() | |
| except Exception: | |
| return gr.update(), gr.update() | |
| # ---------------- Install status ------------------------- | |
| def check_mod(n): | |
| return importlib.util.find_spec(n) is not None | |
| def status_xformers(): | |
| return "✅ xformers" if check_mod("xformers") else "❌ xformers" | |
| def status_sage(): | |
| return "✅ sage-attn" if check_mod("sageattention") else "❌ sage-attn" | |
| def status_flash(): | |
| return "✅ flash-attn" if check_mod("flash_attn") else "⚠️ flash-attn" | |
| def install_pkg(pkg, warn=None): | |
| try: | |
| if warn: | |
| print(warn) | |
| time.sleep(1) | |
| out = subprocess.check_output( | |
| [sys.executable, "-m", "pip", "install", pkg], | |
| stderr=subprocess.STDOUT, text=True | |
| ) | |
| res = f"✅ {pkg}\n{out}\n" | |
| except subprocess.CalledProcessError as e: | |
| res = f"❌ {pkg}\n{e.output}\n" | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[{pkg}] {res}") | |
| return res | |
| install_xformers = lambda: install_pkg("xformers") | |
| install_sage_attn = lambda: install_pkg("sage-attn") | |
| install_flash_attn = lambda: install_pkg("flash-attn","⚠️ long compile") | |
| refresh_logs = lambda: open(INSTALL_LOG).read() | |
| clear_logs = lambda: (open(INSTALL_LOG,'w').close() or "✅ Logs cleared") | |
| # ---------------- LoRA Download and Load ------------------ | |
| def download_lora(repo_id, filename, hf_token): | |
| try: | |
| lora_path = os.path.join(LORA_CACHE, filename) | |
| if not os.path.exists(lora_path): | |
| if get_cuda_free_memory_gb(gpu) < 2: | |
| return "❌ Low VRAM (<2GB). Free up memory.", None | |
| hf_hub_download( | |
| repo_id=repo_id, | |
| filename=filename, | |
| local_dir=LORA_CACHE, | |
| token=hf_token | |
| ) | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[LoRA Download] {repo_id}/{filename} downloaded to {lora_path}\n") | |
| return "✅ LoRA downloaded", lora_path | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[LoRA Download Error] {repo_id}/{filename}: {str(e)}\n") | |
| return f"❌ Download failed: {str(e)}", None | |
| def load_lora(transformer, lora_path, lora_weight): | |
| try: | |
| if lora_path and os.path.exists(lora_path): | |
| if hasattr(transformer, 'load_lora_weights'): | |
| transformer.load_lora_weights( | |
| lora_path, | |
| adapter_name="fastvideo", | |
| weight=lora_weight | |
| ) | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[LoRA Load] {lora_path} loaded with standard method, weight {lora_weight}\n") | |
| return "✅ LoRA loaded" | |
| else: | |
| # Manual LoRA loading | |
| lora_weights = load_file(lora_path) | |
| state_dict = transformer.state_dict() | |
| for key, value in lora_weights.items(): | |
| if key in state_dict: | |
| state_dict[key] = state_dict[key] + lora_weight * value.to(state_dict[key].device) | |
| else: | |
| # Try partial key matching for common transformer layers | |
| for model_key in state_dict: | |
| if key.split('.')[-1] in model_key and ('self_attn' in model_key or 'ffn' in model_key): | |
| state_dict[model_key] = state_dict[model_key] + lora_weight * value.to(state_dict[model_key].device) | |
| break | |
| else: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[LoRA Load Warning] Key {key} not found in model state_dict\n") | |
| transformer.load_state_dict(state_dict) | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[LoRA Load] {lora_path} loaded manually, weight {lora_weight}\n") | |
| return "✅ LoRA loaded manually" | |
| return "❌ No valid LoRA path" | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[LoRA Load Error] {lora_path}: {str(e)}\n") | |
| return f"⚠️ LoRA not supported, using base model: {str(e)}" | |
| def delete_lora(sel): | |
| try: | |
| loras = list_loras() | |
| if sel in [os.path.basename(p) for p in loras]: | |
| pth = loras[[os.path.basename(p) for p in loras].index(sel)] | |
| os.remove(pth) | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[LoRA Delete] {pth} deleted\n") | |
| return "✅ LoRA deleted", gr.update(choices=[os.path.basename(l) for l in list_loras()], value=None) | |
| return "❌ No LoRA selected", gr.update() | |
| except Exception as e: | |
| return f"❌ Delete failed: {str(e)}", gr.update() | |
| # ---------------- Model load ----------------------------- | |
| free_mem = get_cuda_free_memory_gb(gpu) | |
| hv = free_mem > 60 | |
| try: | |
| text_encoder = LlamaModel.from_pretrained( | |
| "hunyuanvideo-community/HunyuanVideo", | |
| subfolder='text_encoder', torch_dtype=torch.float16, token=HF_TOKEN | |
| ).cpu().eval() | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[Model Load Error] text_encoder: {str(e)}\n") | |
| raise gr.Error(f"Failed to load text_encoder: {str(e)}") | |
| try: | |
| text_encoder_2 = CLIPTextModel.from_pretrained( | |
| "hunyuanvideo-community/HunyuanVideo", | |
| subfolder='text_encoder_2', torch_dtype=torch.float16, token=HF_TOKEN | |
| ).cpu().eval() | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[Model Load Error] text_encoder_2: {str(e)}\n") | |
| raise gr.Error(f"Failed to load text_encoder_2: {str(e)}") | |
| try: | |
| tokenizer = LlamaTokenizerFast.from_pretrained( | |
| "hunyuanvideo-community/HunyuanVideo", | |
| subfolder='tokenizer', token=HF_TOKEN | |
| ) | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[Model Load Error] tokenizer: {str(e)}\n") | |
| raise gr.Error(f"Failed to load tokenizer: {str(e)}") | |
| try: | |
| tokenizer_2 = CLIPTokenizer.from_pretrained( | |
| "hunyuanvideo-community/HunyuanVideo", | |
| subfolder='tokenizer_2', token=HF_TOKEN | |
| ) | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[Model Load Error] tokenizer_2: {str(e)}\n") | |
| raise gr.Error(f"Failed to load tokenizer_2: {str(e)}") | |
| try: | |
| vae = AutoencoderKLHunyuanVideo.from_pretrained( | |
| "hunyuanvideo-community/HunyuanVideo", | |
| subfolder='vae', torch_dtype=torch.float16, token=HF_TOKEN | |
| ).cpu().eval() | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[Model Load Error] vae: {str(e)}\n") | |
| raise gr.Error(f"Failed to load vae: {str(e)}") | |
| try: | |
| feature_extractor = SiglipImageProcessor.from_pretrained( | |
| "lllyasviel/flux_redux_bfl", subfolder='feature_extractor', token=HF_TOKEN | |
| ) | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[Model Load Error] feature_extractor: {str(e)}\n") | |
| raise gr.Error(f"Failed to load feature_extractor: {str(e)}") | |
| try: | |
| image_encoder = SiglipVisionModel.from_pretrained( | |
| "lllyasviel/flux_redux_bfl", | |
| subfolder='image_encoder', torch_dtype=torch.float16, token=HF_TOKEN | |
| ).cpu().eval() | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[Model Load Error] image_encoder: {str(e)}\n") | |
| raise gr.Error(f"Failed to load image_encoder: {str(e)}") | |
| try: | |
| transformer = HunyuanVideoTransformer3DModelPacked.from_pretrained( | |
| "lllyasviel/FramePack_F1_I2V_HY_20250503", | |
| torch_dtype=torch.bfloat16, token=HF_TOKEN | |
| ).cpu().eval() | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[Model Load Error] transformer: {str(e)}\n") | |
| raise gr.Error(f"Failed to load transformer: {str(e)}") | |
| if not hv: | |
| vae.enable_slicing() | |
| vae.enable_tiling() | |
| transformer.high_quality_fp32_output_for_inference = True | |
| transformer.to(dtype=torch.bfloat16) | |
| for m in (vae, image_encoder, text_encoder, text_encoder_2): | |
| m.to(dtype=torch.float16) | |
| for m in (vae, image_encoder, text_encoder, text_encoder_2, transformer): | |
| m.requires_grad_(False) | |
| if not hv: | |
| DynamicSwapInstaller.install_model(transformer, device=gpu) | |
| DynamicSwapInstaller.install_model(text_encoder, device=gpu) | |
| else: | |
| for m in (text_encoder, text_encoder_2, image_encoder, vae, transformer): | |
| m.to(gpu) | |
| stream = AsyncStream() | |
| # ---------------- Worker ------------------------------- | |
| def worker(img, prompt, n_p, seed, secs, win, stp, cfg, gsc, rsc, keep, tea, crf, lora_path, lora_weight, disable_prompt_mods): | |
| # Download and load LoRA if specified | |
| lora_msg = "No LoRA specified" | |
| if lora_path: | |
| try: | |
| if lora_path.startswith("http") or lora_path.startswith("Kijai/"): | |
| repo_id = "Kijai/HunyuanVideo_comfy" | |
| filename = "hyvideo_FastVideo_LoRA-fp8.safetensors" | |
| lora_msg, lora_path = download_lora(repo_id, filename, HF_TOKEN) | |
| if not lora_path: | |
| raise gr.Error(lora_msg) | |
| lora_msg = load_lora(transformer, lora_path, lora_weight) | |
| if "⚠️" in lora_msg or "❌" in lora_msg: | |
| print(lora_msg) | |
| else: | |
| stp = 8 # Override steps for FastVideo LoRA | |
| except Exception as e: | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[LoRA Error] {lora_path}: {str(e)}\n") | |
| lora_msg = f"⚠️ LoRA failed, using base model: {str(e)}" | |
| # Validate prompt | |
| try: | |
| if not disable_prompt_mods: | |
| if "stop" not in prompt.lower() and secs > 5: | |
| prompt += " The subject stops moving after 5 seconds." | |
| if "smooth" not in prompt.lower(): | |
| prompt = f"Smooth animation: {prompt}" | |
| if "silent" not in prompt.lower(): | |
| prompt += ", silent" | |
| if len(prompt.split()) > 50: | |
| print("Warning: Complex prompt may slow rendering or cause instability.") | |
| except Exception as e: | |
| raise gr.Error(f"Prompt validation failed: {str(e)}") | |
| # Check VRAM availability | |
| if get_cuda_free_memory_gb(gpu) < 2: | |
| raise gr.Error("Low VRAM (<2GB). Lower 'kee' or 'win'.") | |
| sections = max(round((secs*30)/(win*4)), 1) | |
| jid = generate_timestamp() | |
| try: | |
| with open(PROMPT_LOG, 'a') as f: | |
| f.write(f"{jid}\t{prompt}\t{n_p}\n") | |
| except Exception as e: | |
| print(f"Failed to log prompt: {str(e)}") | |
| stream.output_queue.push(('progress', (None, "", make_progress_bar_html(0, "Start")))) | |
| try: | |
| if not hv: | |
| unload_complete_models(text_encoder, text_encoder_2, image_encoder, vae, transformer) | |
| fake_diffusers_current_device(text_encoder, gpu) | |
| load_model_as_complete(text_encoder_2, gpu) | |
| lv, cp = encode_prompt_conds(prompt, text_encoder, text_encoder_2, tokenizer, tokenizer_2) | |
| if cfg == 1: | |
| lv_n = torch.zeros_like(lv) | |
| cp_n = torch.zeros_like(cp) | |
| else: | |
| lv_n, cp_n = encode_prompt_conds(n_p, text_encoder, text_encoder_2, tokenizer, tokenizer_2) | |
| lv, m = crop_or_pad_yield_mask(lv, 512) | |
| lv_n, m_n = crop_or_pad_yield_mask(lv_n, 512) | |
| lv, cp, lv_n, cp_n = [x.to(torch.bfloat16) for x in (lv, cp, lv_n, cp_n)] | |
| H, W, _ = img.shape | |
| h, w = find_nearest_bucket(H, W, 640) | |
| img_np = resize_and_center_crop(img, w, h) | |
| Image.fromarray(img_np).save(os.path.join(OUT_IMG, f"{jid}.png")) | |
| img_pt = (torch.from_numpy(img_np).float()/127.5-1).permute(2,0,1)[None,:,None] | |
| if not hv: | |
| load_model_as_complete(vae, gpu) | |
| start_lat = vae_encode(img_pt, vae) | |
| if not hv: | |
| load_model_as_complete(image_encoder, gpu) | |
| img_emb = hf_clip_vision_encode(img_np, feature_extractor, image_encoder).last_hidden_state.to(torch.bfloat16) | |
| gen = torch.Generator("cpu").manual_seed(seed) | |
| hist_lat = torch.zeros((1,16,1+2+16,h//8,w//8), dtype=torch.float16).cpu() | |
| hist_px = None | |
| total = 0 | |
| pad_seq = [3] + [2]*(sections-3) + [1,0] if sections>4 else list(reversed(range(sections))) | |
| for pad in pad_seq: | |
| last = pad == 0 | |
| if stream.input_queue.top() == "end": | |
| stream.output_queue.push(("end", None)) | |
| return | |
| pad_sz = pad * win | |
| idx = torch.arange(0, sum([1,pad_sz,win,1,2,16]))[None].to(device=gpu) | |
| a,b,c,d,e,f = idx.split([1,pad_sz,win,1,2,16],1) | |
| clean_idx = torch.cat([a,d],1) | |
| pre = start_lat.to(hist_lat) | |
| post, two, four = hist_lat[:,:,:1+2+16].split([1,2,16],2) | |
| clean = torch.cat([pre, post],2) | |
| if not hv: | |
| unload_complete_models() | |
| move_model_to_device_with_memory_preservation(transformer, gpu, keep) | |
| transformer.initialize_teacache(tea, stp) | |
| def cb(d): | |
| pv = vae_decode_fake(d["denoised"]) | |
| pv = (pv*255).cpu().numpy().clip(0,255).astype(np.uint8) | |
| pv = einops.rearrange(pv, "b c t h w -> (b h) (t w) c") | |
| cur = d["i"]+1 | |
| stream.output_queue.push(('progress', (pv, f"{cur}/{stp}", make_progress_bar_html(int(100*cur/stp), f"{cur}/{stp}")))) | |
| if stream.input_queue.top()=="end": | |
| stream.output_queue.push(("end", None)) | |
| raise KeyboardInterrupt | |
| new_lat = sample_hunyuan( | |
| transformer=transformer, sampler="unipc", width=w, height=h, frames=win*4-3, | |
| real_guidance_scale=cfg, distilled_guidance_scale=gsc, guidance_rescale=rsc, | |
| num_inference_steps=stp, generator=gen, | |
| prompt_embeds=lv, prompt_embeds_mask=m, prompt_poolers=cp, | |
| negative_prompt_embeds=lv_n, negative_prompt_embeds_mask=m_n, negative_prompt_poolers=cp_n, | |
| device=gpu, dtype=torch.bfloat16, image_embeddings=img_emb, | |
| latent_indices=c, clean_latents=clean, clean_latent_indices=clean_idx, | |
| clean_latents_2x=two, clean_latent_2x_indices=e, | |
| clean_latents_4x=four, clean_latent_4x_indices=f, callback=cb | |
| ) | |
| if last: | |
| new_lat = torch.cat([start_lat.to(new_lat), new_lat],2) | |
| total += new_lat.shape[2] | |
| hist_lat = torch.cat([new_lat.to(hist_lat), hist_lat],2) | |
| if not hv: | |
| offload_model_from_device_for_memory_preservation(transformer, gpu, 8) | |
| load_model_as_complete(vae, gpu) | |
| real = hist_lat[:,:,:total] | |
| if hist_px is None: | |
| hist_px = vae_decode(real, vae).cpu() | |
| else: | |
| overlap = win*4-3 | |
| curr = vae_decode(real[:,:,:win*2], vae).cpu() | |
| hist_px = soft_append_bcthw(curr, hist_px, overlap) | |
| if not hv: | |
| unload_complete_models() | |
| tmp = os.path.join(OUT_TMP, f"{jid}_{total}.mp4") | |
| save_bcthw_as_mp4(hist_px, tmp, fps=30, crf=crf) | |
| stream.output_queue.push(('file', tmp)) | |
| if last: | |
| fin = os.path.join(OUT_VID, f"{jid}_{total}.mp4") | |
| os.replace(tmp, fin) | |
| stream.output_queue.push(('complete', fin)) | |
| break | |
| except Exception as e: | |
| traceback.print_exc() | |
| with open(INSTALL_LOG, 'a') as f: | |
| f.write(f"[Worker Error] {str(e)}\n") | |
| stream.output_queue.push(("end", None)) | |
| return lora_msg | |
| # ---------------- Process Function ----------------------- | |
| def process(img, prm, npr, sd, sec, win, stp, cfg, gsc, rsc, kee, tea, crf, lora_path, lora_weight, disable_prompt_mods): | |
| global stream | |
| if img is None: | |
| raise gr.Error("Upload an image") | |
| yield None, None, "", "", gr.update(interactive=False), gr.update(interactive=True), gr.update() | |
| stream = AsyncStream() | |
| lora_msg = async_run(worker, img, prm, npr, sd, sec, win, stp, cfg, gsc, rsc, kee, tea, crf, lora_path, lora_weight, disable_prompt_mods) | |
| out, log = None, "" | |
| while True: | |
| flag, data = stream.output_queue.next() | |
| if flag == "file": | |
| out = data | |
| yield out, gr.update(), gr.update(), log, gr.update(interactive=False), gr.update(interactive=True), gr.update(value=lora_msg) | |
| if flag == "progress": | |
| pv, desc, html = data | |
| log = desc | |
| yield gr.update(), gr.update(visible=True, value=pv), desc, html, gr.update(interactive=False), gr.update(interactive=True), gr.update(value=lora_msg) | |
| if flag in ("complete", "end"): | |
| yield out, gr.update(visible=False), gr.update(), "", gr.update(interactive=True), gr.update(interactive=False), gr.update(value=lora_msg) | |
| break | |
| def end_process(): | |
| stream.input_queue.push("end") | |
| # ------------------- UI ------------------------------ | |
| quick_prompts = [ | |
| ["Smooth animation: A character waves for 3 seconds, then stands still for 2 seconds, static camera, silent."], | |
| ["Smooth animation: A character moves for 5 seconds, static camera, silent."] | |
| ] | |
| css = make_progress_bar_css() + """ | |
| .orange-button{background:#ff6200;color:#fff;border-color:#ff6200;} | |
| .load-button{background:#4CAF50;color:#fff;border-color:#4CAF50;margin-left:10px;} | |
| .big-setting-button{background:#0066cc;color:#fff;border:none;padding:14px 24px;font-size:18px;width:100%;border-radius:6px;margin:8px 0;} | |
| .styled-dropdown{width:250px;padding:5px;border-radius:4px;} | |
| .viewer-column{width:100%;max-width:900px;margin:0 auto;} | |
| .media-preview img,.media-preview video{max-width:100%;height:380px;object-fit:contain;border:1px solid #444;border-radius:6px;} | |
| .media-container{display:flex;gap:20px;align-items:flex-start;} | |
| .control-box{min-width:220px;} | |
| .control-grid{display:grid;grid-template-columns:1fr 1fr;gap:10px;} | |
| .image-gallery{display:grid!important;grid-template-columns:repeat(auto-fit,minmax(300px,1fr))!important;gap:10px;padding:10px!important;overflow-y:auto!important;max-height:360px!important;} | |
| .image-gallery .gallery-item{padding:10px;height:360px!important;width:300px!important;} | |
| .image-gallery img{object-fit:contain;height:360px!important;width:300px!important;} | |
| .video-gallery{display:grid!important;grid-template-columns:repeat(auto-fit,minmax(300px,1fr))!important;gap:10px;padding:10px!important;overflow-y:auto!important;max-height:360px!important;} | |
| .video-gallery .gallery-item{padding:10px;height:360px!important;width:300px!important;} | |
| .video-gallery video{object-fit:contain;height:360px!important;width:300px!important;} | |
| .lora-gallery{display:grid!important;grid-template-columns:repeat(auto-fit,minmax(300px,1fr))!important;gap:10px;padding:10px!important;overflow-y:auto!important;max-height:360px!important;} | |
| .lora-gallery .gallery-item{padding:10px;height:360px!important;width:300px!important;} | |
| .lora-gallery .gallery-item div{text-align:center;font-size:16px;color:#fff;} | |
| """ | |
| blk = gr.Blocks(css=css).queue() | |
| with blk: | |
| gr.Markdown("# 👻 GhostPack F1 Pro") | |
| with gr.Tabs(): | |
| with gr.TabItem("👻 Generate"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| img_in = gr.Image(sources="upload", type="numpy", label="Image", height=320) | |
| prm = gr.Textbox(label="Prompt") | |
| npr = gr.Textbox(label="Negative Prompt", value="low quality, blurry, speaking, talking, moaning, vocalizing, lip movement, mouth animation, sound, dialogue, speech, whispering, shouting, lip sync, facial animation, expressive face, verbal expression, animated mouth") | |
| save_msg = gr.Markdown("") | |
| lora_path = gr.Textbox( | |
| label="FastVideo LoRA Path or HF Repo", | |
| value="Kijai/HunyuanVideo_comfy", | |
| placeholder="e.g., Kijai/HunyuanVideo_comfy/hyvideo_FastVideo_LoRA-fp8.safetensors or /path/to/hyvideo_FastVideo_LoRA-fp8.safetensors" | |
| ) | |
| lora_weight = gr.Slider(label="LoRA Weight", minimum=0.5, maximum=1.5, value=1.0, step=0.1) | |
| disable_prompt_mods = gr.Checkbox(label="Disable Prompt Modifications", value=False) | |
| lora_status_gen = gr.Markdown(value=auto_download_status) | |
| btn_save = gr.Button("Save Prompt") | |
| btn1, btn2, btn3 = gr.Button("Load Most Recent"), gr.Button("Load 2nd Recent"), gr.Button("Load 3rd Recent") | |
| ds = gr.Dataset(samples=quick_prompts, label="Quick List", components=[prm]) | |
| ds.click(lambda x: x[0], [ds], [prm]) | |
| btn_save.click(save_prompt_fn, [prm, npr], [save_msg]) | |
| btn1.click(lambda: load_prompt_fn(0), [], [prm]) | |
| btn2.click(lambda: load_prompt_fn(1), [], [prm]) | |
| btn3.click(lambda: load_prompt_fn(2), [], [prm]) | |
| with gr.Row(): | |
| b_go, b_end = gr.Button("Start"), gr.Button("End", interactive=False) | |
| with gr.Group(): | |
| tea = gr.Checkbox(label="Use TeaCache", value=True) | |
| se = gr.Number(label="Seed", value=31337, precision=0) | |
| sec = gr.Slider(label="Video Length (s)", minimum=1, maximum=120, value=5, step=0.1) | |
| win = gr.Slider(label="Latent Window", minimum=1, maximum=33, value=5, step=1) | |
| stp = gr.Slider(label="Steps", minimum=1, maximum=100, value=8, step=1) | |
| cfg = gr.Slider(label="CFG", minimum=1, maximum=32, value=1, step=0.01, visible=False) | |
| gsc = gr.Slider(label="Distilled CFG", minimum=1, maximum=32, value=5, step=0.01) | |
| rsc = gr.Slider(label="CFG Re-Scale", minimum=0, maximum=1, value=0.5, step=0.01) | |
| kee = gr.Slider(label="GPU Keep (GB)", minimum=4, maximum=free_mem, value=6, step=0.1) | |
| crf = gr.Slider(label="MP4 CRF", minimum=0, maximum=100, value=20, step=1) | |
| with gr.Column(): | |
| pv = gr.Image(label="Next Latents", height=200, visible=False) | |
| vid = gr.Video(label="Finished", autoplay=True, height=500, loop=True, show_share_button=False) | |
| log_md = gr.Markdown("") | |
| bar = gr.HTML("") | |
| b_go.click( | |
| process, | |
| [img_in, prm, npr, se, sec, win, stp, cfg, gsc, rsc, kee, tea, crf, lora_path, lora_weight, disable_prompt_mods], | |
| [vid, pv, log_md, bar, b_go, b_end, lora_status_gen] | |
| ) | |
| b_end.click(end_process) | |
| with gr.TabItem("🖼️ Image Gallery"): | |
| with gr.Row(elem_classes="media-container"): | |
| with gr.Column(scale=3): | |
| image_preview = gr.Image( | |
| label="Viewer", | |
| value=(list_images()[0] if list_images() else None), | |
| interactive=False, elem_classes="media-preview" | |
| ) | |
| with gr.Column(elem_classes="control-box"): | |
| image_dropdown = gr.Dropdown( | |
| choices=[os.path.basename(i) for i in list_images()], | |
| value=(os.path.basename(list_images()[0]) if list_images() else None), | |
| label="Select", elem_classes="styled-dropdown" | |
| ) | |
| with gr.Row(elem_classes="control-grid"): | |
| load_btn = gr.Button("Load", elem_classes="load-button") | |
| next_btn = gr.Button("Next", elem_classes="load-button") | |
| with gr.Row(elem_classes="control-grid"): | |
| refresh_btn = gr.Button("Refresh") | |
| delete_btn = gr.Button("Delete", elem_classes="orange-button") | |
| image_gallery = gr.Gallery( | |
| value=list_images(), label="Thumbnails", columns=6, height=360, | |
| allow_preview=False, type="filepath", elem_classes="image-gallery" | |
| ) | |
| load_btn.click(load_image, [image_dropdown], [image_preview, image_dropdown]) | |
| next_btn.click(next_image_and_load, [image_dropdown], [image_preview, image_dropdown]) | |
| refresh_btn.click(lambda: ( | |
| gr.update(choices=[os.path.basename(i) for i in list_images()], | |
| value=os.path.basename(list_images()[0]) if list_images() else None), | |
| gr.update(value=list_images()[0] if list_images() else None), | |
| gr.update(value=list_images()) | |
| ), [], [image_dropdown, image_preview, image_gallery]) | |
| delete_btn.click(lambda sel: (os.remove(os.path.join(OUT_IMG, sel)) if sel else None) or load_image(""), | |
| [image_dropdown], [image_preview, image_dropdown]) | |
| image_gallery.select(gallery_image_select, [], [image_preview, image_dropdown]) | |
| with gr.TabItem("🎬 Video Gallery"): | |
| with gr.Row(elem_classes="media-container"): | |
| with gr.Column(scale=3): | |
| video_preview = gr.Video( | |
| label="Viewer", | |
| value=(list_videos()[0] if list_videos() else None), | |
| autoplay=True, loop=True, interactive=False, elem_classes="media-preview" | |
| ) | |
| with gr.Column(elem_classes="control-box"): | |
| video_dropdown = gr.Dropdown( | |
| choices=[os.path.basename(v) for v in list_videos()], | |
| value=(os.path.basename(list_videos()[0]) if list_videos() else None), | |
| label="Select", elem_classes="styled-dropdown" | |
| ) | |
| with gr.Row(elem_classes="control-grid"): | |
| load_vbtn = gr.Button("Load", elem_classes="load-button") | |
| next_vbtn = gr.Button("Next", elem_classes="load-button") | |
| with gr.Row(elem_classes="control-grid"): | |
| refresh_v = gr.Button("Refresh") | |
| delete_v = gr.Button("Delete", elem_classes="orange-button") | |
| video_gallery = gr.Gallery( | |
| value=list_videos(), label="Thumbnails", columns=6, height=360, | |
| allow_preview=False, type="filepath", elem_classes="video-gallery" | |
| ) | |
| load_vbtn.click(load_video, [video_dropdown], [video_preview, video_dropdown]) | |
| next_vbtn.click(next_video_and_load, [video_dropdown], [video_preview, video_dropdown]) | |
| refresh_v.click(lambda: ( | |
| gr.update(choices=[os.path.basename(v) for v in list_videos()], | |
| value=os.path.basename(list_videos()[0]) if list_videos() else None), | |
| gr.update(value=list_videos()[0] if list_videos() else None), | |
| gr.update(value=list_videos()) | |
| ), [], [video_dropdown, video_preview, video_gallery]) | |
| delete_v.click(lambda sel: (os.remove(os.path.join(OUT_VID, sel)) if sel else None) or load_video(""), | |
| [video_dropdown], [video_preview, video_dropdown]) | |
| video_gallery.select(gallery_video_select, [], [video_preview, video_dropdown]) | |
| with gr.TabItem("📦 LoRA Management"): | |
| with gr.Row(elem_classes="media-container"): | |
| with gr.Column(scale=3): | |
| lora_status = gr.Markdown("") | |
| with gr.Column(elem_classes="control-box"): | |
| lora_dropdown = gr.Dropdown( | |
| choices=[os.path.basename(l) for l in list_loras()], | |
| value=(os.path.basename(list_loras()[0]) if list_loras() else None), | |
| label="Select LoRA", elem_classes="styled-dropdown" | |
| ) | |
| with gr.Row(elem_classes="control-grid"): | |
| load_lora_btn = gr.Button("Load", elem_classes="load-button") | |
| next_lora_btn = gr.Button("Next", elem_classes="load-button") | |
| with gr.Row(elem_classes="control-grid"): | |
| refresh_lora_btn = gr.Button("Refresh") | |
| delete_lora_btn = gr.Button("Delete", elem_classes="orange-button") | |
| download_fastvideo_btn = gr.Button("Download FastVideo LoRA", elem_classes="big-setting-button") | |
| lora_gallery = gr.Gallery( | |
| value=[(l, os.path.basename(l)) for l in list_loras()], label="LoRA Files", columns=6, height=360, | |
| allow_preview=False, elem_classes="lora-gallery" | |
| ) | |
| load_lora_btn.click(load_lora_select, [lora_dropdown], [lora_path, lora_dropdown]) | |
| next_lora_btn.click(next_lora_and_load, [lora_dropdown], [lora_path, lora_dropdown]) | |
| refresh_lora_btn.click(lambda: ( | |
| gr.update(choices=[os.path.basename(l) for l in list_loras()], | |
| value=os.path.basename(list_loras()[0]) if list_loras() else None), | |
| gr.update(value=[(l, os.path.basename(l)) for l in list_loras()]) | |
| ), [], [lora_dropdown, lora_gallery]) | |
| delete_lora_btn.click(delete_lora, [lora_dropdown], [lora_status, lora_dropdown]) | |
| download_fastvideo_btn.click( | |
| lambda: auto_download_fastvideo_lora(), | |
| [], [lora_status] | |
| ) | |
| lora_gallery.select(gallery_lora_select, [], [lora_path, lora_dropdown]) | |
| with gr.TabItem("👻 About"): | |
| gr.Markdown("## GhostPack F1 Pro") | |
| with gr.Row(): | |
| with gr.Column(): | |
| gr.Markdown("**🛠️ Description**\nImage-to-Video toolkit powered by HunyuanVideo & FramePack-F1") | |
| with gr.Column(): | |
| gr.Markdown("**📦 Version**\n2025-05-03") | |
| with gr.Column(): | |
| gr.Markdown("**✍️ Author**\nGhostAI") | |
| with gr.Column(): | |
| gr.Markdown("**🔗 Repo**\nhttps://huggingface.co/spaces/ghostai1/GhostPack") | |
| with gr.TabItem("⚙️ Settings"): | |
| ct = gr.Button("Clear Temp", elem_classes="big-setting-button") | |
| ctmsg = gr.Markdown("") | |
| co = gr.Button("Clear Old", elem_classes="big-setting-button") | |
| comsg= gr.Markdown("") | |
| ci = gr.Button("Clear Images", elem_classes="big-setting-button") | |
| cimg= gr.Markdown("") | |
| cv = gr.Button("Clear Videos", elem_classes="big-setting-button") | |
| cvid= gr.Markdown("") | |
| ct.click(clear_temp_videos, [], ctmsg) | |
| co.click(clear_old_files, [], comsg) | |
| ci.click(clear_images, [], cimg) | |
| cv.click(clear_videos, [], cvid) | |
| with gr.TabItem("🛠️ Install"): | |
| xs = gr.Textbox(value=status_xformers(), interactive=False, label="xformers") | |
| bx = gr.Button("Install xformers", elem_classes="big-setting-button") | |
| ss = gr.Textbox(value=status_sage(), interactive=False, label="sage-attn") | |
| bs = gr.Button("Install sage-attn", elem_classes="big-setting-button") | |
| fs = gr.Textbox(value=status_flash(),interactive=False, label="flash-attn") | |
| bf = gr.Button("Install flash-attn", elem_classes="big-setting-button") | |
| bx.click(install_xformers, [], xs) | |
| bs.click(install_sage_attn, [], ss) | |
| bf.click(install_flash_attn, [], fs) | |
| with gr.TabItem("📜 Logs"): | |
| logs = gr.Textbox(lines=20, interactive=False, label="Install Logs") | |
| rl = gr.Button("Refresh", elem_classes="big-setting-button") | |
| cl = gr.Button("Clear", elem_classes="big-setting-button") | |
| rl.click(refresh_logs, [], logs) | |
| cl.click(clear_logs, [], logs) | |
| # Force video previews to seek to 2s | |
| gr.HTML("""<script> | |
| document.querySelectorAll('.video-gallery video').forEach(v => { | |
| v.addEventListener('loadedmetadata', () => { | |
| if (v.duration > 2) v.currentTime = 2; | |
| }); | |
| }); | |
| </script>""") | |
| blk.launch( | |
| server_name=args.server, | |
| server_port=args.port, | |
| share=args.share, | |
| inbrowser=args.inbrowser | |
| ) |