Clasificación y ajuste fino

Clasificación de fotografías de fuentes de agua recopiladas mediante la ciencia participativa de GLOBE, utilizando aprendizaje por transferencia con ResNet-50.

# Instalar los paquetes necesarios

!pip install -q tensorflow pillow scikit-learn matplotlib transformers huggingface_hub
import geopandas as gpd
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from PIL import Image
import requests
from io import BytesIO
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
from tensorflow.keras.applications import ResNet50
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix
from sklearn.preprocessing import LabelEncoder
import seaborn as sns
from pathlib import Path

A continuación, cargaremos fotografías de hábitats de mosquitos, es decir, fuentes de agua, a partir de observaciones de NASA GLOBE. Para obtener más información, consulta la lección Introducción a los datos de GLOBE.

# Cargar los datos de hábitats de mosquitos de GLOBE

# Obtener los enlaces de las fotografías incluidas en las observaciones de GLOBE
mosquito = gpd.read_file(
    "https://github.com/geo-di-lab/emerge-lessons/raw/refs/heads/main/docs/data/globe_mosquito.zip"
)
data = mosquito.dropna(
    subset=["WaterSourcePhotoUrls"]
)[["WaterSourcePhotoUrls", "WaterSourceType"]].copy()

# Algunas observaciones incluyen varias URL; separarlas en filas distintas
data["WaterSourcePhotoUrls"] = data["WaterSourcePhotoUrls"].str.split(";")
data = data.explode("WaterSourcePhotoUrls").reset_index(drop=True)
data["WaterSourcePhotoUrls"] = data["WaterSourcePhotoUrls"].str.strip()

# Contar las fotografías de cada clasificación de fuente de agua
class_counts = data["WaterSourceType"].value_counts()
print(f"Total de observaciones con fotografías: {len(data)}")
print(f"\nTipos de fuentes de agua:\n{class_counts}")

A continuación, seleccionaremos un total de 200 fotografías de fuentes de agua, 50 de cada tipo, para realizar el ajuste fino del modelo de clasificación.

# Determinar la cantidad de muestras por clase
# Se utilizarán 200 imágenes en total, distribuidas de manera uniforme
n_classes = len(class_counts)
samples_per_class = 200 // n_classes
print(
    f"\nSeleccionando {samples_per_class} fotografías por clase "
    f"({n_classes} clases)"
)

# Muestreo estratificado: seleccionar la misma cantidad de cada clase
sample_data = data.groupby(
    "WaterSourceType",
    group_keys=False
).apply(
    lambda x: x.sample(
        n=min(samples_per_class, len(x)),
        random_state=42
    )
).reset_index(drop=True)

print(f"\nSe seleccionaron {len(sample_data)} observaciones")
print(
    "Distribución equilibrada de las clases:\n"
    f"{sample_data['WaterSourceType'].value_counts()}"
)

# Descargar y procesar las imágenes
# Las imágenes se redimensionarán a 224 × 224, el tamaño estándar de
# ResNet-50, y sus valores se normalizarán
def download_image(url, target_size=(224, 224)):
    """Descarga y preprocesa una imagen a partir de una URL."""
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        img = Image.open(BytesIO(response.content)).convert("RGB")
        img = img.resize(target_size)
        return np.array(img) / 255.0  # Normalizar al intervalo [0, 1]
    except Exception:
        return None

print("Descargando imágenes. Este proceso puede tardar unos minutos...")
images = []
labels = []

for idx, row in sample_data.iterrows():
    img = download_image(row["WaterSourcePhotoUrls"])
    if img is not None:
        images.append(img)
        labels.append(row["WaterSourceType"])

    if (idx + 1) % 50 == 0:
        print(
            f"Se procesaron {idx + 1}/{len(sample_data)} imágenes"
        )

print(f"\nSe cargaron correctamente {len(images)} imágenes")

Aquí daremos formato a las imágenes y las etiquetas para entrenar el modelo. También dividiremos los datos: el 80 % se utilizará para el entrenamiento y el 20 % para las pruebas.

