Compare commits

..
36 Commits
Author SHA1 Message Date
iuyua9 bc48ba4308 docs: align GFPGAN model filenames (#1910)
README and models/instructions.txt named GFPGANv1.4, but the enhancer loads gfpgan-1024.onnx via onnxruntime. instructions.txt also pointed at a PyTorch .pth that the ONNX path cannot load at all, so the manual-fallback instructions were unusable.
2026-08-29 15:05:23 +08:00
Ihor Kuzmychov b53844eb57 fix: skip setrlimit(RLIMIT_DATA) on macOS (#1849)
setrlimit(RLIMIT_DATA) is rejected by the macOS kernel for the values used here,
crashing the app during startup. Since --max-memory defaults to suggest_max_memory()
(4 on Darwin), max_memory is always set on macOS and every launch reached this call.

Fixes #1848.
2026-08-29 14:53:57 +08:00
Chris G 20cafb0079 docs: change dependency version number (#1914) 2026-08-29 14:28:40 +08:00
Dopan 7ca6d0b202 Merge pull request #1864 from 5uck1ess/pr/webp-support
feat: WEBP source image support
2026-08-26 10:51:56 +08:00
Kenneth Estanislao f7db37679a Update readme
Includes website links and proper redirect to our official website
2026-08-23 05:12:24 +08:00
Kenneth Estanislao bd500d54b4 auto download some models
Retarget download url for safer model controls
2026-08-23 02:55:32 +08:00
cuyua9 987f6b392b fix: extract frames for map faces fallback (#1824)
Verified this fix. Confirmed the bug by reverting just the `modules/core.py` hunk and
re-running the new regression test — with the old code, `process_video`/`create_video`
run against a temp directory that was never populated when `map_faces=True`, since
`create_temp`/`extract_frames` were skipped for that case. That means map-faces video
runs were silently broken (empty or failed output).

The fix removes the `map_faces` guard so extraction always runs before the disk-based
fallback, which is correct for both cases that reach this branch (map_faces=True, and
non-map-faces pipe failures). `create_temp` is idempotent (mkdir exist_ok=True), so the
double-call for the non-map-faces path is harmless.
2026-08-14 06:44:22 +08:00
Dopan 97a44800a2 Merge pull request #1902 from 1ceseismic/fix/linux-camera-device-path-rebase
fix: no cam detected on Arch Linux, use string path for camera capture instead
2026-08-13 22:50:16 +08:00
Vito-M 345fa4a0b0 fix: no cam detected on Arch Linux, use string path for camera capture instead 2026-08-09 19:48:57 +12:00
Gao Yiman bdeeb3ace0 docs: add onnxruntime-openvino/OpenVINO version pairing note (#1893)
Thanks for this — useful reference table, and it lines up with the version-pairing issues we've been fixing (#1879). Merging.
2026-08-08 07:15:21 +08:00
Kenneth Estanislao 230217ec11 update on requirements
some update on what is needed to be updated
2026-07-29 05:20:29 +08:00
Kenneth Estanislao 156321f7a3 Upgrade onnxruntime-gpu to version 1.26.0
Updated onnxruntime-gpu version to 1.26.0 for non-Darwin platforms.
2026-07-29 04:37:54 +08:00
Nguyen Van Nam 8234965ee8 fix: clamp video frame seek index (#1790)
Prevent get_video_frame() from seeking to invalid frame positions.

The default frame_number=0 now resolves to the first frame instead of -1, and oversized frame requests clamp to the final valid frame instead of seeking past the end. Empty or invalid videos now return None safely after releasing the capture.

Affected files: capturer.py

Signed-off-by: Nguyen Van Nam <nam.nv205106@gmail.com>
2026-07-23 22:16:38 +08:00
Dopan ab64c186ec Merge pull request #1879 from dunegym/fix/openvino-dll-loading
fix: resolve OpenVINO DLL loading on Windows for OpenVINOExecutionProvider
2026-07-19 15:12:49 +08:00
Makaru b8e781e539 chore: remove trailing whitespace 2026-07-19 12:20:54 +08:00
KRSHH ff7ee0d219 Revise Quickstart section in README 2026-07-19 00:25:10 +05:30
Nguyen Van Nam 8d727eba3e fix: bound face-cluster count by available embeddings (#1793)
`find_cluster_centroids()` iterates `k` from 1..`max_k` unconditionally. If `len(embeddings) < max_k`, `KMeans(n_clusters=k)` will raise `ValueError` when `k` exceeds the number of samples. This is an unhandled crash path on small datasets.


Affected files: cluster_analysis.py

Signed-off-by: Nguyen Van Nam <nam.nv205106@gmail.com>
2026-07-14 23:55:23 +08:00
Cocoon-Break eba2a958d3 fix: skip empty face clusters in default_target_face (#1757)
Skip face clusters when no best face was detected, preventing a NoneType error while preserving normal face-detection behavior.

Closes #1755
2026-07-14 23:51:43 +08:00
dunegym 14ba4f9c0b fix: centralize OPENVINO_PROVIDER_CONFIG and log SystemExit
Address Sourcery review feedback on PR #1879:

- Move OPENVINO_PROVIDER_CONFIG from _onnx_enhancer.py to
  platform_info.py (a leaf module with no modules.* imports), so
  the enhancer and face_swapper no longer import each other just to
  share a constant. _onnx_enhancer re-exports it; face_swapper now
  imports it at module top level instead of inside get_face_swapper().
- Narrow run.py's SystemExit handling: catch SystemExit separately
  and print a [startup] message so the failure is visible instead
  of being swallowed alongside ImportError/FileNotFoundError.
2026-07-12 15:16:18 +08:00
dunegym 7d2d7fb1f3 fix: address PR review feedback — SystemExit, AUTO device, thread timing
- Catch SystemExit from add_openvino_libs_to_path() so a missing
  OpenVINO installation never causes a hard exit on Windows
- Replace hard-coded GPU+FP16 with AUTO:GPU,NPU,CPU device priority,
  letting OpenVINO pick the best available accelerator
- Extract shared OPENVINO_PROVIDER_CONFIG constant to avoid
  duplication between _onnx_enhancer and face_swapper
- Defer thread-suggestion evaluation until after execution_providers
  is assigned, fixing a latent timing bug that affected OpenVINO,
  CUDA, and DML thread hints
2026-07-11 12:45:00 +08:00
noahximus 57c4c32377 Merge pull request #1876 from ElKhalil19/main 2026-07-07 04:55:43 +08:00
El Khalil d00b09f5d8 docs: update manual installation to use shallow clone (#1866) 2026-07-03 19:54:39 +01:00
dunegym 897dc21da4 fix: resolve OpenVINO DLL loading on Windows for OpenVINOExecutionProvider
- Add add_openvino_libs_to_path() call in run.py before any ONNX
  InferenceSession creation to register openvino.dll directory
- Detect and advertise OpenVINOExecutionProvider in platform_info
  banner and accelerator label
- Prioritize openvino over dml in suggest_default_execution_provider
- Configure OpenVINO EP with GPU + FP16 device options for optimal
  performance (~13 FPS on Intel GPU vs ~1 FPS CPU fallback)
- Set thread hint to 1 when OpenVINO EP is active
2026-06-28 21:46:55 +08:00
Kenneth Estanislao 834092c891 Update Quick Start section to v2.7 RC6 2026-06-24 18:15:40 +08:00
Kenneth Estanislao da0672ad6b Enhance README with details on pre-built versions
Updated the README to clarify the benefits of pre-built versions and optimizations for hardware.
2026-06-24 18:14:59 +08:00
Tym Rabchuk 47dffeb307 fix(webp): finish extension centralization from pre-submission review
- build the video save-dialog filter from VIDEO_EXTENSIONS (_VIDEO_FILE_FILTER)
  instead of a hardcoded "Videos (*.mp4 *.mkv)" — the last filter that still
  drifted from the canonical set
- remove the now-dead file_types list (unused in both the fork and upstream;
  the PySide6 dialogs use the QFileDialog filter strings) and drop it from the
  centralization comment
2026-06-23 19:30:07 -04:00
Tym Rabchuk 9e1f0cc3a5 fix(webp): address review — drop broken GIF, robust ext check, centralize lists
Review feedback on #1831:
- Remove *.gif from the save/output dialog filter (PR had added it there).
  Verified empirically that cv2.imread/imwrite cannot decode OR encode GIF on
  OpenCV 4.10 *or* 4.11 (write raises, read returns None), so GIF silently
  failed on both ends — dropped from every dialog and from has_image_extension.
- has_image_extension now uses os.path.splitext so only the true extension
  counts ('photo.png.bak' / 'clip.webp.mp4' are no longer treated as images).
- Centralize the supported-extension set in modules.globals (IMAGE_EXTENSIONS /
  VIDEO_EXTENSIONS); file_types, all QFileDialog filters and has_image_extension
  now derive from it instead of hand-copied lists that had already drifted.

WEBP itself is unchanged and works (libwebp ships with opencv-python).
2026-06-22 20:33:11 -04:00
Kenneth Estanislao 834bc43768 Support non-ascii characters 2026-06-14 20:18:56 +08:00
Dopan 3b69413d61 Merge pull request #1845 from maxwbuckley/ruff-code-health
Add ruff CI gate and fix deterministic lint issues
2026-06-01 00:50:59 +08:00
Kenneth Estanislao 07e2e960c8 Update Quick Start version from v2.7 RC1 to v2.7 RC2 2026-05-24 18:55:35 +08:00
Max BuckleyandClaude Opus 4.7 ba27b75265 Use astral-sh/ruff-action for inline PR annotations
Swap the manual pip install + ruff check steps for astral-sh/ruff-action@v4.0.0.
Same pinned ruff 0.15.7, but with --output-format=github so violations appear
as inline annotations on the PR diff instead of a flat log.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-23 15:50:44 +02:00
Max BuckleyandClaude Opus 4.7 cfa8123b67 Add ruff CI gate and fix deterministic lint issues
Introduces pyproject.toml + .github/workflows/ruff.yml that gate
E701, E711, E712, F401, F541 on every PR and push to main.

Fixes the existing findings for those rules:
- Remove unused imports (sklearn.silhouette_score, numpy in several
  files, typing.Optional, get_one_face, gpu_cvt_color, sys,
  insightface.face_align)
- Annotate the intentional tkinter_fix side-effect import with
  `# noqa: F401`
- Split multi-statement `if x: y` one-liners onto separate lines
- Replace `state == True` / `state == False` with truthiness checks
- Drop `f` prefix from f-strings with no placeholders

F841 (unused-variable), E402 (module-level-import-not-at-top), and
F821 (undefined-name) are left out of the gate for now — they surface
real findings (including a latent NameError in face_swapper.py) that
require human review to fix safely.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-23 15:44:31 +02:00
Dopan 08b2dd2526 Merge pull request #1844 from hklcf/fix/bugfix-batch
lgtm
2026-05-23 16:54:41 +08:00
hklcf 886e64b320 Fix: resolve 5 confirmed bugs (imwrite_unicode, macOS memory, face_analyser None crash, silent sys.exit, core memory calc) 2026-05-23 10:37:20 +08:00
Kenneth Estanislao aa6f2cbade Update version from v2.7 beta to v2.7 RC1 in README 2026-05-21 05:11:41 +08:00
Tym Rabchuk 0b61ad5c0d feat: webp source image support
Ported from April 2026 Fork:
- has_image_extension() now recognizes .webp/.gif/.bmp
- is_image() checks extension before mimetypes (Windows mimetypes
  doesn't always register webp)
- File dialog filter includes *.webp
2026-05-18 21:00:33 -04:00
30 changed files with 779 additions and 191 deletions
+16
View File
@@ -0,0 +1,16 @@
name: ruff
on:
pull_request:
push:
branches: [main]
jobs:
ruff:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: astral-sh/ruff-action@v4.0.0
with:
version: "0.15.7"
args: "check --output-format=github"
+78 -27
View File
@@ -30,13 +30,45 @@ By using this software, you agree to these terms and commit to using it in a man
Users are expected to use this software responsibly and legally. If using a real person's face, obtain their consent and clearly label any output as a deepfake when sharing online. We are not responsible for end-user actions.
## Exclusive v2.7 beta Quick Start - Pre-built (Windows/Mac Silicon/CPU)
## Pre-built Deep-Live-Cam 2.7 Ultimate!
<a href="https://deeplivecam.net/index.php/quickstart"> <img src="media/Download.png" width="285" height="77" />
<p align="center">
<a href="https://deeplivecam.net/index.php/quickstart">
<img src="https://github.com/user-attachments/assets/fa2cdf79-c933-4b93-844a-b087192261ed" width="100%" alt="Lite / Ultimate Download Banner">
</a>
</p>
##### This is the fastest build you can get if you have a discrete NVIDIA or AMD GPU, CPU or Mac Silicon, And you'll receive special priority support. 2.7 beta is the best you can have with 30+ extra features than the open source version.
###### These Pre-builts are perfect for non-technical users or those who don't have time to, or can't manually install all the requirements. Just a heads-up: this is an open-source project, so you can also install it manually.
<p align="center">
<a href="https://deeplivecam.net/index.php/plans/nvidia-gpu?plan_id=0&group_id=1">
<img src="https://github.com/user-attachments/assets/56b61811-3a1e-4672-9b50-cf7f6e8e6852" width="40" alt="Windows">
</a>
&nbsp;&nbsp;&nbsp;
<a href="https://deeplivecam.net/index.php/plans/nvidia-gpu?plan_id=0&group_id=2">
<img src="https://github.com/user-attachments/assets/6538e3a6-c957-431a-b586-2d6abcf534dc" width="34" alt="Mac Silicon">
</a>
&nbsp;&nbsp;&nbsp;
<a href="https://deeplivecam.net/index.php/plans/nvidia-gpu?plan_id=0&group_id=3">
<img src="https://github.com/user-attachments/assets/ad45142e-426c-4364-a2a9-a512670cc62c" width="40" alt="CPU">
</a>
</p>
<p align="center">
<strong>Windows • Mac Silicon • CPU • NVIDIA • AMD</strong>
</p>
<p align="center">
Builds optimized for your hardware.
</p>
<p align="center">
<a href="https://deeplivecam.net/index.php/quickstart">
<img src="media/Download.png" width="280" alt="Download">
</a>
</p>
> **Ultimate** includes **30+ exclusive features**, performance optimizations, and **priority support** We only have a single official website which is https://deeplivecam.net . Please be careful on where you download other versions of this application aside from that website and this github repo.
Perfect if you want the fastest setup with **zero manual installation**, pre-configured dependencies, and optimized builds for every supported platform.
## TLDR; Live Deepfake in just 3 Clicks
![easysteps](https://github.com/user-attachments/assets/af825228-852c-411b-b787-ffd9aac72fc6)
@@ -109,7 +141,7 @@ This is more likely to work on your computer but will be slower as it utilizes t
**1. Set up Your Platform**
- Python (3.11 recommended)
- Python (3.14 recommended; 3.11-3.14 supported)
- pip
- git
- [ffmpeg](https://www.youtube.com/watch?v=OlNWCpFdVMA) - ```iex (irm ffmpeg.tc.ht)```
@@ -118,13 +150,13 @@ This is more likely to work on your computer but will be slower as it utilizes t
**2. Clone the Repository**
```bash
git clone https://github.com/hacksider/Deep-Live-Cam.git
git clone --depth 1 https://github.com/hacksider/Deep-Live-Cam.git
cd Deep-Live-Cam
```
**3. Download the Models**
1. [GFPGANv1.4](https://huggingface.co/hacksider/deep-live-cam/resolve/main/GFPGANv1.4.onnx)
1. [gfpgan-1024.onnx](https://huggingface.co/hacksider/deep-live-cam/resolve/main/gfpgan-1024.onnx)
2. [inswapper\_128\_fp16.onnx](https://huggingface.co/hacksider/deep-live-cam/resolve/main/inswapper_128_fp16.onnx)
Place these files in the "**models**" folder.
@@ -142,7 +174,7 @@ pip install -r requirements.txt
```
For Linux:
```bash
# Ensure you use the installed Python 3.11
# Ensure you use the installed Python 3.14
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
@@ -150,17 +182,17 @@ pip install -r requirements.txt
**For macOS:**
Apple Silicon (M1/M2/M3) requires specific setup:
Apple Silicon (M1 through M5) requires specific setup:
```bash
# Install Python 3.11 (specific version is important)
brew install python@3.11
# Install Python 3.14
brew install python@3.14
# Install tkinter package (required for the GUI)
brew install python-tk@3.11
brew install python-tk@3.14
# Create and activate virtual environment with Python 3.11
python3.11 -m venv venv
# Create and activate virtual environment with Python 3.14
python3.14 -m venv venv
source venv/bin/activate
# Install dependencies
@@ -201,7 +233,7 @@ pip install git+https://github.com/TencentARC/GFPGAN.git@master
```bash
pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu128
pip uninstall onnxruntime onnxruntime-gpu
pip install onnxruntime-gpu==1.21.0
pip install onnxruntime-gpu==1.26.0
```
3. Usage:
@@ -212,26 +244,29 @@ python run.py --execution-provider cuda
**CoreML Execution Provider (Apple Silicon)**
Apple Silicon (M1/M2/M3) specific installation:
Apple Silicon (M1 through M5) specific installation:
1. Make sure you've completed the macOS setup above using Python 3.11.
2. Install dependencies:
1. Make sure you've completed the macOS setup above using Python 3.14.
2. No extra install step is needed — `requirements.txt` pulls the official
`onnxruntime` build, whose macOS wheels ship the CoreML execution provider.
If you previously installed the unmaintained `onnxruntime-silicon` fork,
remove it first, as it shadows the real package:
```bash
pip uninstall onnxruntime onnxruntime-silicon
pip install onnxruntime-silicon==1.13.1
pip uninstall onnxruntime-silicon
pip install -r requirements.txt
```
3. Usage:
```bash
python3.11 run.py --execution-provider coreml
python3.14 run.py --execution-provider coreml
```
**Important Notes for macOS:**
- You **must** use Python 3.11, not newer versions like 3.13
- Always run with `python3.11` command not just `python` if you have multiple Python versions installed
- If you get error about `_tkinter` missing, reinstall the tkinter package: `brew reinstall python-tk@3.11`
- Python 3.11 is the minimum (onnxruntime dropped 3.10); 3.14 is recommended
- Always run with `python3.14` command not just `python` if you have multiple Python versions installed
- If you get error about `_tkinter` missing, reinstall the tkinter package: `brew reinstall python-tk@3.14`
- If you get model loading errors, check that your models are in the correct folder
- If you encounter conflicts with other Python versions, consider uninstalling them:
```bash
@@ -239,9 +274,9 @@ python3.11 run.py --execution-provider coreml
brew list | grep python
# Uninstall conflicting versions if needed
brew uninstall --ignore-dependencies python@3.13
brew uninstall --ignore-dependencies python@3.11
# Keep only Python 3.11
# Keep only Python 3.14
brew cleanup
```
@@ -284,6 +319,22 @@ pip uninstall onnxruntime onnxruntime-openvino
pip install onnxruntime-openvino==1.21.0
```
**Note:** `onnxruntime-openvino` newer than 1.21.0 must be installed together with `openvino`, and the two versions must correspond one-to-one. The supported pairings are:
| onnxruntime-openvino | OpenVINO |
| --- | --- |
| 1.24.1 | 2025.4.1 |
| 1.23.0 | 2025.3 |
| 1.22.0 | 2025.1 |
```bash
# Example: onnxruntime-openvino 1.24.1 pairs with OpenVINO 2025.4.1
pip install openvino==2025.4.1
pip install onnxruntime-openvino==1.24.1
```
See the [OpenVINO Execution Provider requirements](https://onnxruntime.ai/docs/execution-providers/OpenVINO-ExecutionProvider.html#requirements) for the full version-mapping details.
2. Usage:
```bash
+8 -5
View File
@@ -14,7 +14,6 @@ if sys.platform == "win32":
import insightface
from insightface.app import FaceAnalysis
from insightface.utils import face_align
from modules.processors.frame.face_swapper import _fast_paste_back
from modules import platform_info
@@ -81,10 +80,14 @@ def capture_thread():
try:
capture_queue.put_nowait(frame)
except queue.Full:
try: capture_queue.get_nowait()
except queue.Empty: pass
try: capture_queue.put_nowait(frame)
except queue.Full: pass
try:
capture_queue.get_nowait()
except queue.Empty:
pass
try:
capture_queue.put_nowait(frame)
except queue.Full:
pass
cap_t = threading.Thread(target=capture_thread, daemon=True)
cap_t.start()
+1 -1
View File
@@ -1,4 +1,4 @@
just put the models in this folder -
https://huggingface.co/hacksider/deep-live-cam/resolve/main/inswapper_128_fp16.onnx?download=true
https://github.com/TencentARC/GFPGAN/releases/download/v1.3.4/GFPGANv1.4.pth
https://huggingface.co/hacksider/deep-live-cam/resolve/main/gfpgan-1024.onnx?download=true
+38 -18
View File
@@ -1,18 +1,38 @@
import os
import cv2
import numpy as np
# Utility function to support unicode characters in file paths for reading
def imread_unicode(path, flags=cv2.IMREAD_COLOR):
return cv2.imdecode(np.fromfile(path, dtype=np.uint8), flags)
# Utility function to support unicode characters in file paths for writing
def imwrite_unicode(path, img, params=None):
root, ext = os.path.splitext(path)
if not ext:
ext = ".png"
result, encoded_img = cv2.imencode(ext, img, params if params else [])
result, encoded_img = cv2.imencode(f".{ext}", img, params if params is not None else [])
encoded_img.tofile(path)
return True
return False
import os
import cv2
import numpy as np
# Utility function to support unicode characters in file paths for reading.
# OpenCV's cv2.imread() encodes the path with the locale ANSI code page on
# Windows, so it silently returns None for paths containing non-ASCII
# characters (Chinese, Japanese, Cyrillic, accents, ...). Reading the bytes
# through NumPy (which uses Python's unicode-aware file I/O) and decoding them
# in memory sidesteps that limitation. Returns None on failure, matching
# cv2.imread() so it stays a drop-in replacement.
def imread_unicode(path, flags=cv2.IMREAD_COLOR):
try:
data = np.fromfile(path, dtype=np.uint8)
if data.size == 0:
return None
return cv2.imdecode(data, flags)
except Exception:
return None
# Utility function to support unicode characters in file paths for writing.
# cv2.imwrite() has the same ANSI-path limitation, so we encode the image in
# memory and write the bytes out with NumPy's unicode-aware file I/O. Returns
# True/False like cv2.imwrite() so it stays a drop-in replacement.
def imwrite_unicode(path, img, params=None):
try:
root, ext = os.path.splitext(path)
if not ext:
ext = ".png"
result, encoded_img = cv2.imencode(ext, img, params if params is not None else [])
if not result:
return False
encoded_img.tofile(path)
return True
except Exception:
return False
+7 -2
View File
@@ -14,8 +14,13 @@ def get_video_frame(video_path: str, frame_number: int = 0) -> Any:
if modules.globals.color_correction:
capture.set(cv2.CAP_PROP_CONVERT_RGB, 1)
frame_total = capture.get(cv2.CAP_PROP_FRAME_COUNT)
capture.set(cv2.CAP_PROP_POS_FRAMES, min(frame_total, frame_number - 1))
frame_total = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
if frame_total <= 0:
capture.release()
return None
target_index = 0 if frame_number <= 1 else min(frame_total - 1, frame_number - 1)
capture.set(cv2.CAP_PROP_POS_FRAMES, target_index)
has_frame, frame = capture.read()
if has_frame and modules.globals.color_correction:
+12 -1
View File
@@ -1,10 +1,21 @@
import numpy as np
from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score
from typing import Any
def find_cluster_centroids(embeddings, max_k=10) -> Any:
n_samples = len(embeddings)
if n_samples == 0:
raise ValueError("embeddings must not be empty")
if max_k < 1:
raise ValueError("max_k must be at least 1")
max_k = min(max_k, n_samples)
if max_k == 1:
kmeans = KMeans(n_clusters=1, random_state=0)
kmeans.fit(embeddings)
return kmeans.cluster_centers_
inertia = []
cluster_centroids = []
K = range(1, max_k+1)
+20 -10
View File
@@ -58,7 +58,7 @@ def parse_args() -> None:
program.add_argument('--live-resizable', help='The live camera frame is resizable', dest='live_resizable', action='store_true', default=False)
program.add_argument('--max-memory', help='maximum amount of RAM in GB', dest='max_memory', type=int, default=suggest_max_memory())
program.add_argument('--execution-provider', help='execution provider', dest='execution_provider', default=[suggest_default_execution_provider()], choices=suggest_execution_providers(), nargs='+')
program.add_argument('--execution-threads', help='number of execution threads', dest='execution_threads', type=int, default=suggest_execution_threads())
program.add_argument('--execution-threads', help='number of execution threads', dest='execution_threads', type=int, default=None)
program.add_argument('-v', '--version', action='version', version=f'{modules.metadata.name} {modules.metadata.version}')
# register deprecated args
@@ -90,6 +90,12 @@ def parse_args() -> None:
modules.globals.execution_threads = args.execution_threads
modules.globals.lang = args.lang
# The argparse default (None) avoids evaluating suggest_execution_threads()
# before providers are decoded, and deprecated-arg overrides above may
# have already set execution_threads.
if modules.globals.execution_threads is None:
modules.globals.execution_threads = suggest_execution_threads()
#for ENHANCER tumblers:
for enhancer_key in ('face_enhancer', 'face_enhancer_gpen256', 'face_enhancer_gpen512'):
modules.globals.fp_ui[enhancer_key] = enhancer_key in args.frame_processor
@@ -132,9 +138,9 @@ def suggest_max_memory() -> int:
def suggest_default_execution_provider() -> str:
"""Pick the best available provider: cuda > rocm > coreml > dml > cpu."""
"""Pick the best available provider: cuda > rocm > coreml > openvino > dml > cpu."""
available = encode_execution_providers(onnxruntime.get_available_providers())
for pref in ('cuda', 'rocm', 'coreml', 'dml'):
for pref in ('cuda', 'rocm', 'coreml', 'openvino', 'dml'):
if pref in available:
return pref
return 'cpu'
@@ -157,6 +163,8 @@ def suggest_execution_threads() -> int:
return 1
if 'CUDAExecutionProvider' in modules.globals.execution_providers:
return 2
if 'OpenVINOExecutionProvider' in modules.globals.execution_providers:
return 1
# For CPU execution, use most cores but leave some for system
return max(4, min(cpu_count - 2, 16))
@@ -170,9 +178,11 @@ def limit_resources() -> None:
tensorflow.config.experimental.set_memory_growth(gpu, True)
# limit memory usage
if modules.globals.max_memory:
memory = modules.globals.max_memory * 1024 ** 3
# setrlimit(RLIMIT_DATA) fails with EINVAL on macOS, crashing on launch.
# See https://github.com/hacksider/Deep-Live-Cam/issues/1848
if platform.system().lower() == 'darwin':
memory = modules.globals.max_memory * 1024 ** 6
return
memory = modules.globals.max_memory * 1024 ** 3
if platform.system().lower() == 'windows':
import ctypes
kernel32 = ctypes.windll.kernel32
@@ -270,10 +280,9 @@ def start() -> None:
update_status('Falling back to disk-based processing...')
extraction_start = time.time()
if not modules.globals.map_faces:
create_temp(modules.globals.target_path)
update_status('Extracting frames...')
extract_frames(modules.globals.target_path)
create_temp(modules.globals.target_path)
update_status('Extracting frames...')
extract_frames(modules.globals.target_path)
extraction_time = time.time() - extraction_start
temp_frame_paths = get_temp_frame_paths(modules.globals.target_path)
@@ -324,7 +333,8 @@ def start() -> None:
def destroy(to_quit=True) -> None:
if modules.globals.target_path:
clean_temp(modules.globals.target_path)
if to_quit: quit()
if to_quit:
quit()
def run() -> None:
+16 -7
View File
@@ -4,9 +4,8 @@ from typing import Any
import insightface
import threading
import cv2
import numpy as np
import modules.globals
from modules import imread_unicode, imwrite_unicode
from tqdm import tqdm
from modules.typing import Frame
from modules.cluster_analysis import find_cluster_centroids, find_closest_centroid
@@ -30,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',
@@ -255,8 +257,10 @@ def add_blank_map() -> Any:
def get_unique_faces_from_target_image() -> Any:
try:
modules.globals.source_target_map = []
target_frame = cv2.imread(modules.globals.target_path)
target_frame = imread_unicode(modules.globals.target_path)
many_faces = get_many_faces(target_frame)
if many_faces is None:
return None
i = 0
for face in many_faces:
@@ -289,8 +293,10 @@ def get_unique_faces_from_target_video() -> Any:
i = 0
for temp_frame_path in tqdm(temp_frame_paths, desc="Extracting face embeddings from frames"):
temp_frame = cv2.imread(temp_frame_path)
temp_frame = imread_unicode(temp_frame_path)
many_faces = get_many_faces(temp_frame)
if many_faces is None:
continue
for face in many_faces:
face_embeddings.append(face.normed_embedding)
@@ -332,6 +338,9 @@ def default_target_face():
best_frame = frame
break
if best_face is None:
continue # No faces detected in this cluster — skip
for frame in map['target_faces_in_frame']:
for face in frame['faces']:
if face['det_score'] > best_face['det_score']:
@@ -340,7 +349,7 @@ def default_target_face():
x_min, y_min, x_max, y_max = best_face['bbox']
target_frame = cv2.imread(best_frame['location'])
target_frame = imread_unicode(best_frame['location'])
map['target'] = {
'cv2' : target_frame[int(y_min):int(y_max), int(x_min):int(x_max)],
'face' : best_face
@@ -356,7 +365,7 @@ def dump_faces(centroids: Any, frame_face_embeddings: list):
Path(temp_directory_path + f"/{i}").mkdir(parents=True, exist_ok=True)
for frame in tqdm(frame_face_embeddings, desc=f"Copying faces to temp/./{i}"):
temp_frame = cv2.imread(frame['location'])
temp_frame = imread_unicode(frame['location'])
j = 0
for face in frame['faces']:
@@ -364,5 +373,5 @@ def dump_faces(centroids: Any, frame_face_embeddings: list):
x_min, y_min, x_max, y_max = face['bbox']
if temp_frame[int(y_min):int(y_max), int(x_min):int(x_max)].size > 0:
cv2.imwrite(temp_directory_path + f"/{i}/{frame['frame']}_{j}.png", temp_frame[int(y_min):int(y_max), int(x_min):int(x_max)])
imwrite_unicode(temp_directory_path + f"/{i}/{frame['frame']}_{j}.png", temp_frame[int(y_min):int(y_max), int(x_min):int(x_max)])
j += 1
+7 -4
View File
@@ -6,10 +6,13 @@ from typing import List, Dict, Any
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
WORKFLOW_DIR = os.path.join(ROOT_DIR, "workflow")
file_types = [
("Image", ("*.png", "*.jpg", "*.jpeg", "*.gif", "*.bmp")),
("Video", ("*.mp4", "*.mkv")),
]
# Canonical media extensions, defined once so the file dialogs and
# has_image_extension never drift. GIF is intentionally excluded: OpenCV's
# cv2.imread/imwrite (the only image I/O this app uses) cannot decode or
# encode GIF on 4.10 or 4.11, so offering it would silently fail. WEBP works
# via the libwebp bundled with opencv-python.
IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg", ".bmp", ".webp")
VIDEO_EXTENSIONS = (".mp4", ".mkv")
# Face Mapping Data
source_target_map: List[Dict[str, Any]] = [] # Stores detailed map for image/video processing
+1 -1
View File
@@ -21,7 +21,7 @@ from __future__ import annotations
import os
import cv2
import numpy as np
from typing import Tuple, Optional
from typing import Tuple
# ---------------------------------------------------------------------------
# CUDA availability detection (evaluated once at import time)
+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
+1 -1
View File
@@ -392,7 +392,7 @@ def _decompose_split(model) -> bool:
# Collect all needed boundary constants
for _, (a, b) in splits:
ensure_const(f"_sp_s0", [0])
ensure_const("_sp_s0", [0])
ensure_const(f"_sp_s{a}", [a])
ensure_const(f"_sp_s{a + b}", [a + b])
+11
View File
@@ -40,6 +40,15 @@ ONNX_PROVIDERS: List[str] = _detect_onnx_providers()
HAS_CUDA_PROVIDER: bool = "CUDAExecutionProvider" in ONNX_PROVIDERS
HAS_COREML_PROVIDER: bool = "CoreMLExecutionProvider" in ONNX_PROVIDERS
HAS_DML_PROVIDER: bool = "DmlExecutionProvider" in ONNX_PROVIDERS
HAS_OPENVINO_PROVIDER: bool = "OpenVINOExecutionProvider" in ONNX_PROVIDERS
# OpenVINO execution-provider config shared by every ONNX session builder.
# AUTO:GPU,NPU,CPU lets OpenVINO pick the best available device in priority
# order (Intel GPU → NPU → CPU).
OPENVINO_PROVIDER_CONFIG = (
"OpenVINOExecutionProvider",
{"device_type": "AUTO:GPU,NPU,CPU"},
)
def camera_backends() -> List[Tuple[int, int]]:
@@ -65,6 +74,8 @@ def accelerator_label() -> str:
return "CoreML (Apple Neural Engine)"
if HAS_COREML_PROVIDER:
return "CoreML"
if HAS_OPENVINO_PROVIDER:
return "OpenVINO (Intel)"
if HAS_DML_PROVIDER:
return "DirectML"
return "CPU"
+13
View File
@@ -1,4 +1,17 @@
import importlib.util
import os
import numpy
# Keras 3 defaults to the TensorFlow backend, which has no Python 3.14 wheels.
# opennsfw2 only runs inference, so any installed backend works; pick one that
# is actually present before opennsfw2 imports keras.
if "KERAS_BACKEND" not in os.environ:
for _backend in ("torch", "tensorflow", "jax"):
if importlib.util.find_spec(_backend) is not None:
os.environ["KERAS_BACKEND"] = _backend
break
import opennsfw2
from PIL import Image
import cv2 # Add OpenCV import
@@ -14,6 +14,7 @@ import numpy as np
import onnxruntime
import modules.globals
from modules.platform_info import OPENVINO_PROVIDER_CONFIG
IS_APPLE_SILICON = platform.system() == "Darwin" and platform.machine() == "arm64"
@@ -50,6 +51,9 @@ def build_provider_config(providers=None):
"AllowLowPrecisionAccumulationOnGPU": 1,
},
))
elif p == "OpenVINOExecutionProvider":
# AUTO lets OpenVINO select the best device
config.append(OPENVINO_PROVIDER_CONFIG)
else:
config.append(p)
return config
+5 -4
View File
@@ -37,6 +37,7 @@ def load_frame_processor_module(frame_processor: str) -> Any:
frame_processor_module = importlib.import_module(f'modules.processors.frame.{frame_processor}')
for method_name in FRAME_PROCESSORS_INTERFACE:
if not hasattr(frame_processor_module, method_name):
print(f"Frame processor {frame_processor} is missing required method {method_name}")
sys.exit()
except ImportError:
print(f"Frame processor {frame_processor} not found")
@@ -59,7 +60,7 @@ def set_frame_processors_modules_from_ui(frame_processors: List[str]) -> None:
current_processor_names = [proc.__name__.split('.')[-1] for proc in FRAME_PROCESSORS_MODULES]
for frame_processor, state in modules.globals.fp_ui.items():
if state == True and frame_processor not in current_processor_names:
if state and frame_processor not in current_processor_names:
try:
frame_processor_module = load_frame_processor_module(frame_processor)
FRAME_PROCESSORS_MODULES.append(frame_processor_module)
@@ -70,7 +71,7 @@ def set_frame_processors_modules_from_ui(frame_processors: List[str]) -> None:
except Exception as e:
print(f"Warning: Error loading frame processor {frame_processor} requested by UI state: {e}")
elif state == False and frame_processor in current_processor_names:
elif not state and frame_processor in current_processor_names:
try:
module_to_remove = next((mod for mod in FRAME_PROCESSORS_MODULES if mod.__name__.endswith(f'.{frame_processor}')), None)
if module_to_remove:
@@ -125,7 +126,7 @@ def process_video_in_memory(source_path: str, target_path: str, fps: float) -> b
Returns True on success, False on failure (caller should fall back to the
disk-based pipeline).
"""
import cv2
from modules import imread_unicode
from modules.face_analyser import get_one_face
from modules.utilities import (
get_video_dimensions,
@@ -138,7 +139,7 @@ def process_video_in_memory(source_path: str, target_path: str, fps: float) -> b
# --- Pre-load source face (needed by face_swapper in simple mode) ---
source_face = None
if source_path and os.path.exists(source_path):
source_img = cv2.imread(source_path)
source_img = imread_unicode(source_path)
if source_img is not None:
source_face = get_one_face(source_img)
del source_img
+19 -12
View File
@@ -10,8 +10,9 @@ import onnxruntime
import modules.globals
import modules.processors.frame.core
from modules import imread_unicode, imwrite_unicode
from modules.core import update_status
from modules.face_analyser import get_one_face, get_many_faces
from modules.face_analyser import get_many_faces
from modules.typing import Frame, Face
from modules.utilities import (
is_image,
@@ -22,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(
@@ -43,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
@@ -72,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:
@@ -407,7 +414,7 @@ def process_frames(
progress.update(1)
continue
temp_frame = cv2.imread(temp_frame_path)
temp_frame = imread_unicode(temp_frame_path)
if temp_frame is None:
print(
f"{NAME}: Warning: Failed to read frame {temp_frame_path}, skipping."
@@ -417,7 +424,7 @@ def process_frames(
continue
result_frame = process_frame(None, temp_frame)
cv2.imwrite(temp_frame_path, result_frame)
imwrite_unicode(temp_frame_path, result_frame)
if progress:
progress.update(1)
@@ -426,12 +433,12 @@ def process_image(
source_path: str | None, target_path: str, output_path: str
) -> None:
"""Processes a single image file."""
target_frame = cv2.imread(target_path)
target_frame = imread_unicode(target_path)
if target_frame is None:
print(f"{NAME}: Error: Failed to read target image {target_path}")
return
result_frame = process_frame(None, target_frame)
cv2.imwrite(output_path, result_frame)
imwrite_unicode(output_path, result_frame)
print(f"{NAME}: Enhanced image saved to {output_path}")
@@ -4,11 +4,9 @@ from typing import Any, List
import os
import threading
import cv2
import numpy as np
import modules.globals
import modules.processors.frame.core
from modules import imread_unicode, imwrite_unicode
from modules.core import update_status
from modules.face_analyser import get_one_face
from modules.typing import Frame, Face
@@ -24,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
@@ -36,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
@@ -56,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)
@@ -103,24 +121,24 @@ def process_frames(
source_path: str | None, temp_frame_paths: List[str], progress: Any = None
) -> None:
for temp_frame_path in temp_frame_paths:
temp_frame = cv2.imread(temp_frame_path)
temp_frame = imread_unicode(temp_frame_path)
if temp_frame is None:
if progress:
progress.update(1)
continue
result = process_frame(None, temp_frame)
cv2.imwrite(temp_frame_path, result)
imwrite_unicode(temp_frame_path, result)
if progress:
progress.update(1)
def process_image(source_path: str | None, target_path: str, output_path: str) -> None:
target_frame = cv2.imread(target_path)
target_frame = imread_unicode(target_path)
if target_frame is None:
print(f"{NAME}: Error: Failed to read target image {target_path}")
return
result_frame = process_frame(None, target_frame)
cv2.imwrite(output_path, result_frame)
imwrite_unicode(output_path, result_frame)
print(f"{NAME}: Enhanced image saved to {output_path}")
@@ -4,11 +4,9 @@ from typing import Any, List
import os
import threading
import cv2
import numpy as np
import modules.globals
import modules.processors.frame.core
from modules import imread_unicode, imwrite_unicode
from modules.core import update_status
from modules.face_analyser import get_one_face
from modules.typing import Frame, Face
@@ -24,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
@@ -36,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
@@ -56,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)
@@ -103,24 +121,24 @@ def process_frames(
source_path: str | None, temp_frame_paths: List[str], progress: Any = None
) -> None:
for temp_frame_path in temp_frame_paths:
temp_frame = cv2.imread(temp_frame_path)
temp_frame = imread_unicode(temp_frame_path)
if temp_frame is None:
if progress:
progress.update(1)
continue
result = process_frame(None, temp_frame)
cv2.imwrite(temp_frame_path, result)
imwrite_unicode(temp_frame_path, result)
if progress:
progress.update(1)
def process_image(source_path: str | None, target_path: str, output_path: str) -> None:
target_frame = cv2.imread(target_path)
target_frame = imread_unicode(target_path)
if target_frame is None:
print(f"{NAME}: Error: Failed to read target image {target_path}")
return
result_frame = process_frame(None, target_frame)
cv2.imwrite(output_path, result_frame)
imwrite_unicode(output_path, result_frame)
print(f"{NAME}: Enhanced image saved to {output_path}")
+1 -1
View File
@@ -2,7 +2,7 @@ import cv2
import numpy as np
from modules.typing import Face, Frame
import modules.globals
from modules.gpu_processing import gpu_gaussian_blur, gpu_resize, gpu_cvt_color
from modules.gpu_processing import gpu_gaussian_blur, gpu_resize
def apply_color_transfer(source, target):
"""
+50 -29
View File
@@ -7,16 +7,17 @@ import numpy as np
import platform
import modules.globals
import modules.processors.frame.core
from modules import imread_unicode, imwrite_unicode
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,
)
from modules.cluster_analysis import find_closest_centroid
from modules.gpu_processing import gpu_gaussian_blur, gpu_sharpen, gpu_add_weighted, gpu_resize, gpu_cvt_color
from modules.gpu_processing import gpu_gaussian_blur, gpu_sharpen, gpu_add_weighted, gpu_resize
from modules.platform_info import OPENVINO_PROVIDER_CONFIG
import os
from collections import deque
import time
@@ -190,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
@@ -240,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.
@@ -269,6 +279,8 @@ def get_face_swapper() -> Any:
# Use bare provider — ONNX Runtime defaults are
# fastest on modern GPUs (Blackwell/sm_120).
providers_config.append(p)
elif p == "OpenVINOExecutionProvider":
providers_config.append(OPENVINO_PROVIDER_CONFIG)
else:
providers_config.append(p)
FACE_SWAPPER = insightface.model_zoo.get_model(
@@ -680,7 +692,8 @@ def apply_post_processing(current_frame: Frame, swapped_face_bboxes: List[np.nda
continue
face_region = processed_frame[y1:y2, x1:x2]
if face_region.size == 0: continue
if face_region.size == 0:
continue
# Apply sharpening (GPU-accelerated when CUDA OpenCV is available)
try:
@@ -815,9 +828,11 @@ def process_frame_v2(temp_frame: Frame, temp_frame_path: str = "") -> Frame:
else: # Single face or specific mapping
for map_data in source_target_map:
source_info = map_data.get("source", {})
if not source_info: continue # Skip if no source info
if not source_info:
continue # Skip if no source info
source_face = source_info.get("face")
if not source_face: continue # Skip if no source defined for this map entry
if not source_face:
continue # Skip if no source defined for this map entry
if is_image(modules.globals.target_path):
target_info = map_data.get("target", {})
@@ -854,7 +869,8 @@ def process_frame_v2(temp_frame: Frame, temp_frame_path: str = "") -> Frame:
if len(detected_faces) <= len(target_embeddings):
# More targets defined than detected - match each detected face
for detected_face in detected_faces:
if detected_face.normed_embedding is None: continue
if detected_face.normed_embedding is None:
continue
closest_idx, _ = find_closest_centroid(target_embeddings, detected_face.normed_embedding)
if 0 <= closest_idx < len(source_faces):
source_target_pairs.append((source_faces[closest_idx], detected_face))
@@ -862,7 +878,8 @@ def process_frame_v2(temp_frame: Frame, temp_frame_path: str = "") -> Frame:
# More faces detected than targets defined - match each target embedding to closest detected face
detected_embeddings = [f.normed_embedding for f in detected_faces if f.normed_embedding is not None]
detected_faces_with_embedding = [f for f in detected_faces if f.normed_embedding is not None]
if not detected_embeddings: return processed_frame # No embeddings to match
if not detected_embeddings:
return processed_frame # No embeddings to match
for i, target_embedding in enumerate(target_embeddings):
if 0 <= i < len(source_faces): # Ensure source face exists for this embedding
@@ -912,7 +929,7 @@ def process_frames(
# Log the error but allow proceeding; subsequent check will stop processing.
else:
try:
source_img = cv2.imread(source_path)
source_img = imread_unicode(source_path)
if source_img is None:
# Specific error for file reading failure
update_status(f"Error reading source image file {source_path}. Please check the path and file integrity.", NAME)
@@ -936,7 +953,7 @@ def process_frames(
# --- Stop processing entirely if in Simple Mode and source face is invalid ---
if not use_v2 and source_face is None:
update_status(f"Halting video processing: Invalid or no face detected in source image for simple mode.", NAME)
update_status("Halting video processing: Invalid or no face detected in source image for simple mode.", NAME)
if progress:
# Ensure the progress bar completes if it was started
remaining_updates = total_frames - progress.n if hasattr(progress, 'n') else total_frames
@@ -952,14 +969,16 @@ def process_frames(
# Read the target frame
temp_frame = None
try:
temp_frame = cv2.imread(temp_frame_path)
temp_frame = imread_unicode(temp_frame_path)
if temp_frame is None:
print(f"{NAME}: Error: Could not read frame: {temp_frame_path}, skipping.")
if progress: progress.update(1)
if progress:
progress.update(1)
continue # Skip this frame if read fails
except Exception as read_e:
print(f"{NAME}: Error reading frame {temp_frame_path}: {read_e}, skipping.")
if progress: progress.update(1)
if progress:
progress.update(1)
continue
# Select processing function and execute
@@ -988,7 +1007,7 @@ def process_frames(
# Write the result back to the same frame path with optimized compression
try:
# Use PNG compression level 3 (faster) instead of default 9
write_success = cv2.imwrite(temp_frame_path, result_frame, [cv2.IMWRITE_PNG_COMPRESSION, 3])
write_success = imwrite_unicode(temp_frame_path, result_frame, [cv2.IMWRITE_PNG_COMPRESSION, 3])
if not write_success:
print(f"{NAME}: Error: Failed to write processed frame to {temp_frame_path}")
except Exception as write_e:
@@ -1018,7 +1037,7 @@ def process_image(source_path: str, target_path: str, output_path: str) -> None:
# Read target first
try:
target_frame = cv2.imread(target_path)
target_frame = imread_unicode(target_path)
if target_frame is None:
update_status(f"Error: Could not read target image: {target_path}", NAME)
return
@@ -1037,7 +1056,7 @@ def process_image(source_path: str, target_path: str, output_path: str) -> None:
else: # Simple mode
try:
source_img = cv2.imread(source_path)
source_img = imread_unicode(source_path)
if source_img is None:
update_status(f"Error: Could not read source image: {source_path}", NAME)
return
@@ -1053,7 +1072,7 @@ def process_image(source_path: str, target_path: str, output_path: str) -> None:
# Write the result if processing was successful
if result is not None:
write_success = cv2.imwrite(output_path, result)
write_success = imwrite_unicode(output_path, result)
if write_success:
update_status(f"Output image saved to: {output_path}", NAME)
else:
@@ -1496,7 +1515,8 @@ def apply_color_transfer(source, target):
if len(source.shape) == 2: # Grayscale
source = cv2.cvtColor(source, cv2.COLOR_GRAY2BGR)
source = np.clip(source, 0, 255).astype(np.uint8)
if len(source.shape)!= 3 or source.shape[2]!= 3: raise ValueError("Conversion failed")
if len(source.shape) != 3 or source.shape[2] != 3:
raise ValueError("Conversion failed")
except Exception:
return source
if len(target.shape) != 3 or target.shape[2] != 3 or target.dtype != np.uint8:
@@ -1505,7 +1525,8 @@ def apply_color_transfer(source, target):
if len(target.shape) == 2: # Grayscale
target = cv2.cvtColor(target, cv2.COLOR_GRAY2BGR)
target = np.clip(target, 0, 255).astype(np.uint8)
if len(target.shape)!= 3 or target.shape[2]!= 3: raise ValueError("Conversion failed")
if len(target.shape) != 3 or target.shape[2] != 3:
raise ValueError("Conversion failed")
except Exception:
return source # Return original source if target invalid
+2 -2
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
# Import the tkinter fix to patch the ScreenChanged error
import tkinter_fix
# Import the tkinter fix to patch the ScreenChanged error (module patches Tk on import)
import tkinter_fix # noqa: F401
import core
+24 -11
View File
@@ -73,6 +73,7 @@ from modules.utilities import (
is_image,
is_video,
)
from modules import imread_unicode
from modules.video_capture import VideoCapturer
if platform.system() == "Windows":
@@ -236,6 +237,18 @@ _RECENT_SOURCE_DIR: Optional[str] = None
_RECENT_TARGET_DIR: Optional[str] = None
_RECENT_OUTPUT_DIR: Optional[str] = None
# QFileDialog filter strings, built from the canonical extension sets in
# globals so every dialog stays in sync (no hand-copied lists to drift).
_IMAGE_FILE_FILTER = "Images (" + " ".join(
f"*{ext}" for ext in modules.globals.IMAGE_EXTENSIONS
) + ")"
_MEDIA_FILE_FILTER = "Media (" + " ".join(
f"*{ext}" for ext in (*modules.globals.IMAGE_EXTENSIONS, *modules.globals.VIDEO_EXTENSIONS)
) + ")"
_VIDEO_FILE_FILTER = "Videos (" + " ".join(
f"*{ext}" for ext in modules.globals.VIDEO_EXTENSIONS
) + ")"
# ─── image utilities ─────────────────────────────────────────────────────
@@ -416,7 +429,7 @@ def get_available_cameras() -> Tuple[List[int], List[str]]:
indices: List[int] = []
names: List[str] = []
for i in range(10):
cap = cv2.VideoCapture(i)
cap = cv2.VideoCapture(f"/dev/video{i}")
if cap.isOpened():
indices.append(i)
names.append(f"Camera {i}")
@@ -733,7 +746,7 @@ class MainWindow(QMainWindow):
path, _filter = QFileDialog.getOpenFileName(
self, _("select an source image"),
_RECENT_SOURCE_DIR or "",
"Images (*.png *.jpg *.jpeg *.gif *.bmp)",
_IMAGE_FILE_FILTER,
)
if path and is_image(path):
modules.globals.source_path = path
@@ -754,7 +767,7 @@ class MainWindow(QMainWindow):
path, _filter = QFileDialog.getOpenFileName(
self, _("select an target image or video"),
_RECENT_TARGET_DIR or "",
"Media (*.png *.jpg *.jpeg *.gif *.bmp *.mp4 *.mkv)",
_MEDIA_FILE_FILTER,
)
if not path:
return
@@ -885,13 +898,13 @@ class MainWindow(QMainWindow):
path, _f = QFileDialog.getSaveFileName(
self, _("save image output file"),
os.path.join(_RECENT_OUTPUT_DIR or "", "output.png"),
"Images (*.png *.jpg *.jpeg *.bmp)",
_IMAGE_FILE_FILTER,
)
elif is_video(modules.globals.target_path):
path, _f = QFileDialog.getSaveFileName(
self, _("save video output file"),
os.path.join(_RECENT_OUTPUT_DIR or "", "output.mp4"),
"Videos (*.mp4 *.mkv)",
_VIDEO_FILE_FILTER,
)
else:
return
@@ -988,7 +1001,7 @@ class PreviewWindow(QWidget):
from modules.processors.frame.core import get_frame_processors_modules as _gfpm
for fp in _gfpm(modules.globals.frame_processors):
temp_frame = fp.process_frame(
get_one_face(cv2.imread(modules.globals.source_path)), temp_frame
get_one_face(imread_unicode(modules.globals.source_path)), temp_frame
)
# Fit to current widget size while preserving aspect ratio.
h, w = temp_frame.shape[:2]
@@ -1071,7 +1084,7 @@ class _ProcessingWorker(QThread):
and modules.globals.source_path != last_source_path
):
last_source_path = modules.globals.source_path
source_image = get_one_face(cv2.imread(modules.globals.source_path))
source_image = get_one_face(imread_unicode(modules.globals.source_path))
det_count += 1
if det_count % det_interval == 0:
@@ -1333,11 +1346,11 @@ class MapperDialog(QDialog):
path, _f = QFileDialog.getOpenFileName(
self, _("select an source image"),
_RECENT_SOURCE_DIR or "",
"Images (*.png *.jpg *.jpeg *.gif *.bmp)",
_IMAGE_FILE_FILTER,
)
if not path:
return
cv2_img = cv2.imread(path)
cv2_img = imread_unicode(path)
face = get_one_face(cv2_img)
if face is None:
self.set_status("Face could not be detected in last upload!")
@@ -1438,11 +1451,11 @@ class LiveMapperDialog(QDialog):
path, _f = QFileDialog.getOpenFileName(
self, _("select an source image"),
_RECENT_SOURCE_DIR or "",
"Images (*.png *.jpg *.jpeg *.gif *.bmp)",
_IMAGE_FILE_FILTER,
)
if not path:
return
cv2_img = cv2.imread(path)
cv2_img = imread_unicode(path)
face = get_one_face(cv2_img)
if face is None:
self.set_status("Face could not be detected in last upload!")
+6 -1
View File
@@ -262,11 +262,16 @@ def clean_temp(target_path: str) -> None:
def has_image_extension(image_path: str) -> bool:
return image_path.lower().endswith(("png", "jpg", "jpeg"))
# splitext so only the real extension counts (e.g. "photo.png.bak" is not
# an image); the set is centralized in globals to stay in sync with dialogs.
return os.path.splitext(image_path)[1].lower() in modules.globals.IMAGE_EXTENSIONS
def is_image(image_path: str) -> bool:
if image_path and os.path.isfile(image_path):
# Extension check first — Windows mimetypes doesn't always register webp
if has_image_extension(image_path):
return True
mimetype, _ = mimetypes.guess_type(image_path)
return bool(mimetype and mimetype.startswith("image/"))
return False
+2 -2
View File
@@ -1,6 +1,5 @@
import cv2
import numpy as np
import sys
import time
from typing import Optional, Tuple, Callable
import platform
@@ -72,8 +71,9 @@ class VideoCapturer:
self.cap.release()
except Exception:
continue
elif platform.system() == "Linux":
self.cap = cv2.VideoCapture(f"/dev/video{self.device_index}")
else:
# Unix-like systems (Linux/Mac) capture method
self.cap = cv2.VideoCapture(self.device_index)
if not self.cap or not self.cap.isOpened():
+9
View File
@@ -0,0 +1,9 @@
[tool.ruff]
target-version = "py310"
[tool.ruff.lint]
# Deterministic, low-risk rules enforced in CI. Other rules (F841, E402, F821)
# surface real findings but require human judgement to fix safely, so they are
# left out of the gate for now. Intentional side-effect imports should be
# annotated with `# noqa: F401`.
select = ["E701", "E711", "E712", "F401", "F541"]
+15 -14
View File
@@ -1,17 +1,18 @@
numpy>=1.23.5,<2
typing-extensions>=4.8.0
opencv-python==4.10.0.84
cv2_enumerate_cameras==1.1.15
onnx==1.18.0
numpy>=2.0,<3
typing-extensions>=4.15.0
opencv-python==4.14.0.94
opencv-python-headless==4.14.0.94
cv2_enumerate_cameras==1.3.3
onnx==1.22.0
insightface==0.7.3
psutil==5.9.8
psutil==7.2.2
PySide6>=6.7,<7
pillow==12.1.1
tqdm>=4.65.0
onnxruntime-silicon==1.16.3; sys_platform == 'darwin' and platform_machine == 'arm64'
onnxruntime-gpu==1.23.2; sys_platform != 'darwin'
tensorflow>=2.15.0; sys_platform != 'darwin'
tensorflow>=2.15.0; sys_platform == 'darwin' and python_version < '3.13'
opennsfw2==0.10.2
protobuf==4.25.1
pillow==12.3.0
tqdm>=4.66.3
onnxruntime==1.28.0; sys_platform == 'darwin' and platform_machine == 'arm64'
onnxruntime==1.23.0; sys_platform == 'darwin' and platform_machine != 'arm64'
onnxruntime-gpu==1.26.0; sys_platform != 'darwin'
opennsfw2==0.18.0
keras>=3.0.0
protobuf>=6.33.5,<8
pygrabber; sys_platform == 'win32'
+24
View File
@@ -31,6 +31,30 @@ if sys.platform == "win32":
except (OSError, AttributeError):
pass
# On Windows, register OpenVINO DLL directories so onnxruntime's
# OpenVINOExecutionProvider can find openvino.dll. This must happen
# before any ONNX InferenceSession is created. Failure is non-fatal:
# OpenVINO simply isn't installed, and onnxruntime will fall back to CPU.
try:
from onnxruntime.tools.add_openvino_win_libs import ( # type: ignore[import-untyped] # noqa: E501
add_openvino_libs_to_path,
)
add_openvino_libs_to_path()
except ImportError:
# onnxruntime build without the OpenVINO tooling module — no-op.
pass
except FileNotFoundError:
# OpenVINO site-packages dir absent — no-op.
pass
except SystemExit as exc:
# add_openvino_libs_to_path() calls sys.exit() when OpenVINO libs
# can't be located (e.g. OPENVINO_LIB_PATHS unset). Log the message
# it raised with so the failure is visible, but keep startup alive.
print(
f"[startup] OpenVINO DLL registration skipped: {exc}",
flush=True,
)
# On Linux, pre-load NVIDIA shared libraries (cuDNN, cuBLAS, nvrtc...) shipped
# inside the venv via pip wheels (nvidia-cudnn-cu12, etc.). LD_LIBRARY_PATH
# cannot be set after Python starts, so we use ctypes.CDLL with RTLD_GLOBAL
+137
View File
@@ -0,0 +1,137 @@
import importlib
import sys
import types
import unittest
from contextlib import contextmanager
from unittest.mock import patch
@contextmanager
def _patched_core_import_stubs(calls, pipe_result=False):
class Processor:
NAME = "test_processor"
def pre_start(self):
return True
def pre_check(self):
return True
def process_image(self, *_args, **_kwargs):
raise AssertionError("image path should not be used")
def process_video(self, source_path, frame_paths):
calls.append(("process_video", source_path, tuple(frame_paths)))
stubs = {
"cv2": types.SimpleNamespace(
IMREAD_COLOR=1,
imdecode=lambda *_args, **_kwargs: None,
imencode=lambda *_args, **_kwargs: (
True,
types.SimpleNamespace(tofile=lambda *_a, **_k: None),
),
),
"numpy": types.SimpleNamespace(uint8=object, fromfile=lambda *_args, **_kwargs: b""),
"torch": types.SimpleNamespace(
cuda=types.SimpleNamespace(empty_cache=lambda: None)
),
"onnxruntime": types.SimpleNamespace(
get_available_providers=lambda: ["CPUExecutionProvider"]
),
"tensorflow": types.SimpleNamespace(),
"modules.metadata": types.SimpleNamespace(name="Deep-Live-Cam", version="test"),
"modules.ui": types.SimpleNamespace(
check_and_ignore_nsfw=lambda *_args, **_kwargs: False,
update_status=lambda *_args, **_kwargs: None,
init=lambda *_args, **_kwargs: types.SimpleNamespace(mainloop=lambda: None),
),
"modules.processors.frame.core": types.SimpleNamespace(
get_frame_processors_modules=lambda _names: [Processor()],
process_video_in_memory=lambda *_args, **_kwargs: calls.append(("pipe",))
or pipe_result,
),
"modules.utilities": types.SimpleNamespace(
has_image_extension=lambda _path: False,
is_image=lambda _path: False,
is_video=lambda _path: True,
detect_fps=lambda _path: 24.0,
create_video=lambda target_path, fps: calls.append(
("create_video", target_path, fps)
)
or True,
extract_frames=lambda target_path: calls.append(
("extract_frames", target_path)
),
get_temp_frame_paths=lambda target_path: [f"{target_path}/0001.png"],
restore_audio=lambda *_args, **_kwargs: calls.append(("restore_audio",)),
create_temp=lambda target_path: calls.append(("create_temp", target_path)),
move_temp=lambda target_path, output_path: calls.append(
("move_temp", target_path, output_path)
),
clean_temp=lambda target_path: calls.append(("clean_temp", target_path)),
normalize_output_path=lambda _source, _target, output: output,
),
}
with patch.dict(sys.modules, stubs, clear=False):
sys.modules.pop("modules.core", None)
yield importlib.import_module("modules.core")
sys.modules.pop("modules.core", None)
def _configure_video_run(core, *, map_faces):
core.modules.globals.source_path = "source.jpg"
core.modules.globals.target_path = "target.mp4"
core.modules.globals.output_path = "output.mp4"
core.modules.globals.frame_processors = ["face_swapper"]
core.modules.globals.headless = True
core.modules.globals.keep_fps = False
core.modules.globals.keep_audio = False
core.modules.globals.keep_frames = False
core.modules.globals.map_faces = map_faces
core.modules.globals.nsfw_filter = False
core.modules.globals.execution_threads = 1
core.modules.globals.execution_providers = ["CPUExecutionProvider"]
core.modules.globals.max_memory = None
class MapFacesFallbackTests(unittest.TestCase):
def test_map_faces_disk_fallback_extracts_frames_before_processing(self):
calls = []
with _patched_core_import_stubs(calls, pipe_result=False) as core:
_configure_video_run(core, map_faces=True)
with patch.object(core.os.path, "isfile", return_value=True):
core.start()
self.assertNotIn(("pipe",), calls)
self.assertIn(("create_temp", "target.mp4"), calls)
self.assertIn(("extract_frames", "target.mp4"), calls)
self.assertIn(("process_video", "source.jpg", ("target.mp4/0001.png",)), calls)
self.assertIn(("create_video", "target.mp4", 30.0), calls)
self.assertIn(("move_temp", "target.mp4", "output.mp4"), calls)
step_indices = {}
for index, call in enumerate(calls):
step_indices.setdefault(call[0], index)
self.assertLess(step_indices["create_temp"], step_indices["extract_frames"])
self.assertLess(step_indices["extract_frames"], step_indices["process_video"])
self.assertLess(step_indices["process_video"], step_indices["create_video"])
self.assertLess(step_indices["create_video"], step_indices["move_temp"])
def test_non_map_faces_pipe_success_does_not_extract_frames(self):
calls = []
with _patched_core_import_stubs(calls, pipe_result=True) as core:
_configure_video_run(core, map_faces=False)
with patch.object(core.os.path, "isfile", return_value=True):
core.start()
self.assertIn(("pipe",), calls)
self.assertNotIn(("extract_frames", "target.mp4"), calls)
self.assertNotIn(("process_video", "source.jpg", ("target.mp4/0001.png",)), calls)
if __name__ == "__main__":
unittest.main()