🌦️ Tutorial 34: NOAA Weather & Climate Analytics¶

# Generate synthetic NOAA weather data for development
# Reference: data_generation/generators/federal/noaa_generator.py

import sys
sys.path.append("../../data_generation")

from generators.federal.noaa_generator import NOAAGenerator

generator = NOAAGenerator(
    output_path="/lakehouse/default/Files/raw/noaa/",
    num_observations=100000,       # 100K weather observations
    num_storm_events=5000,         # 5K storm events
    date_range=("2020-01-01", "2024-12-31"),
    regions=["northeast", "southeast", "midwest", "southwest", "pacific"],
    seed=42
)

df_obs, df_storms = generator.generate_batch()
print(f"Generated {df_obs.count():,} weather observations")
print(f"Generated {df_storms.count():,} storm events")

# Connect to real NOAA data sources
import requests
import os

NOAA_TOKEN = os.environ.get("NOAA_API_TOKEN", "YOUR_TOKEN_HERE")

# 1. Weather API - Current Observations
def fetch_weather_observations(station_id: str) -> dict:
    """Fetch latest observation from a weather station."""
    url = f"https://api.weather.gov/stations/{station_id}/observations/latest"
    headers = {"User-Agent": "FabricWeatherAnalytics/1.0", "Accept": "application/geo+json"}
    response = requests.get(url, headers=headers)
    return response.json()

# 2. Storm Events DB - Bulk Download
STORM_EVENTS_URL = "https://www.ncei.noaa.gov/pub/data/swdi/stormevents/csvfiles/"
# Files named: StormEvents_details-ftp_v1.0_d{YEAR}_c{DATE}.csv.gz

# 3. Climate Data Online (CDO) - REST API
def fetch_climate_data(dataset_id: str, station_id: str, start_date: str, end_date: str) -> dict:
    """Fetch historical climate data from CDO."""
    url = "https://www.ncdc.noaa.gov/cdo-web/api/v2/data"
    headers = {"token": NOAA_TOKEN}
    params = {
        "datasetid": dataset_id,
        "stationid": station_id,
        "startdate": start_date,
        "enddate": end_date,
        "limit": 1000,
        "units": "standard",
    }
    response = requests.get(url, headers=headers, params=params)
    return response.json()

# Example: Fetch observations for DCA (Reagan National Airport)
obs = fetch_weather_observations("KDCA")
print(f"Station: KDCA")
print(f"Temperature: {obs['properties']['temperature']['value']}C")
print(f"Wind Speed: {obs['properties']['windSpeed']['value']} km/h")

# Schema for NOAA weather observations (matches noaa_weather_schema.json)
from pyspark.sql.types import *

weather_observation_schema = StructType([
    # Station identification
    StructField("station_id", StringType(), False),           # ICAO/WBAN station code
    StructField("station_name", StringType(), True),          # Station display name
    StructField("observation_time", TimestampType(), False),  # UTC timestamp
    StructField("latitude", DoubleType(), True),
    StructField("longitude", DoubleType(), True),
    StructField("elevation_m", DoubleType(), True),           # Station elevation (meters)
    StructField("state", StringType(), True),                 # 2-letter state code
    StructField("county_fips", StringType(), True),           # 5-digit county FIPS
    StructField("nws_zone", StringType(), True),              # NWS forecast zone

    # Meteorological observations
    StructField("temperature_c", DoubleType(), True),         # Temperature (Celsius)
    StructField("dewpoint_c", DoubleType(), True),            # Dew point (Celsius)
    StructField("relative_humidity_pct", DoubleType(), True), # Relative humidity (%)
    StructField("wind_speed_kmh", DoubleType(), True),        # Wind speed (km/h)
    StructField("wind_direction_deg", IntegerType(), True),   # Wind direction (degrees)
    StructField("wind_gust_kmh", DoubleType(), True),         # Wind gust (km/h)
    StructField("barometric_pressure_hpa", DoubleType(), True), # Pressure (hPa)
    StructField("visibility_km", DoubleType(), True),         # Visibility (km)
    StructField("precipitation_mm", DoubleType(), True),      # Precipitation (mm)
    StructField("snow_depth_cm", DoubleType(), True),         # Snow depth (cm)
    StructField("cloud_cover", StringType(), True),           # CLR, FEW, SCT, BKN, OVC
    StructField("weather_condition", StringType(), True),     # Present weather (rain, snow, etc.)

    # Quality flags
    StructField("qc_flag", StringType(), True),               # NOAA quality control flag
    StructField("data_source", StringType(), True),           # API, CDO, ASOS, AWOS
])