# Convertir las imágenes y las etiquetas en arreglos de NumPy
X = np.array(images)
y = np.array(labels)

# Codificar las etiquetas como números enteros
label_encoder = LabelEncoder()
y_encoded = label_encoder.fit_transform(y)
num_classes = len(label_encoder.classes_)

print(f"\nCantidad de clases: {num_classes}")
print(f"Clases: {label_encoder.classes_}")
print(f"Muestras por clase: {np.bincount(y_encoded)}")

# Visualizar una muestra de las imágenes
fig, axes = plt.subplots(2, 4, figsize=(12, 6))
for i, ax in enumerate(axes.flat):
    if i < len(X):
        ax.imshow(X[i])
        ax.set_title(f"{y[i]}", fontsize=10)
        ax.axis("off")

plt.tight_layout()
plt.show()

# Dividir los datos: 80 % para entrenamiento y 20 % para pruebas
# La división estratificada mantiene una representación proporcional
# de cada clase
X_train, X_test, y_train, y_test = train_test_split(
    X,
    y_encoded,
    test_size=0.2,
    random_state=42,
    stratify=y_encoded
)

print(f"Conjunto de entrenamiento: {len(X_train)} imágenes")
print(f"Conjunto de prueba: {len(X_test)} imágenes")
print(
    "Distribución de clases en el conjunto de entrenamiento: "
    f"{np.bincount(y_train)}"
)
print(
    "Distribución de clases en el conjunto de prueba: "
    f"{np.bincount(y_test)}"
)

Crear el modelo de aprendizaje por transferencia

El aprendizaje por transferencia utiliza un modelo entrenado previamente con millones de imágenes y permite adaptarlo para clasificar las imágenes que nos interesan.

• ResNet-50 ya aprendió a reconocer bordes, texturas, formas y objetos.

• Conservaremos estas características aprendidas al mantener fijos sus pesos.

• Solo entrenaremos las nuevas capas superiores para clasificar las fuentes de agua.

• Este método funciona mucho mejor cuando se dispone de un conjunto de datos pequeño.

# Cargar ResNet-50 con entrenamiento previo
print("Cargando ResNet-50, previamente entrenado con ImageNet...")
base_model = ResNet50(
    weights="imagenet",    # Pesos obtenidos mediante entrenamiento previo
    include_top=False,     # Excluir la capa de clasificación original
    input_shape=(224, 224, 3)
)

# Congelar las capas del modelo base
base_model.trainable = False
print(
    f"Se congelaron {len(base_model.layers)} capas "
    "previamente entrenadas"
)

# Crear el modelo completo
model = keras.Sequential([
    base_model,
    layers.GlobalAveragePooling2D(),
    layers.Dense(128, activation="relu"),
    layers.Dropout(0.5),
    layers.Dense(num_classes, activation="softmax")
])

model.compile(
    optimizer=keras.optimizers.Adam(learning_rate=0.001),
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"]
)

model.summary()

Ahora entrenaremos el modelo con las imágenes de NASA GLOBE.

# Solo se entrenan las capas superiores, mientras que el modelo base
# ResNet-50 proporciona las características aprendidas previamente
history = model.fit(
    X_train,
    y_train,
    epochs=20,
    batch_size=16,
    validation_split=0.2,
    verbose=1
)

# Descongelar y realizar el ajuste fino del modelo base
# Después del entrenamiento inicial, podemos descongelar algunas de las
# capas superiores de ResNet-50 para mejorar los resultados
print(
    "\nAjuste fino: descongelando las capas superiores de ResNet-50..."
)
base_model.trainable = True

# Congelar todas las capas excepto las últimas 10
for layer in base_model.layers[:-10]:
    layer.trainable = False

print(
    "Capas entrenables: "
    f"{sum(1 for layer in model.layers if layer.trainable)}"
)

# Volver a compilar el modelo con una tasa de aprendizaje menor
model.compile(
    optimizer=keras.optimizers.Adam(learning_rate=0.0001),
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"]
)

