Transfer Learning paso a paso con PyTorch

import torchvision.models as models

# Ver todos los modelos disponibles con pesos preentrenados
available = models.list_models(module=models)
print(f"Modelos en torchvision: {len(available)}")
print(available[:15])  # Primeros 15

# Con timm (instalar: pip install timm)
import timm
timm_models = timm.list_models(pretrained=True)
print(f"\nModelos en timm: {len(timm_models)}")

# Buscar modelos de una familia específica
resnet_models = timm.list_models('resnet*', pretrained=True)
efficientnet_models = timm.list_models('efficientnet*', pretrained=True)
print(f"ResNets disponibles: {len(resnet_models)}")
print(f"EfficientNets disponibles: {len(efficientnet_models)}")

# 👆 Haz clic en un modelo de la galería para ver cómo cargarlo
model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V2)

import torch
import torch.nn as nn
import torchvision.models as models
from torchvision.models import ResNet50_Weights

# Cargar ResNet-50 con pesos de ImageNet (V2 = mejores pesos)
weights = ResNet50_Weights.IMAGENET1K_V2
backbone = models.resnet50(weights=weights)

print(f"Modelo cargado: ResNet-50")
print(f"Parámetros totales: {sum(p.numel() for p in backbone.parameters()):,}")
print(f"Output del backbone: {backbone.fc.in_features} features")
print(f"Clases originales: {backbone.fc.out_features}")

# Las transforms recomendadas para este modelo
preprocess = weights.transforms()
print(f"\nTransforms recomendadas:\n{preprocess}")

from PIL import Image
from torchvision.transforms import functional as F

# Descargar una imagen de ejemplo (o usa una tuya)
import urllib.request
url = "https://upload.wikimedia.org/wikipedia/commons/thumb/2/26/YellowLabradorLooking_new.jpg/1200px-YellowLabradorLooking_new.jpg"
urllib.request.urlretrieve(url, "test_dog.jpg")
img = Image.open("test_dog.jpg")

# Preprocesar con las transforms del modelo
batch = preprocess(img).unsqueeze(0)  # (3,224,224) → (1,3,224,224)

# Predecir
backbone.eval()
with torch.no_grad():
    logits = backbone(batch)
    probs = torch.softmax(logits, dim=1)
    top5 = torch.topk(probs, 5)

# Mostrar top-5 predicciones
categories = weights.meta["categories"]
for i in range(5):
    idx = top5.indices[0][i].item()
    prob = top5.values[0][i].item() * 100
    print(f"  {categories[idx]:30s} → {prob:.1f}%")

import timm

# timm tiene una API unificada para todos los modelos
model_timm = timm.create_model('resnet50', pretrained=True)

# Para ver la configuración del modelo
data_config = timm.data.resolve_model_data_config(model_timm)
print(data_config)
# {'input_size': (3, 224, 224), 'interpolation': 'bicubic',
#  'mean': (0.485, 0.456, 0.406), 'std': (0.229, 0.224, 0.225), ...}

# También puedes crear el modelo ya sin la cabeza de clasificación
backbone_timm = timm.create_model('resnet50', pretrained=True, num_classes=0)
# Ahora backbone_timm produce directamente un vector de 2048 features

# Ver las capas principales de ResNet-50
for name, module in backbone.named_children():
    params = sum(p.numel() for p in module.parameters())
    print(f"{name:12s} → {params:>10,} params  ({type(module).__name__})")

NUM_CLASSES = 2  # perros vs gatos (o tu número de clases)

# El head original
print(f"Head original: {backbone.fc}")

# Opción A: head simple (una sola capa lineal)
backbone.fc = nn.Linear(backbone.fc.in_features, NUM_CLASSES)

# Opción B: head con más capacidad (recomendado para few-shot)
backbone.fc = nn.Sequential(
    nn.Linear(2048, 512),
    nn.ReLU(),
    nn.Dropout(0.3),
    nn.Linear(512, NUM_CLASSES),
)

print(f"Head nuevo: {backbone.fc}")
print(f"Parámetros del head: {sum(p.numel() for p in backbone.fc.parameters()):,}")