print(f"Weather observation schema: {len(weather_observation_schema.fields)} fields")

# Schema for NOAA Storm Events Database
storm_event_schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("event_type", StringType(), True),            # Tornado, Hail, Flash Flood, etc.
    StructField("begin_date", DateType(), True),
    StructField("end_date", DateType(), True),
    StructField("begin_time", StringType(), True),            # HHMM local time
    StructField("state", StringType(), True),
    StructField("state_fips", StringType(), True),
    StructField("county_fips", StringType(), True),
    StructField("cz_name", StringType(), True),               # County/zone name
    StructField("cz_type", StringType(), True),               # C=county, Z=zone

    # Impact
    StructField("injuries_direct", IntegerType(), True),
    StructField("injuries_indirect", IntegerType(), True),
    StructField("deaths_direct", IntegerType(), True),
    StructField("deaths_indirect", IntegerType(), True),
    StructField("damage_property", DoubleType(), True),       # Property damage ($)
    StructField("damage_crops", DoubleType(), True),          # Crop damage ($)

    # Event details
    StructField("magnitude", DoubleType(), True),             # EF scale, hail size, wind speed
    StructField("magnitude_type", StringType(), True),        # EF, inches, knots
    StructField("tor_f_scale", StringType(), True),           # EF0-EF5 for tornadoes
    StructField("flood_cause", StringType(), True),           # Heavy Rain, Dam Break, etc.
    StructField("episode_narrative", StringType(), True),     # Episode description
    StructField("event_narrative", StringType(), True),       # Event description

    # Geography
    StructField("begin_lat", DoubleType(), True),
    StructField("begin_lon", DoubleType(), True),
    StructField("end_lat", DoubleType(), True),
    StructField("end_lon", DoubleType(), True),

    StructField("source", StringType(), True),                # Trained Spotter, ASOS, etc.
    StructField("year", IntegerType(), True),
])

print(f"Storm event schema: {len(storm_event_schema.fields)} fields")

# Ingest weather observations and storm events into Bronze
from pyspark.sql.functions import *
from datetime import datetime
from uuid import uuid4

BATCH_ID = f"noaa-batch-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}"
RUN_ID = str(uuid4())

# Read weather observations
source_path = "/lakehouse/default/Files/raw/noaa/"
df_obs_raw = spark.read.schema(weather_observation_schema).parquet(f"{source_path}observations/")
df_storm_raw = spark.read.schema(storm_event_schema).parquet(f"{source_path}storm_events/")

# Add Bronze metadata
def add_bronze_metadata(df):
    return df \
        .withColumn("_ingested_at", current_timestamp()) \
        .withColumn("_source_file", input_file_name()) \
        .withColumn("_batch_id", lit(BATCH_ID)) \
        .withColumn("_run_id", lit(RUN_ID)) \
        .withColumn("_load_date", current_timestamp().cast("date"))

df_obs_bronze = add_bronze_metadata(df_obs_raw)
df_storm_bronze = add_bronze_metadata(df_storm_raw)

# Write observations (partition by state and load date)
df_obs_bronze.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .partitionBy("state", "_load_date") \
    .saveAsTable("lh_bronze.bronze_noaa_observations")

# Write storm events (partition by year)
df_storm_bronze.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .partitionBy("year") \
    .saveAsTable("lh_bronze.bronze_noaa_storm_events")

print(f"Bronze observations: {df_obs_bronze.count():,} records")
print(f"Bronze storm events: {df_storm_bronze.count():,} records")

# Unit conversion constants
KMH_TO_MPH = 0.621371
KMH_TO_KNOTS = 0.539957
MM_TO_INCHES = 0.0393701
CM_TO_INCHES = 0.393701
C_TO_F = lambda c: (c * 9/5) + 32 if c is not None else None

# Standardize weather observations
df_std = df_obs_bronze \
    .withColumn("temperature_f", round(col("temperature_c") * 9 / 5 + 32, 1)) \
    .withColumn("dewpoint_f", round(col("dewpoint_c") * 9 / 5 + 32, 1)) \
    .withColumn("wind_speed_mph", round(col("wind_speed_kmh") * KMH_TO_MPH, 1)) \
    .withColumn("wind_speed_knots", round(col("wind_speed_kmh") * KMH_TO_KNOTS, 1)) \
    .withColumn("wind_gust_mph", round(col("wind_gust_kmh") * KMH_TO_MPH, 1)) \
    .withColumn("precipitation_in", round(col("precipitation_mm") * MM_TO_INCHES, 2)) \
    .withColumn("snow_depth_in", round(col("snow_depth_cm") * CM_TO_INCHES, 1)) \
    .withColumn("visibility_mi", round(col("visibility_km") * 0.621371, 1)) \
    .withColumn("pressure_inHg", round(col("barometric_pressure_hpa") * 0.02953, 2))

