From bd500d54b456e6c6eb789c1a97afc9497e8b44dc Mon Sep 17 00:00:00 2001 From: Kenneth Estanislao Date: Sun, 23 Aug 2026 02:55:32 +0800 Subject: [PATCH] auto download some models Retarget download url for safer model controls --- modules/face_analyser.py | 3 + modules/model_downloader.py | 178 ++++++++++++++++++ modules/processors/frame/face_enhancer.py | 20 +- .../processors/frame/face_enhancer_gpen256.py | 44 +++-- .../processors/frame/face_enhancer_gpen512.py | 44 +++-- modules/processors/frame/face_swapper.py | 32 ++-- 6 files changed, 278 insertions(+), 43 deletions(-) create mode 100644 modules/model_downloader.py diff --git a/modules/face_analyser.py b/modules/face_analyser.py index 51fa5ea..e961ad3 100644 --- a/modules/face_analyser.py +++ b/modules/face_analyser.py @@ -29,6 +29,9 @@ def get_face_analyser() -> Any: from modules.processors.frame._onnx_enhancer import ( build_provider_config, ) + from modules.model_downloader import ensure_insightface_pack + + ensure_insightface_pack('buffalo_l') providers = build_provider_config() FACE_ANALYSER = insightface.app.FaceAnalysis( name='buffalo_l', diff --git a/modules/model_downloader.py b/modules/model_downloader.py new file mode 100644 index 0000000..0b84be6 --- /dev/null +++ b/modules/model_downloader.py @@ -0,0 +1,178 @@ +import os +import platform +import ssl +import threading +import urllib.error +import urllib.request +from typing import Dict, List, Optional + +from tqdm import tqdm + +from modules.paths import MODELS_DIR + +HF_REPO_ID = "hacksider/deep-live-cam" +HF_RESOLVE_BASE = f"https://huggingface.co/{HF_REPO_ID}/resolve/main/" + +MODEL_SIZES: Dict[str, int] = { + "inswapper_128.onnx": 554253681, + "inswapper_128_fp16.onnx": 277680638, + "gfpgan-1024.onnx": 365875079, + "GPEN-BFR-256.onnx": 75715262, + "GPEN-BFR-512.onnx": 284244491, + "buffalo_l/buffalo_l/1k3d68.onnx": 143607619, + "buffalo_l/buffalo_l/2d106det.onnx": 5030888, + "buffalo_l/buffalo_l/det_10g.onnx": 16923827, + "buffalo_l/buffalo_l/genderage.onnx": 1322532, + "buffalo_l/buffalo_l/w600k_r50.onnx": 174383860, +} + +_LOCKS: Dict[str, threading.Lock] = {} +_LOCKS_GUARD = threading.Lock() + +CHUNK_SIZE = 1024 * 256 + + +def _ssl_context(): + if platform.system().lower() == "darwin": + return ssl._create_unverified_context() + return None + + +def _lock_for(key: str) -> threading.Lock: + with _LOCKS_GUARD: + if key not in _LOCKS: + _LOCKS[key] = threading.Lock() + return _LOCKS[key] + + +def resolve_url(name: str) -> str: + return HF_RESOLVE_BASE + name.replace(os.sep, "/") + + +def local_path(name: str, dest_dir: Optional[str] = None) -> str: + if dest_dir is not None: + return os.path.join(dest_dir, os.path.basename(name)) + return os.path.join(MODELS_DIR, *name.replace("/", os.sep).split(os.sep)) + + +def expected_size(name: str) -> Optional[int]: + return MODEL_SIZES.get(name.replace(os.sep, "/")) + + +def is_present(name: str, dest_dir: Optional[str] = None) -> bool: + path = local_path(name, dest_dir) + return os.path.isfile(path) and os.path.getsize(path) > 0 + + +def _download(name: str, url: str, target: str, size: Optional[int]) -> bool: + os.makedirs(os.path.dirname(target) or MODELS_DIR, exist_ok=True) + partial = target + ".part" + resume_from = os.path.getsize(partial) if os.path.isfile(partial) else 0 + + headers = {"User-Agent": "Deep-Live-Cam"} + if resume_from: + headers["Range"] = f"bytes={resume_from}-" + + try: + request = urllib.request.Request(url, headers=headers) + response = urllib.request.urlopen(request, context=_ssl_context(), timeout=60) + except urllib.error.HTTPError as error: + if resume_from and error.code in (416, 501): + try: + os.remove(partial) + except OSError: + pass + return _download(name, url, target, size) + print(f"[DLC.MODELS] Failed to download {name}: HTTP {error.code}") + return False + except (urllib.error.URLError, OSError) as error: + print(f"[DLC.MODELS] Failed to download {name}: {error}") + return False + + with response: + if resume_from and getattr(response, "status", 200) != 206: + resume_from = 0 + remaining = int(response.headers.get("Content-Length", 0) or 0) + total = size or (resume_from + remaining) or None + mode = "ab" if resume_from else "wb" + try: + with open(partial, mode) as handle: + with tqdm( + total=total, + initial=resume_from, + desc=f"Downloading {os.path.basename(name)}", + unit="B", + unit_scale=True, + unit_divisor=1024, + ) as progress: + while True: + buffer = response.read(CHUNK_SIZE) + if not buffer: + break + handle.write(buffer) + progress.update(len(buffer)) + except (urllib.error.URLError, OSError) as error: + print(f"[DLC.MODELS] Download of {name} interrupted: {error}") + return False + + downloaded = os.path.getsize(partial) + if size is not None and downloaded != size: + print(f"[DLC.MODELS] {name} is {downloaded} bytes, expected {size}. Discarding.") + try: + os.remove(partial) + except OSError: + pass + return False + + try: + os.replace(partial, target) + except OSError as error: + print(f"[DLC.MODELS] Could not finalise {name}: {error}") + return False + return True + + +def ensure_model( + name: str, quiet: bool = False, dest_dir: Optional[str] = None +) -> Optional[str]: + name = name.replace(os.sep, "/") + target = local_path(name, dest_dir) + + with _lock_for(target): + if is_present(name, dest_dir): + return target + if not quiet: + print(f"[DLC.MODELS] {name} not found in models folder, downloading...") + if _download(name, resolve_url(name), target, expected_size(name)): + return target + return None + + +def ensure_any(names: List[str]) -> Optional[str]: + for name in names: + if is_present(name): + return local_path(name) + for name in names: + path = ensure_model(name) + if path is not None: + return path + return None + + +def ensure_insightface_pack(name: str = "buffalo_l") -> bool: + members = [n for n in MODEL_SIZES if n.startswith(f"{name}/")] + if not members: + return False + + dest_dir = os.path.join(os.path.expanduser("~"), ".insightface", "models", name) + if all(is_present(member, dest_dir) for member in members): + return True + + print(f"[DLC.MODELS] insightface pack '{name}' is missing, downloading...") + ok = True + for member in members: + if ensure_model(member, quiet=True, dest_dir=dest_dir) is None: + ok = False + if not ok: + print(f"[DLC.MODELS] Could not pre-fill '{name}'; insightface will retry.") + return ok diff --git a/modules/processors/frame/face_enhancer.py b/modules/processors/frame/face_enhancer.py index eff313a..8650a93 100644 --- a/modules/processors/frame/face_enhancer.py +++ b/modules/processors/frame/face_enhancer.py @@ -23,6 +23,7 @@ FACE_ENHANCER = None THREAD_SEMAPHORE = threading.Semaphore() THREAD_LOCK = threading.Lock() NAME = "DLC.FACE-ENHANCER" +MODEL_FILE = "gfpgan-1024.onnx" abs_dir = os.path.dirname(os.path.abspath(__file__)) models_dir = os.path.join( @@ -44,11 +45,12 @@ FFHQ_TEMPLATE_512 = np.array( def pre_check() -> bool: - model_path = os.path.join(models_dir, "gfpgan-1024.onnx") - if not os.path.exists(model_path): + from modules.model_downloader import ensure_model + + if ensure_model(MODEL_FILE) is None: update_status( - f"GFPGAN ONNX model not found at {model_path}. " - "Please place gfpgan-1024.onnx in the models folder.", + f"Could not obtain {MODEL_FILE}. Place it in the models folder " + "manually or check your internet connection.", NAME, ) return False @@ -73,11 +75,15 @@ def get_face_enhancer() -> onnxruntime.InferenceSession: with THREAD_LOCK: if FACE_ENHANCER is None: - model_path = os.path.join(models_dir, "gfpgan-1024.onnx") + from modules.model_downloader import ensure_model - if not os.path.exists(model_path): + model_path = ensure_model(MODEL_FILE) + + if model_path is None: raise FileNotFoundError( - f"{NAME}: Model not found at {model_path}" + f"{NAME}: Model not found at " + f"{os.path.join(models_dir, MODEL_FILE)} and could not be " + "downloaded" ) try: diff --git a/modules/processors/frame/face_enhancer_gpen256.py b/modules/processors/frame/face_enhancer_gpen256.py index 7d1fae7..1d77cd6 100644 --- a/modules/processors/frame/face_enhancer_gpen256.py +++ b/modules/processors/frame/face_enhancer_gpen256.py @@ -22,7 +22,7 @@ from modules.processors.frame._onnx_enhancer import ( NAME = "DLC.FACE-ENHANCER-GPEN256" INPUT_SIZE = 256 -MODEL_URL = "https://github.com/harisreedhar/Face-Upscalers-ONNX/releases/download/GPEN-BFR/GPEN-BFR-256.onnx" +MODEL_MIRROR_URL = "https://github.com/harisreedhar/Face-Upscalers-ONNX/releases/download/GPEN-BFR/GPEN-BFR-256.onnx" MODEL_FILE = "GPEN-BFR-256.onnx" ENHANCER = None @@ -34,12 +34,33 @@ models_dir = os.path.join( ) +def _obtain_model(): + from modules.model_downloader import ensure_model + + model_path = ensure_model(MODEL_FILE) + if model_path is not None: + return model_path + + update_status(f"Retrying {MODEL_FILE} from the mirror...", NAME) + from modules.utilities import conditional_download + + try: + conditional_download(models_dir, [MODEL_MIRROR_URL]) + except Exception as error: + update_status(f"Mirror download failed: {error}", NAME) + return None + fallback = os.path.join(models_dir, MODEL_FILE) + return fallback if os.path.exists(fallback) else None + + def pre_check() -> bool: - model_path = os.path.join(models_dir, MODEL_FILE) - if not os.path.exists(model_path): - update_status(f"Downloading {MODEL_FILE}...", NAME) - from modules.utilities import conditional_download - conditional_download(models_dir, [MODEL_URL]) + if _obtain_model() is None: + update_status( + f"Could not obtain {MODEL_FILE}. Place it in the models folder " + "manually or check your internet connection.", + NAME, + ) + return False return True @@ -54,12 +75,11 @@ def get_enhancer() -> Any: global ENHANCER with THREAD_LOCK: if ENHANCER is None: - model_path = os.path.join(models_dir, MODEL_FILE) - if not os.path.exists(model_path): - from modules.utilities import conditional_download - conditional_download(models_dir, [MODEL_URL]) - if not os.path.exists(model_path): - raise FileNotFoundError(f"Model file not found: {model_path}") + model_path = _obtain_model() + if model_path is None: + raise FileNotFoundError( + f"Model file not found: {os.path.join(models_dir, MODEL_FILE)}" + ) print(f"{NAME}: Loading ONNX model from {model_path}") ENHANCER = create_onnx_session(model_path) warmup_session(ENHANCER) diff --git a/modules/processors/frame/face_enhancer_gpen512.py b/modules/processors/frame/face_enhancer_gpen512.py index 36168be..41296c9 100644 --- a/modules/processors/frame/face_enhancer_gpen512.py +++ b/modules/processors/frame/face_enhancer_gpen512.py @@ -22,7 +22,7 @@ from modules.processors.frame._onnx_enhancer import ( NAME = "DLC.FACE-ENHANCER-GPEN512" INPUT_SIZE = 512 -MODEL_URL = "https://github.com/harisreedhar/Face-Upscalers-ONNX/releases/download/GPEN-BFR/GPEN-BFR-512.onnx" +MODEL_MIRROR_URL = "https://github.com/harisreedhar/Face-Upscalers-ONNX/releases/download/GPEN-BFR/GPEN-BFR-512.onnx" MODEL_FILE = "GPEN-BFR-512.onnx" ENHANCER = None @@ -34,12 +34,33 @@ models_dir = os.path.join( ) +def _obtain_model(): + from modules.model_downloader import ensure_model + + model_path = ensure_model(MODEL_FILE) + if model_path is not None: + return model_path + + update_status(f"Retrying {MODEL_FILE} from the mirror...", NAME) + from modules.utilities import conditional_download + + try: + conditional_download(models_dir, [MODEL_MIRROR_URL]) + except Exception as error: + update_status(f"Mirror download failed: {error}", NAME) + return None + fallback = os.path.join(models_dir, MODEL_FILE) + return fallback if os.path.exists(fallback) else None + + def pre_check() -> bool: - model_path = os.path.join(models_dir, MODEL_FILE) - if not os.path.exists(model_path): - update_status(f"Downloading {MODEL_FILE}...", NAME) - from modules.utilities import conditional_download - conditional_download(models_dir, [MODEL_URL]) + if _obtain_model() is None: + update_status( + f"Could not obtain {MODEL_FILE}. Place it in the models folder " + "manually or check your internet connection.", + NAME, + ) + return False return True @@ -54,12 +75,11 @@ def get_enhancer() -> Any: global ENHANCER with THREAD_LOCK: if ENHANCER is None: - model_path = os.path.join(models_dir, MODEL_FILE) - if not os.path.exists(model_path): - from modules.utilities import conditional_download - conditional_download(models_dir, [MODEL_URL]) - if not os.path.exists(model_path): - raise FileNotFoundError(f"Model file not found: {model_path}") + model_path = _obtain_model() + if model_path is None: + raise FileNotFoundError( + f"Model file not found: {os.path.join(models_dir, MODEL_FILE)}" + ) print(f"{NAME}: Loading ONNX model from {model_path}") ENHANCER = create_onnx_session(model_path) warmup_session(ENHANCER) diff --git a/modules/processors/frame/face_swapper.py b/modules/processors/frame/face_swapper.py index 0005950..e25a14e 100644 --- a/modules/processors/frame/face_swapper.py +++ b/modules/processors/frame/face_swapper.py @@ -12,7 +12,6 @@ from modules.core import update_status from modules.face_analyser import get_one_face, get_many_faces, default_source_face from modules.typing import Face, Frame from modules.utilities import ( - conditional_download, is_image, is_video, ) @@ -192,21 +191,26 @@ models_dir = os.path.join( def pre_check() -> bool: # Use models_dir instead of abs_dir to save to the correct location download_directory_path = models_dir - + # Make sure the models directory exists, catch permission errors if they occur try: os.makedirs(download_directory_path, exist_ok=True) except OSError as e: logging.error(f"Failed to create directory {download_directory_path} due to permission error: {e}") return False - - # Use the direct download URL from Hugging Face (FP32 model for broad GPU compatibility) - conditional_download( - download_directory_path, - [ - "https://huggingface.co/hacksider/deep-live-cam/resolve/main/inswapper_128.onnx" - ], - ) + + from modules.model_downloader import ensure_any + + variants = ["inswapper_128.onnx", "inswapper_128_fp16.onnx"] + if _HAS_TORCH_CUDA: + variants.reverse() + if ensure_any(variants) is None: + update_status( + "Could not obtain the inswapper model. Place inswapper_128.onnx in " + "the models folder manually or check your internet connection.", + NAME, + ) + return False return True @@ -242,8 +246,12 @@ def get_face_swapper() -> Any: elif os.path.exists(fp32_path): model_path = fp32_path else: - update_status(f"No inswapper model found in {models_dir}.", NAME) - return None + if not pre_check(): + return None + model_path = fp16_path if os.path.exists(fp16_path) else fp32_path + if not os.path.exists(model_path): + update_status(f"No inswapper model found in {models_dir}.", NAME) + return None # On Apple Silicon, rewrite Pad(reflect) → Slice+Concat so # CoreML can run the entire model in a single partition on # the Neural Engine instead of bouncing between CPU and ANE.