34. Ejemplo completo con Kaggle, Tkinter y PyTorch

{
  "username": "tu_usuario",
  "key": "tu_token_privado"
}

import os

os.environ["KAGGLE_USERNAME"] = "diegomoissetde"
os.environ["KAGGLE_KEY"] = "KGAT_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

MAX_GATOS = 2000
MAX_PERROS = 2000

import os

# ==========================================================
# OPCIÓN 1: CARGAR KAGGLE DIRECTAMENTE EN EL PROGRAMA
# ==========================================================
os.environ["KAGGLE_USERNAME"] = "diegomoissetde"
os.environ["KAGGLE_KEY"] = "KGAT_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

import random
import shutil
import threading
import zipfile
from pathlib import Path

import tkinter as tk
from tkinter import ttk, messagebox, filedialog

from PIL import Image, ImageTk

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, random_split
from torchvision import datasets, transforms


# ==========================================================
# CONFIGURACIÓN GENERAL
# ==========================================================
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")

DATASET_SLUG = "salader/dogsvscats"

CARPETA_BASE = Path("cats_vs_dogs_kaggle")
CARPETA_ZIP = CARPETA_BASE / "dataset.zip"
CARPETA_EXTRAIDA = CARPETA_BASE / "extraido"
CARPETA_PREPARADA = CARPETA_BASE / "preparado"

MODELO_PATH = Path("modelo_cats_vs_dogs_reducido.pth")

BATCH_SIZE = 32
EPOCHS = 5
LEARNING_RATE = 0.001
IMAGE_SIZE = 128

MAX_GATOS = 2000
MAX_PERROS = 2000

CLASES = ["cat", "dog"]


# ==========================================================
# MODELO CNN
# ==========================================================
class RedGatosPerros(nn.Module):
    def __init__(self):
        super().__init__()

        self.convolucional = nn.Sequential(
            nn.Conv2d(3, 16, kernel_size=3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),

            nn.Conv2d(16, 32, kernel_size=3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2),

            nn.Conv2d(32, 64, kernel_size=3, padding=1),
            nn.ReLU(),
            nn.MaxPool2d(2)
        )

        self.clasificador = nn.Sequential(
            nn.Flatten(),
            nn.Linear(64 * 16 * 16, 128),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(128, 2)
        )

    def forward(self, x):
        x = self.convolucional(x)
        x = self.clasificador(x)
        return x