dataset/
├── train/
│   ├── gatos/        # Clase 0
│   │   ├── cat_001.jpg
│   │   ├── cat_002.jpg
│   │   └── ...
│   └── perros/       # Clase 1
│       ├── dog_001.jpg
│       ├── dog_002.jpg
│       └── ...
├── val/
│   ├── gatos/
│   └── perros/
└── test/
    ├── gatos/
    └── perros/

from torchvision import transforms, datasets
from torch.utils.data import DataLoader

# ── Transforms para ENTRENAMIENTO (con augmentation) ─────
train_transforms = transforms.Compose([
    transforms.RandomResizedCrop(224, scale=(0.8, 1.0)),
    transforms.RandomHorizontalFlip(p=0.5),
    transforms.RandomRotation(15),
    transforms.ColorJitter(brightness=0.2, contrast=0.2,
                           saturation=0.2, hue=0.1),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],
                         std=[0.229, 0.224, 0.225]),
])

# ── Transforms para VALIDACIÓN / TEST (sin augmentation) ─
val_transforms = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],
                         std=[0.229, 0.224, 0.225]),
])

# ── Crear datasets ────────────────────────────────────────
train_dataset = datasets.ImageFolder("dataset/train", transform=train_transforms)
val_dataset   = datasets.ImageFolder("dataset/val",   transform=val_transforms)
test_dataset  = datasets.ImageFolder("dataset/test",  transform=val_transforms)

print(f"Clases encontradas: {train_dataset.classes}")
print(f"Mapping: {train_dataset.class_to_idx}")
print(f"Train: {len(train_dataset)} | Val: {len(val_dataset)} | Test: {len(test_dataset)}")

# ── Crear DataLoaders ─────────────────────────────────────
BATCH_SIZE = 32

train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE,
                          shuffle=True, num_workers=4, pin_memory=True)
val_loader   = DataLoader(val_dataset, batch_size=BATCH_SIZE,
                          shuffle=False, num_workers=4, pin_memory=True)
test_loader  = DataLoader(test_dataset, batch_size=BATCH_SIZE,
                          shuffle=False, num_workers=4, pin_memory=True)

# Alternativa rápida: usar un dataset de torchvision
from torchvision.datasets import OxfordIIITPet

# Descarga automáticamente ~800 MB
train_dataset = OxfordIIITPet(root="./data", split="trainval",
                              target_types="category",
                              transform=train_transforms, download=True)
test_dataset  = OxfordIIITPet(root="./data", split="test",
                              target_types="category",
                              transform=val_transforms, download=True)

NUM_CLASSES = 37  # 37 razas de perros y gatos
print(f"Clases: {len(train_dataset.classes)} razas de mascotas")

import torch
import torch.nn as nn
import torchvision.models as models
from torchvision.models import ResNet50_Weights

# 1. Cargar modelo preentrenado
model = models.resnet50(weights=ResNet50_Weights.IMAGENET1K_V2)

# 2. CONGELAR todo el backbone
for param in model.parameters():
    param.requires_grad = False

# 3. Reemplazar el head (se crea con requires_grad=True por defecto)
NUM_CLASSES = 2
model.fc = nn.Sequential(
    nn.Linear(2048, 512),
    nn.ReLU(),
    nn.Dropout(0.3),
    nn.Linear(512, NUM_CLASSES),
)

# Verificar qué se entrena y qué no
total_params = sum(p.numel() for p in model.parameters())
trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
frozen_params = total_params - trainable_params
print(f"Parámetros totales:     {total_params:>12,}")
print(f"Parámetros congelados:  {frozen_params:>12,} ({frozen_params/total_params*100:.1f}%)")
print(f"Parámetros entrenables: {trainable_params:>12,} ({trainable_params/total_params*100:.1f}%)")

# Mover a GPU
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)

import torch.optim as optim

# Solo los parámetros que requieren gradiente
optimizer = optim.Adam(
    filter(lambda p: p.requires_grad, model.parameters()),
    lr=1e-3,              # Learning rate relativamente alto para el head
    weight_decay=1e-4,    # Regularización L2
)

# Loss function
criterion = nn.CrossEntropyLoss()

# Scheduler: reducir LR cuando se estanca la mejora
scheduler = optim.lr_scheduler.ReduceLROnPlateau(
    optimizer, mode='min', factor=0.5, patience=3, verbose=True
)

