Compare commits

..
41 Commits
Author SHA1 Message Date
Dopan d759e31b11 Merge pull request #1922 from huyua9/config-dependabot-pip-root
ci: configure Dependabot for root requirements
2026-09-15 19:03:47 +08:00
Kenneth Estanislao 55d306d5ae Revert "chore: update requirements.txt"
This reverts commit 7895c547a6.
2026-09-09 09:37:17 +08:00
Kenneth Estanislao 7895c547a6 chore: update requirements.txt 2026-09-08 23:58:53 +08:00
Kenneth Estanislao 19f4270e30 Update version number in README for Deep-Live-Cam 2026-09-08 00:43:02 +08:00
ming 327f66154a ci: configure Dependabot for root requirements 2026-08-31 12:51:30 +08:00
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
31 changed files with 785 additions and 191 deletions
+6
View File
@@ -0,0 +1,6 @@
version: 2
updates:
- package-ecosystem: pip
directory: "/"
schedule:
interval: weekly
+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. 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.5 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. <p align="center">
<a href="https://deeplivecam.net/index.php/plans/nvidia-gpu?plan_id=0&group_id=1">
###### 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. <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 ## TLDR; Live Deepfake in just 3 Clicks
![easysteps](https://github.com/user-attachments/assets/af825228-852c-411b-b787-ffd9aac72fc6) ![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** **1. Set up Your Platform**
- Python (3.11 recommended) - Python (3.14 recommended; 3.11-3.14 supported)
- pip - pip
- git - git
- [ffmpeg](https://www.youtube.com/watch?v=OlNWCpFdVMA) - ```iex (irm ffmpeg.tc.ht)``` - [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** **2. Clone the Repository**
```bash ```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 cd Deep-Live-Cam
``` ```
**3. Download the Models** **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) 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. Place these files in the "**models**" folder.
@@ -142,7 +174,7 @@ pip install -r requirements.txt
``` ```
For Linux: For Linux:
```bash ```bash
# Ensure you use the installed Python 3.11 # Ensure you use the installed Python 3.14
python3 -m venv venv python3 -m venv venv
source venv/bin/activate source venv/bin/activate
pip install -r requirements.txt pip install -r requirements.txt
@@ -150,17 +182,17 @@ pip install -r requirements.txt
**For macOS:** **For macOS:**
Apple Silicon (M1/M2/M3) requires specific setup: Apple Silicon (M1 through M5) requires specific setup:
```bash ```bash
# Install Python 3.11 (specific version is important) # Install Python 3.14
brew install python@3.11 brew install python@3.14
# Install tkinter package (required for the GUI) # 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 # Create and activate virtual environment with Python 3.14
python3.11 -m venv venv python3.14 -m venv venv
source venv/bin/activate source venv/bin/activate
# Install dependencies # Install dependencies
@@ -201,7 +233,7 @@ pip install git+https://github.com/TencentARC/GFPGAN.git@master
```bash ```bash
pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu128 pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu128
pip uninstall onnxruntime onnxruntime-gpu pip uninstall onnxruntime onnxruntime-gpu
pip install onnxruntime-gpu==1.21.0 pip install onnxruntime-gpu==1.26.0
``` ```
3. Usage: 3. Usage:
@@ -212,26 +244,29 @@ python run.py --execution-provider cuda
**CoreML Execution Provider (Apple Silicon)** **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. 1. Make sure you've completed the macOS setup above using Python 3.14.
2. Install dependencies: 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 ```bash
pip uninstall onnxruntime onnxruntime-silicon pip uninstall onnxruntime-silicon
pip install onnxruntime-silicon==1.13.1 pip install -r requirements.txt
``` ```
3. Usage: 3. Usage:
```bash ```bash
python3.11 run.py --execution-provider coreml python3.14 run.py --execution-provider coreml
``` ```
**Important Notes for macOS:** **Important Notes for macOS:**
- You **must** use Python 3.11, not newer versions like 3.13 - Python 3.11 is the minimum (onnxruntime dropped 3.10); 3.14 is recommended
- Always run with `python3.11` command not just `python` if you have multiple Python versions installed - 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.11` - 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 get model loading errors, check that your models are in the correct folder
- If you encounter conflicts with other Python versions, consider uninstalling them: - If you encounter conflicts with other Python versions, consider uninstalling them:
```bash ```bash
@@ -239,9 +274,9 @@ python3.11 run.py --execution-provider coreml
brew list | grep python brew list | grep python
# Uninstall conflicting versions if needed # 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 brew cleanup
``` ```
@@ -284,6 +319,22 @@ pip uninstall onnxruntime onnxruntime-openvino
pip install onnxruntime-openvino==1.21.0 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: 2. Usage:
```bash ```bash
+8 -5
View File
@@ -14,7 +14,6 @@ if sys.platform == "win32":
import insightface import insightface
from insightface.app import FaceAnalysis from insightface.app import FaceAnalysis
from insightface.utils import face_align
from modules.processors.frame.face_swapper import _fast_paste_back from modules.processors.frame.face_swapper import _fast_paste_back
from modules import platform_info from modules import platform_info
@@ -81,10 +80,14 @@ def capture_thread():
try: try:
capture_queue.put_nowait(frame) capture_queue.put_nowait(frame)
except queue.Full: except queue.Full:
try: capture_queue.get_nowait() try:
except queue.Empty: pass capture_queue.get_nowait()
try: capture_queue.put_nowait(frame) except queue.Empty:
except queue.Full: pass pass
try:
capture_queue.put_nowait(frame)
except queue.Full:
pass
cap_t = threading.Thread(target=capture_thread, daemon=True) cap_t = threading.Thread(target=capture_thread, daemon=True)
cap_t.start() cap_t.start()
+1 -1
View File
@@ -1,4 +1,4 @@
just put the models in this folder - just put the models in this folder -
https://huggingface.co/hacksider/deep-live-cam/resolve/main/inswapper_128_fp16.onnx?download=true 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 os
import cv2 import cv2
import numpy as np import numpy as np
# Utility function to support unicode characters in file paths for reading
def imread_unicode(path, flags=cv2.IMREAD_COLOR): # Utility function to support unicode characters in file paths for reading.
return cv2.imdecode(np.fromfile(path, dtype=np.uint8), flags) # 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
# Utility function to support unicode characters in file paths for writing # characters (Chinese, Japanese, Cyrillic, accents, ...). Reading the bytes
def imwrite_unicode(path, img, params=None): # through NumPy (which uses Python's unicode-aware file I/O) and decoding them
root, ext = os.path.splitext(path) # in memory sidesteps that limitation. Returns None on failure, matching
if not ext: # cv2.imread() so it stays a drop-in replacement.
ext = ".png" def imread_unicode(path, flags=cv2.IMREAD_COLOR):
result, encoded_img = cv2.imencode(ext, img, params if params else []) try:
result, encoded_img = cv2.imencode(f".{ext}", img, params if params is not None else []) data = np.fromfile(path, dtype=np.uint8)
encoded_img.tofile(path) if data.size == 0:
return True return None
return False 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: if modules.globals.color_correction:
capture.set(cv2.CAP_PROP_CONVERT_RGB, 1) capture.set(cv2.CAP_PROP_CONVERT_RGB, 1)
frame_total = capture.get(cv2.CAP_PROP_FRAME_COUNT) frame_total = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
capture.set(cv2.CAP_PROP_POS_FRAMES, min(frame_total, frame_number - 1)) 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() has_frame, frame = capture.read()
if has_frame and modules.globals.color_correction: if has_frame and modules.globals.color_correction:
+12 -1
View File
@@ -1,10 +1,21 @@
import numpy as np import numpy as np
from sklearn.cluster import KMeans from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score
from typing import Any from typing import Any
def find_cluster_centroids(embeddings, max_k=10) -> 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 = [] inertia = []
cluster_centroids = [] cluster_centroids = []
K = range(1, max_k+1) 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('--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('--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-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}') program.add_argument('-v', '--version', action='version', version=f'{modules.metadata.name} {modules.metadata.version}')
# register deprecated args # register deprecated args
@@ -90,6 +90,12 @@ def parse_args() -> None:
modules.globals.execution_threads = args.execution_threads modules.globals.execution_threads = args.execution_threads
modules.globals.lang = args.lang 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 tumblers:
for enhancer_key in ('face_enhancer', 'face_enhancer_gpen256', 'face_enhancer_gpen512'): for enhancer_key in ('face_enhancer', 'face_enhancer_gpen256', 'face_enhancer_gpen512'):
modules.globals.fp_ui[enhancer_key] = enhancer_key in args.frame_processor 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: 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()) 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: if pref in available:
return pref return pref
return 'cpu' return 'cpu'
@@ -157,6 +163,8 @@ def suggest_execution_threads() -> int:
return 1 return 1
if 'CUDAExecutionProvider' in modules.globals.execution_providers: if 'CUDAExecutionProvider' in modules.globals.execution_providers:
return 2 return 2
if 'OpenVINOExecutionProvider' in modules.globals.execution_providers:
return 1
# For CPU execution, use most cores but leave some for system # For CPU execution, use most cores but leave some for system
return max(4, min(cpu_count - 2, 16)) return max(4, min(cpu_count - 2, 16))
@@ -170,9 +178,11 @@ def limit_resources() -> None:
tensorflow.config.experimental.set_memory_growth(gpu, True) tensorflow.config.experimental.set_memory_growth(gpu, True)
# limit memory usage # limit memory usage
if modules.globals.max_memory: 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': 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': if platform.system().lower() == 'windows':
import ctypes import ctypes
kernel32 = ctypes.windll.kernel32 kernel32 = ctypes.windll.kernel32
@@ -270,10 +280,9 @@ def start() -> None:
update_status('Falling back to disk-based processing...') update_status('Falling back to disk-based processing...')
extraction_start = time.time() extraction_start = time.time()
if not modules.globals.map_faces: create_temp(modules.globals.target_path)
create_temp(modules.globals.target_path) update_status('Extracting frames...')
update_status('Extracting frames...') extract_frames(modules.globals.target_path)
extract_frames(modules.globals.target_path)
extraction_time = time.time() - extraction_start extraction_time = time.time() - extraction_start
temp_frame_paths = get_temp_frame_paths(modules.globals.target_path) temp_frame_paths = get_temp_frame_paths(modules.globals.target_path)
@@ -324,7 +333,8 @@ def start() -> None:
def destroy(to_quit=True) -> None: def destroy(to_quit=True) -> None:
if modules.globals.target_path: if modules.globals.target_path:
clean_temp(modules.globals.target_path) clean_temp(modules.globals.target_path)
if to_quit: quit() if to_quit:
quit()
def run() -> None: def run() -> None:
+16 -7
View File
@@ -4,9 +4,8 @@ from typing import Any
import insightface import insightface
import threading import threading
import cv2
import numpy as np
import modules.globals import modules.globals
from modules import imread_unicode, imwrite_unicode
from tqdm import tqdm from tqdm import tqdm
from modules.typing import Frame from modules.typing import Frame
from modules.cluster_analysis import find_cluster_centroids, find_closest_centroid 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 ( from modules.processors.frame._onnx_enhancer import (
build_provider_config, build_provider_config,
) )
from modules.model_downloader import ensure_insightface_pack
ensure_insightface_pack('buffalo_l')
providers = build_provider_config() providers = build_provider_config()
FACE_ANALYSER = insightface.app.FaceAnalysis( FACE_ANALYSER = insightface.app.FaceAnalysis(
name='buffalo_l', name='buffalo_l',
@@ -255,8 +257,10 @@ def add_blank_map() -> Any:
def get_unique_faces_from_target_image() -> Any: def get_unique_faces_from_target_image() -> Any:
try: try:
modules.globals.source_target_map = [] 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) many_faces = get_many_faces(target_frame)
if many_faces is None:
return None
i = 0 i = 0
for face in many_faces: for face in many_faces:
@@ -289,8 +293,10 @@ def get_unique_faces_from_target_video() -> Any:
i = 0 i = 0
for temp_frame_path in tqdm(temp_frame_paths, desc="Extracting face embeddings from frames"): 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) many_faces = get_many_faces(temp_frame)
if many_faces is None:
continue
for face in many_faces: for face in many_faces:
face_embeddings.append(face.normed_embedding) face_embeddings.append(face.normed_embedding)
@@ -332,6 +338,9 @@ def default_target_face():
best_frame = frame best_frame = frame
break break
if best_face is None:
continue # No faces detected in this cluster — skip
for frame in map['target_faces_in_frame']: for frame in map['target_faces_in_frame']:
for face in frame['faces']: for face in frame['faces']:
if face['det_score'] > best_face['det_score']: 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'] 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'] = { map['target'] = {
'cv2' : target_frame[int(y_min):int(y_max), int(x_min):int(x_max)], 'cv2' : target_frame[int(y_min):int(y_max), int(x_min):int(x_max)],
'face' : best_face '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) 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}"): 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 j = 0
for face in frame['faces']: 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'] 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: 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 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__)) ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
WORKFLOW_DIR = os.path.join(ROOT_DIR, "workflow") WORKFLOW_DIR = os.path.join(ROOT_DIR, "workflow")
file_types = [ # Canonical media extensions, defined once so the file dialogs and
("Image", ("*.png", "*.jpg", "*.jpeg", "*.gif", "*.bmp")), # has_image_extension never drift. GIF is intentionally excluded: OpenCV's
("Video", ("*.mp4", "*.mkv")), # 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 # Face Mapping Data
source_target_map: List[Dict[str, Any]] = [] # Stores detailed map for image/video processing 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 os
import cv2 import cv2
import numpy as np import numpy as np
from typing import Tuple, Optional from typing import Tuple
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# CUDA availability detection (evaluated once at import time) # 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 # Collect all needed boundary constants
for _, (a, b) in splits: 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}", [a])
ensure_const(f"_sp_s{a + b}", [a + b]) 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_CUDA_PROVIDER: bool = "CUDAExecutionProvider" in ONNX_PROVIDERS
HAS_COREML_PROVIDER: bool = "CoreMLExecutionProvider" in ONNX_PROVIDERS HAS_COREML_PROVIDER: bool = "CoreMLExecutionProvider" in ONNX_PROVIDERS
HAS_DML_PROVIDER: bool = "DmlExecutionProvider" 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]]: def camera_backends() -> List[Tuple[int, int]]:
@@ -65,6 +74,8 @@ def accelerator_label() -> str:
return "CoreML (Apple Neural Engine)" return "CoreML (Apple Neural Engine)"
if HAS_COREML_PROVIDER: if HAS_COREML_PROVIDER:
return "CoreML" return "CoreML"
if HAS_OPENVINO_PROVIDER:
return "OpenVINO (Intel)"
if HAS_DML_PROVIDER: if HAS_DML_PROVIDER:
return "DirectML" return "DirectML"
return "CPU" return "CPU"
+13
View File
@@ -1,4 +1,17 @@
import importlib.util
import os
import numpy 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 import opennsfw2
from PIL import Image from PIL import Image
import cv2 # Add OpenCV import import cv2 # Add OpenCV import
@@ -14,6 +14,7 @@ import numpy as np
import onnxruntime import onnxruntime
import modules.globals import modules.globals
from modules.platform_info import OPENVINO_PROVIDER_CONFIG
IS_APPLE_SILICON = platform.system() == "Darwin" and platform.machine() == "arm64" IS_APPLE_SILICON = platform.system() == "Darwin" and platform.machine() == "arm64"
@@ -50,6 +51,9 @@ def build_provider_config(providers=None):
"AllowLowPrecisionAccumulationOnGPU": 1, "AllowLowPrecisionAccumulationOnGPU": 1,
}, },
)) ))
elif p == "OpenVINOExecutionProvider":
# AUTO lets OpenVINO select the best device
config.append(OPENVINO_PROVIDER_CONFIG)
else: else:
config.append(p) config.append(p)
return config 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}') frame_processor_module = importlib.import_module(f'modules.processors.frame.{frame_processor}')
for method_name in FRAME_PROCESSORS_INTERFACE: for method_name in FRAME_PROCESSORS_INTERFACE:
if not hasattr(frame_processor_module, method_name): if not hasattr(frame_processor_module, method_name):
print(f"Frame processor {frame_processor} is missing required method {method_name}")
sys.exit() sys.exit()
except ImportError: except ImportError:
print(f"Frame processor {frame_processor} not found") 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] current_processor_names = [proc.__name__.split('.')[-1] for proc in FRAME_PROCESSORS_MODULES]
for frame_processor, state in modules.globals.fp_ui.items(): 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: try:
frame_processor_module = load_frame_processor_module(frame_processor) frame_processor_module = load_frame_processor_module(frame_processor)
FRAME_PROCESSORS_MODULES.append(frame_processor_module) 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: except Exception as e:
print(f"Warning: Error loading frame processor {frame_processor} requested by UI state: {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: try:
module_to_remove = next((mod for mod in FRAME_PROCESSORS_MODULES if mod.__name__.endswith(f'.{frame_processor}')), None) module_to_remove = next((mod for mod in FRAME_PROCESSORS_MODULES if mod.__name__.endswith(f'.{frame_processor}')), None)
if module_to_remove: 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 Returns True on success, False on failure (caller should fall back to the
disk-based pipeline). disk-based pipeline).
""" """
import cv2 from modules import imread_unicode
from modules.face_analyser import get_one_face from modules.face_analyser import get_one_face
from modules.utilities import ( from modules.utilities import (
get_video_dimensions, 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) --- # --- Pre-load source face (needed by face_swapper in simple mode) ---
source_face = None source_face = None
if source_path and os.path.exists(source_path): 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: if source_img is not None:
source_face = get_one_face(source_img) source_face = get_one_face(source_img)
del source_img del source_img
+19 -12
View File
@@ -10,8 +10,9 @@ import onnxruntime
import modules.globals import modules.globals
import modules.processors.frame.core import modules.processors.frame.core
from modules import imread_unicode, imwrite_unicode
from modules.core import update_status 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.typing import Frame, Face
from modules.utilities import ( from modules.utilities import (
is_image, is_image,
@@ -22,6 +23,7 @@ FACE_ENHANCER = None
THREAD_SEMAPHORE = threading.Semaphore() THREAD_SEMAPHORE = threading.Semaphore()
THREAD_LOCK = threading.Lock() THREAD_LOCK = threading.Lock()
NAME = "DLC.FACE-ENHANCER" NAME = "DLC.FACE-ENHANCER"
MODEL_FILE = "gfpgan-1024.onnx"
abs_dir = os.path.dirname(os.path.abspath(__file__)) abs_dir = os.path.dirname(os.path.abspath(__file__))
models_dir = os.path.join( models_dir = os.path.join(
@@ -43,11 +45,12 @@ FFHQ_TEMPLATE_512 = np.array(
def pre_check() -> bool: def pre_check() -> bool:
model_path = os.path.join(models_dir, "gfpgan-1024.onnx") from modules.model_downloader import ensure_model
if not os.path.exists(model_path):
if ensure_model(MODEL_FILE) is None:
update_status( update_status(
f"GFPGAN ONNX model not found at {model_path}. " f"Could not obtain {MODEL_FILE}. Place it in the models folder "
"Please place gfpgan-1024.onnx in the models folder.", "manually or check your internet connection.",
NAME, NAME,
) )
return False return False
@@ -72,11 +75,15 @@ def get_face_enhancer() -> onnxruntime.InferenceSession:
with THREAD_LOCK: with THREAD_LOCK:
if FACE_ENHANCER is None: 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( 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: try:
@@ -407,7 +414,7 @@ def process_frames(
progress.update(1) progress.update(1)
continue continue
temp_frame = cv2.imread(temp_frame_path) temp_frame = imread_unicode(temp_frame_path)
if temp_frame is None: if temp_frame is None:
print( print(
f"{NAME}: Warning: Failed to read frame {temp_frame_path}, skipping." f"{NAME}: Warning: Failed to read frame {temp_frame_path}, skipping."
@@ -417,7 +424,7 @@ def process_frames(
continue continue
result_frame = process_frame(None, temp_frame) result_frame = process_frame(None, temp_frame)
cv2.imwrite(temp_frame_path, result_frame) imwrite_unicode(temp_frame_path, result_frame)
if progress: if progress:
progress.update(1) progress.update(1)
@@ -426,12 +433,12 @@ def process_image(
source_path: str | None, target_path: str, output_path: str source_path: str | None, target_path: str, output_path: str
) -> None: ) -> None:
"""Processes a single image file.""" """Processes a single image file."""
target_frame = cv2.imread(target_path) target_frame = imread_unicode(target_path)
if target_frame is None: if target_frame is None:
print(f"{NAME}: Error: Failed to read target image {target_path}") print(f"{NAME}: Error: Failed to read target image {target_path}")
return return
result_frame = process_frame(None, target_frame) 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}") print(f"{NAME}: Enhanced image saved to {output_path}")
@@ -4,11 +4,9 @@ from typing import Any, List
import os import os
import threading import threading
import cv2
import numpy as np
import modules.globals import modules.globals
import modules.processors.frame.core import modules.processors.frame.core
from modules import imread_unicode, imwrite_unicode
from modules.core import update_status from modules.core import update_status
from modules.face_analyser import get_one_face from modules.face_analyser import get_one_face
from modules.typing import Frame, Face from modules.typing import Frame, Face
@@ -24,7 +22,7 @@ from modules.processors.frame._onnx_enhancer import (
NAME = "DLC.FACE-ENHANCER-GPEN256" NAME = "DLC.FACE-ENHANCER-GPEN256"
INPUT_SIZE = 256 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" MODEL_FILE = "GPEN-BFR-256.onnx"
ENHANCER = None 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: def pre_check() -> bool:
model_path = os.path.join(models_dir, MODEL_FILE) if _obtain_model() is None:
if not os.path.exists(model_path): update_status(
update_status(f"Downloading {MODEL_FILE}...", NAME) f"Could not obtain {MODEL_FILE}. Place it in the models folder "
from modules.utilities import conditional_download "manually or check your internet connection.",
conditional_download(models_dir, [MODEL_URL]) NAME,
)
return False
return True return True
@@ -56,12 +75,11 @@ def get_enhancer() -> Any:
global ENHANCER global ENHANCER
with THREAD_LOCK: with THREAD_LOCK:
if ENHANCER is None: if ENHANCER is None:
model_path = os.path.join(models_dir, MODEL_FILE) model_path = _obtain_model()
if not os.path.exists(model_path): if model_path is None:
from modules.utilities import conditional_download raise FileNotFoundError(
conditional_download(models_dir, [MODEL_URL]) f"Model file not found: {os.path.join(models_dir, MODEL_FILE)}"
if not os.path.exists(model_path): )
raise FileNotFoundError(f"Model file not found: {model_path}")
print(f"{NAME}: Loading ONNX model from {model_path}") print(f"{NAME}: Loading ONNX model from {model_path}")
ENHANCER = create_onnx_session(model_path) ENHANCER = create_onnx_session(model_path)
warmup_session(ENHANCER) warmup_session(ENHANCER)
@@ -103,24 +121,24 @@ def process_frames(
source_path: str | None, temp_frame_paths: List[str], progress: Any = None source_path: str | None, temp_frame_paths: List[str], progress: Any = None
) -> None: ) -> None:
for temp_frame_path in temp_frame_paths: 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 temp_frame is None:
if progress: if progress:
progress.update(1) progress.update(1)
continue continue
result = process_frame(None, temp_frame) result = process_frame(None, temp_frame)
cv2.imwrite(temp_frame_path, result) imwrite_unicode(temp_frame_path, result)
if progress: if progress:
progress.update(1) progress.update(1)
def process_image(source_path: str | None, target_path: str, output_path: str) -> None: 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: if target_frame is None:
print(f"{NAME}: Error: Failed to read target image {target_path}") print(f"{NAME}: Error: Failed to read target image {target_path}")
return return
result_frame = process_frame(None, target_frame) 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}") print(f"{NAME}: Enhanced image saved to {output_path}")
@@ -4,11 +4,9 @@ from typing import Any, List
import os import os
import threading import threading
import cv2
import numpy as np
import modules.globals import modules.globals
import modules.processors.frame.core import modules.processors.frame.core
from modules import imread_unicode, imwrite_unicode
from modules.core import update_status from modules.core import update_status
from modules.face_analyser import get_one_face from modules.face_analyser import get_one_face
from modules.typing import Frame, Face from modules.typing import Frame, Face
@@ -24,7 +22,7 @@ from modules.processors.frame._onnx_enhancer import (
NAME = "DLC.FACE-ENHANCER-GPEN512" NAME = "DLC.FACE-ENHANCER-GPEN512"
INPUT_SIZE = 512 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" MODEL_FILE = "GPEN-BFR-512.onnx"
ENHANCER = None 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: def pre_check() -> bool:
model_path = os.path.join(models_dir, MODEL_FILE) if _obtain_model() is None:
if not os.path.exists(model_path): update_status(
update_status(f"Downloading {MODEL_FILE}...", NAME) f"Could not obtain {MODEL_FILE}. Place it in the models folder "
from modules.utilities import conditional_download "manually or check your internet connection.",
conditional_download(models_dir, [MODEL_URL]) NAME,
)
return False
return True return True
@@ -56,12 +75,11 @@ def get_enhancer() -> Any:
global ENHANCER global ENHANCER
with THREAD_LOCK: with THREAD_LOCK:
if ENHANCER is None: if ENHANCER is None:
model_path = os.path.join(models_dir, MODEL_FILE) model_path = _obtain_model()
if not os.path.exists(model_path): if model_path is None:
from modules.utilities import conditional_download raise FileNotFoundError(
conditional_download(models_dir, [MODEL_URL]) f"Model file not found: {os.path.join(models_dir, MODEL_FILE)}"
if not os.path.exists(model_path): )
raise FileNotFoundError(f"Model file not found: {model_path}")
print(f"{NAME}: Loading ONNX model from {model_path}") print(f"{NAME}: Loading ONNX model from {model_path}")
ENHANCER = create_onnx_session(model_path) ENHANCER = create_onnx_session(model_path)
warmup_session(ENHANCER) warmup_session(ENHANCER)
@@ -103,24 +121,24 @@ def process_frames(
source_path: str | None, temp_frame_paths: List[str], progress: Any = None source_path: str | None, temp_frame_paths: List[str], progress: Any = None
) -> None: ) -> None:
for temp_frame_path in temp_frame_paths: 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 temp_frame is None:
if progress: if progress:
progress.update(1) progress.update(1)
continue continue
result = process_frame(None, temp_frame) result = process_frame(None, temp_frame)
cv2.imwrite(temp_frame_path, result) imwrite_unicode(temp_frame_path, result)
if progress: if progress:
progress.update(1) progress.update(1)
def process_image(source_path: str | None, target_path: str, output_path: str) -> None: 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: if target_frame is None:
print(f"{NAME}: Error: Failed to read target image {target_path}") print(f"{NAME}: Error: Failed to read target image {target_path}")
return return
result_frame = process_frame(None, target_frame) 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}") print(f"{NAME}: Enhanced image saved to {output_path}")
+1 -1
View File
@@ -2,7 +2,7 @@ import cv2
import numpy as np import numpy as np
from modules.typing import Face, Frame from modules.typing import Face, Frame
import modules.globals 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): def apply_color_transfer(source, target):
""" """
+50 -29
View File
@@ -7,16 +7,17 @@ import numpy as np
import platform import platform
import modules.globals import modules.globals
import modules.processors.frame.core import modules.processors.frame.core
from modules import imread_unicode, imwrite_unicode
from modules.core import update_status from modules.core import update_status
from modules.face_analyser import get_one_face, get_many_faces, default_source_face from modules.face_analyser import get_one_face, get_many_faces, default_source_face
from modules.typing import Face, Frame from modules.typing import Face, Frame
from modules.utilities import ( from modules.utilities import (
conditional_download,
is_image, is_image,
is_video, is_video,
) )
from modules.cluster_analysis import find_closest_centroid 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 import os
from collections import deque from collections import deque
import time import time
@@ -190,21 +191,26 @@ models_dir = os.path.join(
def pre_check() -> bool: def pre_check() -> bool:
# Use models_dir instead of abs_dir to save to the correct location # Use models_dir instead of abs_dir to save to the correct location
download_directory_path = models_dir download_directory_path = models_dir
# Make sure the models directory exists, catch permission errors if they occur # Make sure the models directory exists, catch permission errors if they occur
try: try:
os.makedirs(download_directory_path, exist_ok=True) os.makedirs(download_directory_path, exist_ok=True)
except OSError as e: except OSError as e:
logging.error(f"Failed to create directory {download_directory_path} due to permission error: {e}") logging.error(f"Failed to create directory {download_directory_path} due to permission error: {e}")
return False return False
# Use the direct download URL from Hugging Face (FP32 model for broad GPU compatibility) from modules.model_downloader import ensure_any
conditional_download(
download_directory_path, variants = ["inswapper_128.onnx", "inswapper_128_fp16.onnx"]
[ if _HAS_TORCH_CUDA:
"https://huggingface.co/hacksider/deep-live-cam/resolve/main/inswapper_128.onnx" 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 return True
@@ -240,8 +246,12 @@ def get_face_swapper() -> Any:
elif os.path.exists(fp32_path): elif os.path.exists(fp32_path):
model_path = fp32_path model_path = fp32_path
else: else:
update_status(f"No inswapper model found in {models_dir}.", NAME) if not pre_check():
return None 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 # On Apple Silicon, rewrite Pad(reflect) → Slice+Concat so
# CoreML can run the entire model in a single partition on # CoreML can run the entire model in a single partition on
# the Neural Engine instead of bouncing between CPU and ANE. # 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 # Use bare provider — ONNX Runtime defaults are
# fastest on modern GPUs (Blackwell/sm_120). # fastest on modern GPUs (Blackwell/sm_120).
providers_config.append(p) providers_config.append(p)
elif p == "OpenVINOExecutionProvider":
providers_config.append(OPENVINO_PROVIDER_CONFIG)
else: else:
providers_config.append(p) providers_config.append(p)
FACE_SWAPPER = insightface.model_zoo.get_model( 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 continue
face_region = processed_frame[y1:y2, x1:x2] 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) # Apply sharpening (GPU-accelerated when CUDA OpenCV is available)
try: try:
@@ -815,9 +828,11 @@ def process_frame_v2(temp_frame: Frame, temp_frame_path: str = "") -> Frame:
else: # Single face or specific mapping else: # Single face or specific mapping
for map_data in source_target_map: for map_data in source_target_map:
source_info = map_data.get("source", {}) 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") 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): if is_image(modules.globals.target_path):
target_info = map_data.get("target", {}) 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): if len(detected_faces) <= len(target_embeddings):
# More targets defined than detected - match each detected face # More targets defined than detected - match each detected face
for detected_face in detected_faces: 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) closest_idx, _ = find_closest_centroid(target_embeddings, detected_face.normed_embedding)
if 0 <= closest_idx < len(source_faces): if 0 <= closest_idx < len(source_faces):
source_target_pairs.append((source_faces[closest_idx], detected_face)) 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 # 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_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] 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): for i, target_embedding in enumerate(target_embeddings):
if 0 <= i < len(source_faces): # Ensure source face exists for this embedding 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. # Log the error but allow proceeding; subsequent check will stop processing.
else: else:
try: try:
source_img = cv2.imread(source_path) source_img = imread_unicode(source_path)
if source_img is None: if source_img is None:
# Specific error for file reading failure # Specific error for file reading failure
update_status(f"Error reading source image file {source_path}. Please check the path and file integrity.", NAME) 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 --- # --- Stop processing entirely if in Simple Mode and source face is invalid ---
if not use_v2 and source_face is None: 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: if progress:
# Ensure the progress bar completes if it was started # Ensure the progress bar completes if it was started
remaining_updates = total_frames - progress.n if hasattr(progress, 'n') else total_frames remaining_updates = total_frames - progress.n if hasattr(progress, 'n') else total_frames
@@ -952,14 +969,16 @@ def process_frames(
# Read the target frame # Read the target frame
temp_frame = None temp_frame = None
try: try:
temp_frame = cv2.imread(temp_frame_path) temp_frame = imread_unicode(temp_frame_path)
if temp_frame is None: if temp_frame is None:
print(f"{NAME}: Error: Could not read frame: {temp_frame_path}, skipping.") 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 continue # Skip this frame if read fails
except Exception as read_e: except Exception as read_e:
print(f"{NAME}: Error reading frame {temp_frame_path}: {read_e}, skipping.") print(f"{NAME}: Error reading frame {temp_frame_path}: {read_e}, skipping.")
if progress: progress.update(1) if progress:
progress.update(1)
continue continue
# Select processing function and execute # Select processing function and execute
@@ -988,7 +1007,7 @@ def process_frames(
# Write the result back to the same frame path with optimized compression # Write the result back to the same frame path with optimized compression
try: try:
# Use PNG compression level 3 (faster) instead of default 9 # 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: if not write_success:
print(f"{NAME}: Error: Failed to write processed frame to {temp_frame_path}") print(f"{NAME}: Error: Failed to write processed frame to {temp_frame_path}")
except Exception as write_e: 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 # Read target first
try: try:
target_frame = cv2.imread(target_path) target_frame = imread_unicode(target_path)
if target_frame is None: if target_frame is None:
update_status(f"Error: Could not read target image: {target_path}", NAME) update_status(f"Error: Could not read target image: {target_path}", NAME)
return return
@@ -1037,7 +1056,7 @@ def process_image(source_path: str, target_path: str, output_path: str) -> None:
else: # Simple mode else: # Simple mode
try: try:
source_img = cv2.imread(source_path) source_img = imread_unicode(source_path)
if source_img is None: if source_img is None:
update_status(f"Error: Could not read source image: {source_path}", NAME) update_status(f"Error: Could not read source image: {source_path}", NAME)
return 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 # Write the result if processing was successful
if result is not None: if result is not None:
write_success = cv2.imwrite(output_path, result) write_success = imwrite_unicode(output_path, result)
if write_success: if write_success:
update_status(f"Output image saved to: {output_path}", NAME) update_status(f"Output image saved to: {output_path}", NAME)
else: else:
@@ -1496,7 +1515,8 @@ def apply_color_transfer(source, target):
if len(source.shape) == 2: # Grayscale if len(source.shape) == 2: # Grayscale
source = cv2.cvtColor(source, cv2.COLOR_GRAY2BGR) source = cv2.cvtColor(source, cv2.COLOR_GRAY2BGR)
source = np.clip(source, 0, 255).astype(np.uint8) 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: except Exception:
return source return source
if len(target.shape) != 3 or target.shape[2] != 3 or target.dtype != np.uint8: 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 if len(target.shape) == 2: # Grayscale
target = cv2.cvtColor(target, cv2.COLOR_GRAY2BGR) target = cv2.cvtColor(target, cv2.COLOR_GRAY2BGR)
target = np.clip(target, 0, 255).astype(np.uint8) 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: except Exception:
return source # Return original source if target invalid return source # Return original source if target invalid
+2 -2
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# Import the tkinter fix to patch the ScreenChanged error # Import the tkinter fix to patch the ScreenChanged error (module patches Tk on import)
import tkinter_fix import tkinter_fix # noqa: F401
import core import core
+24 -11
View File
@@ -73,6 +73,7 @@ from modules.utilities import (
is_image, is_image,
is_video, is_video,
) )
from modules import imread_unicode
from modules.video_capture import VideoCapturer from modules.video_capture import VideoCapturer
if platform.system() == "Windows": if platform.system() == "Windows":
@@ -236,6 +237,18 @@ _RECENT_SOURCE_DIR: Optional[str] = None
_RECENT_TARGET_DIR: Optional[str] = None _RECENT_TARGET_DIR: Optional[str] = None
_RECENT_OUTPUT_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 ───────────────────────────────────────────────────── # ─── image utilities ─────────────────────────────────────────────────────
@@ -416,7 +429,7 @@ def get_available_cameras() -> Tuple[List[int], List[str]]:
indices: List[int] = [] indices: List[int] = []
names: List[str] = [] names: List[str] = []
for i in range(10): for i in range(10):
cap = cv2.VideoCapture(i) cap = cv2.VideoCapture(f"/dev/video{i}")
if cap.isOpened(): if cap.isOpened():
indices.append(i) indices.append(i)
names.append(f"Camera {i}") names.append(f"Camera {i}")
@@ -733,7 +746,7 @@ class MainWindow(QMainWindow):
path, _filter = QFileDialog.getOpenFileName( path, _filter = QFileDialog.getOpenFileName(
self, _("select an source image"), self, _("select an source image"),
_RECENT_SOURCE_DIR or "", _RECENT_SOURCE_DIR or "",
"Images (*.png *.jpg *.jpeg *.gif *.bmp)", _IMAGE_FILE_FILTER,
) )
if path and is_image(path): if path and is_image(path):
modules.globals.source_path = path modules.globals.source_path = path
@@ -754,7 +767,7 @@ class MainWindow(QMainWindow):
path, _filter = QFileDialog.getOpenFileName( path, _filter = QFileDialog.getOpenFileName(
self, _("select an target image or video"), self, _("select an target image or video"),
_RECENT_TARGET_DIR or "", _RECENT_TARGET_DIR or "",
"Media (*.png *.jpg *.jpeg *.gif *.bmp *.mp4 *.mkv)", _MEDIA_FILE_FILTER,
) )
if not path: if not path:
return return
@@ -885,13 +898,13 @@ class MainWindow(QMainWindow):
path, _f = QFileDialog.getSaveFileName( path, _f = QFileDialog.getSaveFileName(
self, _("save image output file"), self, _("save image output file"),
os.path.join(_RECENT_OUTPUT_DIR or "", "output.png"), os.path.join(_RECENT_OUTPUT_DIR or "", "output.png"),
"Images (*.png *.jpg *.jpeg *.bmp)", _IMAGE_FILE_FILTER,
) )
elif is_video(modules.globals.target_path): elif is_video(modules.globals.target_path):
path, _f = QFileDialog.getSaveFileName( path, _f = QFileDialog.getSaveFileName(
self, _("save video output file"), self, _("save video output file"),
os.path.join(_RECENT_OUTPUT_DIR or "", "output.mp4"), os.path.join(_RECENT_OUTPUT_DIR or "", "output.mp4"),
"Videos (*.mp4 *.mkv)", _VIDEO_FILE_FILTER,
) )
else: else:
return return
@@ -988,7 +1001,7 @@ class PreviewWindow(QWidget):
from modules.processors.frame.core import get_frame_processors_modules as _gfpm from modules.processors.frame.core import get_frame_processors_modules as _gfpm
for fp in _gfpm(modules.globals.frame_processors): for fp in _gfpm(modules.globals.frame_processors):
temp_frame = fp.process_frame( 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. # Fit to current widget size while preserving aspect ratio.
h, w = temp_frame.shape[:2] h, w = temp_frame.shape[:2]
@@ -1071,7 +1084,7 @@ class _ProcessingWorker(QThread):
and modules.globals.source_path != last_source_path and modules.globals.source_path != last_source_path
): ):
last_source_path = modules.globals.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 det_count += 1
if det_count % det_interval == 0: if det_count % det_interval == 0:
@@ -1333,11 +1346,11 @@ class MapperDialog(QDialog):
path, _f = QFileDialog.getOpenFileName( path, _f = QFileDialog.getOpenFileName(
self, _("select an source image"), self, _("select an source image"),
_RECENT_SOURCE_DIR or "", _RECENT_SOURCE_DIR or "",
"Images (*.png *.jpg *.jpeg *.gif *.bmp)", _IMAGE_FILE_FILTER,
) )
if not path: if not path:
return return
cv2_img = cv2.imread(path) cv2_img = imread_unicode(path)
face = get_one_face(cv2_img) face = get_one_face(cv2_img)
if face is None: if face is None:
self.set_status("Face could not be detected in last upload!") self.set_status("Face could not be detected in last upload!")
@@ -1438,11 +1451,11 @@ class LiveMapperDialog(QDialog):
path, _f = QFileDialog.getOpenFileName( path, _f = QFileDialog.getOpenFileName(
self, _("select an source image"), self, _("select an source image"),
_RECENT_SOURCE_DIR or "", _RECENT_SOURCE_DIR or "",
"Images (*.png *.jpg *.jpeg *.gif *.bmp)", _IMAGE_FILE_FILTER,
) )
if not path: if not path:
return return
cv2_img = cv2.imread(path) cv2_img = imread_unicode(path)
face = get_one_face(cv2_img) face = get_one_face(cv2_img)
if face is None: if face is None:
self.set_status("Face could not be detected in last upload!") 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: 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: def is_image(image_path: str) -> bool:
if image_path and os.path.isfile(image_path): 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) mimetype, _ = mimetypes.guess_type(image_path)
return bool(mimetype and mimetype.startswith("image/")) return bool(mimetype and mimetype.startswith("image/"))
return False return False
+2 -2
View File
@@ -1,6 +1,5 @@
import cv2 import cv2
import numpy as np import numpy as np
import sys
import time import time
from typing import Optional, Tuple, Callable from typing import Optional, Tuple, Callable
import platform import platform
@@ -72,8 +71,9 @@ class VideoCapturer:
self.cap.release() self.cap.release()
except Exception: except Exception:
continue continue
elif platform.system() == "Linux":
self.cap = cv2.VideoCapture(f"/dev/video{self.device_index}")
else: else:
# Unix-like systems (Linux/Mac) capture method
self.cap = cv2.VideoCapture(self.device_index) self.cap = cv2.VideoCapture(self.device_index)
if not self.cap or not self.cap.isOpened(): 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 numpy>=2.0,<3
typing-extensions>=4.8.0 typing-extensions>=4.15.0
opencv-python==4.10.0.84 opencv-python==4.14.0.94
cv2_enumerate_cameras==1.1.15 opencv-python-headless==4.14.0.94
onnx==1.18.0 cv2_enumerate_cameras==1.3.3
onnx==1.22.0
insightface==0.7.3 insightface==0.7.3
psutil==5.9.8 psutil==7.2.2
PySide6>=6.7,<7 PySide6>=6.7,<7
pillow==12.1.1 pillow==12.3.0
tqdm>=4.65.0 tqdm>=4.66.3
onnxruntime-silicon==1.16.3; sys_platform == 'darwin' and platform_machine == 'arm64' onnxruntime==1.28.0; sys_platform == 'darwin' and platform_machine == 'arm64'
onnxruntime-gpu==1.23.2; sys_platform != 'darwin' onnxruntime==1.23.0; sys_platform == 'darwin' and platform_machine != 'arm64'
tensorflow>=2.15.0; sys_platform != 'darwin' onnxruntime-gpu==1.26.0; sys_platform != 'darwin'
tensorflow>=2.15.0; sys_platform == 'darwin' and python_version < '3.13' opennsfw2==0.18.0
opennsfw2==0.10.2 keras>=3.0.0
protobuf==4.25.1 protobuf>=6.33.5,<8
pygrabber; sys_platform == 'win32' pygrabber; sys_platform == 'win32'
+24
View File
@@ -31,6 +31,30 @@ if sys.platform == "win32":
except (OSError, AttributeError): except (OSError, AttributeError):
pass 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 # On Linux, pre-load NVIDIA shared libraries (cuDNN, cuBLAS, nvrtc...) shipped
# inside the venv via pip wheels (nvidia-cudnn-cu12, etc.). LD_LIBRARY_PATH # 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 # 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()