# Meteorological range validation
VALID_RANGES = {
    "temperature_c": (-89.2, 56.7),     # World record extremes
    "wind_speed_kmh": (0, 410),          # Up to 410 km/h (Hurricane Patricia)
    "barometric_pressure_hpa": (870, 1084),  # Record low to high
    "relative_humidity_pct": (0, 100),
    "visibility_km": (0, 400),
    "precipitation_mm": (0, 1825),       # 24-hour world record
    "wind_direction_deg": (0, 360),
}

df_validated = df_std
for col_name, (min_val, max_val) in VALID_RANGES.items():
    df_validated = df_validated \
        .withColumn(f"{col_name}_valid",
            when(
                (col(col_name) >= min_val) & (col(col_name) <= max_val),
                lit(True)
            ).when(col(col_name).isNull(), lit(None))
            .otherwise(lit(False))
        )

# Quality score for observations
df_validated = df_validated \
    .withColumn("_dq_score",
        when(col("station_id").isNotNull(), lit(15)).otherwise(lit(0)) +
        when(col("observation_time").isNotNull(), lit(15)).otherwise(lit(0)) +
        when(col("temperature_c_valid") == True, lit(15)).otherwise(lit(0)) +
        when(col("wind_speed_kmh_valid") == True, lit(10)).otherwise(lit(0)) +
        when(col("barometric_pressure_hpa_valid") == True, lit(10)).otherwise(lit(0)) +
        when(col("qc_flag").isin("V", "S"), lit(15)).otherwise(lit(5)) +
        when(col("latitude").isNotNull() & col("longitude").isNotNull(), lit(10)).otherwise(lit(0)) +
        when(col("state").isNotNull(), lit(5)).otherwise(lit(0)) +
        when(col("county_fips").isNotNull(), lit(5)).otherwise(lit(0))
    )

# Validation summary
for col_name in VALID_RANGES.keys():
    invalid = df_validated.filter(col(f"{col_name}_valid") == False).count()
    if invalid > 0:
        print(f"  {col_name}: {invalid:,} out-of-range values")

# Enrich storm events with severity classification
df_storm_silver = df_storm_bronze \
    .withColumn("total_injuries", col("injuries_direct") + col("injuries_indirect")) \
    .withColumn("total_deaths", col("deaths_direct") + col("deaths_indirect")) \
    .withColumn("total_damage", col("damage_property") + col("damage_crops")) \
    .withColumn("severity",
        when(
            (col("total_deaths") > 0) | (col("total_damage") > 1000000),
            lit("Catastrophic")
        ).when(
            (col("total_injuries") > 10) | (col("total_damage") > 100000),
            lit("Severe")
        ).when(
            (col("total_injuries") > 0) | (col("total_damage") > 10000),
            lit("Significant")
        ).otherwise(lit("Minor"))
    ) \
    .withColumn("event_type_std", upper(trim(col("event_type")))) \
    .withColumn("state_std", upper(trim(col("state")))) \
    .withColumn("decade", (floor(col("year") / 10) * 10).cast("string"))

# Deduplication
df_storm_deduped = df_storm_silver.dropDuplicates(["event_id"])

# Write Silver observations
df_obs_silver = df_validated.withColumn("_silver_timestamp", current_timestamp())

df_obs_silver.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .partitionBy("state") \
    .saveAsTable("lh_silver.silver_noaa_observations")

spark.sql("OPTIMIZE lh_silver.silver_noaa_observations ZORDER BY (station_id, observation_time)")

# Write Silver storm events
df_storm_deduped.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .partitionBy("year") \
    .saveAsTable("lh_silver.silver_noaa_storm_events")

spark.sql("OPTIMIZE lh_silver.silver_noaa_storm_events ZORDER BY (state_std, event_type_std)")

print(f"Silver observations: {df_obs_silver.count():,}")
print(f"Silver storm events: {df_storm_deduped.count():,}")

# Eventstream ingestion: NOAA Weather API -> Eventhouse
# This code runs as a Fabric Eventstream custom source

import json
import requests
import time
from datetime import datetime

# NOAA Weather API stations to monitor
MONITORED_STATIONS = [
    "KDCA", "KJFK", "KLAX", "KORD", "KDFW",
    "KATL", "KDEN", "KSFO", "KMIA", "KSEA",
]

WEATHER_API_BASE = "https://api.weather.gov/stations"

