mirror of
https://github.com/hacksider/Deep-Live-Cam.git
synced 2026-09-05 15:26:38 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bc48ba4308 | ||
|
|
b53844eb57 | ||
|
|
20cafb0079 | ||
|
|
7ca6d0b202 | ||
|
|
f7db37679a | ||
|
|
bd500d54b4 | ||
|
|
987f6b392b | ||
|
|
97a44800a2 | ||
|
|
345fa4a0b0 | ||
|
|
bdeeb3ace0 | ||
|
|
230217ec11 | ||
|
|
156321f7a3 | ||
|
|
8234965ee8 | ||
|
|
ab64c186ec | ||
|
|
b8e781e539 | ||
|
|
ff7ee0d219 | ||
|
|
8d727eba3e | ||
|
|
eba2a958d3 | ||
|
|
14ba4f9c0b | ||
|
|
7d2d7fb1f3 | ||
|
|
57c4c32377 | ||
|
|
d00b09f5d8 | ||
|
|
897dc21da4 | ||
|
|
834092c891 | ||
|
|
da0672ad6b | ||
|
|
47dffeb307 | ||
|
|
9e1f0cc3a5 | ||
|
|
834bc43768 | ||
|
|
3b69413d61 | ||
|
|
07e2e960c8 | ||
|
|
ba27b75265 | ||
|
|
cfa8123b67 | ||
|
|
08b2dd2526 | ||
|
|
886e64b320 | ||
|
|
aa6f2cbade | ||
|
|
0b61ad5c0d |
@@ -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"
|
||||
@@ -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>
|
||||
|
||||
<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>
|
||||
|
||||
<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
|
||||

|
||||
@@ -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
|
||||
|
||||
@@ -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,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
@@ -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
@@ -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:
|
||||
|
||||
@@ -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
@@ -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:
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -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])
|
||||
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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}")
|
||||
|
||||
|
||||
|
||||
@@ -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):
|
||||
"""
|
||||
|
||||
@@ -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
@@ -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
@@ -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!")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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
@@ -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'
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user