USER
Do i need the process folder function or could it be combined with another function? Code: import ctypes
import logging
import os
import shutil
import subprocess
import threading
import time
import tkinter as tk
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
from pathlib import Path
from queue import Queue
from tkinter import filedialog
from tkinter import ttk
from typing import Any, Callable, Dict, Iterable, List, Set, Tuple, Optional
from dataclasses import dataclass, field
import tempfile
import numpy as np
from PIL import Image, ImageOps
# --- Configuration ---
@dataclass
class SharedState:
"""Represents the shared state across the application."""
processed_files_lock: threading.Lock = field(default_factory=threading.Lock)
processed_files_count: int = 0
total_files_count: int = 0
log_message_queue: Queue = field(default_factory=Queue)
tab_messages: Dict[str, List[str]] = field(default_factory=lambda: {
"Log": [],
"Black": [],
"White": [],
"Color": []
})
stop_event: threading.Event = field(default_factory=threading.Event)
class Config:
"""Configuration settings for the image processing application."""
SCRIPT_DIR: Path = Path(__file__).parent.resolve()
INPUT_FOLDER: Path = Path(os.getenv('INPUT_FOLDER', SCRIPT_DIR / '1. Input'))
OUTPUT_FOLDER: Path = Path(os.getenv('OUTPUT_FOLDER', SCRIPT_DIR / '2. Output'))
DEFAULT_INPUT_FOLDER: Path = INPUT_FOLDER
DEFAULT_OUTPUT_FOLDER: Path = OUTPUT_FOLDER
# Executable Paths
IMAGEMAGICK: Path = Path(os.getenv('IMAGEMAGICK', 'D:/Bilder/Python/0. Code/Requirements/ImageMagick.exe'))
FFMPEG: Path = Path(os.getenv('FFMPEG', 'D:/Bilder/Python/0. Code/Requirements/ffmpeg.exe'))
# Supported File Extensions
AUDIO_EXTENSIONS: Set[str] = {'.mka', '.mpga', '.mp3', '.aac', '.flac', '.opus'}
IMAGE_EXTENSIONS: Set[str] = {'.png', '.jpg', '.jpeg', '.webp'}
VIDEO_EXTENSIONS: Set[str] = {'.mkv', '.mp4'}
# Display and Formatting Constants
SEPARATOR_LENGTH: int = 310
PADDING_LENGTH: int = 82
PROGRESS_BAR_LENGTH: int = 100
# Default Resize Options
DEFAULT_RESIZE_MODE: str = "Shortest Side"
DEFAULT_RESIZE_WIDTH: int = 0
DEFAULT_RESIZE_HEIGHT: int = 0
DEFAULT_RESIZE_SIDE: int = 1600
DEFAULT_RESIZE_FORMAT: str = "png"
# Color Detection Thresholds
COLORDIFF_THRESHOLD: int = 5
FRACTION_COLORED_PIXELS_THRESHOLD: float = 0.03
@classmethod
def setup(cls) -> None:
"""Create output directory and configure logging settings."""
cls.OUTPUT_FOLDER.mkdir(parents=True, exist_ok=True)
logging.getLogger("PIL").setLevel(logging.WARNING)
@classmethod
def initialize(cls) -> SharedState:
"""Initialize configuration and return shared state."""
cls.setup()
return SharedState()
@contextmanager
def create_temp_folder() -> Path:
"""Yield a temporary folder within the output directory."""
with tempfile.TemporaryDirectory(dir=Config.OUTPUT_FOLDER) as temp_dir:
yield Path(temp_dir)
# --- Logging ---
def log_errors(error_queue):
"""Log errors from the error queue."""
error_dict = defaultdict(list)
while not error_queue.empty():
file_path, error_messages = error_queue.get()
error_dict[file_path].extend(error_messages)
if error_dict:
shared_state.log_message_queue.put("UPDATE_LOG")
for file_path in sorted(error_dict.keys()):
shared_state.log_message_queue.put(f"{file_path.resolve()}")
shared_state.log_message_queue.put("")
for error_message in sorted(error_dict[file_path]):
log_error(error_message)
shared_state.log_message_queue.put("")
def log_and_execute(choice: str, label: str, function: callable, *args):
"""Log the execution of a function and run it in a separate thread."""
def run_in_thread():
start_time = time.time()
separator = '-' * Config.SEPARATOR_LENGTH
shared_state.log_message_queue.put(f"{separator}\n{label}\n{separator}\n")
try:
total_files = function(*args)
except FileNotFoundError as e:
log_error(str(e))
total_files = 0
except (OSError, ValueError) as e:
log_error(f"An error occurred during '{label}': {e}")
total_files = 0
end_time = time.time()
execution_time = end_time - start_time
shared_state.log_message_queue.put("")
image_processor.write_results_to_files(Config.OUTPUT_FOLDER, choice)
app.after(0, app.update_final_status, execution_time)
app.after(0, app.update_text_area)
thread = threading.Thread(target=run_in_thread)
thread.start()
# --- Utility Functions ---
def log_error(message: str) -> None:
"""Log an error message to the shared state queue."""
shared_state.log_message_queue.put(f"- {message}")
def clear_console() -> None:
"""Clear the console screen."""
os.system('cls' if os.name == 'nt' else 'clear')
def count_image_files(folder: Path) -> int:
"""Count the number of image files in a folder and its subfolders."""
return sum(1 for path in folder.rglob('*') if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS)
def validate_and_load_image(path: Path) -> Optional[Image.Image]:
"""Validate and load image, return None if invalid."""
try:
with Image.open(path) as img:
match img.mode:
case 'P':
return img.convert('L')
case 'L' | 'RGB' | 'RGBA':
return img.copy()
case _:
return img.convert('RGB')
except (OSError, ValueError) as e:
log_error(f"Error reading image {path}: {e}")
return None
def convert_png_to_jpg(input_path: Path, output_path: Path) -> None:
"""Convert a PNG image to JPG format."""
try:
with Image.open(input_path) as img:
if img.mode not in ('RGB', 'L'):
img = img.convert('RGB')
jpg_path = output_path.with_suffix('.jpg')
img.save(jpg_path, format='JPEG', quality=80, subsampling=0)
input_path.unlink()
except Exception as e:
log_error(f"Error converting {input_path} to JPG: {e}")
def check_executable(executable: Path, name: str) -> None:
"""Check if a given executable is available in the system."""
if not shutil.which(str(executable)):
raise FileNotFoundError(f"- {name} executable not found at: {executable}")
def increment_processed_count(start_time: float, total_files: int) -> None:
"""Safely increment the processed files count and update the status."""
with shared_state.processed_files_lock:
shared_state.processed_files_count += 1
update_status(start_time, shared_state.processed_files_count, total_files)
# --- Process Bar and Multithreading ---
def update_status(start_time: float, processed_count: int, total_files: int) -> None:
"""Update the application's status text with processing details."""
elapsed_time = time.time() - start_time
minutes, seconds = divmod(int(elapsed_time), 60)
elapsed_time_str = f"{minutes:02d}:{seconds:02d}"
start_time_str = time.strftime('%H:%M:%S', time.localtime(start_time))
status_text = (
f"Start Time: {start_time_str} - "
f"Processed Files: {processed_count}/{total_files} - "
f"Elapsed Time: {elapsed_time_str} Minutes"
)
padded_status_text = status_text.ljust(Config.PADDING_LENGTH)
app.update_status_text(padded_status_text)
def update_elapsed_time(start_time: float, total_files: int, stop_event: threading.Event) -> None:
"""Continuously update the elapsed time until the stop event is set."""
while not stop_event.is_set():
update_status(start_time, shared_state.processed_files_count, total_files)
time.sleep(1)
def start_elapsed_time_thread(start_time: float, total_files: int, stop_event: threading.Event) -> threading.Thread:
"""Start a thread to update the elapsed time."""
thread = threading.Thread(
target=update_elapsed_time,
args=(start_time, total_files, stop_event),
daemon=True
)
thread.start()
return thread
def run_with_thread_pool(tasks: Iterable[Any], process_func: Callable[..., Any], *args) -> int:
"""Execute tasks using a thread pool, updating progress and handling errors."""
shared_state.processed_files_count = 0
shared_state.total_files_count = total = len(tasks)
start_time = time.time()
app.update_progress(0)
stop_event = threading.Event()
elapsed_time_thread = start_elapsed_time_thread(start_time, total, stop_event)
try:
with ThreadPoolExecutor() as executor:
futures: list[Future] = [
executor.submit(process_func, task, *args) for task in tasks
]
for index, future in enumerate(futures, start=1):
if shared_state.stop_event.is_set():
for f in futures:
f.cancel()
break
try:
future.result()
except Exception as e:
log_error(f"Error during processing: {e}")
finally:
progress = (index / total) * 100
app.update_progress(progress)
finally:
stop_event.set()
elapsed_time_thread.join()
return total
# --- Color Detection And Adjustment ---
def is_color_image(
img: np.ndarray,
diff_threshold: int = Config.COLORDIFF_THRESHOLD,
fraction_threshold: float = Config.FRACTION_COLORED_PIXELS_THRESHOLD
) -> bool:
"""Determine if an image is color based on RGB channel differences."""
# Check if the image has three channels (RGB)
if img.ndim != 3 or img.shape[2] != 3:
return False
# Calculate the peak-to-peak (max - min) difference across the RGB channels for each pixel
max_diff = np.ptp(img, axis=2)
# Compute the fraction of pixels where the max difference exceeds the threshold
fraction_colored = np.mean(max_diff > diff_threshold)
# Determine if the image meets or exceeds the required fraction of colored pixels
return fraction_colored >= fraction_threshold
def adjust_values(img: np.ndarray) -> Tuple[np.ndarray, int, int]:
"""Adjust the values of a grayscale image."""
try:
# Calculate modal white value in the range [250, 256)
hist_white, bins_white = np.histogram(img[img >= 250], bins=6, range=(250, 256))
modal_white = int(bins_white[np.argmax(hist_white)])
# Calculate modal black value in the range [0, 61)
hist_black, bins_black = np.histogram(img[img < 61], bins=61, range=(0, 61))
modal_black = int(bins_black[np.argmax(hist_black)])
# Convert image to float for processing
adjusted = img.astype(np.float64)
# Scale image if modal white is not at maximum intensity
if modal_white != 255:
adjusted *= 255.0 / modal_white
adjusted = np.clip(adjusted, 0, 255)
# Adjust image based on modal black value
if modal_black:
adjusted = (adjusted - modal_black) * (255.0 / (255 - modal_black))
adjusted = np.clip(adjusted, 0, 255)
gamma = 1.0 - (modal_black / 255.0)
adjusted = np.power(adjusted / 255.0, gamma) * 255
# Preserve original extreme values
adjusted[img == 0] = 0
adjusted[img == 255] = 255
# Convert back to unsigned 8-bit integer
adjusted = adjusted.astype(np.uint8)
return adjusted, modal_black, modal_white
except ValueError as e:
log_error(f"Error adjusting values for image: {e}")
return img, 0, 255
# --- Image Analysis ---
def analyze_images(image_processor: 'ImageProcessor') -> None:
"""Analyze images in the input folder for color properties."""
image_processor.reset_results()
image_files = [
path for path in Config.INPUT_FOLDER.rglob('*')
if path.is_file() and path.suffix.lower() in Config.IMAGE_EXTENSIONS
]
if not image_files:
log_error(f"No image files found in '{Config.INPUT_FOLDER}'")
return
with create_temp_folder() as temp_folder:
start_time = time.time()
run_with_thread_pool(image_files, analyze_file, start_time, len(image_files))
def analyze_file(path: Path, start_time: float, total: int) -> None:
"""Analyze an image to classify it as color or grayscale."""
img = validate_and_load_image(path)
if img is None:
return
try:
img_array = np.array(img)
subfolder = path.parent.relative_to(Config.INPUT_FOLDER).as_posix()
filename = path.name
if is_color_image(img_array):
image_processor.color_results[subfolder].append(filename)
else:
adjusted, modal_black, modal_white = adjust_values(img_array)
if modal_black:
image_processor.black_results[subfolder].append((filename, modal_black))
if modal_white != 255:
image_processor.white_results[subfolder].append((filename, modal_white))
except (OSError, ValueError) as e:
log_error(f"Error processing image {path}: {e}")
finally:
increment_processed_count(start_time, total)
def find_color_images(image_processor: "ImageProcessor") -> None:
"""Find and move color images to the output folder."""
image_processor.reset_results()
image_extensions: set[str] = {ext.lower() for ext in Config.IMAGE_EXTENSIONS}
tasks: list[Path]
if not (tasks := [
path
for path in Config.INPUT_FOLDER.rglob("*")
if path.is_file() and path.suffix.lower() in image_extensions
]):
log_error(f"No image files found in '{Config.INPUT_FOLDER}'")
return
with create_temp_folder():
run_with_thread_pool(
tasks, move_color_image_file, time.time(), len(tasks), image_processor
)
def move_color_image_file(path: Path, start_time: float, total_files: int, image_processor: 'ImageProcessor') -> None:
"""Move a color image file to the output folder and update processing status."""
img = validate_and_load_image(path)
if img is None:
with shared_state.processed_files_lock:
shared_state.processed_files_count += 1
update_status(start_time, shared_state.processed_files_count, total_files)
return
try:
img_array = np.array(img)
if is_color_image(img_array):
rel_path = path.relative_to(Config.INPUT_FOLDER)
out_path = Config.OUTPUT_FOLDER / rel_path
out_path.parent.mkdir(parents=True, exist_ok=True)
shutil.move(str(path), str(out_path))
subfolder = str(path.parent.relative_to(Config.INPUT_FOLDER))
image_processor.color_results[subfolder].append(path.name)
except (OSError, ValueError) as e:
log_error(f"Error processing image {path}: {e}")
finally:
increment_processed_count(start_time, total_files)
# --- Corrupted Files ---
def check_for_corrupted_files() -> None:
"""Check for corrupted files in the input folder using FFmpeg."""
image_processor.reset_results()
try:
check_executable(Config.FFMPEG, "FFmpeg")
except FileNotFoundError as error:
log_error(str(error))
return
valid_extensions: set[str] = (
Config.VIDEO_EXTENSIONS
| Config.AUDIO_EXTENSIONS
| Config.IMAGE_EXTENSIONS
)
tasks: List[Path] = [
file
for file in Config.INPUT_FOLDER.rglob("*")
if file.is_file() and file.suffix.lower() in valid_extensions
]
if not tasks:
log_error(f"No valid media files found in '{Config.INPUT_FOLDER}'")
return
error_queue: Queue[Tuple[Path, List[str]]] = Queue()
start_time: float = time.time()
total_files: int = len(tasks)
def check_file(file_path: Path) -> None:
try:
base_command: List[str] = [
str(Config.FFMPEG),
"-v", "error",
"-i", str(file_path)
]
if file_path.suffix.lower() in Config.VIDEO_EXTENSIONS:
base_command += ["-map", "0:a:0"]
base_command += ["-f", "null", "-"]
result: subprocess.CompletedProcess = subprocess.run(
base_command,
capture_output=True,
text=True,
encoding="utf-8",
errors="ignore"
)
combined_output: str = f"{result.stdout}\n{result.stderr}"
error_messages: List[str] = [
line.strip()
for line in combined_output.splitlines()
if "error" in line.lower()
]
if error_messages:
error_queue.put((file_path.resolve(), error_messages))
except Exception as e:
error_queue.put((file_path.resolve(), [f"Error processing file {file_path}: {e}"]))
finally:
increment_processed_count(start_time, total_files)
run_with_thread_pool(tasks, check_file)
log_errors(error_queue)
# --- Resize Images ---
def resize_images(input_folder: Path, output_folder: Path, image_processor: 'ImageProcessor'):
"""Resize images in the input folder and save them to the output folder."""
try:
check_executable(Config.IMAGEMAGICK, "ImageMagick")
except FileNotFoundError as e:
log_error(str(e))
return
tasks = [input_path for input_path in input_folder.rglob('*')
if input_path.is_file() and input_path.suffix.lower() in Config.IMAGE_EXTENSIONS]
if not tasks:
log_error(f"No image files found in '{input_folder}'")
return
with create_temp_folder() as temp_folder:
try:
run_with_thread_pool(tasks, process_file, time.time(), len(tasks), image_processor, temp_folder, True)
finally:
pass
def resize_image(input_path: Path, output_path: Path) -> Path:
"""Resize an image using ImageMagick."""
output_path = output_path.with_suffix('.png')
identify_command = [
str(Config.IMAGEMAGICK),
str(input_path),
'-format', '%wx%h',
'info:'
]
try:
result = subprocess.run(
identify_command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=True
)
current_width, current_height = map(int, result.stdout.strip().split('x'))
except subprocess.CalledProcessError as e:
log_error(f"ImageMagick identify failed for {input_path}: {e.stderr.strip()}")
return output_path
resize_mode = app.RESIZE_MODE.get()
maintain_aspect = app.maintain_aspect_ratio.get()
if resize_mode == "Fit":
target_width = int(app.RESIZE_WIDTH.get() or 0)
target_height = int(app.RESIZE_HEIGHT.get() or 0)
if maintain_aspect:
if target_width == 0 and target_height > 0:
target_width = int(target_height * current_width / current_height)
elif target_height == 0 and target_width > 0:
target_height = int(target_width * current_height / current_width)
resize_arg = f'{target_width}x{target_height}!'
else: # Shortest Side
target_size = int(app.RESIZE_SIDE.get())
resize_arg = f'{target_size}x{target_size}^'
is_upscaling = (
(resize_mode == "Fit" and (target_width > current_width or target_height > current_height)) or
(resize_mode == "Shortest Side" and target_size > min(current_width, current_height))
)
command = [
str(Config.IMAGEMAGICK),
str(input_path),
]
if is_upscaling:
command += [
'-filter', 'LanczosSharp',
'-distort', 'Resize', resize_arg,
]
else:
command += [
'-colorspace', 'RGB',
'-filter', 'Lanczos2Sharp',
'-resize', resize_arg,
'-colorspace', 'sRGB',
]
command += [
'-define', 'png:compression-level=5',
str(output_path)
]
try:
result = subprocess.run(
command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=True
)
if result.stderr:
log_error(f"ImageMagick: {result.stderr.strip()}")
except subprocess.CalledProcessError as e:
log_error(f"ImageMagick resize failed for {input_path}: {e.stderr.strip()}")
return output_path
# --- Processing ---
def process_file(
path: Path,
start_time,
total_files,
image_processor: 'ImageProcessor',
temp_folder: Path,
resize=False,
auto_correct_and_resize=False
) -> None:
"""Process a single image file."""
if shared_state.stop_event.is_set():
return
img = validate_and_load_image(path)
if img is None:
log_error(f"Invalid image file: {path}")
with shared_state.processed_files_lock:
shared_state.processed_files_count += 1
update_status(start_time, shared_state.processed_files_count, total_files)
return
try:
img_array = np.array(img)
except (OSError, ValueError) as e:
log_error(f"Error reading image {path}: {e}")
with shared_state.processed_files_lock:
shared_state.processed_files_count += 1
update_status(start_time, shared_state.processed_files_count, total_files)
return
relative_path = path.relative_to(Config.INPUT_FOLDER)
temp_output_path = temp_folder / relative_path.with_suffix('.png')
temp_output_path.parent.mkdir(parents=True, exist_ok=True)
if auto_correct_and_resize:
# Adjust and save image, then resize
intermediate_path, _, _ = adjust_and_save_image(img_array, path, image_processor, temp_folder)
temp_output_path = resize_image(intermediate_path, temp_output_path)
elif resize:
# Only resize the original image
temp_output_path = resize_image(path, temp_output_path)
else:
# Adjust and save image without resizing
temp_output_path, _, _ = adjust_and_save_image(img_array, path, image_processor, temp_folder)
final_output_path = Config.OUTPUT_FOLDER / relative_path.with_suffix(
'.jpg' if app.RESIZE_FORMAT.get() == 'jpg' else '.png'
)
final_output_path.parent.mkdir(parents=True, exist_ok=True)
if app.RESIZE_FORMAT.get() == 'jpg':
convert_png_to_jpg(temp_output_path, final_output_path)
else:
shutil.move(str(temp_output_path), str(final_output_path))
increment_processed_count(start_time, total_files)
def process_folder(folder: Path, image_processor: 'ImageProcessor', auto_correct_and_resize=False) -> None:
"""Process all images in a folder."""
image_processor.reset_results()
if auto_correct_and_resize:
try:
check_executable(Config.IMAGEMAGICK, "ImageMagick")
except FileNotFoundError as e:
log_error(str(e))
return
tasks = [path for path in folder.rglob('*')
if path.is_file() and path.suffix.lower() in Config.IMAGE_EXTENSIONS]
if not tasks:
log_error(f"No image files found in '{folder}'")
return
with create_temp_folder() as temp_folder:
try:
run_with_thread_pool(tasks, process_file, time.time(), len(tasks), image_processor, temp_folder, False, auto_correct_and_resize)
finally:
pass
def adjust_and_save_image(
img: np.ndarray,
path: Path,
image_processor: 'ImageProcessor',
temp_folder: Path,
diff_threshold: int = Config.COLORDIFF_THRESHOLD,
fraction_threshold: float = Config.FRACTION_COLORED_PIXELS_THRESHOLD
) -> Tuple[Path, Optional[int], Optional[int]]:
"""Process an image, adjusting and saving depending on whether it is color or grayscale."""
try:
is_color = is_color_image(img, diff_threshold, fraction_threshold)
relative_parent = path.parent.relative_to(Config.INPUT_FOLDER)
out_path = temp_folder / relative_parent / f"{path.stem}.png"
out_path.parent.mkdir(parents=True, exist_ok=True)
subfolder = relative_parent.as_posix()
filename = path.name
if is_color:
# Save color image as PNG
Image.fromarray(img).save(out_path, format='PNG', compress_level=5)
image_processor.color_results[subfolder].append(filename)
return out_path, None, None
else:
# Adjust grayscale values
adjusted_img, modal_black, modal_white = adjust_values(img)
Image.fromarray(adjusted_img).convert('L').save(out_path, format='PNG', compress_level=5)
if modal_black:
image_processor.black_results[subfolder].append((filename, modal_black))
if modal_white != 255:
image_processor.white_results[subfolder].append((filename, modal_white))
return out_path, modal_black, modal_white
except (OSError, ValueError) as e:
log_error(f"Error processing image {path}: {e}")
return path, None, None
# --- ImageProcessor Class ---
class ImageProcessor:
"""Class to handle image processing results."""
def __init__(self):
self.white_results = defaultdict(list)
self.black_results = defaultdict(list)
self.color_results = defaultdict(list)
def write_results_to_files(self, output_folder: Path, choice: str):
"""Write processing results to files."""
results_map = {
'2': ("2. Analyze Images", "Modal Value Black", "Modal Value White"),
'3': ("3. Find Color Images", "Filename"),
'5': ("5. Automatic Color Correction", "Modal Value Black", "Modal Value White"),
'6': ("6. Automatic Color Correction + Resize", "Modal Value Black", "Modal Value White")
}
headers = results_map.get(choice)
if headers:
if choice == '3':
self._write_results("Color", self.color_results, headers[0], headers[1])
else:
self._write_results("Black", self.black_results, headers[0], headers[1])
self._write_results("White", self.white_results, headers[0], headers[2])
self._write_results("Color", self.color_results, headers[0], "Filename")
def _write_results(self, file_path: str, results: Dict[str, List[Tuple[str, int]]], header: str,
value_name: str):
"""Write results to a specific file."""
separator = '-' * Config.SEPARATOR_LENGTH
content = f"{separator}\n{header}\n{separator}\n"
if results:
for subfolder, subfolder_results in sorted(results.items()):
full_subfolder_path = Config.INPUT_FOLDER / subfolder
content += f"\n{full_subfolder_path}\n\n"
if value_name == "Filename":
subfolder_results.sort()
content += '\n'.join(f" - {filename}" for filename in subfolder_results)
else:
subfolder_results.sort(key=lambda x: x[1], reverse=True)
content += '\n'.join(
f" - {filename}, {value_name}: {value}" for filename, value in subfolder_results)
content += "\n"
shared_state.tab_messages[file_path].append(content)
def reset_results(self):
"""Reset all results."""
self.white_results = defaultdict(list)
self.black_results = defaultdict(list)
self.color_results = defaultdict(list)
# --- Tkinter App ---
class App(tk.Tk):
def __init__(self):
super().__init__()
self.setup_window()
self.create_variables()
self.create_widgets()
self.setup_logging()
self.process_start_time = 0
def setup_window(self):
self.set_dpi_awareness()
self.title("Image Processor")
self.geometry("2100x1180")
self.default_font = ("Roboto Flex", 16)
self.apply_default_font()
def set_dpi_awareness(self):
try:
ctypes.windll.shcore.SetProcessDpiAwareness(1)
except Exception as e:
print(f"Could not set DPI awareness: {e}")
def apply_default_font(self):
self.option_add("*Font", self.default_font)
self.style = ttk.Style()
for widget_type in ["TButton", "TLabel", "TEntry", "TRadiobutton", "TNotebook.Tab"]:
self.style.configure(widget_type, font=self.default_font)
self.style.configure("TFrame", background="#fdfdfd")
self.style.configure("TLabel", background="#fdfdfd")
self.style.configure("TNotebook", background="#f8f8f8")
self.style.configure("TRadiobutton", background="#fdfdfd")
def create_variables(self):
self.RESIZE_MODE = tk.StringVar(value=Config.DEFAULT_RESIZE_MODE)
self.RESIZE_WIDTH = tk.StringVar(value=str(Config.DEFAULT_RESIZE_WIDTH))
self.RESIZE_HEIGHT = tk.StringVar(value=str(Config.DEFAULT_RESIZE_HEIGHT))
self.RESIZE_SIDE = tk.StringVar(value=str(Config.DEFAULT_RESIZE_SIDE))
self.RESIZE_FORMAT = tk.StringVar(value=Config.DEFAULT_RESIZE_FORMAT)
self.option_var = tk.StringVar(value="4")
self.status_text = tk.StringVar(value="")
def create_widgets(self):
main_frame = ttk.Frame(self, padding="20")
main_frame.pack(fill=tk.BOTH, expand=True)
self.create_folder_selection_frames(main_frame)
self.create_resize_options_frame(main_frame)
self.create_options_and_start_frame(main_frame)
self.create_text_frame(main_frame)
self.create_progress_bar(main_frame)
self.create_status_label(main_frame)
def create_folder_selection_frames(self, parent):
self.create_folder_frame(parent, "Input", Config.INPUT_FOLDER, self.set_default_input, self.select_input_folder)
self.create_folder_frame(parent, "Output", Config.OUTPUT_FOLDER, self.set_default_output, self.select_output_folder)
def create_folder_frame(self, parent, label, default_folder, default_command, select_command):
frame = ttk.Frame(parent)
frame.pack(fill=tk.X, pady=(0, 10))
ttk.Button(frame, text="Default", command=default_command, width=10).pack(side=tk.LEFT, padx=(0, 10))
entry = ttk.Entry(frame)
entry.insert(0, str(default_folder))
entry.pack(side=tk.LEFT, expand=True, fill=tk.X)
ttk.Button(frame, text=f"Select {label} Folder", command=select_command, width=20).pack(side=tk.LEFT, padx=(10, 0))
setattr(self, f"{label.lower()}_entry", entry)
def create_resize_options_frame(self, parent):
resize_frame = ttk.Frame(parent)
resize_frame.pack(fill=tk.X, pady=(0, 10))
ttk.Button(resize_frame, text="Default", command=self.reset_resize_options, width=10).pack(side=tk.LEFT, padx=(0, 10))
self.create_resize_options(resize_frame)
self.start_button = ttk.Button(resize_frame, text="Start", command=self.start_processing, width=20)
self.start_button.pack(side=tk.RIGHT, padx=(10, 0))
def create_resize_options(self, frame):
ttk.Label(frame, text="Format:").pack(side=tk.LEFT, padx=(0, 5))
ttk.Combobox(frame, textvariable=self.RESIZE_FORMAT, values=["png", "jpg"], state="readonly", width=4).pack(side=tk.LEFT, padx=(0, 10))
ttk.Label(frame, text="Resize:").pack(side=tk.LEFT, padx=(0, 5))
mode_combo = ttk.Combobox(frame, textvariable=self.RESIZE_MODE, values=["Shortest Side", "Fit"], state="readonly", width=12)
mode_combo.pack(side=tk.LEFT, padx=(0, 10))
mode_combo.bind("<<ComboboxSelected>>", self.update_resize_options)
self.width_label = ttk.Label(frame, text="Width:")
self.width_entry = ttk.Entry(frame, textvariable=self.RESIZE_WIDTH, width=5)
self.height_label = ttk.Label(frame, text="Height:")
self.height_entry = ttk.Entry(frame, textvariable=self.RESIZE_HEIGHT, width=5)
self.maintain_aspect_ratio = tk.BooleanVar(value=False)
self.maintain_aspect_ratio_label = ttk.Label(frame, text="Maintain Aspect Ratio", font=self.default_font)
self.maintain_aspect_ratio_checkbox = ttk.Checkbutton(frame, variable=self.maintain_aspect_ratio)
self.side_label = ttk.Label(frame, text="Side:")
self.side_entry = ttk.Entry(frame, textvariable=self.RESIZE_SIDE, width=5)
self.update_resize_options()
def create_options_and_start_frame(self, parent):
frame = ttk.Frame(parent)
frame.pack(fill=tk.X, pady=(0, 10))
self.create_options_frame(frame)
self.create_stop_button(frame)
def create_options_frame(self, parent):
options_frame = ttk.Frame(parent)
options_frame.pack(side=tk.LEFT, fill=tk.X, expand=True)
options = [
("Check For Corrupted Files", "4"),
("Analyze Images", "2"),
("Find Color Images", "3"),
("Resize Images", "1"),
("Automatic Color Correction", "5"),
("Automatic Color Correction + Resize", "6")
]
for col, (text, value) in enumerate(options):
ttk.Radiobutton(options_frame, text=text, variable=self.option_var, value=value).grid(row=0, column=col, sticky="ew")
options_frame.columnconfigure(col, weight=1)
def create_stop_button(self, parent):
self.stop_button = ttk.Button(parent, text="Stop", command=self.stop_processing, width=20, state=tk.DISABLED)
self.stop_button.pack(side=tk.RIGHT, padx=(10, 0))
def create_text_frame(self, parent):
text_frame = ttk.Frame(parent)
text_frame.pack(fill=tk.BOTH, expand=True)
self.notebook = ttk.Notebook(text_frame)
self.notebook.pack(fill=tk.BOTH, expand=True)
self.text_areas = {}
for tab in ["Log", "Black", "White", "Color"]:
frame = ttk.Frame(self.notebook)
self.notebook.add(frame, text=tab)
self.text_areas[tab] = self.create_text_area(frame)
def create_text_area(self, parent):
frame = ttk.Frame(parent)
frame.pack(fill=tk.BOTH, expand=True)
text_area = tk.Text(frame, wrap=tk.NONE, bg="#ffffff", state='disabled', padx=7, pady=7, fg="#000000", font=self.default_font)
scrollbar_y = ttk.Scrollbar(frame, command=text_area.yview)
scrollbar_x = ttk.Scrollbar(frame, command=text_area.xview, orient='horizontal')
scrollbar_y.pack(side=tk.RIGHT, fill=tk.Y)
text_area.pack(side=tk.TOP, fill=tk.BOTH, expand=True)
scrollbar_x.pack(side=tk.BOTTOM, fill=tk.X)
text_area.config(yscrollcommand=scrollbar_y.set, xscrollcommand=scrollbar_x.set)
return text_area
def create_progress_bar(self, parent):
self.progress = ttk.Progressbar(parent, mode="determinate", maximum=Config.PROGRESS_BAR_LENGTH)
self.progress.pack(pady=(30, 0), padx=20, fill="x")
self.progress.config(value=0)
def create_status_label(self, parent):
ttk.Label(parent, textvariable=self.status_text).pack(pady=(10, 15))
def setup_logging(self):
self.queue_thread = threading.Thread(target=self.update_text_area_from_queue, daemon=True)
self.queue_thread.start()
def set_default_folder(self, entry, default_folder):
entry.delete(0, tk.END)
entry.insert(0, str(default_folder))
return Path(default_folder)
def set_default_input(self):
Config.INPUT_FOLDER = self.set_default_folder(self.input_entry, Config.DEFAULT_INPUT_FOLDER)
def set_default_output(self):
Config.OUTPUT_FOLDER = self.set_default_folder(self.output_entry, Config.DEFAULT_OUTPUT_FOLDER)
def select_folder(self, entry: ttk.Entry, config_attr: str) -> None:
selected_folder = filedialog.askdirectory()
if selected_folder:
path = Path(selected_folder).resolve()
entry.delete(0, tk.END)
entry.insert(0, str(path))
setattr(Config, config_attr, path)
def select_input_folder(self):
self.select_folder(self.input_entry, 'INPUT_FOLDER')
def select_output_folder(self):
self.select_folder(self.output_entry, 'OUTPUT_FOLDER')
def update_status_text(self, text):
self.status_text.set(text)
self.update_idletasks()
def update_progress(self, value):
self.progress['value'] = min(value, Config.PROGRESS_BAR_LENGTH)
self.update_idletasks()
def update_text_area(self, content_key=None, message=None):
for tab_key, text_area in self.text_areas.items():
text_area.configure(state='normal')
text_area.delete("1.0", tk.END)
content = "\n".join(shared_state.tab_messages[tab_key])
text_area.insert(tk.END, content)
text_area.configure(state='disabled')
def start_processing(self):
shared_state.stop_event.clear()
self.start_button.config(state=tk.DISABLED)
self.stop_button.config(state=tk.NORMAL)
self.process_start_time = time.time()
if hasattr(self, 'stopped_by_user'):
delattr(self, 'stopped_by_user')
choice = self.option_var.get()
actions = {
"1": ("1. Resize Images", resize_images, Config.INPUT_FOLDER, Config.OUTPUT_FOLDER, image_processor),
"2": ("2. Analyze Images", analyze_images, image_processor),
"3": ("3. Find Color Images", find_color_images, image_processor),
"4": ("4. Check For Corrupted Files", check_for_corrupted_files),
"5": ("5. Automatic Color Correction", process_folder, Config.INPUT_FOLDER, image_processor),
"6": ("6. Automatic Color Correction + Resize", process_folder, Config.INPUT_FOLDER, image_processor, True),
}
label, func, *args = actions.get(choice, ("Unknown Action", lambda: None))
log_and_execute(choice, label, func, *args)
def stop_processing(self):
shared_state.stop_event.set()
self.stopped_by_user = True
self.stop_button.config(state=tk.DISABLED)
self.start_button.config(state=tk.NORMAL)
def update_resize_options(self, event=None):
mode = self.RESIZE_MODE.get()
if mode == "Fit":
self.side_label.pack_forget()
self.side_entry.pack_forget()
self.width_label.pack(side=tk.LEFT, padx=(10, 5))
self.width_entry.pack(side=tk.LEFT, padx=(0, 10))
self.height_label.pack(side=tk.LEFT, padx=(0, 5))
self.height_entry.pack(side=tk.LEFT, padx=(0, 10))
self.maintain_aspect_ratio_label.pack(side=tk.LEFT, padx=(10, 0))
self.maintain_aspect_ratio_checkbox.pack(side=tk.LEFT, padx=(0, 10))
else: # Shortest Side
self.width_label.pack_forget()
self.width_entry.pack_forget()
self.height_label.pack_forget()
self.height_entry.pack_forget()
self.maintain_aspect_ratio_checkbox.pack_forget()
self.maintain_aspect_ratio_label.pack_forget()
self.side_label.pack(side=tk.LEFT, padx=(10, 5))
self.side_entry.pack(side=tk.LEFT, padx=(0, 10))
def reset_resize_options(self):
self.RESIZE_MODE.set(Config.DEFAULT_RESIZE_MODE)
self.RESIZE_WIDTH.set(str(Config.DEFAULT_RESIZE_WIDTH))
self.RESIZE_HEIGHT.set(str(Config.DEFAULT_RESIZE_HEIGHT))
self.RESIZE_SIDE.set(str(Config.DEFAULT_RESIZE_SIDE))
self.RESIZE_FORMAT.set(Config.DEFAULT_RESIZE_FORMAT)
self.update_resize_options()
def update_text_area_from_queue(self):
while True:
if not shared_state.log_message_queue.empty():
try:
item = shared_state.log_message_queue.get(block=True, timeout=0.1)
self.process_log_message(item)
except:
pass
else:
time.sleep(0.1)
def process_log_message(self, item):
update_tabs = {"UPDATE_LOG": "Log", "UPDATE_BLACK": "Black", "UPDATE_WHITE": "White", "UPDATE_COLOR": "Color"}
if item in update_tabs:
self.after(0, self.update_text_area, update_tabs[item])
else:
self.update_tab_messages(item)
def update_tab_messages(self, item):
if "Modal Value Black" in item:
shared_state.tab_messages["Black"].append(item)
if "Modal Value White" in item:
shared_state.tab_messages["White"].append(item)
if "Filename" in item:
shared_state.tab_messages["Color"].append(item)
shared_state.tab_messages["Log"].append(item)
for tab in ["Log", "Black", "White", "Color"]:
self.after(0, self.update_text_area, tab)
def update_final_status(self, execution_time):
self.update_progress(100)
end_time = time.time()
start_time = self.process_start_time
processed_count = shared_state.processed_files_count
total_files = shared_state.total_files_count
start_time_str = time.strftime('%H:%M:%S', time.localtime(start_time))
elapsed_minutes = int(execution_time // 60)
elapsed_seconds = int(execution_time % 60)
elapsed_time_str = f"{elapsed_minutes:02d}:{elapsed_seconds:02d}"
status_text = (
f"Start Time: {start_time_str} - "
f"Processed Files: {processed_count}/{total_files} - "
f"Elapsed Time: {elapsed_time_str} Minutes"
)
if hasattr(self, 'stopped_by_user') and self.stopped_by_user:
status_text += " - Processing stopped by user"
delattr(self, 'stopped_by_user')
padded_status_text = status_text.ljust(Config.PADDING_LENGTH)
self.update_status_text(padded_status_text)
self.reset_gui()
def reset_gui(self):
self.start_button.config(state=tk.NORMAL)
self.stop_button.config(state=tk.DISABLED)
# --- Main Execution ---
if __name__ == "__main__":
shared_state = Config.initialize()
image_processor = ImageProcessor()
app = App()
app.mainloop()
ASSISTANT
Yes, you can combine the `process_folder` function with another function to reduce redundancy and make your code cleaner. Specifically, the `process_folder` function and the `resize_images` function have similar structures and purposes—they both process images in a folder and use the `process_file` function to handle individual files.
By refactoring these functions, you can create a more generic function that handles the processing of images with different options, such as resizing, auto-correction, or both. Below, I will guide you through the steps to combine these functions effectively.
### Understanding the Current Structure
First, let's look at the existing `process_folder` and `resize_images` functions:
**`process_folder` Function:**
```python
def process_folder(
folder: Path,
image_processor: 'ImageProcessor',
auto_correct_and_resize=False
) -> None:
"""Process all images in a folder."""
image_processor.reset_results()
if auto_correct_and_resize:
try:
check_executable(Config.IMAGEMAGICK, "ImageMagick")
except FileNotFoundError as e:
log_error(str(e))
return
tasks = [path for path in folder.rglob('*')
if path.is_file() and path.suffix.lower() in Config.IMAGE_EXTENSIONS]
if not tasks:
log_error(f"No image files found in '{folder}'")
return
with create_temp_folder() as temp_folder:
try:
run_with_thread_pool(
tasks,
process_file,
time.time(),
len(tasks),
image_processor,
temp_folder,
resize=False,
auto_correct_and_resize=auto_correct_and_resize
)
finally:
pass
```
**`resize_images` Function:**
```python
def resize_images(
input_folder: Path,
output_folder: Path,
image_processor: 'ImageProcessor'
):
"""Resize images in the input folder and save them to the output folder."""
try:
check_executable(Config.IMAGEMAGICK, "ImageMagick")
except FileNotFoundError as e:
log_error(str(e))
return
tasks = [input_path for input_path in input_folder.rglob('*')
if input_path.is_file() and input_path.suffix.lower() in Config.IMAGE_EXTENSIONS]
if not tasks:
log_error(f"No image files found in '{input_folder}'")
return
with create_temp_folder() as temp_folder:
try:
run_with_thread_pool(
tasks,
process_file,
time.time(),
len(tasks),
image_processor,
temp_folder,
resize=True
)
finally:
pass
```
As you can see, both functions:
- Collect image file paths from a folder.
- Check for the presence of ImageMagick if needed.
- Utilize `run_with_thread_pool` to process the images using the `process_file` function.
- Differ primarily in the parameters passed to `process_file`.
### Combining the Functions
To combine these functions, you can create a single, more general function called `process_images_in_folder` that accepts parameters to control the behavior (e.g., whether to resize, auto-correct, or both):
```python
def process_images_in_folder(
folder: Path,
image_processor: 'ImageProcessor',
resize=False,
auto_correct=False
) -> None:
"""Process all images in a folder with specified options."""
image_processor.reset_results()
if resize or auto_correct:
try:
check_executable(Config.IMAGEMAGICK, "ImageMagick")
except FileNotFoundError as e:
log_error(str(e))
return
tasks = [path for path in folder.rglob('*')
if path.is_file() and path.suffix.lower() in Config.IMAGE_EXTENSIONS]
if not tasks:
log_error(f"No image files found in '{folder}'")
return
with create_temp_folder() as temp_folder:
try:
run_with_thread_pool(
tasks,
process_file,
time.time(),
len(tasks),
image_processor,
temp_folder,
resize=resize,
auto_correct=auto_correct
)
finally:
pass
```
Now, you can replace both `process_folder` and `resize_images` with this new function.
### Updating the `process_file` Function
Ensure that the `process_file` function can handle both resizing and auto-correction based on the parameters:
```python
def process_file(
path: Path,
start_time,
total_files,
image_processor: 'ImageProcessor',
temp_folder: Path,
resize=False,
auto_correct=False
) -> None:
"""Process a single image file."""
if shared_state.stop_event.is_set():
return
img = validate_and_load_image(path)
if img is None:
log_error(f"Invalid image file: {path}")
increment_processed_count(start_time, total_files)
return
try:
img_array = np.array(img)
except (OSError, ValueError) as e:
log_error(f"Error reading image {path}: {e}")
increment_processed_count(start_time, total_files)
return
relative_path = path.relative_to(Config.INPUT_FOLDER)
temp_output_path = temp_folder / relative_path.with_suffix('.png')
temp_output_path.parent.mkdir(parents=True, exist_ok=True)
if auto_correct and resize:
# Adjust and save image, then resize
intermediate_path, _, _ = adjust_and_save_image(
img_array, path, image_processor, temp_folder
)
temp_output_path = resize_image(intermediate_path, temp_output_path)
elif auto_correct:
# Adjust and save image without resizing
temp_output_path, _, _ = adjust_and_save_image(
img_array, path, image_processor, temp_folder
)
elif resize:
# Only resize the original image
temp_output_path = resize_image(path, temp_output_path)
else:
# Simply copy the image to the output path
shutil.copy(str(path), str(temp_output_path))
final_output_path = Config.OUTPUT_FOLDER / relative_path.with_suffix(
'.jpg' if app.RESIZE_FORMAT.get() == 'jpg' else '.png'
)
final_output_path.parent.mkdir(parents=True, exist_ok=True)
if app.RESIZE_FORMAT.get() == 'jpg':
convert_png_to_jpg(temp_output_path, final_output_path)
else:
shutil.move(str(temp_output_path), str(final_output_path))
increment_processed_count(start_time, total_files)
```
This `process_file` function now checks the combination of `resize` and `auto_correct` flags and processes the image accordingly.
### Adjusting the Application Logic
Update the `start_processing` method in your Tkinter application to use the new function:
```python
def start_processing(self):
shared_state.stop_event.clear()
self.start_button.config(state=tk.DISABLED)
self.stop_button.config(state=tk.NORMAL)
self.process_start_time = time.time()
if hasattr(self, 'stopped_by_user'):
delattr(self, 'stopped_by_user')
choice = self.option_var.get()
actions = {
"1": (
"1. Resize Images",
process_images_in_folder,
Config.INPUT_FOLDER,
image_processor,
True, # resize
False # auto_correct
),
"5": (
"5. Automatic Color Correction",
process_images_in_folder,
Config.INPUT_FOLDER,
image_processor,
False, # resize
True # auto_correct
),
"6": (
"6. Automatic Color Correction + Resize",
process_images_in_folder,
Config.INPUT_FOLDER,
image_processor,
True, # resize
True # auto_correct
),
# ... other actions remain unchanged ...
}
label, func, *args = actions.get(choice, ("Unknown Action", lambda: None))
log_and_execute(choice, label, func, *args)
```
Now, when the user selects an option, the `process_images_in_folder` function is called with the appropriate flags for resizing and auto-correction.
### Benefits of Combining the Functions
Combining `process_folder` and `resize_images` into a single function provides several advantages:
- **Code Reusability:** You avoid duplicating code that collects tasks and manages threading.
- **Maintainability:** Having a single function simplifies updates and bug fixes.
- **Flexibility:** The new function can handle various combinations of processing options.
- **Readability:** Reduces the overall length of your code and makes it easier to understand.
### Conclusion
You can safely combine the `process_folder` function with another function like `resize_images` by creating a generalized function. This approach reduces redundancy and enhances the flexibility of your code. By carefully adjusting the parameters and logic, you ensure that all processing options are still available while maintaining a cleaner codebase.