def fetch_and_emit_observation(station_id: str) -> dict:
    """Fetch latest observation and format for Eventstream."""
    url = f"{WEATHER_API_BASE}/{station_id}/observations/latest"
    headers = {
        "User-Agent": "FabricNOAAAnalytics/1.0",
        "Accept": "application/geo+json"
    }
    response = requests.get(url, headers=headers, timeout=10)

    if response.status_code != 200:
        return None

    obs = response.json()["properties"]
    return {
        "station_id": station_id,
        "observation_time": obs["timestamp"],
        "temperature_c": obs["temperature"]["value"],
        "dewpoint_c": obs["dewpoint"]["value"],
        "wind_speed_kmh": obs["windSpeed"]["value"],
        "wind_direction_deg": obs["windDirection"]["value"],
        "barometric_pressure_hpa": obs["barometricPressure"]["value"],
        "visibility_km": (obs["visibility"]["value"] or 0) / 1000,
        "relative_humidity_pct": obs["relativeHumidity"]["value"],
        "weather_condition": obs["textDescription"],
        "event_time": datetime.utcnow().isoformat(),
    }

# Polling loop (runs inside Eventstream custom source)
while True:
    for station in MONITORED_STATIONS:
        obs = fetch_and_emit_observation(station)
        if obs:
            # emit_event() is the Eventstream output function
            emit_event(json.dumps(obs))
    time.sleep(300)  # 5-minute polling interval (respect API limits)

# Fabric Activator alert for severe weather conditions
activator_rules = {
    "high_wind_alert": {
        "condition": "wind_speed_kmh > 100",  # ~62 mph
        "trigger": "Any station reports sustained winds > 100 km/h",
        "action": "Send Teams notification to weather-monitoring channel",
        "cooldown_minutes": 30,
    },
    "extreme_cold_alert": {
        "condition": "temperature_c < -35",
        "trigger": "Any station reports temperature below -35C",
        "action": "Send email to operations@contoso.com",
        "cooldown_minutes": 60,
    },
    "pressure_drop_alert": {
        "condition": "barometric_pressure_hpa < 970",
        "trigger": "Station pressure below 970 hPa (intense cyclone)",
        "action": "Send Teams notification + email to emergency-ops",
        "cooldown_minutes": 15,
    },
}

for alert_name, config in activator_rules.items():
    print(f"\n{alert_name}:")
    for key, value in config.items():
        print(f"  {key}: {value}")

# Daily weather summaries by station and state
df_weather_summary = df_obs_silver \
    .withColumn("observation_date", col("observation_time").cast("date")) \
    .groupBy("station_id", "station_name", "state", "observation_date") \
    .agg(
        avg("temperature_c").alias("avg_temp_c"),
        min("temperature_c").alias("min_temp_c"),
        max("temperature_c").alias("max_temp_c"),
        avg("temperature_f").alias("avg_temp_f"),
        min("temperature_f").alias("min_temp_f"),
        max("temperature_f").alias("max_temp_f"),

        sum("precipitation_mm").alias("total_precip_mm"),
        sum("precipitation_in").alias("total_precip_in"),
        max("snow_depth_cm").alias("max_snow_depth_cm"),

        avg("wind_speed_mph").alias("avg_wind_mph"),
        max("wind_gust_mph").alias("max_gust_mph"),
        avg("barometric_pressure_hpa").alias("avg_pressure_hpa"),
        avg("relative_humidity_pct").alias("avg_humidity_pct"),

        count("*").alias("observation_count"),
        avg("_dq_score").alias("avg_dq_score"),
    )

df_weather_summary.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("state") \
    .saveAsTable("lh_gold.gold_noaa_weather_summary")

# Storm damage aggregation by state, event type, and decade
df_storm_damage = df_storm_deduped \
    .groupBy("state_std", "event_type_std", "decade", "severity") \
    .agg(
        count("*").alias("event_count"),
        sum("damage_property").alias("total_property_damage"),
        sum("damage_crops").alias("total_crop_damage"),
        sum("total_damage").alias("total_combined_damage"),
        sum("total_injuries").alias("total_injuries"),
        sum("total_deaths").alias("total_fatalities"),
        avg("magnitude").alias("avg_magnitude"),
        max("magnitude").alias("max_magnitude"),
    ) \
    .withColumn("damage_per_event",
        when(col("event_count") > 0,
            round(col("total_combined_damage") / col("event_count"), 0)
        ).otherwise(lit(0))
    )

df_storm_damage.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("lh_gold.gold_noaa_storm_damage")

