Proyecto BermejaPi — Entrada 16: main.py definitivo para Astro Pi, del simulador a la ISS
En las entradas 11 a 15 construimos, módulo a módulo, el diagnóstico de un planeta desconocido. Pero eran scripts sueltos: no había un programa único que se ejecutara solo, respetara el límite de 10 minutos y guardara medidas reales. En esta entrada juntamos todo en el main.py que enviaremos a la ESA.
Idea clave: la Tierra es nuestro planeta de calibración. Como conocemos las respuestas correctas, podemos comprobar si el método funciona antes de aplicarlo a un mundo nuevo.
1. Qué cambia respecto a las entradas 5–15
| Antes | Ahora |
|---|---|
| Si fallaba el sensor, se inventaban valores (18,45 / −22,10 / 35,80 µT) sin avisar | En vuelo la celda queda vacía con una marca en flags. La simulación solo existe en modo desarrollo y queda marcada (sim_camera, emu_sense) en cada fila y en summary.json |
| El informe final (entrada 15) usaba un diccionario escrito a mano | summary.json se calcula con las medidas reales del vuelo |
Rutas relativas (Path("data.csv")) | Todo cuelga de Path(__file__).parent.resolve() |
| Una sola lectura del magnetómetro | 3 lecturas por ciclo, con posición ISS (latitud/longitud) para poder buscar la SAA en tierra |
| Velocidad = media de las 50 mejores coincidencias | Mediana con filtro de valores atípicos (MAD) y rango físico 5–9 km/s |
| VARI (cámara RGB visible) | BNDVI (cámara NoIR con filtro azul: el canal rojo es infrarrojo cercano) |
| Fotos nocturnas guardadas (negras) | Se detectan y se borran; el programa sigue midiendo |
| Temperatura llamada “ambiente” | Se llama sensor_temp_c: es la del Sense HAT dentro de la estación, no la del planeta |
| Sin cámara ni Sense HAT no hacía nada útil en tu máquina virtual | Modo desarrollo (ALLOW_SIMULATION = True): usa el emulador sense_emu y una cámara sintética, como en las entradas 4–9 |
| Comentarios y mensajes en español | En inglés (el reglamento exige que todo el envío esté en inglés) |
2. Checklist del reglamento 2026/27
| Requisito | Cómo lo cumple |
|---|---|
Fichero main.py, Python 3.13 | Sí. Se ejecuta con python3 main.py |
| Para antes de 10 minutos | Reloj time.monotonic() desde la primera línea. No empieza ciclos después de 530 s; margen de ~70 s |
| Máx. 42 imágenes al final | Se conservan como máximo 30 (MAX_IMAGES_KEPT) |
| Máx. 250 MB | Control adicional de 200 MB (MAX_STORAGE_MB) |
Solo guarda en la carpeta de main.py | Path(__file__).parent.resolve(); sin carpetas nuevas ni rutas absolutas |
| Nombres de fichero válidos | data.csv, summary.json, mission.log, img_001.jpg… |
| Tipos de salida permitidos | .csv, .json, .log, .jpg |
| Sin red, procesos ni hilos | No se usa subprocess, socket, threading |
| Sin excepciones sin capturar | Cada etapa está protegida; el finally cierra la cámara y escribe el resumen |
| Usa Sense HAT o cámara | Usa ambos |
| Sin librerías fuera de la guía | sense_emu solo se importa con ALLOW_SIMULATION = True; en la versión final está desactivado |
| Zip ≤ 3 MB | main.py + README.txt ocupan unos 8 KB |
3. Modo desarrollo y modo vuelo
| Situación | Cámara | Sense HAT | Filas marcadas |
|---|---|---|---|
| ISS y Replay Tool | Real | Real | (sin marca) |
| Máquina virtual de Raspberry Pi OS | Sintética | Emulador sense_emu | sim_camera, emu_sense |
Versión final (ALLOW_SIMULATION = False) | Real o nada | Real o nada | no_camera, no_sense |
Antes de crear el zip, cambia ALLOW_SIMULATION a False. Así, si en la ISS fallara algo, el programa nunca guardaría datos de mentira.
3b. Cómo funciona un ciclo (cada 15 s)
- Lee la posición de la ISS (
orbit) y el Sense HAT (temperatura, presión, humedad, magnetómetro ×3). - Hace una foto y la reduce a 1014 px de ancho para analizarla rápido.
- Si la foto es casi negra (lado nocturno), la marca como
night, la borra y sigue. - Calcula nubes (HSV) y BNDVI (vegetación) sobre los píxeles que no son nube.
- Compara con la foto anterior con ORB y calcula la velocidad (con la altitud real de la ISS).
- Escribe una fila en
data.csv, purga el almacenamiento y espera al siguiente ciclo.
Al terminar, summary.json incluye la velocidad mediana, la nubosidad, el rango del campo magnético (mín, máx, cociente máx/mín y dónde fue más débil) y, a partir de la velocidad, la masa, la gravedad y la velocidad de escape del planeta de calibración.
4. Columnas de data.csv
| Columna | Descripción |
|---|---|
timestamp_utc, elapsed_s | Hora UTC y segundos desde el inicio |
lat_deg, lon_deg, alt_km | Posición de la ISS |
sensor_temp_c, pressure_hpa, humidity_pct | Sense HAT (interior de la estación) |
mag_x_ut, mag_y_ut, mag_z_ut, mag_total_ut | Campo magnético (µT) |
image, brightness | Foto conservada y brillo medio |
dt_s, shift_px, inliers, speed_kms | Medición de velocidad |
cloud_pct, bndvi_mean, veg_pct | Nubes y señal de vegetación |
flags | night, no_speed, bad_speed, capture_error, mag_error, no_sense, no_camera… |
5. Código fuente (main.py)
"""
BermejaPi - Astro Pi Mission Space Lab 2026/27
==============================================
Experiment
----------
The ISS flies over a planet whose properties we treat as *unknown*. Using only
what the Astro Pi can measure in a single 10-minute pass, the program tries to
characterise that planet. Earth is used as a "calibration planet": because we
know the true answers, we can check whether the method works.
What is measured (every CYCLE_S seconds)
* ISS position (latitude, longitude, altitude) -> `orbit` library
* Sense HAT: temperature, pressure, humidity, magnetometer (3 axes)
* One photo from the Astro Pi camera, from which we compute
- ground speed of the ISS (ORB feature matching between photos)
- cloud coverage (HSV threshold)
- vegetation signal with BNDVI (NoIR camera + blue filter:
the red channel records near-infrared)
Outputs (all written next to this file, as required by the rulebook)
data.csv one row per cycle (raw measurements, no simulated values)
summary.json statistics and derived planet properties
mission.log events and errors
img_XXX.jpg at most MAX_IMAGES_KEPT photos are kept
Development mode (ALLOW_SIMULATION)
On a computer without the real hardware (for example the Raspberry Pi OS
virtual machine) the program falls back to the Sense HAT emulator
(sense_emu) and to a synthetic camera that produces a moving landscape.
Every affected row is flagged ("emu_sense" / "sim_camera") and
summary.json states the data source. On the ISS and in the Astro Pi
Replay Tool the real hardware exists, so the fallback is never used.
Set ALLOW_SIMULATION = False before creating the final zip.
Design rules (see the Mission Space Lab rulebook checklist)
* Stops well before the 10-minute limit (see SAFE_STOP_S).
* Never raises an unhandled exception: every stage is protected.
* If a sensor or the camera is not available (and simulation is not
allowed) the row is written with empty values and a flag.
Values are never silently invented.
* Only libraries from the Mission Space Lab creator guide are used.
* No threads, no networking, no subprocesses, no absolute paths.
"""
import time
# The clock starts here, before the heavy imports, so that the total run time
# (including start-up) is always the one we monitor.
START = time.monotonic()
import csv
import json
import logging
import math
import sys
from datetime import datetime, timezone
from pathlib import Path
import cv2
import numpy as np
# --------------------------------------------------------------------------
# Optional hardware / mission libraries. If one is missing the program keeps
# running and flags the missing data instead of failing.
# --------------------------------------------------------------------------
try:
from picamera2 import Picamera2
except Exception: # ImportError or any environment problem
Picamera2 = None
try:
from sense_hat import SenseHat
except Exception:
SenseHat = None
try:
from orbit import ISS
except Exception:
ISS = None
# --------------------------------------------------------------------------
# Development mode. Set to False for the version that is submitted to ESA.
# --------------------------------------------------------------------------
ALLOW_SIMULATION = True
EmulatedSenseHat = None
if ALLOW_SIMULATION:
try:
from sense_emu import SenseHat as EmulatedSenseHat # Raspberry Pi OS emulator
except Exception:
EmulatedSenseHat = None
# Where the data of this run really comes from (filled in by init_*)
SOURCES = {"camera": "none", "sense_hat": "none"}
# --------------------------------------------------------------------------
# Paths (rulebook: only the folder of main.py, built from __file__)
# --------------------------------------------------------------------------
BASE_DIR = Path(__file__).parent.resolve()
CSV_PATH = BASE_DIR / "data.csv"
SUMMARY_PATH = BASE_DIR / "summary.json"
LOG_PATH = BASE_DIR / "mission.log"
# --------------------------------------------------------------------------
# Timing (seconds since START)
# --------------------------------------------------------------------------
TIME_LIMIT_S = 600 # hard rulebook limit (never reached on purpose)
LAST_CYCLE_START_S = 530 # do not start a new cycle after this moment
SAFE_STOP_S = 560 # abort any remaining work after this moment
CYCLE_S = 15 # target time between two photos
# --------------------------------------------------------------------------
# Storage (rulebook: <= 42 images kept at the end, <= 250 MB in total)
# --------------------------------------------------------------------------
MAX_IMAGES_KEPT = 30
MAX_STORAGE_MB = 200
# --------------------------------------------------------------------------
# Camera geometry (Raspberry Pi HQ Camera, Astro Pi lens).
# Verify these values against the creator guide before submitting.
# GSD [m/px] = altitude_m * sensor_width_mm / (focal_length_mm * image_width_px)
# --------------------------------------------------------------------------
SENSOR_WIDTH_MM = 6.287
FOCAL_LENGTH_MM = 5.0
DEFAULT_ALTITUDE_KM = 420.0 # used only if the `orbit` library is unavailable
# --------------------------------------------------------------------------
# Image analysis (calibrate these thresholds with the Astro Pi Replay Tool)
# --------------------------------------------------------------------------
ANALYSIS_WIDTH_PX = 1014 # photos are downscaled to this width for speed
VALID_MIN_V = 25 # HSV value below this = too dark to analyse
MIN_VALID_FRACTION = 0.10 # less lit pixels than this -> "night" image
CLOUD_MAX_S = 50 # cloud: low saturation ...
CLOUD_MIN_V = 180 # ... and high brightness
BNDVI_VEG_THRESHOLD = 0.15 # BNDVI above this counts as vegetation signal
# --------------------------------------------------------------------------
# Speed estimation (ORB feature matching between consecutive photos)
# --------------------------------------------------------------------------
ORB_FEATURES = 1500
MAX_MATCHES_USED = 300
MIN_INLIERS = 10
SPEED_MIN_KMS = 5.0 # physically plausible range for the ISS
SPEED_MAX_KMS = 9.0
# --------------------------------------------------------------------------
# Magnetometer
# --------------------------------------------------------------------------
MAG_SAMPLES = 3 # readings averaged in each cycle
# --------------------------------------------------------------------------
# Synthetic camera (development mode only)
# --------------------------------------------------------------------------
SIM_WIDTH_PX = 2028
SIM_HEIGHT_PX = 1520
SIM_SPEED_KMS = 7.3 # ground speed the fake landscape moves at
# --------------------------------------------------------------------------
# Physical constants and modelling assumptions used in summary.json
# --------------------------------------------------------------------------
G = 6.67430e-11 # gravitational constant [m^3 kg^-1 s^-2]
SIGMA = 5.670374e-8 # Stefan-Boltzmann constant [W m^-2 K^-4]
SOLAR_CONSTANT = 1361.0 # [W m^-2]
ASSUMED_RADIUS_KM = 6371.0 # radius of the calibration planet (an input)
ALBEDO_CLOUD = 0.70 # assumed reflectivity of clouds
ALBEDO_SURFACE = 0.15 # assumed reflectivity of land/ocean
CSV_COLUMNS = [
"timestamp_utc", "elapsed_s", "lat_deg", "lon_deg", "alt_km",
"sensor_temp_c", "pressure_hpa", "humidity_pct",
"mag_x_ut", "mag_y_ut", "mag_z_ut", "mag_total_ut",
"image", "brightness", "dt_s", "shift_px", "inliers", "speed_kms",
"cloud_pct", "bndvi_mean", "veg_pct", "flags",
]
def elapsed():
"""Seconds since the program started."""
return time.monotonic() - START
# --------------------------------------------------------------------------
# Small helpers
# --------------------------------------------------------------------------
def clean(value, digits=3):
"""Round numbers; turn None / NaN / inf into None so they are stored empty."""
if value is None:
return None
try:
value = float(value)
except (TypeError, ValueError):
return None
if math.isnan(value) or math.isinf(value):
return None
return round(value, digits)
def median_of(values):
values = [v for v in values if v is not None]
return float(np.median(values)) if values else None
# --------------------------------------------------------------------------
# Hardware initialisation
# --------------------------------------------------------------------------
class SimulatedCamera:
"""Development-only camera: a synthetic landscape scrolling at SIM_SPEED_KMS.
Ocean (blue), land with a strong red/NIR response and clouds are drawn on a
tileable texture. Each capture moves the view according to the real time
elapsed since the previous capture, so the speed estimation can be tested.
"""
def __init__(self):
rng = np.random.default_rng(7)
self.width, self.height = SIM_WIDTH_PX, SIM_HEIGHT_PX
self.period = 2 * self.width
small = rng.integers(0, 255, (self.height // 8, self.period // 8, 3),
dtype=np.uint8)
base = cv2.resize(small, (self.period, self.height),
interpolation=cv2.INTER_CUBIC)
base = (cv2.GaussianBlur(base, (0, 0), 2) * 0.4).astype(np.uint8)
base[:, : self.period // 3, 0] = np.clip(
base[:, : self.period // 3, 0].astype(int) + 120, 0, 255)
base[200:900, self.period // 3: self.period // 2, 2] = np.clip(
base[200:900, self.period // 3: self.period // 2, 2].astype(int)
+ 150, 0, 255)
for cx, cy, radius in ((700, 500, 160), (2400, 900, 220),
(3600, 400, 180)):
cv2.circle(base, (cx, cy), radius, (235, 235, 235), -1)
self.base = cv2.GaussianBlur(base, (0, 0), 1.2)
self.gsd_m = (DEFAULT_ALTITUDE_KM * 1000.0 * SENSOR_WIDTH_MM /
(FOCAL_LENGTH_MM * self.width))
self.offset = 0.0
self.last = None
def capture_file(self, path):
now = time.monotonic()
if self.last is not None:
self.offset += SIM_SPEED_KMS * 1000.0 * (now - self.last) / self.gsd_m
self.last = now
columns = (int(self.offset) + np.arange(self.width)) % self.period
if not cv2.imwrite(str(path), self.base[:, columns]):
raise OSError("could not write synthetic image")
def stop(self):
pass
def close(self):
pass
def init_sense():
"""Return a SenseHat object (real, emulated or None) and record its source."""
if SenseHat is not None:
try:
sense = SenseHat()
SOURCES["sense_hat"] = "real"
return sense
except Exception as exc:
logging.warning("Sense HAT could not be initialised: %s", exc)
else:
logging.warning("sense_hat library not available")
if EmulatedSenseHat is not None:
try:
sense = EmulatedSenseHat()
SOURCES["sense_hat"] = "emulator"
logging.warning("Using the Sense HAT EMULATOR (development mode)")
return sense
except Exception as exc:
logging.warning("Sense HAT emulator failed: %s", exc)
return None
def init_camera():
"""Return a started camera (real, synthetic or None) and record its source."""
if Picamera2 is not None:
try:
camera = Picamera2()
camera.configure(camera.create_still_configuration())
try:
camera.options["quality"] = 85 # keeps JPEG files small
except Exception:
pass
camera.start()
time.sleep(1.0) # let exposure settle
SOURCES["camera"] = "real"
return camera
except Exception as exc:
logging.warning("Camera could not be initialised: %s", exc)
else:
logging.warning("picamera2 library not available")
if ALLOW_SIMULATION:
SOURCES["camera"] = "simulated"
logging.warning("Using the SYNTHETIC camera (development mode)")
return SimulatedCamera()
return None
# --------------------------------------------------------------------------
# Measurements
# --------------------------------------------------------------------------
def read_position():
"""Return (lat_deg, lon_deg, alt_km) from the orbit library or None."""
if ISS is None:
return None
try:
point = ISS.coordinates()
return (point.latitude.degrees,
point.longitude.degrees,
point.elevation.km)
except Exception as exc:
logging.warning("ISS position unavailable: %s", exc)
return None
def read_sensors(sense):
"""Read the Sense HAT. Missing values stay None (never simulated)."""
data = {"temp": None, "pressure": None, "humidity": None,
"mx": None, "my": None, "mz": None, "mtotal": None}
if sense is None:
return data
for key, getter in (("temp", "get_temperature"),
("pressure", "get_pressure"),
("humidity", "get_humidity")):
try:
data[key] = float(getattr(sense, getter)())
except Exception as exc:
logging.warning("Sense HAT %s failed: %s", getter, exc)
samples = []
for _ in range(MAG_SAMPLES):
try:
raw = sense.get_compass_raw()
samples.append((float(raw["x"]), float(raw["y"]), float(raw["z"])))
except Exception as exc:
logging.warning("Magnetometer read failed: %s", exc)
break
time.sleep(0.05)
if samples:
arr = np.array(samples)
data["mx"], data["my"], data["mz"] = (float(v) for v in arr.mean(axis=0))
data["mtotal"] = float(np.linalg.norm(arr, axis=1).mean())
return data
def load_image(path):
"""Read a photo and return (downscaled BGR image, scale, original width)."""
image = cv2.imread(str(path), cv2.IMREAD_COLOR)
if image is None:
return None, 1.0, 0
height, width = image.shape[:2]
if width > ANALYSIS_WIDTH_PX:
scale = width / ANALYSIS_WIDTH_PX
size = (ANALYSIS_WIDTH_PX, int(round(height / scale)))
small = cv2.resize(image, size, interpolation=cv2.INTER_AREA)
else:
scale = 1.0
small = image
return small, scale, width
def analyse_colour(small):
"""Cloud coverage (HSV) and vegetation signal (BNDVI) of one photo.
Returns None if the photo is (almost) dark, i.e. the ISS is in the
night side of the orbit.
NoIR camera with blue filter: channel 0 (B) is blue and channel 2 (R)
records near-infrared (NIR), so BNDVI = (NIR - Blue) / (NIR + Blue).
"""
hsv = cv2.cvtColor(small, cv2.COLOR_BGR2HSV)
saturation = hsv[:, :, 1]
value = hsv[:, :, 2]
valid = value >= VALID_MIN_V
n_valid = int(np.count_nonzero(valid))
if n_valid < MIN_VALID_FRACTION * valid.size:
return None
cloud = valid & (saturation <= CLOUD_MAX_S) & (value >= CLOUD_MIN_V)
result = {
"brightness": float(value[valid].mean()) / 255.0,
"cloud_pct": 100.0 * np.count_nonzero(cloud) / n_valid,
"bndvi_mean": None,
"veg_pct": None,
}
blue = small[:, :, 0].astype(np.float32)
nir = small[:, :, 2].astype(np.float32)
denominator = nir + blue
usable = valid & ~cloud & (denominator > 0)
n_usable = int(np.count_nonzero(usable))
if n_usable > 0:
bndvi = (nir[usable] - blue[usable]) / denominator[usable]
result["bndvi_mean"] = float(bndvi.mean())
result["veg_pct"] = 100.0 * np.count_nonzero(
bndvi > BNDVI_VEG_THRESHOLD) / n_usable
return result
def compute_features(orb, small):
"""ORB keypoints and descriptors of a downscaled photo."""
gray = cv2.cvtColor(small, cv2.COLOR_BGR2GRAY)
keypoints, descriptors = orb.detectAndCompute(gray, None)
return keypoints, descriptors
def estimate_shift(matcher, prev_features, features):
"""Median displacement (px, in analysis resolution) between two photos.
Uses the best matches, then removes outliers with a median/MAD filter.
Returns (shift_px, n_inliers) or (None, 0) if not reliable.
"""
kp_prev, des_prev = prev_features
kp_now, des_now = features
if des_prev is None or des_now is None:
return None, 0
if len(kp_prev) < MIN_INLIERS or len(kp_now) < MIN_INLIERS:
return None, 0
matches = matcher.match(des_prev, des_now)
if len(matches) < MIN_INLIERS:
return None, 0
matches = sorted(matches, key=lambda m: m.distance)[:MAX_MATCHES_USED]
dx = np.array([kp_now[m.trainIdx].pt[0] - kp_prev[m.queryIdx].pt[0]
for m in matches])
dy = np.array([kp_now[m.trainIdx].pt[1] - kp_prev[m.queryIdx].pt[1]
for m in matches])
mdx, mdy = np.median(dx), np.median(dy)
deviation = np.hypot(dx - mdx, dy - mdy)
limit = max(3.0, 3.0 * float(np.median(deviation)))
inliers = deviation <= limit
n_inliers = int(np.count_nonzero(inliers))
if n_inliers < MIN_INLIERS:
return None, n_inliers
shift = math.hypot(float(np.median(dx[inliers])),
float(np.median(dy[inliers])))
return shift, n_inliers
# --------------------------------------------------------------------------
# Storage control
# --------------------------------------------------------------------------
def purge_storage():
"""Keep at most MAX_IMAGES_KEPT photos and MAX_STORAGE_MB in total."""
try:
images = sorted(BASE_DIR.glob("img_*.jpg"))
while len(images) > MAX_IMAGES_KEPT:
images.pop(0).unlink()
def total_mb():
return sum(f.stat().st_size for f in BASE_DIR.iterdir()
if f.is_file()) / (1024 * 1024)
while images and total_mb() > MAX_STORAGE_MB:
images.pop(0).unlink()
except Exception as exc:
logging.warning("Storage purge failed: %s", exc)
# --------------------------------------------------------------------------
# Summary
# --------------------------------------------------------------------------
def derive_planet_properties(speed_kms, altitude_km, cloud_pct):
"""Physical properties deduced from the measured speed and cloud cover."""
if speed_kms is None:
return None
r_orbit = (ASSUMED_RADIUS_KM + altitude_km) * 1000.0
r_surface = ASSUMED_RADIUS_KM * 1000.0
v = speed_kms * 1000.0
mass = v * v * r_orbit / G # centripetal = gravity
gravity = G * mass / r_surface ** 2
escape = math.sqrt(2.0 * G * mass / r_surface) / 1000.0
properties = {
"mass_kg": mass,
"surface_gravity_ms2": gravity,
"escape_velocity_kms": escape,
}
if cloud_pct is not None:
fraction = cloud_pct / 100.0
albedo = fraction * ALBEDO_CLOUD + (1.0 - fraction) * ALBEDO_SURFACE
t_eq = (SOLAR_CONSTANT * (1.0 - albedo) / (4.0 * SIGMA)) ** 0.25
properties["albedo_estimate"] = albedo
properties["equilibrium_temp_c"] = t_eq - 273.15
return properties
def build_summary(rows):
"""Statistics over all rows plus the derived planet properties."""
def column(name):
return [r[name] for r in rows if r.get(name) is not None]
speeds = column("speed_kms")
altitudes = column("alt_km")
clouds = column("cloud_pct")
mags = column("mag_total_ut")
speed = median_of(speeds)
altitude = median_of(altitudes)
if altitude is None:
altitude = DEFAULT_ALTITUDE_KM
cloud = median_of(clouds)
summary = {
"data_source": dict(SOURCES),
"program_runtime_s": clean(elapsed(), 1),
"cycles": len(rows),
"photos_analysed": len(column("brightness")),
"speed_samples": len(speeds),
"speed_median_kms": clean(speed),
"altitude_median_km": clean(altitude),
"cloud_pct_median": clean(cloud),
"bndvi_mean_median": clean(median_of(column("bndvi_mean"))),
"veg_pct_median": clean(median_of(column("veg_pct"))),
"magnetic_field": None,
"calibration_planet": None,
"assumptions": {
"radius_km": ASSUMED_RADIUS_KM,
"albedo_cloud": ALBEDO_CLOUD,
"albedo_surface": ALBEDO_SURFACE,
"solar_constant_wm2": SOLAR_CONSTANT,
},
}
if mags:
low, high = min(mags), max(mags)
summary["magnetic_field"] = {
"samples": len(mags),
"min_ut": clean(low),
"max_ut": clean(high),
"mean_ut": clean(float(np.mean(mags))),
"std_ut": clean(float(np.std(mags))),
"max_over_min": clean(high / low) if low > 0 else None,
}
# Where was the field weakest? (candidate South Atlantic Anomaly)
weakest = min((r for r in rows if r.get("mag_total_ut") is not None),
key=lambda r: r["mag_total_ut"])
summary["magnetic_field"]["weakest_at"] = {
"lat_deg": clean(weakest.get("lat_deg")),
"lon_deg": clean(weakest.get("lon_deg")),
}
planet = derive_planet_properties(speed, altitude, cloud)
if planet is not None:
summary["calibration_planet"] = {
k: clean(v, 4) if k != "mass_kg" else float(f"{v:.4e}")
for k, v in planet.items()
}
return summary
def write_summary(rows):
try:
with open(SUMMARY_PATH, "w", encoding="utf-8") as handle:
json.dump(build_summary(rows), handle, indent=2)
except Exception as exc:
logging.error("Could not write summary: %s", exc)
# --------------------------------------------------------------------------
# One measurement cycle
# --------------------------------------------------------------------------
def run_cycle(index, sense, camera, orb, matcher, state):
"""Measure everything once. Returns the row as a dictionary."""
flags = []
row = {name: None for name in CSV_COLUMNS}
row["timestamp_utc"] = datetime.now(timezone.utc).strftime(
"%Y-%m-%dT%H:%M:%SZ")
row["elapsed_s"] = elapsed()
# ---- position -------------------------------------------------------
position = read_position()
if position is not None:
row["lat_deg"], row["lon_deg"], row["alt_km"] = position
else:
flags.append("no_position")
# ---- Sense HAT ------------------------------------------------------
if sense is None:
flags.append("no_sense")
sensors = read_sensors(sense)
row["sensor_temp_c"] = sensors["temp"]
row["pressure_hpa"] = sensors["pressure"]
row["humidity_pct"] = sensors["humidity"]
row["mag_x_ut"] = sensors["mx"]
row["mag_y_ut"] = sensors["my"]
row["mag_z_ut"] = sensors["mz"]
row["mag_total_ut"] = sensors["mtotal"]
if sense is not None and sensors["mtotal"] is None:
flags.append("mag_error")
# ---- camera ---------------------------------------------------------
if camera is None:
flags.append("no_camera")
state["prev"] = None
elif elapsed() < SAFE_STOP_S:
image_path = BASE_DIR / f"img_{index:03d}.jpg"
t_before = time.monotonic()
try:
camera.capture_file(str(image_path))
t_capture = (t_before + time.monotonic()) / 2.0
except Exception as exc:
logging.warning("Capture %d failed: %s", index, exc)
flags.append("capture_error")
state["prev"] = None
t_capture = None
if t_capture is not None:
small, scale, original_width = load_image(image_path)
if small is None:
flags.append("image_unreadable")
state["prev"] = None
else:
try:
colour = analyse_colour(small)
except Exception as exc:
logging.warning("Colour analysis failed: %s", exc)
colour = None
flags.append("analysis_error")
if colour is None:
if "analysis_error" not in flags:
flags.append("night")
state["prev"] = None
try:
image_path.unlink() # a dark photo has no value
except Exception:
pass
else:
row["image"] = image_path.name
row["brightness"] = colour["brightness"]
row["cloud_pct"] = colour["cloud_pct"]
row["bndvi_mean"] = colour["bndvi_mean"]
row["veg_pct"] = colour["veg_pct"]
_update_speed(row, flags, orb, matcher, small, scale,
original_width, t_capture, state)
else:
flags.append("skipped_time")
if SOURCES["camera"] == "simulated":
flags.append("sim_camera")
if SOURCES["sense_hat"] == "emulator":
flags.append("emu_sense")
row["flags"] = "|".join(flags)
return row
def _update_speed(row, flags, orb, matcher, small, scale, original_width,
t_capture, state):
"""Compare with the previous valid photo and fill the speed columns."""
try:
features = compute_features(orb, small)
previous = state.get("prev")
if previous is None:
flags.append("no_speed")
else:
dt = t_capture - previous["time"]
shift, n_inliers = estimate_shift(matcher, previous["features"],
features)
row["dt_s"] = dt
row["inliers"] = n_inliers
if shift is None or dt <= 0:
flags.append("no_speed")
else:
altitude = row["alt_km"] if row["alt_km"] else \
DEFAULT_ALTITUDE_KM
gsd = (altitude * 1000.0 * SENSOR_WIDTH_MM /
(FOCAL_LENGTH_MM * original_width))
shift_full_px = shift * scale
speed = shift_full_px * gsd / dt / 1000.0
row["shift_px"] = shift_full_px
if SPEED_MIN_KMS <= speed <= SPEED_MAX_KMS:
row["speed_kms"] = speed
else:
flags.append("bad_speed")
state["prev"] = {"features": features, "time": t_capture}
except Exception as exc:
logging.warning("Speed estimation failed: %s", exc)
flags.append("speed_error")
state["prev"] = None
# --------------------------------------------------------------------------
# Main program
# --------------------------------------------------------------------------
def main():
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
handlers=[logging.FileHandler(str(LOG_PATH), encoding="utf-8"),
logging.StreamHandler(sys.stdout)])
logging.info("BermejaPi started")
rows = []
camera = None
try:
sense = init_sense()
camera = init_camera()
with open(CSV_PATH, "w", newline="", encoding="utf-8") as handle:
writer = csv.writer(handle)
writer.writerow(CSV_COLUMNS)
handle.flush()
if sense is None and camera is None:
logging.error("Neither Sense HAT nor camera available; stopping")
return
orb = cv2.ORB_create(nfeatures=ORB_FEATURES)
matcher = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=True)
state = {"prev": None}
index = 0
next_cycle = time.monotonic()
while elapsed() < LAST_CYCLE_START_S:
pause = next_cycle - time.monotonic()
if pause > 0:
time.sleep(min(pause, max(0.0, LAST_CYCLE_START_S -
elapsed())))
if elapsed() >= LAST_CYCLE_START_S:
break
index += 1
try:
row = run_cycle(index, sense, camera, orb, matcher, state)
except Exception as exc: # last line of defence
logging.error("Cycle %d failed: %s", index, exc)
state["prev"] = None
next_cycle = time.monotonic() + CYCLE_S
continue
rows.append(row)
writer.writerow([_cell(row[name]) for name in CSV_COLUMNS])
handle.flush()
purge_storage()
logging.info("[Cycle %d | %.0f s] %s", index, elapsed(),
describe(row))
next_cycle += CYCLE_S
if next_cycle < time.monotonic():
next_cycle = time.monotonic()
except Exception as exc:
logging.error("Unexpected error: %s", exc)
finally:
if camera is not None:
try:
camera.stop()
camera.close()
except Exception:
pass
write_summary(rows)
purge_storage()
logging.info("BermejaPi finished after %.1f s (limit %d s)",
elapsed(), TIME_LIMIT_S)
logging.shutdown()
def describe(row):
"""One readable line per cycle (console and log)."""
def fmt(value, template):
return "-" if value is None else template.format(value)
return ("B=%s uT | v=%s km/s | cloud=%s%% | veg=%s%% | flags=%s" % (
fmt(row["mag_total_ut"], "{:.1f}"), fmt(row["speed_kms"], "{:.2f}"),
fmt(row["cloud_pct"], "{:.1f}"), fmt(row["veg_pct"], "{:.1f}"),
row["flags"] or "ok"))
def _cell(value):
"""Format a value for the CSV file (None -> empty)."""
if value is None:
return ""
if isinstance(value, float):
cleaned = clean(value, 4)
return "" if cleaned is None else cleaned
return value
if __name__ == "__main__":
main()6. Cómo probarlo
- Replay Tool: ejecútalo igual que en la entrada 1 y revisa
data.csv,summary.jsonymission.log. - Calibra los umbrales con las imágenes del Replay Tool:
CLOUD_MAX_S,CLOUD_MIN_VyBNDVI_VEG_THRESHOLD. Con un filtro azul las imágenes tienen otra dominante de color y los valores de las entradas 7 y 12 pueden no servir tal cual. - En tu máquina virtual:
python3 main.pyfunciona sin hardware (dura ~9 minutos). Para una prueba rápida bajaLAST_CYCLE_START_Sa 60 ySAFE_STOP_Sa 90 mientras pruebas, y devuélvelos a 530 y 560 antes del zip. - Pruebas automáticas (fuera del zip):
python3 run_sim.py normal /tmp/pruebacon los escenariosnormal,nosense,camfail,vm,vm_emuyflight_nohw.
7. Empaquetado
zip bermejapi.zip main.py README.txtAntes: ALLOW_SIMULATION = False, LAST_CYCLE_START_S = 530, SAFE_STOP_S = 560. main.py debe quedar en la raíz del zip. Si copias código de otros autores, añade un LICENSE.txt (con extensión, porque el reglamento solo admite ciertos tipos de fichero).
8. Fe de erratas de las entradas 12–15
- Entrada 15: con los valores publicados, la función
calcular_indice_habitabilidaddevuelve 79,94 %, no 88,35 %. Con ese resultado el planeta queda en «habitable con restricciones», no en Clase M. - Entradas 11–12: la temperatura del Sense HAT (21,5 °C) es la de la cabina de la ISS, no la superficial. La ΔT de +39,5 °C no se puede interpretar como efecto invernadero.
- Entrada 14: un umbral fijo de 20–70 µT toma valores de la Tierra para concluir que es la Tierra. Un campo dipolar se reconoce porque varía a lo largo de la órbita; por eso ahora guardamos la serie completa y su posición.
Reto para la clase: Entrada 17
Con los datos que devuelva el Replay Tool, ¿cómo cambia el índice de habitabilidad si usamos las medidas reales? ¿Aparece un mínimo del campo magnético cerca de la Anomalía del Atlántico Sur (latitud negativa, longitud entre −90° y 0°)?
Etiqueta:AIDARAC, Astro Pi, BermejaPi, BNDVI, ies monterroso, main.py, Mission Space Lab, OpenCV, raspberry pi iss, Replay Tool, Rulebook ESA, Sense HAT
También te puede interesar
Proyecto BermejaPi — Entrada 16: La anomalía (Detectando lo desconocido)
septiembre 19, 2026
Proyecto BermejaPi — Entrada 15: Informe de Misión e Índice de Habitabilidad
septiembre 18, 2026