print(f"Optimizador: Adam (lr=1e-3)")
print(f"Parámetros en el optimizer: {sum(p.numel() for group in optimizer.param_groups for p in group['params']):,}")

import time
from pathlib import Path

def train_one_epoch(model, loader, criterion, optimizer, device):
    """Entrena un epoch completo y devuelve loss y accuracy."""
    model.train()
    running_loss, correct, total = 0.0, 0, 0

    for images, labels in loader:
        images, labels = images.to(device), labels.to(device)

        outputs = model(images)
        loss = criterion(outputs, labels)

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        running_loss += loss.item() * images.size(0)
        correct += outputs.argmax(1).eq(labels).sum().item()
        total += labels.size(0)

    return running_loss / total, correct / total


@torch.no_grad()
def validate(model, loader, criterion, device):
    """Evalúa en el conjunto de validación."""
    model.eval()
    running_loss, correct, total = 0.0, 0, 0

    for images, labels in loader:
        images, labels = images.to(device), labels.to(device)
        outputs = model(images)
        loss = criterion(outputs, labels)

        running_loss += loss.item() * images.size(0)
        correct += outputs.argmax(1).eq(labels).sum().item()
        total += labels.size(0)

    return running_loss / total, correct / total


# ── Entrenamiento ─────────────────────────────────────────
EPOCHS = 15
best_val_loss = float('inf')
patience_counter = 0
PATIENCE = 5  # early stopping: parar si no mejora en 5 epochs
history = {'train_loss': [], 'val_loss': [], 'train_acc': [], 'val_acc': []}

save_dir = Path("checkpoints")
save_dir.mkdir(exist_ok=True)

print("Epoch  Train Loss  Train Acc  Val Loss   Val Acc   LR        Tiempo")
print("─" * 75)

for epoch in range(EPOCHS):
    t0 = time.time()

    train_loss, train_acc = train_one_epoch(
        model, train_loader, criterion, optimizer, device
    )
    val_loss, val_acc = validate(
        model, val_loader, criterion, device
    )

    # Learning rate scheduler
    scheduler.step(val_loss)
    current_lr = optimizer.param_groups[0]['lr']

    elapsed = time.time() - t0

    # Guardar historial
    history['train_loss'].append(train_loss)
    history['val_loss'].append(val_loss)
    history['train_acc'].append(train_acc)
    history['val_acc'].append(val_acc)

    # Early stopping + guardar mejor modelo
    marker = ""
    if val_loss < best_val_loss:
        best_val_loss = val_loss
        patience_counter = 0
        torch.save(model.state_dict(), save_dir / "best_model.pth")
        marker = " ★ saved"
    else:
        patience_counter += 1
        if patience_counter >= PATIENCE:
            print(f"\n⏹  Early stopping en epoch {epoch+1} (sin mejora en {PATIENCE} epochs)")
            break

    print(f"{epoch+1:3d}/{EPOCHS}  {train_loss:.4f}      {train_acc:.4f}     "
          f"{val_loss:.4f}     {val_acc:.4f}    {current_lr:.1e}  {elapsed:.1f}s{marker}")

# ── Fase 2: Fine-tuning ───────────────────────────────────

# 1. Cargar el mejor modelo de feature extraction
model.load_state_dict(torch.load("checkpoints/best_model.pth", weights_only=True))

# 2. Descongelar las últimas capas del backbone
#    Estrategia: descongelar layer4 (y opcionalmente layer3)
for name, param in model.named_parameters():
    param.requires_grad = False  # Congelar todo primero

# Descongelar layer4 + head
for name, param in model.named_parameters():
    if "layer4" in name or "fc" in name:
        param.requires_grad = True

# Verificar
trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
total = sum(p.numel() for p in model.parameters())
print(f"Entrenables ahora: {trainable:,} / {total:,} ({trainable/total*100:.1f}%)")

# 3. Optimizador con LEARNING RATES DIFERENCIADOS
#    - Backbone (layer4): lr bajo para no destruir features
#    - Head: lr más alto para que siga aprendiendo
optimizer_ft = optim.Adam([
    {"params": [p for n, p in model.named_parameters()
                if "layer4" in n and p.requires_grad],
     "lr": 1e-5},      # ← LR bajo para el backbone
    {"params": model.fc.parameters(),
     "lr": 5e-4},      # ← LR más alto para el head
], weight_decay=1e-4)

