Entrada 7: Proyecto BermejaPi — Misión Científica: Estimación del Albedo Terrestre y Cobertura Nubosa

Tras superar el ensayo general de vuelo (Dry Run), el proyecto BermejaPi evoluciona de un simple programa de pruebas a una misión de observación terrestre con objetivo científico para la Agencia Espacial Europea (ESA).

En esta entrega explicamos qué es el albedo planetario, cómo procesar las imágenes capturadas desde la Estación Espacial Internacional (ISS) utilizando el espacio de color HSV y cómo integrar este algoritmo en main.py.

1. El Objetivo Científico: Medir el Albedo Terrestre

El albedo es la medida de la cantidad de radiación solar que refleja una superficie en comparación con la que recibe. Explicado para alumnos de secundaria o bachillerato:

• Efecto camiseta: Una camiseta negra al sol absorbe la luz y se calienta rápidamente (albedo bajo); una camiseta blanca la refleja y se mantiene fresca (albedo alto).
• Escudo planetario: El hielo, la nieve y las nubes reflejan más del 60-80% de la luz solar hacia el espacio. Los océanos y los bosques oscuros absorben más del 90%.
• Importancia climática: Medir el porcentaje de masa nubosa desde el espacio ayuda a calibrar los modelos de balance energético del planeta y a comprender el impacto del cambio climático.

2. Visión por Computador: ¿Por qué HSV en lugar de RGB?

Para estimar el albedo necesitamos detectar qué porcentaje de cada imagen tomada desde la ISS corresponde a nubes brillantes. Procesar imágenes en formato estándar RGB suele fallar debido a sombras y cambios de luz solar. Por eso convertimos cada fotograma al espacio de color HSV:

Modelo	Componentes	¿Por qué lo usamos?
RGB	Red (Rojo), Green (Verde), Blue (Azul)	Falla con facilidad en el espacio porque las variaciones de iluminación modifican los tres canales a la vez.
HSV	Hue (Tono/Color), Saturation (Saturación/Pureza), Value (Brillo/Valor)	Separa el color de la luz. Las nubes se identifican fácilmente por tener muy poco color (Saturación baja) y un brillo elevado (Valor alto).

3. Algoritmo Matemático de Filtrado

1. Conversión de espacio: Pasamos la imagen de BGR (formato nativo de OpenCV) a HSV.
2. Máscara binaria (Thresholding): Filtramos los píxeles cuyas coordenadas cumplan:
   $$\text{Saturación} \in [0, 50] \quad \text{y} \quad \text{Valor} \in [180, 255]$$
3. Cálculo de cobertura: Todos los píxeles que cumplen el criterio se convierten en blancos (nube, valor 255) y el resto en negros (terreno/océano, valor 0).

La fórmula del albedo estimado queda expresada como:

$$\text{Albedo}_{\text{estimado}} (\%) = \left( \frac{\text{Píxeles de Nubes}}{\text{Píxeles Totales}} \right) \times 100$$

4. Código Fuente Definitivo (main.py)

Esta versión incluye la función calcular_albedo_y_nubes(), la lectura de sensores del Sense HAT, el cálculo de velocidad orbital (ORB) y el borrado automático de imágenes antiguas para no superar la regla de las 42 fotos de la ESA:

Python

import csv
import time
from datetime import datetime, timedelta
from pathlib import Path
import cv2
import numpy as np

# Cargar módulo de cámara oficial
try:
    from picamera2 import Picamera2
except ImportError:
    Picamera2 = None

# Cargar Sense HAT (hardware real o emulador)
try:
    from sense_hat import SenseHat
    sense = SenseHat()
except Exception:
    try:
        from sense_emu import SenseHat
        sense = SenseHat()
    except Exception:
        sense = None

# ============================================
# CONSTANTES DE MISIÓN Y REGLAS ESA
# ============================================
DIR_BASE = Path(__file__).parent.resolve()
FICHERO_CSV = DIR_BASE / "data.csv"

DURACION_MINUTOS = 10
TIEMPO_LIMITE_SEG = DURACION_MINUTOS * 60  # 600 segundos
INTERVALO_CAPTURA_SEG = 15                  # 600s / 15s = 40 capturas max.

MAX_IMAGENES_PERMITIDAS = 42
LIMITE_ALMACENAMIENTO_MB = 240              # Margen de seguridad sobre los 250 MB

ALTURA_ISS_M = 400000.0                     # Altura orbital promedio (400 km)
ANCHO_SENSOR_MM = 7.564                     # HQ Camera
FOCAL_MM = 6.0
ANCHO_IMAGEN_PX = 4056