# ==========================================================
# APLICACIÓN TKINTER
# ==========================================================
class AplicacionCatsDogs:
    def __init__(self, root):
        self.root = root
        self.root.title("Cats vs Dogs - Tkinter + PyTorch + Kaggle")
        self.root.geometry("1320x820")

        self.modelo = None
        self.train_loader = None
        self.val_loader = None
        self.entrenando = False
        self.imagen_tk = None

        self.transformacion_dataset = transforms.Compose([
            transforms.Resize((IMAGE_SIZE, IMAGE_SIZE)),
            transforms.ToTensor(),
            transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
        ])

        self.transformacion_prediccion = transforms.Compose([
            transforms.Resize((IMAGE_SIZE, IMAGE_SIZE)),
            transforms.ToTensor(),
            transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
        ])

        self.crear_interfaz()
        self.intentar_cargar_modelo_existente()

    def crear_interfaz(self):
        contenedor = ttk.Frame(self.root, padding=10)
        contenedor.pack(fill="both", expand=True)

        panel_izq = ttk.Frame(contenedor)
        panel_izq.pack(side="left", fill="y", padx=(0, 10))

        panel_der = ttk.Frame(contenedor)
        panel_der.pack(side="left", fill="both", expand=True)

        ttk.Label(
            panel_izq,
            text="Cats vs Dogs - Versión didáctica",
            font=("Arial", 15, "bold")
        ).pack(anchor="w", pady=(0, 8))

        ttk.Label(
            panel_izq,
            text=f"Muestra reducida: {MAX_GATOS} gatos + {MAX_PERROS} perros",
            font=("Arial", 10)
        ).pack(anchor="w", pady=(0, 12))

        self.btn_descargar = ttk.Button(
            panel_izq,
            text="1. Descargar dataset",
            command=self.iniciar_descarga
        )
        self.btn_descargar.pack(fill="x", pady=4)

        self.btn_preparar = ttk.Button(
            panel_izq,
            text="2. Preparar dataset reducido",
            command=self.preparar_dataset
        )
        self.btn_preparar.pack(fill="x", pady=4)

        self.btn_cargar = ttk.Button(
            panel_izq,
            text="3. Cargar datos",
            command=self.cargar_datos
        )
        self.btn_cargar.pack(fill="x", pady=4)

        self.btn_entrenar = ttk.Button(
            panel_izq,
            text="4. Entrenar modelo",
            command=self.iniciar_entrenamiento
        )
        self.btn_entrenar.pack(fill="x", pady=4)

        self.btn_imagen = ttk.Button(
            panel_izq,
            text="5. Seleccionar imagen y clasificar",
            command=self.seleccionar_imagen
        )
        self.btn_imagen.pack(fill="x", pady=(12, 4))

        self.barra = ttk.Progressbar(
            panel_izq,
            orient="horizontal",
            mode="determinate",
            length=260
        )
        self.barra.pack(fill="x", pady=(15, 5))

        self.lbl_estado = ttk.Label(panel_izq, text="Estado: esperando acción")
        self.lbl_estado.pack(anchor="w", pady=4)

        self.lbl_device = ttk.Label(panel_izq, text=f"Dispositivo: {DEVICE}")
        self.lbl_device.pack(anchor="w", pady=4)

        self.lbl_modelo = ttk.Label(panel_izq, text="Modelo: no cargado")
        self.lbl_modelo.pack(anchor="w", pady=4)

        marco_info = ttk.LabelFrame(panel_izq, text="Información", padding=10)
        marco_info.pack(fill="x", pady=(18, 0))

        self.lbl_epoca = ttk.Label(marco_info, text="Época: -")
        self.lbl_epoca.pack(anchor="w")

        self.lbl_batch = ttk.Label(marco_info, text="Batch: -")
        self.lbl_batch.pack(anchor="w")

        self.lbl_loss = ttk.Label(marco_info, text="Loss: -")
        self.lbl_loss.pack(anchor="w")

        self.lbl_acc = ttk.Label(marco_info, text="Accuracy entrenamiento: -")
        self.lbl_acc.pack(anchor="w")

        self.lbl_val = ttk.Label(marco_info, text="Accuracy validación: -")
        self.lbl_val.pack(anchor="w")

        self.lbl_pred = ttk.Label(
            panel_izq,
            text="Predicción: -",
            font=("Arial", 12, "bold")
        )
        self.lbl_pred.pack(anchor="w", pady=(20, 4))

        self.lbl_prob = ttk.Label(panel_izq, text="Probabilidad: -")
        self.lbl_prob.pack(anchor="w", pady=4)

        marco_imagen = ttk.LabelFrame(panel_der, text="Visualización", padding=10)
        marco_imagen.pack(fill="both", expand=True)

        self.canvas = tk.Canvas(marco_imagen, width=780, height=680, bg="black")
        self.canvas.pack(fill="both", expand=True)

        self.lbl_imagen = ttk.Label(
            panel_der,
            text="Imagen actual: -",
            font=("Arial", 11, "bold")
        )
        self.lbl_imagen.pack(pady=8)

    def actualizar_estado(self, texto):
        self.root.after(0, lambda: self.lbl_estado.config(text=f"Estado: {texto}"))

    def actualizar_barra(self, valor, maximo=100):
        def _actualizar():
            self.barra["maximum"] = maximo
            self.barra["value"] = valor
        self.root.after(0, _actualizar)

    def mostrar_pil_en_canvas(self, imagen_pil, texto_label):
        ancho_canvas = self.canvas.winfo_width()
        alto_canvas = self.canvas.winfo_height()

        if ancho_canvas < 10:
            ancho_canvas = 780
        if alto_canvas < 10:
            alto_canvas = 680

        copia = imagen_pil.copy()
        copia.thumbnail((ancho_canvas - 20, alto_canvas - 20))
        self.imagen_tk = ImageTk.PhotoImage(copia)

        def _dibujar():
            self.canvas.delete("all")
            self.canvas.create_image(
                ancho_canvas // 2,
                alto_canvas // 2,
                image=self.imagen_tk
            )
            self.lbl_imagen.config(text=texto_label)

        self.root.after(0, _dibujar)

    def mostrar_imagen_tensor(self, tensor, etiqueta_real):
        imagen = tensor.detach().cpu()
        media = torch.tensor([0.5, 0.5, 0.5]).view(3, 1, 1)
        desv = torch.tensor([0.5, 0.5, 0.5]).view(3, 1, 1)
        imagen = imagen * desv + media
        imagen = torch.clamp(imagen, 0, 1)
        pil = transforms.ToPILImage()(imagen)
        self.mostrar_pil_en_canvas(pil, f"Imagen actual: {CLASES[etiqueta_real]}")

    def autenticar_kaggle(self):
        from kaggle.api.kaggle_api_extended import KaggleApi

        username = os.environ.get("KAGGLE_USERNAME")
        key = os.environ.get("KAGGLE_KEY")

        if not username or not key:
            raise RuntimeError(
                "No se encontraron KAGGLE_USERNAME y KAGGLE_KEY dentro del programa."
            )

        api = KaggleApi()
        api.authenticate()
        return api

    def dataset_ya_preparado(self):
        carpeta_cat = CARPETA_PREPARADA / "cat"
        carpeta_dog = CARPETA_PREPARADA / "dog"

        if not carpeta_cat.exists() or not carpeta_dog.exists():
            return False

        cantidad_cat = len(list(carpeta_cat.glob("*.jpg")))
        cantidad_dog = len(list(carpeta_dog.glob("*.jpg")))

        return cantidad_cat >= 1 and cantidad_dog >= 1

    def intentar_cargar_modelo_existente(self):
        if MODELO_PATH.exists():
            try:
                self.modelo = RedGatosPerros().to(DEVICE)
                self.modelo.load_state_dict(torch.load(MODELO_PATH, map_location=DEVICE))
                self.modelo.eval()
                self.lbl_modelo.config(text=f"Modelo: cargado desde {MODELO_PATH.name}")
                self.actualizar_estado("modelo existente cargado")
            except Exception:
                self.lbl_modelo.config(text="Modelo: encontrado pero no se pudo cargar")

    def iniciar_descarga(self):
        hilo = threading.Thread(target=self.descargar_dataset, daemon=True)
        hilo.start()

    def descargar_dataset(self):
        try:
            CARPETA_BASE.mkdir(exist_ok=True)

            if CARPETA_ZIP.exists():
                self.actualizar_estado("el zip ya existe, no se vuelve a descargar")
                messagebox.showinfo(
                    "Descarga",
                    f"El archivo ya existe:\n{CARPETA_ZIP}\n\nNo se volvió a descargar."
                )
                return

            if self.dataset_ya_preparado():
                self.actualizar_estado("el dataset ya está preparado")
                messagebox.showinfo(
                    "Descarga",
                    "El dataset reducido ya está preparado.\nNo hace falta volver a descargarlo."
                )
                return

            self.actualizar_estado("descargando dataset desde Kaggle...")
            self.actualizar_barra(0, 100)

            api = self.autenticar_kaggle()

            self.actualizar_barra(15, 100)

            api.dataset_download_files(
                DATASET_SLUG,
                path=str(CARPETA_BASE),
                unzip=False,
                quiet=False
            )

            archivos_zip = list(CARPETA_BASE.glob("*.zip"))
            if not archivos_zip:
                raise RuntimeError("No se encontró el archivo zip descargado.")

            archivo_bajado = archivos_zip[0]

            if archivo_bajado != CARPETA_ZIP:
                if CARPETA_ZIP.exists():
                    CARPETA_ZIP.unlink()
                archivo_bajado.rename(CARPETA_ZIP)

            self.actualizar_barra(100, 100)
            self.actualizar_estado("dataset descargado correctamente")
            messagebox.showinfo("Descarga", "El dataset se descargó correctamente.")

        except Exception as e:
            self.actualizar_estado("error en la descarga")
            messagebox.showerror("Error", f"No se pudo descargar el dataset.\n\n{e}")

    def preparar_dataset(self):
        hilo = threading.Thread(target=self._preparar_dataset_hilo, daemon=True)
        hilo.start()

    def _preparar_dataset_hilo(self):
        try:
            if self.dataset_ya_preparado():
                self.actualizar_estado("dataset reducido ya preparado")
                messagebox.showinfo(
                    "Preparación",
                    "El dataset reducido ya existe.\nNo se volvió a preparar."
                )
                return

            if not CARPETA_ZIP.exists():
                raise RuntimeError("Primero debes descargar el dataset.")

            self.actualizar_estado("extrayendo zip...")
            self.actualizar_barra(0, 100)

            if CARPETA_EXTRAIDA.exists():
                shutil.rmtree(CARPETA_EXTRAIDA)
            CARPETA_EXTRAIDA.mkdir(parents=True, exist_ok=True)

            with zipfile.ZipFile(CARPETA_ZIP, "r") as archivo_zip:
                archivo_zip.extractall(CARPETA_EXTRAIDA)

            self.actualizar_barra(20, 100)
            self.actualizar_estado("buscando imágenes de gatos y perros...")

            if CARPETA_PREPARADA.exists():
                shutil.rmtree(CARPETA_PREPARADA)

            (CARPETA_PREPARADA / "cat").mkdir(parents=True, exist_ok=True)
            (CARPETA_PREPARADA / "dog").mkdir(parents=True, exist_ok=True)

            archivos_gatos = []
            archivos_perros = []

            for archivo in CARPETA_EXTRAIDA.rglob("*.jpg"):
                nombre = archivo.name.lower()

                if nombre.startswith("cat"):
                    archivos_gatos.append(archivo)
                elif nombre.startswith("dog"):
                    archivos_perros.append(archivo)

            if not archivos_gatos or not archivos_perros:
                raise RuntimeError("No se encontraron imágenes de gatos o perros.")

            random.shuffle(archivos_gatos)
            random.shuffle(archivos_perros)

            archivos_gatos = archivos_gatos[:MAX_GATOS]
            archivos_perros = archivos_perros[:MAX_PERROS]

            total_a_copiar = len(archivos_gatos) + len(archivos_perros)
            copiados = 0

            self.actualizar_estado("copiando gatos...")
            for archivo in archivos_gatos:
                destino = CARPETA_PREPARADA / "cat" / archivo.name
                shutil.copy2(archivo, destino)
                copiados += 1
                self.actualizar_barra(copiados, total_a_copiar)

            self.actualizar_estado("copiando perros...")
            for archivo in archivos_perros:
                destino = CARPETA_PREPARADA / "dog" / archivo.name
                shutil.copy2(archivo, destino)
                copiados += 1
                self.actualizar_barra(copiados, total_a_copiar)

            self.actualizar_estado("dataset reducido preparado")
            messagebox.showinfo(
                "Preparación",
                f"Dataset reducido listo.\n\n"
                f"Gatos: {len(archivos_gatos)}\n"
                f"Perros: {len(archivos_perros)}\n"
                f"Total: {total_a_copiar}"
            )

        except Exception as e:
            self.actualizar_estado("error al preparar dataset")
            messagebox.showerror("Error", f"No se pudo preparar el dataset.\n\n{e}")

    def cargar_datos(self):
        try:
            if not self.dataset_ya_preparado():
                raise RuntimeError("Primero debes preparar el dataset reducido.")

            self.actualizar_estado("cargando datos en PyTorch...")

            dataset_completo = datasets.ImageFolder(
                root=str(CARPETA_PREPARADA),
                transform=self.transformacion_dataset
            )

            cantidad_total = len(dataset_completo)
            cantidad_train = int(cantidad_total * 0.8)
            cantidad_val = cantidad_total - cantidad_train

            generador = torch.Generator().manual_seed(42)

            train_dataset, val_dataset = random_split(
                dataset_completo,
                [cantidad_train, cantidad_val],
                generator=generador
            )

            self.train_loader = DataLoader(
                train_dataset,
                batch_size=BATCH_SIZE,
                shuffle=True
            )

            self.val_loader = DataLoader(
                val_dataset,
                batch_size=BATCH_SIZE,
                shuffle=False
            )

            self.actualizar_estado(f"datos cargados: {cantidad_total} imágenes")
            messagebox.showinfo(
                "Carga exitosa",
                f"Total: {cantidad_total}\n"
                f"Entrenamiento: {cantidad_train}\n"
                f"Validación: {cantidad_val}"
            )

        except Exception as e:
            self.actualizar_estado("error al cargar datos")
            messagebox.showerror("Error", str(e))

    def iniciar_entrenamiento(self):
        if self.train_loader is None or self.val_loader is None:
            messagebox.showwarning("Aviso", "Primero debes cargar los datos.")
            return

        if self.entrenando:
            return

        self.entrenando = True
        hilo = threading.Thread(target=self.entrenar_modelo, daemon=True)
        hilo.start()

    def evaluar_validacion(self):
        self.modelo.eval()
        correctos = 0
        total = 0

        with torch.no_grad():
            for imagenes, etiquetas in self.val_loader:
                imagenes = imagenes.to(DEVICE)
                etiquetas = etiquetas.to(DEVICE)

                salidas = self.modelo(imagenes)
                predicciones = torch.argmax(salidas, dim=1)

                total += etiquetas.size(0)
                correctos += (predicciones == etiquetas).sum().item()

        if total == 0:
            return 0.0
        return 100.0 * correctos / total

    def entrenar_modelo(self):
        try:
            self.modelo = RedGatosPerros().to(DEVICE)

            criterio = nn.CrossEntropyLoss()
            optimizador = optim.Adam(self.modelo.parameters(), lr=LEARNING_RATE)

            total_batches = len(self.train_loader)
            self.actualizar_barra(0, total_batches * EPOCHS)

            for epoca in range(EPOCHS):
                self.modelo.train()

                correctos = 0
                total = 0
                loss_acumulada = 0.0

                for indice_batch, (imagenes, etiquetas) in enumerate(self.train_loader, start=1):
                    imagenes = imagenes.to(DEVICE)
                    etiquetas = etiquetas.to(DEVICE)

                    optimizador.zero_grad()
                    salidas = self.modelo(imagenes)
                    loss = criterio(salidas, etiquetas)
                    loss.backward()
                    optimizador.step()

                    loss_acumulada += loss.item()

                    predicciones = torch.argmax(salidas, dim=1)
                    total += etiquetas.size(0)
                    correctos += (predicciones == etiquetas).sum().item()

                    acc_actual = 100.0 * correctos / total if total > 0 else 0.0

                    indice_mostrar = random.randint(0, imagenes.size(0) - 1)
                    self.mostrar_imagen_tensor(
                        imagenes[indice_mostrar],
                        etiquetas[indice_mostrar].item()
                    )

                    progreso_global = epoca * total_batches + indice_batch
                    self.actualizar_barra(progreso_global, total_batches * EPOCHS)

                    self.root.after(
                        0,
                        lambda e=epoca + 1: self.lbl_epoca.config(text=f"Época: {e}/{EPOCHS}")
                    )
                    self.root.after(
                        0,
                        lambda b=indice_batch, tb=total_batches: self.lbl_batch.config(text=f"Batch: {b}/{tb}")
                    )
                    self.root.after(
                        0,
                        lambda l=loss.item(): self.lbl_loss.config(text=f"Loss: {l:.4f}")
                    )
                    self.root.after(
                        0,
                        lambda a=acc_actual: self.lbl_acc.config(text=f"Accuracy entrenamiento: {a:.2f}%")
                    )

                acc_val = self.evaluar_validacion()
                loss_promedio = loss_acumulada / total_batches

                self.root.after(
                    0,
                    lambda v=acc_val: self.lbl_val.config(text=f"Accuracy validación: {v:.2f}%")
                )

                self.actualizar_estado(
                    f"época {epoca + 1} finalizada - loss {loss_promedio:.4f} - val {acc_val:.2f}%"
                )

            torch.save(self.modelo.state_dict(), MODELO_PATH)
            self.modelo.eval()

            self.root.after(
                0,
                lambda: self.lbl_modelo.config(text=f"Modelo: guardado en {MODELO_PATH.name}")
            )

            self.actualizar_estado("entrenamiento finalizado")
            messagebox.showinfo(
                "Entrenamiento finalizado",
                f"Modelo guardado en:\n{MODELO_PATH}"
            )

        except Exception as e:
            messagebox.showerror("Error en entrenamiento", str(e))
            self.actualizar_estado("error en entrenamiento")
        finally:
            self.entrenando = False

    def seleccionar_imagen(self):
        ruta = filedialog.askopenfilename(
            title="Seleccionar imagen",
            filetypes=[
                ("Imágenes", "*.jpg *.jpeg *.png *.bmp *.webp"),
                ("Todos los archivos", "*.*")
            ]
        )

        if not ruta:
            return

        try:
            if self.modelo is None:
                if MODELO_PATH.exists():
                    self.intentar_cargar_modelo_existente()

            if self.modelo is None:
                raise RuntimeError(
                    "No hay modelo cargado.\n"
                    "Primero entrena el modelo o asegúrate de que exista el archivo guardado."
                )

            imagen = Image.open(ruta).convert("RGB")
            self.mostrar_pil_en_canvas(imagen, f"Imagen seleccionada: {Path(ruta).name}")

            tensor = self.transformacion_prediccion(imagen).unsqueeze(0).to(DEVICE)

            self.modelo.eval()
            with torch.no_grad():
                salida = self.modelo(tensor)
                probabilidades = torch.softmax(salida, dim=1)[0]
                indice = int(torch.argmax(probabilidades).item())
                prob = float(probabilidades[indice].item()) * 100.0

            clase = CLASES[indice]

            self.lbl_pred.config(text=f"Predicción: {clase}")
            self.lbl_prob.config(text=f"Probabilidad: {prob:.2f}%")
            self.actualizar_estado("predicción realizada")

        except Exception as e:
            messagebox.showerror("Error", f"No se pudo clasificar la imagen.\n\n{e}")


# ==========================================================
# PROGRAMA PRINCIPAL
# ==========================================================
if __name__ == "__main__":
    root = tk.Tk()
    app = AplicacionCatsDogs(root)
    root.mainloop()

DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")

api.dataset_download_files(
    DATASET_SLUG,
    path=str(CARPETA_BASE),
    unzip=False,
    quiet=False
)

preparado/
  cat/
  dog/

torch.save(self.modelo.state_dict(), MODELO_PATH)