Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 38 additions & 18 deletions modules/__init__.py
Original file line number Diff line number Diff line change
@@ -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 is not None else [])
if not result:
return False
encoded_img.tofile(path)
return True
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
12 changes: 6 additions & 6 deletions modules/face_analyser.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@
import insightface
import threading

import cv2
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
Expand Down Expand Up @@ -254,7 +254,7 @@ 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
Expand Down Expand Up @@ -290,7 +290,7 @@ 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
Expand Down Expand Up @@ -343,7 +343,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
Expand All @@ -359,13 +359,13 @@ 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']:
if face['target_centroid'] == i:
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
4 changes: 2 additions & 2 deletions modules/processors/frame/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,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,
Expand All @@ -139,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
Expand Down
9 changes: 5 additions & 4 deletions modules/processors/frame/face_enhancer.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

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_many_faces
from modules.typing import Frame, Face
Expand Down Expand Up @@ -407,7 +408,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."
Expand All @@ -417,7 +418,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)

Expand All @@ -426,12 +427,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}")


Expand Down
11 changes: 5 additions & 6 deletions modules/processors/frame/face_enhancer_gpen256.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,9 @@
import os
import threading

import cv2

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
Expand Down Expand Up @@ -102,24 +101,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}")


Expand Down
11 changes: 5 additions & 6 deletions modules/processors/frame/face_enhancer_gpen512.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,9 @@
import os
import threading

import cv2

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
Expand Down Expand Up @@ -102,24 +101,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}")


Expand Down
13 changes: 7 additions & 6 deletions modules/processors/frame/face_swapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
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
Expand Down Expand Up @@ -917,7 +918,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)
Expand Down Expand Up @@ -957,7 +958,7 @@ 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:
Expand Down Expand Up @@ -995,7 +996,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:
Expand Down Expand Up @@ -1025,7 +1026,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
Expand All @@ -1044,7 +1045,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
Expand All @@ -1060,7 +1061,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:
Expand Down
9 changes: 5 additions & 4 deletions modules/ui.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@
is_image,
is_video,
)
from modules import imread_unicode
from modules.video_capture import VideoCapturer

if platform.system() == "Windows":
Expand Down Expand Up @@ -988,7 +989,7 @@ def refresh_frame(self, frame_number: int = 0) -> None:
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]
Expand Down Expand Up @@ -1071,7 +1072,7 @@ def run(self) -> None:
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:
Expand Down Expand Up @@ -1337,7 +1338,7 @@ def _select_source(self, row: int) -> None:
)
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!")
Expand Down Expand Up @@ -1442,7 +1443,7 @@ def _select_face(self, row: int, kind: str) -> None:
)
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!")
Expand Down
Loading