def inicializar_csv():
    """Crea el archivo data.csv con las cabeceras requeridas si no existe."""
    if not FICHERO_CSV.exists():
        with open(FICHERO_CSV, mode="w", newline="", encoding="utf-8") as f:
            escritor = csv.writer(f)
            escritor.writerow([
                "timestamp", 
                "temp_c", 
                "presion_hpa", 
                "humedad_pct", 
                "delta_t_s", 
                "desplazamiento_px", 
                "velocidad_kms",
                "albedo_pct"
            ])


def leer_telemetria_sensores():
    """Lee temperatura, presión y humedad del Sense HAT o genera valores simulados."""
    if sense is not None:
        try:
            t = sense.get_temperature()
            p = sense.get_pressure()
            h = sense.get_humidity()
            return round(t, 2), round(p, 2), round(h, 2)
        except Exception:
            pass
    return 21.5, 1013.25, 42.0


def capturar_imagen(ruta, camara=None):
    """Captura foto con la cámara o genera terreno sintético con nubes simuladas para VM."""
    if camara is not None:
        camara.capture_file(str(ruta))
    else:
        h, w = 1080, 1920
        # Terreno en BGR de 3 canales
        terreno_base = np.random.randint(60, 140, (h, w, 3), dtype=np.uint8)
        terreno = cv2.GaussianBlur(terreno_base, (31, 31), 0)
        
        # Simulación de nubes brillantes para validar el algoritmo
        cv2.circle(terreno, (600, 400), 180, (245, 245, 245), -1)
        cv2.circle(terreno, (1200, 700), 250, (250, 250, 250), -1)
        
        cv2.imwrite(str(ruta), terreno)


def calcular_albedo_y_nubes(ruta_imagen):
    """Calcula el porcentaje de cobertura nubosa (albedo estimado) usando el espacio HSV."""
    img = cv2.imread(str(ruta_imagen))
    if img is None:
        return 0.0

    # 1. Convertir a espacio de color HSV
    hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)

    # 2. Definir rango para nubes (baja saturación, alto brillo/valor)
    umbral_inferior = np.array([0, 0, 180])
    umbral_superior = np.array([180, 50, 255])

    # 3. Crear máscara binaria (Blanco = Nube, Negro = Terreno/Océano)
    mascara_nubes = cv2.inRange(hsv, umbral_inferior, umbral_superior)

    # 4. Calcular proporción de píxeles
    pixeles_nubes = cv2.countNonZero(mascara_nubes)
    pixeles_totales = img.shape[0] * img.shape[1]

    porcentaje_albedo = (pixeles_nubes / pixeles_totales) * 100.0
    return round(porcentaje_albedo, 2)


def calcular_velocidad(ruta1, ruta2, delta_t):
    """Calcula desplazamiento y velocidad orbital mediante ORB de OpenCV."""
    img1 = cv2.imread(str(ruta1), cv2.IMREAD_GRAYSCALE)
    img2 = cv2.imread(str(ruta2), cv2.IMREAD_GRAYSCALE)

    if img1 is None or img2 is None:
        return 0.0, 0.0

    orb = cv2.ORB_create(nfeatures=1000)
    kp1, des1 = orb.detectAndCompute(img1, None)
    kp2, des2 = orb.detectAndCompute(img2, None)

    if des1 is None or des2 is None:
        return 0.0, 0.0

    bf = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=True)
    coincidencias = bf.match(des1, des2)
    
    if not coincidencias:
        return 0.0, 0.0

    coincidencias = sorted(coincidencias, key=lambda x: x.distance)[:50]

    distancias_px = [
        np.sqrt((kp2[m.trainIdx].pt[0] - kp1[m.queryIdx].pt[0])**2 + 
                (kp2[m.trainIdx].pt[1] - kp1[m.queryIdx].pt[1])**2)
        for m in coincidencias
    ]

    desplazamiento_px = float(np.mean(distancias_px))
    gsd = (ALTURA_ISS_M * ANCHO_SENSOR_MM) / (FOCAL_MM * ANCHO_IMAGEN_PX)
    distancia_m = desplazamiento_px * gsd
    velocidad_kms = (distancia_m / delta_t) / 1000.0 if delta_t > 0 else 0.0

    return desplazamiento_px, velocidad_kms