# 4. Nuevo scheduler
scheduler_ft = optim.lr_scheduler.CosineAnnealingLR(
    optimizer_ft, T_max=10, eta_min=1e-6
)

print("\nGrupos del optimizador:")
for i, group in enumerate(optimizer_ft.param_groups):
    n_params = sum(p.numel() for p in group['params'])
    print(f"  Grupo {i}: lr={group['lr']:.1e}, params={n_params:,}")

# ── Fase 2: Training loop de fine-tuning ──────────────────
FT_EPOCHS = 10
best_val_loss_ft = float('inf')
patience_counter_ft = 0

print("\n📌 Fine-tuning (layer4 + head)")
print("Epoch  Train Loss  Train Acc  Val Loss   Val Acc   Tiempo")
print("─" * 65)

for epoch in range(FT_EPOCHS):
    t0 = time.time()

    train_loss, train_acc = train_one_epoch(
        model, train_loader, criterion, optimizer_ft, device
    )
    val_loss, val_acc = validate(model, val_loader, criterion, device)

    scheduler_ft.step()
    elapsed = time.time() - t0

    marker = ""
    if val_loss < best_val_loss_ft:
        best_val_loss_ft = val_loss
        patience_counter_ft = 0
        torch.save(model.state_dict(), "checkpoints/best_model_finetuned.pth")
        marker = " ★ saved"
    else:
        patience_counter_ft += 1
        if patience_counter_ft >= 5:
            print(f"\n⏹  Early stopping en epoch {epoch+1}")
            break

    print(f"{epoch+1:3d}/{FT_EPOCHS}  {train_loss:.4f}      {train_acc:.4f}     "
          f"{val_loss:.4f}     {val_acc:.4f}    {elapsed:.1f}s{marker}")

import numpy as np
from sklearn.metrics import classification_report, confusion_matrix

# 1. Cargar el mejor modelo fine-tuned
model.load_state_dict(
    torch.load("checkpoints/best_model_finetuned.pth", weights_only=True)
)
model.eval()

# 2. Recoger todas las predicciones del test set
all_preds = []
all_labels = []
all_probs = []

with torch.no_grad():
    for images, labels in test_loader:
        images = images.to(device)
        outputs = model(images)
        probs = torch.softmax(outputs, dim=1)

        all_preds.extend(outputs.argmax(1).cpu().numpy())
        all_labels.extend(labels.numpy())
        all_probs.extend(probs.cpu().numpy())

all_preds = np.array(all_preds)
all_labels = np.array(all_labels)
all_probs = np.array(all_probs)

# 3. Accuracy global
test_acc = (all_preds == all_labels).mean()
print(f"🎯 Test Accuracy: {test_acc:.4f} ({test_acc*100:.1f}%)")
print(f"   Correctas: {(all_preds == all_labels).sum()} / {len(all_labels)}")

# 4. Classification report (precision, recall, F1)
class_names = test_dataset.classes  # ['gatos', 'perros']
print(f"\n📊 Classification Report:")
print(classification_report(all_labels, all_preds, target_names=class_names))

# 5. Matriz de confusión
cm = confusion_matrix(all_labels, all_preds)
print(f"Confusion Matrix:")
print(cm)

# Encontrar las predicciones incorrectas
errors_idx = np.where(all_preds != all_labels)[0]
print(f"Errores totales: {len(errors_idx)} / {len(all_labels)}")

# Encontrar las predicciones con menor confianza
confidence = all_probs.max(axis=1)
least_confident_idx = np.argsort(confidence)[:10]  # 10 menos confiables

print(f"\n🔍 Top-10 predicciones menos confiables:")
for idx in least_confident_idx:
    pred_class = class_names[all_preds[idx]]
    real_class = class_names[all_labels[idx]]
    conf = confidence[idx] * 100
    status = "✅" if all_preds[idx] == all_labels[idx] else "❌"
    print(f"  {status} Imagen {idx}: pred={pred_class} ({conf:.1f}%) | real={real_class}")

