Iván Eduardo Chavez Ayub
402 lines
16 KiB
Python
402 lines
16 KiB
Python
import logging
|
|
import os
|
|
import queue
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
import customtkinter as ctk
|
|
import cv2
|
|
import numpy as np
|
|
from PIL import Image
|
|
|
|
|
|
# --- CONFIGURACIÓN Y ESTILOS ---
|
|
class Theme:
|
|
BG_COLOR = "#101010" # Fondo general un poco más oscuro
|
|
ITEM_BG = "#3A3A3A" # Fondo de cada tarjeta
|
|
TEXT_MAIN = "#FFFFFF"
|
|
TEXT_SEC = "#FDEF2F"
|
|
ACCENT = "#F7F7F7" # Dorado original
|
|
ERROR = "#CF6679" # Rojo suave para errores
|
|
BTN_HOVER_DANGER = "#B00020"
|
|
BTN_HOVER_NORMAL = "#4A4A4A"
|
|
BORDER_COLOR = "#555555"
|
|
|
|
FONT_MAIN = ("Roboto Medium", 13)
|
|
FONT_SUB = ("Roboto", 11)
|
|
FONT_MONO = ("Consolas", 10)
|
|
|
|
|
|
# Configuración de logging para debug
|
|
logging.basicConfig(level=logging.ERROR,
|
|
format='%(asctime)s - %(levelname)s - %(message)s')
|
|
|
|
|
|
class QueueItem:
|
|
"""Clase de datos para manejar el estado de cada archivo individualmente."""
|
|
|
|
def __init__(self, path: str):
|
|
self.path = path
|
|
self.name = os.path.basename(path)
|
|
self.info_text = "Cargando info..."
|
|
self.thumbnail: Optional[ctk.CTkImage] = None
|
|
self.is_loaded = False
|
|
self.has_error = False
|
|
self.id = id(self) # Identificador único para referenciar widgets
|
|
|
|
|
|
class FileQueueManager(ctk.CTkScrollableFrame):
|
|
def __init__(self, master, clear_icon=None, on_queue_empty_callback=None, **kwargs):
|
|
super().__init__(master, fg_color=Theme.BG_COLOR, **kwargs)
|
|
|
|
self.on_queue_empty_callback = on_queue_empty_callback
|
|
self.clear_icon = clear_icon
|
|
|
|
# Estado interno
|
|
self.queue_items: List[QueueItem] = []
|
|
# Mapa para acceder a widgets por ID de item
|
|
self._widget_refs: Dict[int, dict] = {}
|
|
|
|
# Factores de redimensionado
|
|
self.upscale_factor = 1
|
|
self.input_resize_factor = 0
|
|
self.output_resize_factor = 0
|
|
|
|
# Sistema de hilos para no congelar la GUI
|
|
self.executor = ThreadPoolExecutor(max_workers=2)
|
|
self.msg_queue = queue.Queue()
|
|
|
|
# Grid layout básico
|
|
self.grid_columnconfigure(0, weight=1)
|
|
|
|
# Iniciar loop de verificación de mensajes (para actualizaciones desde hilos)
|
|
self._check_queue()
|
|
|
|
def _check_queue(self):
|
|
"""Revisa la cola de mensajes para actualizar la UI desde el hilo principal."""
|
|
try:
|
|
while True:
|
|
task_type, item_id, data = self.msg_queue.get_nowait()
|
|
if task_type == "UPDATE_ITEM":
|
|
self._update_item_ui(item_id, data)
|
|
except queue.Empty:
|
|
pass
|
|
finally:
|
|
self.after(100, self._check_queue)
|
|
|
|
def add_files(self, file_paths: List[str]):
|
|
"""Agrega archivos y lanza la carga en segundo plano."""
|
|
added_any = False
|
|
existing_paths = {item.path for item in self.queue_items}
|
|
|
|
for path in file_paths:
|
|
norm_path = os.path.normpath(path)
|
|
if norm_path not in existing_paths and os.path.exists(path):
|
|
new_item = QueueItem(norm_path)
|
|
self.queue_items.append(new_item)
|
|
added_any = True
|
|
|
|
# Renderizar placeholder inmediatamente
|
|
self._render_item(new_item, index=len(self.queue_items)-1)
|
|
|
|
# Lanzar tarea pesada en background
|
|
self.executor.submit(self._worker_load_info, new_item)
|
|
|
|
if added_any:
|
|
self._update_layout_indices()
|
|
|
|
def _worker_load_info(self, item: QueueItem):
|
|
"""Método que corre en otro hilo (Thread) para procesar OpenCV."""
|
|
try:
|
|
info_text, pil_img = self._extract_file_data(item.path)
|
|
|
|
# Crear CTkImage debe hacerse preferiblemente al volver al main,
|
|
# pero PIL Image es seguro pasarlo.
|
|
item.info_text = info_text
|
|
item.is_loaded = True
|
|
|
|
# Enviamos resultados al hilo principal
|
|
self.msg_queue.put(
|
|
("UPDATE_ITEM", item.id, {"info": info_text, "img": pil_img}))
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error procesando {item.path}: {e}")
|
|
item.has_error = True
|
|
item.info_text = "Error al leer archivo"
|
|
self.msg_queue.put(("UPDATE_ITEM", item.id, {"error": True}))
|
|
|
|
def _update_item_ui(self, item_id, data):
|
|
"""Actualiza un widget específico una vez que el hilo terminó."""
|
|
widgets = self._widget_refs.get(item_id)
|
|
if not widgets:
|
|
return
|
|
|
|
if "error" in data:
|
|
widgets['lbl_info'].configure(
|
|
text="⚠️ Archivo corrupto o ilegible", text_color=Theme.ERROR)
|
|
return
|
|
|
|
# Actualizar texto
|
|
widgets['lbl_info'].configure(text=data["info"])
|
|
|
|
# Actualizar imagen si existe
|
|
if data["img"]:
|
|
ctk_img = ctk.CTkImage(data["img"], size=data["img"].size)
|
|
# Guardamos referencia en el objeto item para que no se pierda
|
|
item = next((x for x in self.queue_items if x.id == item_id), None)
|
|
if item:
|
|
item.thumbnail = ctk_img
|
|
widgets['lbl_icon'].configure(
|
|
image=ctk_img, text="") # Quitar texto placeholder
|
|
|
|
def _render_item(self, item: QueueItem, index: int):
|
|
"""Dibuja una tarjeta para el archivo."""
|
|
row_idx = index + 1 # +1 para dejar espacio al botón Clean arriba si fuera necesario
|
|
|
|
# Frame contenedor (Tarjeta)
|
|
card = ctk.CTkFrame(self, fg_color=Theme.ITEM_BG, corner_radius=8,
|
|
border_width=1, border_color=Theme.BORDER_COLOR)
|
|
card.grid(row=row_idx, column=0, sticky="ew", padx=10, pady=5)
|
|
card.grid_columnconfigure(1, weight=1)
|
|
|
|
# 1. Icono / Thumbnail
|
|
lbl_icon = ctk.CTkLabel(card, text="⏳", width=64,
|
|
height=64, fg_color="#222", corner_radius=6)
|
|
lbl_icon.grid(row=0, column=0, rowspan=2, padx=8, pady=8)
|
|
|
|
# Si ya teníamos la imagen (ej: reordenando lista), la ponemos directo
|
|
if item.thumbnail:
|
|
lbl_icon.configure(image=item.thumbnail, text="")
|
|
|
|
# 2. Información
|
|
lbl_name = ctk.CTkLabel(
|
|
card, text=item.name, font=Theme.FONT_MAIN, text_color=Theme.ACCENT, anchor="w")
|
|
lbl_name.grid(row=0, column=1, sticky="w", padx=5, pady=(8, 0))
|
|
|
|
lbl_info = ctk.CTkLabel(card, text=item.info_text, font=Theme.FONT_MONO,
|
|
text_color=Theme.TEXT_SEC, justify="left", anchor="w")
|
|
lbl_info.grid(row=1, column=1, sticky="w", padx=5, pady=(0, 8))
|
|
|
|
# 3. Controles (Frame derecho)
|
|
ctrl_frame = ctk.CTkFrame(card, fg_color="transparent")
|
|
ctrl_frame.grid(row=0, column=2, rowspan=2, padx=8, sticky="e")
|
|
|
|
# Botón Subir
|
|
btn_up = ctk.CTkButton(ctrl_frame, text="▲", width=28, height=28,
|
|
fg_color="transparent", border_width=1, border_color=Theme.TEXT_SEC,
|
|
hover_color=Theme.BTN_HOVER_NORMAL, text_color=Theme.TEXT_MAIN,
|
|
font=("Arial", 12),
|
|
command=lambda: self.move_up(item))
|
|
btn_up.pack(side="left", padx=2)
|
|
|
|
# Botón Eliminar
|
|
btn_del = ctk.CTkButton(ctrl_frame, text="✕", width=28, height=28,
|
|
fg_color="transparent", border_width=1, border_color=Theme.ERROR,
|
|
text_color=Theme.ERROR, hover_color=Theme.BTN_HOVER_DANGER,
|
|
font=("Arial", 12, "bold"),
|
|
command=lambda: self.remove_item(item))
|
|
btn_del.pack(side="left", padx=2)
|
|
|
|
# Guardar referencias para actualizaciones asíncronas
|
|
self._widget_refs[item.id] = {
|
|
"card": card,
|
|
"lbl_info": lbl_info,
|
|
"lbl_icon": lbl_icon,
|
|
"btn_up": btn_up # Guardamos por si queremos deshabilitar el primero
|
|
}
|
|
|
|
# Renderizar encabezado si es el primer elemento
|
|
if index == 0:
|
|
self._render_header()
|
|
|
|
def _render_header(self):
|
|
"""Dibuja el botón de limpiar todo si hay elementos."""
|
|
# Limpiar header anterior si existe
|
|
for w in self.grid_slaves(row=0):
|
|
w.destroy()
|
|
|
|
if self.queue_items:
|
|
header_frame = ctk.CTkFrame(self, fg_color="transparent")
|
|
header_frame.grid(row=0, column=0, sticky="ew",
|
|
padx=10, pady=(5, 0))
|
|
|
|
ctk.CTkLabel(header_frame, text=f"Queue: {len(self.queue_items)} Files",
|
|
font=Theme.FONT_SUB, text_color=Theme.TEXT_SEC).pack(side="left")
|
|
|
|
ctk.CTkButton(header_frame, text="CLEAN ALL", image=self.clear_icon,
|
|
font=Theme.FONT_SUB, height=24, width=100,
|
|
fg_color="transparent", border_width=1, border_color=Theme.ACCENT, text_color=Theme.ACCENT,
|
|
hover_color=Theme.BTN_HOVER_NORMAL,
|
|
command=self.clean_file_list).pack(side="right")
|
|
|
|
def _update_layout_indices(self):
|
|
"""Re-dibuja toda la lista. Útil para reordenar o borrar."""
|
|
# Nota: En una app muy grande esto sería ineficiente, pero para <100 archivos está bien
|
|
# destruir y recrear widgets garantiza orden visual correcto.
|
|
|
|
# Limpiar UI actual
|
|
for widget in self.winfo_children():
|
|
widget.destroy()
|
|
|
|
self._widget_refs.clear()
|
|
|
|
# Redibujar
|
|
if self.queue_items:
|
|
self._render_header()
|
|
for i, item in enumerate(self.queue_items):
|
|
self._render_item(item, i)
|
|
else:
|
|
# Lista vacía
|
|
if self.on_queue_empty_callback:
|
|
self.on_queue_empty_callback()
|
|
|
|
# --- LÓGICA DE DATOS ---
|
|
|
|
def move_up(self, item: QueueItem):
|
|
try:
|
|
idx = self.queue_items.index(item)
|
|
if idx > 0:
|
|
self.queue_items[idx], self.queue_items[idx -
|
|
1] = self.queue_items[idx - 1], self.queue_items[idx]
|
|
self._update_layout_indices()
|
|
except ValueError:
|
|
pass
|
|
|
|
def remove_item(self, item: QueueItem):
|
|
if item in self.queue_items:
|
|
self.queue_items.remove(item)
|
|
# Eliminar referencias para liberar memoria
|
|
if item.id in self._widget_refs:
|
|
del self._widget_refs[item.id]
|
|
self._update_layout_indices()
|
|
|
|
def clean_file_list(self):
|
|
self.queue_items.clear()
|
|
self._widget_refs.clear()
|
|
self._update_layout_indices()
|
|
|
|
def get_selected_file_list(self) -> List[str]:
|
|
return [item.path for item in self.queue_items if not item.has_error]
|
|
|
|
def regenerate_all_info(self):
|
|
"""Recalcula el texto de redimensionado sin recargar imágenes."""
|
|
for item in self.queue_items:
|
|
if not item.has_error and item.is_loaded:
|
|
# Recalculamos solo el texto basado en lo que ya sabemos (o recargamos dims rápidas)
|
|
# Para optimizar, asumimos que podemos volver a lanzar el worker o
|
|
# simplemente recalcular si guardáramos width/height en el objeto.
|
|
# Por simplicidad y robustez, relanzamos el worker ligero.
|
|
self.executor.submit(self._worker_load_info, item)
|
|
|
|
# --- SETTERS ---
|
|
def set_upscale_factor(self, f):
|
|
self.upscale_factor = f
|
|
|
|
def set_input_resize_factor(self, f):
|
|
self.input_resize_factor = f
|
|
|
|
def set_output_resize_factor(self, f):
|
|
self.output_resize_factor = f
|
|
|
|
# --- UTILIDADES ESTÁTICAS (Lógica pura) ---
|
|
|
|
def _extract_file_data(self, file_path) -> Tuple[str, Optional[Image.Image]]:
|
|
"""Extrae info y genera thumbnail PIL (Seguro para Threads)."""
|
|
info_str = "Info no disponible"
|
|
pil_image = None
|
|
|
|
try:
|
|
exts_video = ['.mp4', '.avi', '.mkv',
|
|
'.mov', '.wmv', '.flv', '.webm']
|
|
is_video = any(file_path.lower().endswith(ext)
|
|
for ext in exts_video)
|
|
|
|
width, height = 0, 0
|
|
|
|
if is_video:
|
|
cap = cv2.VideoCapture(file_path)
|
|
if cap.isOpened():
|
|
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
|
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
|
frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
|
|
fps = cap.get(cv2.CAP_PROP_FPS)
|
|
|
|
# Leer frame central para thumbnail
|
|
try:
|
|
if frames > 100:
|
|
cap.set(cv2.CAP_PROP_POS_FRAMES, 50)
|
|
ret, frame = cap.read()
|
|
if ret:
|
|
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
|
pil_image = self._process_thumbnail(frame)
|
|
except:
|
|
pass # Fallo silencioso solo en thumbnail
|
|
|
|
dur = frames/fps if fps > 0 else 0
|
|
m, s = divmod(int(dur), 60)
|
|
info_str = f"Video: {m}m:{s:02d}s • {frames} frames • {width}x{height}\n"
|
|
cap.release()
|
|
else:
|
|
# Imagen
|
|
# cv2.imdecode es mejor para rutas con caracteres especiales en Windows
|
|
img_array = np.fromfile(file_path, np.uint8)
|
|
img = cv2.imdecode(img_array, cv2.IMREAD_COLOR)
|
|
|
|
if img is not None:
|
|
height, width = img.shape[:2]
|
|
img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
|
|
pil_image = self._process_thumbnail(img)
|
|
info_str = f"Image: {width}x{height}\n"
|
|
|
|
# Añadir datos de redimensionado calculado
|
|
if width > 0 and height > 0:
|
|
info_str += self._calculate_resize_text(width, height)
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error interno extract: {e}")
|
|
raise e
|
|
|
|
return info_str, pil_image
|
|
|
|
def _process_thumbnail(self, img_array) -> Image.Image:
|
|
"""Redimensiona y recorta imagen para thumbnail."""
|
|
h, w = img_array.shape[:2]
|
|
target_size = 64
|
|
|
|
# Mantener aspect ratio dentro de 64x64
|
|
scale = target_size / max(h, w)
|
|
new_w, new_h = int(w * scale), int(h * scale)
|
|
|
|
resized = cv2.resize(img_array, (new_w, new_h),
|
|
interpolation=cv2.INTER_AREA)
|
|
return Image.fromarray(resized)
|
|
|
|
def _calculate_resize_text(self, w, h) -> str:
|
|
"""Lógica pura de cálculo de dimensiones."""
|
|
if self.upscale_factor <= 0:
|
|
return ""
|
|
|
|
iw = int(w * (self.input_resize_factor/100)
|
|
) if self.input_resize_factor else w
|
|
ih = int(h * (self.input_resize_factor/100)
|
|
) if self.input_resize_factor else h
|
|
|
|
uw, uh = int(iw * self.upscale_factor), int(ih * self.upscale_factor)
|
|
|
|
ow = int(uw * (self.output_resize_factor/100)
|
|
) if self.output_resize_factor else uw
|
|
oh = int(uh * (self.output_resize_factor/100)
|
|
) if self.output_resize_factor else uh
|
|
|
|
txt = ""
|
|
if self.input_resize_factor:
|
|
txt += f"Input AI ({int(self.input_resize_factor)}%) ➜ {iw}x{ih}\n"
|
|
|
|
txt += f"Output AI (x{self.upscale_factor}) ➜ {uw}x{uh}"
|
|
|
|
if self.output_resize_factor:
|
|
txt += f"\nFinal Output ({int(self.output_resize_factor)}%) ➜ {ow}x{oh}"
|
|
|
|
return txt
|