diff --git a/modules/__init__.py b/modules/__init__.py index 1e5a2ad..29b1ee7 100644 --- a/modules/__init__.py +++ b/modules/__init__.py @@ -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 \ No newline at end of file +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 diff --git a/modules/face_analyser.py b/modules/face_analyser.py index ccc438a..b2e48f7 100644 --- a/modules/face_analyser.py +++ b/modules/face_analyser.py @@ -4,8 +4,8 @@ from typing import Any 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 @@ -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 @@ -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 @@ -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 @@ -359,7 +359,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']: @@ -367,5 +367,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 diff --git a/modules/processors/frame/core.py b/modules/processors/frame/core.py index 64330c6..7645815 100644 --- a/modules/processors/frame/core.py +++ b/modules/processors/frame/core.py @@ -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, @@ -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 diff --git a/modules/processors/frame/face_enhancer.py b/modules/processors/frame/face_enhancer.py index 12e65aa..eff313a 100644 --- a/modules/processors/frame/face_enhancer.py +++ b/modules/processors/frame/face_enhancer.py @@ -10,6 +10,7 @@ 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_many_faces from modules.typing import Frame, Face @@ -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." @@ -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) @@ -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}") diff --git a/modules/processors/frame/face_enhancer_gpen256.py b/modules/processors/frame/face_enhancer_gpen256.py index 9814ebd..7d1fae7 100644 --- a/modules/processors/frame/face_enhancer_gpen256.py +++ b/modules/processors/frame/face_enhancer_gpen256.py @@ -4,10 +4,9 @@ from typing import Any, List 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 @@ -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}") diff --git a/modules/processors/frame/face_enhancer_gpen512.py b/modules/processors/frame/face_enhancer_gpen512.py index b6b792a..36168be 100644 --- a/modules/processors/frame/face_enhancer_gpen512.py +++ b/modules/processors/frame/face_enhancer_gpen512.py @@ -4,10 +4,9 @@ from typing import Any, List 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 @@ -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}") diff --git a/modules/processors/frame/face_swapper.py b/modules/processors/frame/face_swapper.py index 109c359..7559af6 100644 --- a/modules/processors/frame/face_swapper.py +++ b/modules/processors/frame/face_swapper.py @@ -7,6 +7,7 @@ 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 @@ -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) @@ -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: @@ -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: @@ -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 @@ -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 @@ -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: diff --git a/modules/ui.py b/modules/ui.py index cade5f5..fca9f30 100644 --- a/modules/ui.py +++ b/modules/ui.py @@ -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": @@ -988,7 +989,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 +1072,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: @@ -1337,7 +1338,7 @@ class MapperDialog(QDialog): ) 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!") @@ -1442,7 +1443,7 @@ class LiveMapperDialog(QDialog): ) 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!")