# Continuar el entrenamiento
history_fine = model.fit(
    X_train,
    y_train,
    epochs=10,
    batch_size=16,
    validation_split=0.2,
    verbose=1
)

# Combinar el historial del entrenamiento inicial y el ajuste fino
for key in history.history.keys():
    history.history[key].extend(history_fine.history[key])

# Visualizar el progreso del entrenamiento
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4))

# Exactitud
ax1.plot(history.history["accuracy"], label="Entrenamiento")
ax1.plot(history.history["val_accuracy"], label="Validación")
ax1.axvline(
    x=20,
    color="gray",
    linestyle="--",
    alpha=0.5,
    label="Comienza el ajuste fino"
)
ax1.set_title("Exactitud del modelo")
ax1.set_xlabel("Época")
ax1.set_ylabel("Exactitud")
ax1.legend()
ax1.grid(True, alpha=0.3)

# Pérdida
ax2.plot(history.history["loss"], label="Entrenamiento")
ax2.plot(history.history["val_loss"], label="Validación")
ax2.axvline(
    x=20,
    color="gray",
    linestyle="--",
    alpha=0.5,
    label="Comienza el ajuste fino"
)
ax2.set_title("Pérdida del modelo")
ax2.set_xlabel("Época")
ax2.set_ylabel("Pérdida")
ax2.legend()
ax2.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

A continuación, evaluaremos la exactitud del modelo con el conjunto de prueba, es decir, el 20 % de las 200 imágenes que reservamos al principio. El rendimiento con este conjunto muestra qué tan bien puede generalizar el modelo a imágenes nuevas que no utilizó durante el entrenamiento.

# Evaluar el modelo con el conjunto de prueba
test_loss, test_accuracy = model.evaluate(
    X_test,
    y_test,
    verbose=0
)
print(f"Exactitud en el conjunto de prueba: {test_accuracy:.2%}")
print(f"Pérdida en el conjunto de prueba: {test_loss:.4f}")

# Obtener las predicciones
y_pred = np.argmax(model.predict(X_test), axis=1)

# Informe de clasificación
print("\nInforme de clasificación:")
print(
    classification_report(
        y_test,
        y_pred,
        target_names=label_encoder.classes_
    )
)

# Matriz de confusión
cm = confusion_matrix(y_test, y_pred)
plt.figure(figsize=(10, 8))
sns.heatmap(
    cm,
    annot=True,
    fmt="d",
    cmap="Blues",
    xticklabels=label_encoder.classes_,
    yticklabels=label_encoder.classes_
)
plt.title("Matriz de confusión")
plt.ylabel("Etiqueta verdadera")
plt.xlabel("Etiqueta predicha")
plt.tight_layout()
plt.show()

Por último, visualizaremos las clasificaciones del modelo en algunas imágenes del conjunto de prueba.

# Visualizar las predicciones para algunas imágenes de prueba
fig, axes = plt.subplots(3, 4, figsize=(14, 10))
indices = np.random.choice(
    len(X_test),
    size=min(12, len(X_test)),
    replace=False
)

for i, ax in enumerate(axes.flat):
    if i < len(indices):
        idx = indices[i]
        ax.imshow(X_test[idx])

        # Obtener las probabilidades de predicción
        pred_probs = model.predict(
            X_test[idx:idx + 1],
            verbose=0
        )[0]
        pred_class = np.argmax(pred_probs)
        confidence = pred_probs[pred_class]

        true_label = label_encoder.inverse_transform(
            [y_test[idx]]
        )[0]
        pred_label = label_encoder.inverse_transform(
            [pred_class]
        )[0]
        color = "green" if true_label == pred_label else "red"

        ax.set_title(
            f"Real: {true_label}\n"
            f"Predicción: {pred_label} ({confidence:.1%})",
            color=color,
            fontsize=9
        )
        ax.axis("off")

plt.tight_layout()
plt.show()