Objetivo del proyecto de ingeniería de datos

Desarrollar un ecosistema de datos que centralice el histórico climatológico de una ubicación específica, transformando registros crudos en un dashboard analítico en Looker Studio para optimizar la toma de decisiones basada en variables ambientales.

Flujo de procesamiento y consumo de API (Workflow)

El proceso de obtención de datos sigue un modelo de solicitud y redirección técnica, estructurado en dos fases:

Fase A: Petición de acceso y orquestación

https://opendata.aemet.es/opendata/api/valores/climatologicos/mensualesanuales/datos/anioini/{AÑO}/aniofin/{AÑO}/estacion/{ID_ESTACION}/?api_key={TU_API_KEY}

Al realizar la consulta inicial al endpoint de AEMET, el servidor no devuelve los datos climatológicos directamente, sino un objeto de respuesta de control en formato JSON.

Este objeto actúa como un puente de seguridad y eficiencia:

• Validación de estado: Se verifica que el atributo "estado": 200 y "descripcion": "exito" confirmen la disponibilidad de la información.
• Localización de recursos: El sistema identifica el atributo "datos", el cual contiene una URL dinámica y temporal generada específicamente para la consulta.

{
  "descripcion" : "exito",
  "estado" : 200,
  "datos" : "https://opendata.aemet.es/opendata/sh/4780eaa8",
  "metadatos" : "https://opendata.aemet.es/opendata/sh/997c0034"
}

Fase B: Extracción del dataset final
Una vez obtenida la URL del atributo "datos", el workflow ejecuta una segunda llamada automática a dicha dirección. Esta respuesta final entrega el JSON enriquecido con las variables meteorológicas (temperaturas, humedad, viento, etc.) de la estación y año seleccionados.

[ {
  “indicativo” : “3129”,
  “p_max” : “21.3(11)”,
  “n_cub” : “8”,
  “hr” : “75”,
  “n_gra” : “0”,
  “n_fog” : “8”,
  “inso” : “5.0”,
  “q_max” : “959.0(02)”,
  “q_mar” : “1017.5”,
  “q_med” : “948.5”,
  “tm_min” : “2.6”,
  “ta_max” : “20.8(01)”,
  “ts_min” : “12.4”,
  “nt_30” : “0”,
  “nv_0050” : “0”,
  “n_des” : “6”,
  “np_100” : “3”,
  “n_nub” : “16”,
  “p_sol” : “50”,
  “np_001” : “9”,
  “ta_min” : “-3.8(28)”,
  “e” : “79”,
  “np_300” : “0”,
  “nv_1000” : “0”,
  “evap” : “690”,
  “p_mes” : “66.8”,
  “n_llu” : “11”,
  “n_tor” : “0”,
  “w_med” : “8”,
  “nt_00” : “7”,
  “ti_max” : “5.0”,
  “n_nie” : “0”,
  “tm_mes” : “7.5”,
  “tm_max” : “12.4”,
  “nv_0100” : “0”,
  “q_min” : “931.0(23)”,
  “np_010” : “6”,
  “fecha” : “1969-11”
}, {...
},{...
},{...}
,...
]

La fecha detalla el año y el mes. En el json se ve un objeto con fecha AAAA-13 que hace referencia al resumen del año. Por ejemplo: 1969-13.

Para comprender el resto de indicadores, consulta esta URL de metadatos.

Desarrollo del script de ingesta y automatización

Se desarrollará un script robusto en Python diseñado para la extracción cíclica de datos, gestionando las restricciones de la API y garantizando la integridad de la base de datos resultante.

Lógica de extracción (Proceso de doble petición)

El script ejecutará un flujo de trabajo automatizado por cada periodo anual:

1. Request Nivel 1: Consulta al endpoint de AEMET para obtener el enlace temporal de descarga.
2. Extracción de URL: Análisis del JSON de control para capturar el atributo datos.
3. Request Nivel 2: Descarga del dataset final con los registros climatológicos detallados.

Control de latencia y Rate limiting