# Climate trend: annual temperature averages compared to 30-year normals
df_climate_trends = df_obs_silver \
    .withColumn("year", year(col("observation_time"))) \
    .withColumn("month", month(col("observation_time"))) \
    .groupBy("state", "year", "month") \
    .agg(
        avg("temperature_c").alias("avg_temp_c"),
        avg("temperature_f").alias("avg_temp_f"),
        sum("precipitation_mm").alias("total_precip_mm"),
        count("*").alias("observation_count"),
    )

# Join with climate normals (if available)
# Climate normals represent the 1991-2020 30-year average
df_climate_trends = df_climate_trends \
    .withColumn("temp_anomaly_c", col("avg_temp_c") - lit(15.0))  # Placeholder for actual normals

df_climate_trends.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("lh_gold.gold_noaa_climate_trends")

# Severe weather risk index by state (composite of frequency, damage, and fatalities)
df_risk = df_storm_deduped \
    .filter(col("year") >= 2010) \
    .groupBy("state_std") \
    .agg(
        count("*").alias("total_events"),
        sum("total_damage").alias("total_damage"),
        sum("total_deaths").alias("total_fatalities"),
        sum("total_injuries").alias("total_injuries"),
        countDistinct("event_type_std").alias("event_type_diversity"),
        sum(when(col("severity") == "Catastrophic", 1).otherwise(0)).alias("catastrophic_events"),
    )

# Normalize to 0-100 risk score
from pyspark.sql.window import Window

w = Window.orderBy(col("total_events").desc())
df_risk = df_risk \
    .withColumn("event_freq_rank", percent_rank().over(w)) \
    .withColumn("damage_rank", percent_rank().over(Window.orderBy(col("total_damage").desc()))) \
    .withColumn("fatality_rank", percent_rank().over(Window.orderBy(col("total_fatalities").desc()))) \
    .withColumn("risk_index",
        round((col("event_freq_rank") * 40 + col("damage_rank") * 35 + col("fatality_rank") * 25) * 100, 1)
    )

df_risk.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("lh_gold.gold_noaa_severe_weather_risk")

# Summary
for table in ["gold_noaa_weather_summary", "gold_noaa_storm_damage",
              "gold_noaa_climate_trends", "gold_noaa_severe_weather_risk"]:
    count = spark.table(f"lh_gold.{table}").count()
    print(f"  {table}: {count:,} rows")

# Data quality checks specific to weather observations
from great_expectations.core import ExpectationSuite

noaa_expectations = ExpectationSuite(expectation_suite_name="noaa_weather_suite")

# Temperature within physical limits
noaa_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "temperature_c", "min_value": -89.2, "max_value": 56.7, "mostly": 0.999}
})

# Wind speed non-negative
noaa_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "wind_speed_kmh", "min_value": 0, "max_value": 410, "mostly": 0.999}
})

# Pressure within recorded range
noaa_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "barometric_pressure_hpa", "min_value": 870, "max_value": 1084, "mostly": 0.999}
})

# Station ID format (ICAO 4-letter code)
noaa_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_match_regex",
    "kwargs": {"column": "station_id", "regex": r"^K?[A-Z0-9]{3,4}$", "mostly": 0.95}
})

# No future observation timestamps
noaa_expectations.add_expectation({
    "expectation_type": "expect_column_max_to_be_between",
    "kwargs": {"column": "observation_time", "max_value": datetime.utcnow().isoformat()}
})

print(f"NOAA validation suite: {len(noaa_expectations.expectations)} expectations")

# Purview configuration for NOAA data
purview_config = {
    "lh_bronze.bronze_noaa_observations": {
        "sensitivity_label": "Public - Federal Open Data",
        "attribution": "NOAA / National Weather Service",
        "data_source_url": "https://api.weather.gov/",
        "update_frequency": "Continuous (real-time) + daily batch",
        "retention_policy": "Indefinite (public scientific record)",
    },
    "lh_bronze.bronze_noaa_storm_events": {
        "sensitivity_label": "Public - Federal Open Data",
        "attribution": "NOAA / National Centers for Environmental Information (NCEI)",
        "data_source_url": "https://www.ncei.noaa.gov/pub/data/swdi/stormevents/",
        "update_frequency": "Monthly bulk updates",
    },
    "lh_gold.gold_noaa_severe_weather_risk": {
        "sensitivity_label": "Internal - Derived Analytics",
        "data_lineage": "bronze -> silver -> risk calculation (composite index)",
        "disclaimer": "Risk index is a derived metric, not an official NOAA product",
    },
}

for table, config in purview_config.items():
    print(f"\n{table}:")
    for key, value in config.items():
        print(f"  {key}: {value}")