auto download some models

Retarget download url for safer model controls
This commit is contained in:
Kenneth Estanislao
2026-08-23 02:55:32 +08:00
parent 987f6b392b
commit bd500d54b4
6 changed files with 278 additions and 43 deletions
+3
View File
@@ -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',
+178
View File
@@ -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
+13 -7
View File
@@ -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:
@@ -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)
@@ -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)
+20 -12
View File
@@ -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.