Para cumplir con las políticas de uso de AEMET OpenData y evitar bloqueos de la IP:

• Pausa Programada: Se implementará un retardo de 10 segundos entre iteraciones mediante la librería time.
• Iteración Temporal: El ciclo iniciará en el año 1965 y finalizará en 2025, procesando un año por ciclo de ejecución.

Gestión dinámica del dataset (Estructura del fichero)

La salida de datos se consolidará en un archivo Excel (XLSX/CSV) con las siguientes capacidades:

• Mapeo de atributos: Cada clave del JSON se transformará automáticamente en una columna del dataset.
• Esquema evolutivo (Dynamic schema): El script comparará los atributos de cada nueva llamada con las columnas existentes. Si se detectan nuevas variables meteorológicas en años recientes, el script realizará una actualización del esquema, añadiendo las columnas necesarias sin perder la información histórica.

Sistema de gestión de errores

El sistema implementa una política de resiliencia ante errores de red o rate limiting basada en dos fases:

1. Reintentos de Nivel 1 (Burst): Ante un fallo, se ejecutan hasta 6 intentos con un intervalo fijo de 10 segundos entre ellos para mitigar latencias temporales.
2. Pausa de Enfriamiento (Cool-down): Si el error persiste tras el tercer intento, el sistema aplica un backoff de 60 segundos para asegurar el refresco de la ventana de cuota (peticiones por minuto) antes de reiniciar el ciclo.

Objetivo: Maximizar la tasa de éxito sin saturar el servicio externo y evitar bloqueos por exceder límites de tráfico.

Gestión multi-estación: Iteración y agregación de datos

El sistema permite el procesamiento concurrente o secuencial de múltiples puntos de medición mediante una arquitectura basada en mapeo de identificadores.

• Identificación (Mapping): Se utiliza un diccionario MAPEO_ESTACIONES para vincular los códigos alfanuméricos de la API con etiquetas legibles (ej. "4452" → "Badajoz").
• Procesamiento en lote (Batch processing): El programa itera sobre un array de estaciones, ejecutando llamadas independientes por cada identificador definido.
• Agregación de datos: Los resultados individuales se consolidan en un fichero general unificado, facilitando el análisis comparativo transaccional entre diferentes ubicaciones geográficas.

Resumen técnico del stack

• Lenguaje: Python 3.x
• Librerías Clave: * Requests: Para la gestión de peticiones HTTP.
  • Pandas: Para el procesamiento de datos y exportación a Excel.
  • Time: Para el control del intervalo de seguridad (Rate Limit).

import requests
import time
import pandas as pd
import os

# --- Configuración del Proyecto ---
API_KEY = "{API_KEY}"
# Diccionario de mapeo: ID de estación -> Nombre de la ciudad
MAPEO_ESTACIONES = {
    "4452": "Badajoz",
    "5530E": "Granada",
    "1428": "Santiago de Compostela",
    "3129": "Madrid"
}

ANIO_INICIO = 1965
ANIO_FIN = 2025
FICHERO_SALIDA = r"C:\Users\{nombre_usuario}\Desktop\Proyecto clima\datos_clima_multiestacion.xlsx"

# Configuración de Reintentos
MAX_REINTENTOS = 3
ESPERA_ERROR = 10 