# Métricas adicionales
from sklearn.metrics import roc_auc_score
if NUM_CLASSES == 2:
    auc = roc_auc_score(all_labels, all_probs[:, 1])
    print(f"\n📈 AUC-ROC: {auc:.4f}")

from PIL import Image

def predict_image(model, image_path, transform, class_names, device):
    """Clasifica una imagen y devuelve clase y confianza."""
    img = Image.open(image_path).convert("RGB")
    img_tensor = transform(img).unsqueeze(0).to(device)  # (1, 3, 224, 224)

    model.eval()
    with torch.no_grad():
        logits = model(img_tensor)
        probs = torch.softmax(logits, dim=1)
        confidence, pred_idx = probs.max(dim=1)

    pred_class = class_names[pred_idx.item()]
    confidence = confidence.item() * 100
    return pred_class, confidence, probs[0].cpu().numpy()


# Usar el modelo
class_names = ['gatos', 'perros']
pred, conf, probs = predict_image(
    model, "nueva_imagen.jpg", val_transforms, class_names, device
)
print(f"Predicción: {pred} (confianza: {conf:.1f}%)")
print(f"Probabilidades: {dict(zip(class_names, [f'{p:.3f}' for p in probs]))}")

from pathlib import Path

def predict_batch(model, image_dir, transform, class_names, device):
    """Clasifica todas las imágenes de un directorio."""
    model.eval()
    results = []

    image_paths = list(Path(image_dir).glob("*.jpg")) + \
                  list(Path(image_dir).glob("*.png"))

    for path in image_paths:
        img = Image.open(path).convert("RGB")
        img_tensor = transform(img).unsqueeze(0).to(device)

        with torch.no_grad():
            logits = model(img_tensor)
            probs = torch.softmax(logits, dim=1)
            confidence, pred_idx = probs.max(dim=1)

        results.append({
            "file": path.name,
            "class": class_names[pred_idx.item()],
            "confidence": confidence.item() * 100,
        })

    return results

# Ejemplo
results = predict_batch(model, "nuevas_imagenes/", val_transforms, class_names, device)
for r in results:
    print(f"  {r['file']:25s} → {r['class']} ({r['confidence']:.1f}%)")

# Checkpoint completo: para poder reentrenar o compartir
checkpoint = {
    "model_state_dict": model.state_dict(),
    "class_names": class_names,
    "num_classes": NUM_CLASSES,
    "backbone": "resnet50",
    "input_size": 224,
    "normalize_mean": [0.485, 0.456, 0.406],
    "normalize_std": [0.229, 0.224, 0.225],
    "test_accuracy": test_acc,
    "training_info": {
        "feature_extraction_epochs": EPOCHS,
        "finetuning_epochs": FT_EPOCHS,
        "strategy": "progressive_unfreezing",
    }
}
torch.save(checkpoint, "model_final.pth")
print("✅ Checkpoint completo guardado")

# Para cargar después:
ckpt = torch.load("model_final.pth", weights_only=False)
model_loaded = models.resnet50(weights=None)  # Sin pesos preentrenados
model_loaded.fc = nn.Sequential(
    nn.Linear(2048, 512), nn.ReLU(), nn.Dropout(0.3),
    nn.Linear(512, ckpt["num_classes"]),
)
model_loaded.load_state_dict(ckpt["model_state_dict"])
model_loaded.eval()
print(f"✅ Modelo cargado: {ckpt['backbone']}, acc={ckpt['test_accuracy']:.3f}")

# Exportar a ONNX
model.eval()
dummy_input = torch.randn(1, 3, 224, 224).to(device)

torch.onnx.export(
    model,
    dummy_input,
    "model_transfer.onnx",
    input_names=["image"],
    output_names=["logits"],
    dynamic_axes={
        "image": {0: "batch_size"},
        "logits": {0: "batch_size"},
    },
    opset_version=17,
)
print("✅ Modelo exportado a ONNX")

# Verificar con ONNX Runtime
# pip install onnxruntime
import onnxruntime as ort

session = ort.InferenceSession("model_transfer.onnx")
ort_input = {"image": dummy_input.cpu().numpy()}
ort_output = session.run(None, ort_input)
print(f"Output shape: {ort_output[0].shape}")  # (1, 2)