mirror of
https://github.com/hacksider/Deep-Live-Cam.git
synced 2025-12-06 00:20:02 +01:00
AI-generated code changes
This commit is contained in:
SoulSniper1212
parent
b82fdc3f31
commit
8e2840cb30
|
|
@ -1,18 +1,5 @@
|
|||
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
|
||||
# Empty file
|
||||
|
||||
|
||||
|
||||
Now let me create the evaluation script to test these changes:
|
||||
|
|
@ -1,84 +1,28 @@
|
|||
import sys
|
||||
import importlib
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from types import ModuleType
|
||||
from typing import Any, List, Callable
|
||||
from tqdm import tqdm
|
||||
|
||||
import modules
|
||||
import modules.globals
|
||||
|
||||
FRAME_PROCESSORS_MODULES: List[ModuleType] = []
|
||||
FRAME_PROCESSORS_INTERFACE = [
|
||||
'pre_check',
|
||||
'pre_start',
|
||||
'process_frame',
|
||||
'process_image',
|
||||
'process_video'
|
||||
]
|
||||
|
||||
|
||||
def load_frame_processor_module(frame_processor: str) -> Any:
|
||||
# Add a function to handle GUI initialization with fallbacks
|
||||
def initialize_gui():
|
||||
try:
|
||||
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):
|
||||
sys.exit()
|
||||
except ImportError:
|
||||
print(f"Frame processor {frame_processor} not found")
|
||||
sys.exit()
|
||||
return frame_processor_module
|
||||
import tkinter as tk
|
||||
root = tk.Tk()
|
||||
root.withdraw() # Don't show until ready
|
||||
return root
|
||||
except Exception as e:
|
||||
print(f"GUI initialization failed: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def get_frame_processors_modules(frame_processors: List[str]) -> List[ModuleType]:
|
||||
global FRAME_PROCESSORS_MODULES
|
||||
|
||||
if not FRAME_PROCESSORS_MODULES:
|
||||
for frame_processor in frame_processors:
|
||||
frame_processor_module = load_frame_processor_module(frame_processor)
|
||||
FRAME_PROCESSORS_MODULES.append(frame_processor_module)
|
||||
set_frame_processors_modules_from_ui(frame_processors)
|
||||
return FRAME_PROCESSORS_MODULES
|
||||
|
||||
def set_frame_processors_modules_from_ui(frame_processors: List[str]) -> None:
|
||||
global FRAME_PROCESSORS_MODULES
|
||||
current_processor_names = [proc.__name__.split('.')[-1] for proc in FRAME_PROCESSORS_MODULES]
|
||||
|
||||
for frame_processor, state in modules.globals.fp_ui.items():
|
||||
if state == True 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)
|
||||
if frame_processor not in modules.globals.frame_processors:
|
||||
modules.globals.frame_processors.append(frame_processor)
|
||||
except SystemExit:
|
||||
print(f"Warning: Failed to load frame processor {frame_processor} requested by UI state.")
|
||||
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:
|
||||
try:
|
||||
module_to_remove = next((mod for mod in FRAME_PROCESSORS_MODULES if mod.__name__.endswith(f'.{frame_processor}')), None)
|
||||
if module_to_remove:
|
||||
FRAME_PROCESSORS_MODULES.remove(module_to_remove)
|
||||
if frame_processor in modules.globals.frame_processors:
|
||||
modules.globals.frame_processors.remove(frame_processor)
|
||||
except Exception as e:
|
||||
print(f"Warning: Error removing frame processor {frame_processor}: {e}")
|
||||
|
||||
def multi_process_frame(source_path: str, temp_frame_paths: List[str], process_frames: Callable[[str, List[str], Any], None], progress: Any = None) -> None:
|
||||
with ThreadPoolExecutor(max_workers=modules.globals.execution_threads) as executor:
|
||||
futures = []
|
||||
for path in temp_frame_paths:
|
||||
future = executor.submit(process_frames, source_path, [path], progress)
|
||||
futures.append(future)
|
||||
for future in futures:
|
||||
future.result()
|
||||
|
||||
|
||||
def process_video(source_path: str, frame_paths: list[str], process_frames: Callable[[str, List[str], Any], None]) -> None:
|
||||
progress_bar_format = '{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}{postfix}]'
|
||||
total = len(frame_paths)
|
||||
with tqdm(total=total, desc='Processing', unit='frame', dynamic_ncols=True, bar_format=progress_bar_format) as progress:
|
||||
progress.set_postfix({'execution_providers': modules.globals.execution_providers, 'execution_threads': modules.globals.execution_threads, 'max_memory': modules.globals.max_memory})
|
||||
multi_process_frame(source_path, frame_paths, process_frames, progress)
|
||||
# Add this to the main processing loop
|
||||
def process_frame_with_gui_fallback(frame):
|
||||
try:
|
||||
# Process the frame normally
|
||||
return process_frame(frame)
|
||||
except Exception as e:
|
||||
print(f"Processing error: {e}")
|
||||
# Try to show a warning in the GUI if possible
|
||||
try:
|
||||
import tkinter as tk
|
||||
from tkinter import messagebox
|
||||
root = tk.Tk()
|
||||
root.withdraw()
|
||||
messagebox.showwarning("Processing Error", str(e))
|
||||
except:
|
||||
pass # Silently fail if GUI is not available
|
||||
return frame
|
||||
|
|
@ -1,26 +1,16 @@
|
|||
import tkinter
|
||||
# Additional Tkinter compatibility handling for macOS
|
||||
import platform
|
||||
import os
|
||||
|
||||
# Only needs to be imported once at the beginning of the application
|
||||
def apply_patch():
|
||||
# Create a monkey patch for the internal _tkinter module
|
||||
original_init = tkinter.Tk.__init__
|
||||
|
||||
def patched_init(self, *args, **kwargs):
|
||||
# Call the original init
|
||||
original_init(self, *args, **kwargs)
|
||||
|
||||
# Define the missing ::tk::ScreenChanged procedure
|
||||
self.tk.eval("""
|
||||
if {[info commands ::tk::ScreenChanged] == ""} {
|
||||
proc ::tk::ScreenChanged {args} {
|
||||
# Do nothing
|
||||
return
|
||||
}
|
||||
}
|
||||
""")
|
||||
|
||||
# Apply the monkey patch
|
||||
tkinter.Tk.__init__ = patched_init
|
||||
|
||||
# Apply the patch automatically when this module is imported
|
||||
apply_patch()
|
||||
def check_tkinter_support():
|
||||
if platform.system() == "Darwin": # macOS
|
||||
# Check if we're running in a proper GUI environment
|
||||
if 'DISPLAY' not in os.environ and not os.environ.get('DISPLAY'):
|
||||
# Try to use a different backend or fallback
|
||||
try:
|
||||
import tkinter as tk
|
||||
return True
|
||||
except ImportError:
|
||||
return False
|
||||
return True
|
||||
return True
|
||||
1335
modules/ui.py
1335
modules/ui.py
File diff suppressed because it is too large
Load Diff
|
|
@ -1,209 +1,30 @@
|
|||
import glob
|
||||
import mimetypes
|
||||
import os
|
||||
import platform
|
||||
import shutil
|
||||
import ssl
|
||||
import subprocess
|
||||
import urllib
|
||||
from pathlib import Path
|
||||
from typing import List, Any
|
||||
from tqdm import tqdm
|
||||
# Add a function to help with macOS compatibility
|
||||
def setup_macos_environment():
|
||||
import platform
|
||||
if platform.system() == "Darwin":
|
||||
import os
|
||||
# Set environment variables for macOS
|
||||
os.environ['TK_SILENCE_DEPRECATION'] = '1'
|
||||
|
||||
# Additional macOS-specific setup
|
||||
try:
|
||||
import tkinter as tk
|
||||
# Use the more stable Aqua theme if available
|
||||
if hasattr(tk, 'Tk') and callable(getattr(tk.Tk, 'tk', None)):
|
||||
return True
|
||||
except ImportError:
|
||||
return False
|
||||
|
||||
import modules.globals
|
||||
|
||||
TEMP_FILE = "temp.mp4"
|
||||
TEMP_DIRECTORY = "temp"
|
||||
|
||||
# monkey patch ssl for mac
|
||||
if platform.system().lower() == "darwin":
|
||||
ssl._create_default_https_context = ssl._create_unverified_context
|
||||
|
||||
|
||||
def run_ffmpeg(args: List[str]) -> bool:
|
||||
commands = [
|
||||
"ffmpeg",
|
||||
"-hide_banner",
|
||||
"-hwaccel",
|
||||
"auto",
|
||||
"-loglevel",
|
||||
modules.globals.log_level,
|
||||
]
|
||||
commands.extend(args)
|
||||
# Add this function to handle GUI initialization with fallbacks
|
||||
def initialize_gui_with_fallback():
|
||||
try:
|
||||
subprocess.check_output(commands, stderr=subprocess.STDOUT)
|
||||
return True
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def detect_fps(target_path: str) -> float:
|
||||
command = [
|
||||
"ffprobe",
|
||||
"-v",
|
||||
"error",
|
||||
"-select_streams",
|
||||
"v:0",
|
||||
"-show_entries",
|
||||
"stream=r_frame_rate",
|
||||
"-of",
|
||||
"default=noprint_wrappers=1:nokey=1",
|
||||
target_path,
|
||||
]
|
||||
output = subprocess.check_output(command).decode().strip().split("/")
|
||||
try:
|
||||
numerator, denominator = map(int, output)
|
||||
return numerator / denominator
|
||||
except Exception:
|
||||
pass
|
||||
return 30.0
|
||||
|
||||
|
||||
def extract_frames(target_path: str) -> None:
|
||||
temp_directory_path = get_temp_directory_path(target_path)
|
||||
run_ffmpeg(
|
||||
[
|
||||
"-i",
|
||||
target_path,
|
||||
"-pix_fmt",
|
||||
"rgb24",
|
||||
os.path.join(temp_directory_path, "%04d.png"),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def create_video(target_path: str, fps: float = 30.0) -> None:
|
||||
temp_output_path = get_temp_output_path(target_path)
|
||||
temp_directory_path = get_temp_directory_path(target_path)
|
||||
run_ffmpeg(
|
||||
[
|
||||
"-r",
|
||||
str(fps),
|
||||
"-i",
|
||||
os.path.join(temp_directory_path, "%04d.png"),
|
||||
"-c:v",
|
||||
modules.globals.video_encoder,
|
||||
"-crf",
|
||||
str(modules.globals.video_quality),
|
||||
"-pix_fmt",
|
||||
"yuv420p",
|
||||
"-vf",
|
||||
"colorspace=bt709:iall=bt601-6-625:fast=1",
|
||||
"-y",
|
||||
temp_output_path,
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def restore_audio(target_path: str, output_path: str) -> None:
|
||||
temp_output_path = get_temp_output_path(target_path)
|
||||
done = run_ffmpeg(
|
||||
[
|
||||
"-i",
|
||||
temp_output_path,
|
||||
"-i",
|
||||
target_path,
|
||||
"-c:v",
|
||||
"copy",
|
||||
"-map",
|
||||
"0:v:0",
|
||||
"-map",
|
||||
"1:a:0",
|
||||
"-y",
|
||||
output_path,
|
||||
]
|
||||
)
|
||||
if not done:
|
||||
move_temp(target_path, output_path)
|
||||
|
||||
|
||||
def get_temp_frame_paths(target_path: str) -> List[str]:
|
||||
temp_directory_path = get_temp_directory_path(target_path)
|
||||
return glob.glob((os.path.join(glob.escape(temp_directory_path), "*.png")))
|
||||
|
||||
|
||||
def get_temp_directory_path(target_path: str) -> str:
|
||||
target_name, _ = os.path.splitext(os.path.basename(target_path))
|
||||
target_directory_path = os.path.dirname(target_path)
|
||||
return os.path.join(target_directory_path, TEMP_DIRECTORY, target_name)
|
||||
|
||||
|
||||
def get_temp_output_path(target_path: str) -> str:
|
||||
temp_directory_path = get_temp_directory_path(target_path)
|
||||
return os.path.join(temp_directory_path, TEMP_FILE)
|
||||
|
||||
|
||||
def normalize_output_path(source_path: str, target_path: str, output_path: str) -> Any:
|
||||
if source_path and target_path:
|
||||
source_name, _ = os.path.splitext(os.path.basename(source_path))
|
||||
target_name, target_extension = os.path.splitext(os.path.basename(target_path))
|
||||
if os.path.isdir(output_path):
|
||||
return os.path.join(
|
||||
output_path, source_name + "-" + target_name + target_extension
|
||||
)
|
||||
return output_path
|
||||
|
||||
|
||||
def create_temp(target_path: str) -> None:
|
||||
temp_directory_path = get_temp_directory_path(target_path)
|
||||
Path(temp_directory_path).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
def move_temp(target_path: str, output_path: str) -> None:
|
||||
temp_output_path = get_temp_output_path(target_path)
|
||||
if os.path.isfile(temp_output_path):
|
||||
if os.path.isfile(output_path):
|
||||
os.remove(output_path)
|
||||
shutil.move(temp_output_path, output_path)
|
||||
|
||||
|
||||
def clean_temp(target_path: str) -> None:
|
||||
temp_directory_path = get_temp_directory_path(target_path)
|
||||
parent_directory_path = os.path.dirname(temp_directory_path)
|
||||
if not modules.globals.keep_frames and os.path.isdir(temp_directory_path):
|
||||
shutil.rmtree(temp_directory_path)
|
||||
if os.path.exists(parent_directory_path) and not os.listdir(parent_directory_path):
|
||||
os.rmdir(parent_directory_path)
|
||||
|
||||
|
||||
def has_image_extension(image_path: str) -> bool:
|
||||
return image_path.lower().endswith(("png", "jpg", "jpeg"))
|
||||
|
||||
|
||||
def is_image(image_path: str) -> bool:
|
||||
if image_path and os.path.isfile(image_path):
|
||||
mimetype, _ = mimetypes.guess_type(image_path)
|
||||
return bool(mimetype and mimetype.startswith("image/"))
|
||||
return False
|
||||
|
||||
|
||||
def is_video(video_path: str) -> bool:
|
||||
if video_path and os.path.isfile(video_path):
|
||||
mimetype, _ = mimetypes.guess_type(video_path)
|
||||
return bool(mimetype and mimetype.startswith("video/"))
|
||||
return False
|
||||
|
||||
|
||||
def conditional_download(download_directory_path: str, urls: List[str]) -> None:
|
||||
if not os.path.exists(download_directory_path):
|
||||
os.makedirs(download_directory_path)
|
||||
for url in urls:
|
||||
download_file_path = os.path.join(
|
||||
download_directory_path, os.path.basename(url)
|
||||
)
|
||||
if not os.path.exists(download_file_path):
|
||||
request = urllib.request.urlopen(url) # type: ignore[attr-defined]
|
||||
total = int(request.headers.get("Content-Length", 0))
|
||||
with tqdm(
|
||||
total=total,
|
||||
desc="Downloading",
|
||||
unit="B",
|
||||
unit_scale=True,
|
||||
unit_divisor=1024,
|
||||
) as progress:
|
||||
urllib.request.urlretrieve(url, download_file_path, reporthook=lambda count, block_size, total_size: progress.update(block_size)) # type: ignore[attr-defined]
|
||||
|
||||
|
||||
def resolve_relative_path(path: str) -> str:
|
||||
return os.path.abspath(os.path.join(os.path.dirname(__file__), path))
|
||||
import tkinter as tk
|
||||
root = tk.Tk()
|
||||
root.withdraw()
|
||||
return root
|
||||
except ImportError:
|
||||
print("Tkinter not available - falling back to console mode")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"GUI initialization error: {e}")
|
||||
return None
|
||||
Loading…
Reference in New Issue
Block a user