Entrada 6: Proyecto BermejaPi — Simulación de Vuelo (Dry Run), Sense HAT y Verificación de Reglas ESA

En esta entrega ejecutamos un ensayo de vuelo completo (Dry Run) para auditar el comportamiento del script de vuelo main.py bajo las restricciones oficiales de la Agencia Espacial Europea (ESA) para la competición Astro Pi Mission Space Lab. Esta prueba intermedia permite validar la estabilidad del entorno, el control de almacenamiento y la lectura de sensores antes de abordar la fase de análisis de datos e investigación avanzada.

1. Propósito del Ensayo General (Dry Run)

Antes de considerar la entrega final, la ESA exige comprobar que el software es capaz de operar de manera 100% autónoma en la Estación Espacial Internacional (ISS) sin intervención humana ni fallos del sistema. Los objetivos de este simulacro son:

• Validar la resiliencia: Garantizar que el programa capture excepciones sin detenerse ante fallos de hardware simulados.
• Comprobar la purga automática: Verificar que el almacenamiento acumulado no supere en ningún momento las 42 imágenes ni la cuota de 250 MB.
• Registrar telemetría ambiental: Incorporar datos del Sense HAT (temperatura, presión y humedad) al registro .csv junto con los datos de velocidad orbital.

2. Código Fuente para la Prueba de Vuelo (main.py)

A continuación se presenta el código ajustado con la lógica de purga de imágenes, control estricto de tiempo (10 minutos) y gestión de fallbacks para hardware no presente:

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"
            ])


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 para VM."""
    if camara is not None:
        camara.capture_file(str(ruta))
    else:
        h, w = 1080, 1920
        terreno = np.random.randint(80, 180, (h, w), dtype=np.uint8)
        terreno = cv2.GaussianBlur(terreno, (31, 31), 0)
        cv2.circle(terreno, (600, 400), 80, (230), -1)
        cv2.imwrite(str(ruta), terreno)


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")))
    
    # 1. Purga por número máximo de imágenes (Regla ESA <= 42)
    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}")

    # 2. Control de espacio total en megabytes
    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):
    """Guarda la fila de datos 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)
        ])


def bucle_principal_mision():
    print("=== INICIANDO ENSAYO DE VUELO (DRY RUN - REGLAS ESA) ===")
    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()
    # Detener el programa 10 segundos antes del límite de 600 segundos
    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}] Leyendo sensores y analizando terreno...")
            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

            # Lectura de sensores
            temp, presion, humedad = leer_telemetria_sensores()

            # Cálculo de visión por computador
            px, v_kms = calcular_velocidad(foto_anterior, foto_actual, delta_t)

            print(f"Sensores -> T: {temp}°C | P: {presion} hPa | H: {humedad}%")
            print(f"Orbital  -> Delta t: {delta_t:.1f}s | Shift: {px:.1f}px | V: {v_kms:.2f} km/s")

            # Registro de datos
            registrar_telemetria(temp, presion, humedad, delta_t, px, v_kms)

            # Control de purga y almacenamiento según reglas ESA
            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=== ENSAYO DE VUELO COMPLETADO CORRECTAMENTE ===")


if __name__ == "__main__":
    bucle_principal_mision()

3. Lista de Comprobación para el Envío de Prueba

Antes de empaquetar la versión preliminar, verifica los siguientes puntos técnicos en la máquina virtual:

• Estructura plana del .zip: Al comprimir, main.py debe quedar situado en la raíz del paquete, sin subcarpetas contenedoras.
• Limpieza de archivos residuales: Elimina las imágenes foto_*.jpg de prueba y el archivo data.csv antes de generar el paquete final.
• Comando de empaquetado:
  Bashzip -r bermejapi_dryrun.zip main.py
• Comprobación de tamaño: Asegúrate de que el archivo .zip tenga un tamaño inferior a 3 MB.

Extracto (Resumen)

En esta sexta entrega del proyecto BermejaPi, ejecutamos una prueba de ensayo general (Dry Run) para validar el cumplimiento de la normativa oficial de la ESA en el concurso Astro Pi Mission Space Lab. Implementamos un sistema de lectura ambiental con Sense HAT, procesamiento ORB con OpenCV, parada automática antes de los 10 minutos y un algoritmo dinámico de purga de imágenes para garantizar un límite máximo de 42 archivos y 250 MB en disco, dejando el software listo para posteriores fases de experimentación.