def ejecutar_proyecto_clima():
    df_acumulado = pd.DataFrame()

    for estacion_id, ciudad in MAPEO_ESTACIONES.items():
        print(f"\n========== PROCESANDO: {ciudad} ==========")
        
        for anio in range(ANIO_INICIO, ANIO_FIN + 1):
            print(f"--- Año {anio} ({ciudad}) ---")
            
            reintento_actual = 0
            exito_anio = False

            while reintento_actual < MAX_REINTENTOS and not exito_anio:
                try:
                    url_api = f"https://opendata.aemet.es/opendata/api/valores/climatologicos/mensualesanuales/datos/anioini/{anio}/aniofin/{anio}/estacion/{estacion_id}/?api_key={API_KEY}"
                    
                    respuesta_primaria = requests.get(url_api, timeout=30)
                    respuesta_primaria.raise_for_status()
                    resultado_api = respuesta_primaria.json()

                    if resultado_api.get("estado") == 200:
                        url_datos_reales = resultado_api.get("datos")
                        
                        respuesta_datos = requests.get(url_datos_reales, timeout=30)
                        respuesta_datos.raise_for_status()
                        json_datos_clima = respuesta_datos.json()

                        df_anio = pd.DataFrame(json_datos_clima)
                        
                        # --- CAMBIO SOLICITADO ---
                        # Forzamos que la columna 'indicativo' contenga el nombre de la ciudad
                        df_anio['indicativo'] = ciudad 
                        
                        df_acumulado = pd.concat([df_acumulado, df_anio], ignore_index=True, sort=False)
                        
                        # Guardar progreso parcial
                        df_acumulado.to_excel(FICHERO_SALIDA, index=False)
                        print(f"Éxito: {ciudad} - {anio}.")
                        exito_anio = True 
                    
                    elif resultado_api.get("estado") == 429:
                        print(f"Límite alcanzado (429). Esperando 60s...")
                        time.sleep(60) 
                        reintento_actual += 1
                    else:
                        print(f"Aviso: {resultado_api.get('descripcion')} ({ciudad}-{anio})")
                        break 

                except Exception as e:
                    reintento_actual += 1
                    print(f"Error {reintento_actual}/{MAX_REINTENTOS} para {ciudad}-{anio}: {e}")
                    if reintento_actual < MAX_REINTENTOS:
                        time.sleep(ESPERA_ERROR)

            # Pausa de seguridad para evitar bloqueos
            time.sleep(5)

    print(f"\n¡Proyecto finalizado! Los datos de las {len(MAPEO_ESTACIONES)} ciudades están en '{FICHERO_SALIDA}'.")

if __name__ == "__main__":
    ejecutar_proyecto_clima()

Configuración del entorno y ejecución

Para el correcto funcionamiento del script, se requiere la preparación del entorno local y la configuración de las rutas de salida.

Gestión de dependencias

El proyecto utiliza librerías externas para la gestión de peticiones HTTP (requests) y el tratamiento avanzado de datos (pandas, openpyxl). Para configurar el entorno, ejecute el siguiente comando en la terminal:

python -m pip install requests pandas openpyxl

Configuración del almacenamiento

Para asegurar la persistencia de los datos, el script debe apuntar a un libro de Excel preexistente que actuará como base de datos centralizada. Se debe definir la ruta absoluta del archivo mediante la variable de configuración FICHERO_SALIDA:

# Configuración de la ruta local del dataset
FICHERO_SALIDA = r"C:\Users\{nombre_usuario}\Desktop\Proyecto clima\datos_clima.xlsx"

Nota técnica: Se recomienda ubicar el archivo .xlsx en el mismo directorio que el script para facilitar la portabilidad, aunque el uso de rutas absolutas garantiza que el proceso de escritura no falle por problemas de contexto de ejecución.

Conexión con Looker Studio a través de Google Sheet

El proceso para dar vida a los datos es bastante directo:

1. Limpieza en Google Sheets: Primero, llevamos los datos del script de Python a una hoja de cálculo. Para que Looker Studio no tenga problemas con los números, hacemos un rápido “buscar y reemplazar” de puntos por comas. Con este ajuste de formato, los datos quedan listos para procesar.
2. Conexión con Looker Studio: Creamos el informe y conectamos el Google Sheet como nuestra fuente principal. Es un paso rápido que nos abre la puerta a toda la parte visual.
3. Diseño y métricas a medida: Aquí viene lo divertido. Empezamos a “jugar” con los gráficos para que la información entre por los ojos y, sobre todo, creamos campos calculados. Estos nos permiten ir más allá de los datos brutos y obtener los KPIs exactos que queremos mostrar.

Enlaces de interés

• OpenData AEMET
• Swagger AEMET
• Python para descargar