def verificar_y_purgar_almacenamiento():
    """Garantiza no superar 42 fotos ni el límite de espacio asignado por la ESA."""
    imagenes_guardadas = sorted(list(DIR_BASE.glob("foto_*.jpg")))
    
    while len(imagenes_guardadas) > MAX_IMAGENES_PERMITIDAS:
        foto_antigua = imagenes_guardadas.pop(0)
        try:
            foto_antigua.unlink()
            print(f"[Purga ESA] Eliminada imagen antigua: {foto_antigua.name}")
        except Exception as e:
            print(f"Error al eliminar {foto_antigua.name}: {e}")

    tamano_total_bytes = sum(f.stat().st_size for f in DIR_BASE.glob("*") if f.is_file())
    tamano_total_mb = tamano_total_bytes / (1024 * 1024)

    if tamano_total_mb > LIMITE_ALMACENAMIENTO_MB and imagenes_guardadas:
        foto_a_borrar = imagenes_guardadas.pop(0)
        try:
            foto_a_borrar.unlink()
            print(f"[Alerta Almacenamiento] Eliminada por espacio: {foto_a_borrar.name}")
        except Exception as e:
            print(f"Error al liberar espacio: {e}")


def registrar_telemetria(temp, presion, humedad, delta_t, px, v_kms, albedo_pct):
    """Guarda la fila de datos completa en data.csv."""
    ahora = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with open(FICHERO_CSV, mode="a", newline="", encoding="utf-8") as f:
        escritor = csv.writer(f)
        escritor.writerow([
            ahora, 
            temp, 
            presion, 
            humedad, 
            round(delta_t, 2), 
            round(px, 2), 
            round(v_kms, 2),
            albedo_pct
        ])


def bucle_principal_mision():
    print("=== INICIANDO MISIÓN BERMEJAPI (ANÁLISIS DE ALBEDO Y VELOCIDAD) ===")
    inicializar_csv()

    camara = None
    if Picamera2 is not None:
        try:
            camara = Picamera2()
            camara.start()
        except Exception as e:
            print(f"Modo simulación cámara activo: {e}")

    hora_inicio = datetime.now()
    hora_fin = hora_inicio + timedelta(seconds=TIEMPO_LIMITE_SEG - 10)

    contador_fotos = 1
    foto_anterior = DIR_BASE / f"foto_{contador_fotos:02d}.jpg"
    
    capturar_imagen(foto_anterior, camara)
    tiempo_anterior = time.time()

    ciclo = 1
    while datetime.now() < hora_fin:
        try:
            print(f"\n[Ciclo {ciclo}] Procesando telemetría, albedo y velocidad...")
            time.sleep(INTERVALO_CAPTURA_SEG)

            contador_fotos += 1
            foto_actual = DIR_BASE / f"foto_{contador_fotos:02d}.jpg"

            capturar_imagen(foto_actual, camara)
            tiempo_actual = time.time()
            delta_t = tiempo_actual - tiempo_anterior

            # Telemetría del Sense HAT
            temp, presion, humedad = leer_telemetria_sensores()

            # Visión por computador: Velocidad (ORB) y Albedo (HSV)
            px, v_kms = calcular_velocidad(foto_anterior, foto_actual, delta_t)
            albedo = calcular_albedo_y_nubes(foto_actual)

            print(f"Sensores -> T: {temp}°C | P: {presion} hPa | H: {humedad}%")
            print(f"Visión   -> Shift: {px:.1f}px | V: {v_kms:.2f} km/s | Albedo: {albedo}%")

            # Registro CSV
            registrar_telemetria(temp, presion, humedad, delta_t, px, v_kms, albedo)

            # Purga de almacenamiento
            verificar_y_purgar_almacenamiento()

            foto_anterior = foto_actual
            tiempo_anterior = tiempo_actual
            ciclo += 1

        except Exception as e:
            print(f"Excepción controlada en ciclo {ciclo}: {e}")
            time.sleep(2)

    if camara is not None:
        camara.stop()

    verificar_y_purgar_almacenamiento()
    print("\n=== PRUEBA DE MISIÓN COMPLETADA CORRECTAMENTE ===")


if __name__ == "__main__":
    bucle_principal_mision()

5. Verificación de Resultados

Borra cualquier archivo data.csv antiguo y ejecuta el programa en la consola de Raspberry Pi OS:

Bash

rm -f data.csv
python3 main.py

Comprueba los datos generados abriendo el archivo de registros:

Bash

cat data.csv

La última columna reflejará el porcentaje de albedo simulado en cada iteración, consolidando la telemetría ambiental, la dinámica orbital y el análisis fotométrico en un único informe CSV.