🏔️ Tutorial 36: DOI Natural Resources Analytics¶

# Generate synthetic DOI data for development
import sys
sys.path.append("../../data_generation")

from generators.federal.doi_generator import DOIGenerator

generator = DOIGenerator(
    output_path="/lakehouse/default/Files/raw/doi/",
    num_earthquakes=50000,          # 50K earthquake events
    num_water_measurements=100000,  # 100K water readings
    num_park_records=5000,          # 5K visitation records
    num_species_records=3000,       # 3K species records
    date_range=("2019-01-01", "2024-12-31"),
    seed=42
)

datasets = generator.generate_batch()
for name, df in datasets.items():
    print(f"Generated {name}: {df.count():,} records")

# Connect to real DOI data sources -- all open, no API keys needed
import requests

# 1. USGS Earthquake API (GeoJSON)
def fetch_earthquakes(start_time: str, end_time: str, min_magnitude: float = 2.5) -> dict:
    """Fetch earthquake events from USGS."""
    url = "https://earthquake.usgs.gov/fdsnws/event/1/query"
    params = {
        "format": "geojson",
        "starttime": start_time,
        "endtime": end_time,
        "minmagnitude": min_magnitude,
        "orderby": "time-asc",
    }
    response = requests.get(url, params=params)
    return response.json()

# 2. USGS NWIS Water Data
def fetch_water_data(site_no: str, param_cd: str = "00060",
                     start_date: str = "2024-01-01") -> dict:
    """Fetch streamflow data from USGS NWIS. param_cd 00060 = discharge (cfs)."""
    url = "https://waterservices.usgs.gov/nwis/iv/"
    params = {
        "format": "json",
        "sites": site_no,
        "parameterCd": param_cd,
        "startDT": start_date,
    }
    response = requests.get(url, params=params)
    return response.json()

# 3. NPS Visitation (CSV download)
NPS_STATS_URL = "https://irma.nps.gov/Stats/SSRSReports/National%20Reports/Query%20Builder%20for%20Public%20Use%20Statistics"

# 4. FWS ECOS Species Data
def fetch_species_list(status: str = "E") -> dict:
    """Fetch endangered (E) or threatened (T) species from ECOS."""
    url = "https://ecos.fws.gov/ecp/report/species-listings-by-tax-group"
    # Note: ECOS provides bulk downloads; API is limited
    return {}

# Example: Fetch recent earthquakes
eq_data = fetch_earthquakes("2024-01-01", "2024-01-31", min_magnitude=4.0)
print(f"Earthquakes (M4.0+, Jan 2024): {eq_data['metadata']['count']}")
for feature in eq_data["features"][:5]:
    props = feature["properties"]
    coords = feature["geometry"]["coordinates"]
    print(f"  M{props['mag']:.1f} - {props['place']} ({coords[1]:.2f}, {coords[0]:.2f})")

# Schema for USGS earthquake events (matches doi_earthquake_schema.json)
from pyspark.sql.types import *

earthquake_schema = StructType([
    StructField("event_id", StringType(), False),             # USGS event ID (e.g., us7000abcd)
    StructField("event_time", TimestampType(), False),        # UTC origin time
    StructField("magnitude", DoubleType(), True),             # Event magnitude
    StructField("magnitude_type", StringType(), True),        # ml, mb, mw, md
    StructField("depth_km", DoubleType(), True),              # Hypocenter depth (km)
    StructField("latitude", DoubleType(), True),              # Epicenter latitude
    StructField("longitude", DoubleType(), True),             # Epicenter longitude

    # Location details
    StructField("place_description", StringType(), True),     # "5km NW of City, State"
    StructField("state", StringType(), True),                 # US state (if applicable)
    StructField("country", StringType(), True),
    StructField("region", StringType(), True),                # Seismic region

    # Impact and quality
    StructField("felt_reports", IntegerType(), True),         # DYFI felt reports count
    StructField("cdi", DoubleType(), True),                   # Community Decimal Intensity
    StructField("mmi", DoubleType(), True),                   # Modified Mercalli Intensity
    StructField("alert_level", StringType(), True),           # green, yellow, orange, red
    StructField("tsunami", BooleanType(), True),              # Tsunami flag
    StructField("significance", IntegerType(), True),         # USGS significance (0-1000+)

    # Metadata
    StructField("status", StringType(), True),                # automatic, reviewed
    StructField("net", StringType(), True),                   # Contributing network
    StructField("nst", IntegerType(), True),                  # Number of stations
    StructField("gap", DoubleType(), True),                   # Azimuthal gap (degrees)
    StructField("rms", DoubleType(), True),                   # Travel time residual (seconds)
    StructField("updated_at", TimestampType(), True),
])

print(f"Earthquake schema: {len(earthquake_schema.fields)} fields")

# Schema for USGS NWIS water data
water_data_schema = StructType([
    StructField("site_no", StringType(), False),              # USGS site number (8-15 digits)
    StructField("site_name", StringType(), True),
    StructField("measurement_time", TimestampType(), False),
    StructField("parameter_code", StringType(), True),        # 00060=discharge, 00065=gage height
    StructField("parameter_name", StringType(), True),
    StructField("value", DoubleType(), True),                 # Measurement value
    StructField("unit", StringType(), True),                  # cfs, ft, deg C, mg/L
    StructField("qualification_code", StringType(), True),    # A=approved, P=provisional

    # Site geography
    StructField("latitude", DoubleType(), True),
    StructField("longitude", DoubleType(), True),
    StructField("state_code", StringType(), True),
    StructField("county_code", StringType(), True),
    StructField("huc_code", StringType(), True),              # Hydrologic Unit Code (watershed)
    StructField("drainage_area_sq_mi", DoubleType(), True),
    StructField("site_type", StringType(), True),             # Stream, Well, Spring, Lake
])

print(f"Water data schema: {len(water_data_schema.fields)} fields")

# NPS Park Visitation schema
nps_visitation_schema = StructType([
    StructField("park_code", StringType(), False),            # 4-letter NPS unit code
    StructField("park_name", StringType(), True),
    StructField("year", IntegerType(), True),
    StructField("month", IntegerType(), True),
    StructField("recreation_visitors", LongType(), True),     # Recreational visits
    StructField("non_recreation_visitors", LongType(), True),
    StructField("total_visitors", LongType(), True),

    # Park details
    StructField("park_type", StringType(), True),             # National Park, Monument, etc.
    StructField("region", StringType(), True),                # NPS region
    StructField("state", StringType(), True),
    StructField("latitude", DoubleType(), True),
    StructField("longitude", DoubleType(), True),
    StructField("acres", DoubleType(), True),                 # Park area
])

# FWS Species schema
fws_species_schema = StructType([
    StructField("species_id", StringType(), False),
    StructField("scientific_name", StringType(), True),
    StructField("common_name", StringType(), True),
    StructField("taxonomic_group", StringType(), True),       # Mammals, Birds, Fish, Plants, etc.
    StructField("listing_status", StringType(), True),        # Endangered, Threatened
    StructField("listing_date", DateType(), True),
    StructField("critical_habitat", BooleanType(), True),
    StructField("recovery_plan", BooleanType(), True),

    # Geography
    StructField("range_states", StringType(), True),          # Comma-separated state codes
    StructField("lead_region", StringType(), True),           # FWS region responsible
    StructField("population_description", StringType(), True),
])

print(f"NPS visitation schema: {len(nps_visitation_schema.fields)} fields")
print(f"FWS species schema: {len(fws_species_schema.fields)} fields")

# Multi-domain DOI Bronze ingestion
from pyspark.sql.functions import *
from datetime import datetime
from uuid import uuid4

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

source_path = "/lakehouse/default/Files/raw/doi/"

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

# Earthquakes
df_eq = spark.read.schema(earthquake_schema).parquet(f"{source_path}earthquakes/")
df_eq_bronze = add_bronze_metadata(df_eq)
df_eq_bronze.write.format("delta").mode("append") \
    .partitionBy("_load_date") \
    .saveAsTable("lh_bronze.bronze_doi_earthquakes")

# Water data
df_water = spark.read.schema(water_data_schema).parquet(f"{source_path}water/")
df_water_bronze = add_bronze_metadata(df_water)
df_water_bronze.write.format("delta").mode("append") \
    .partitionBy("state_code", "_load_date") \
    .saveAsTable("lh_bronze.bronze_doi_water_data")

# NPS visitation
df_nps = spark.read.schema(nps_visitation_schema).parquet(f"{source_path}nps/")
df_nps_bronze = add_bronze_metadata(df_nps)
df_nps_bronze.write.format("delta").mode("append") \
    .partitionBy("year") \
    .saveAsTable("lh_bronze.bronze_doi_nps_visitation")

# FWS species
df_fws = spark.read.schema(fws_species_schema).parquet(f"{source_path}species/")
df_fws_bronze = add_bronze_metadata(df_fws)
df_fws_bronze.write.format("delta").mode("append") \
    .saveAsTable("lh_bronze.bronze_doi_fws_species")

print(f"Bronze earthquakes: {df_eq_bronze.count():,}")
print(f"Bronze water data: {df_water_bronze.count():,}")
print(f"Bronze NPS visitation: {df_nps_bronze.count():,}")
print(f"Bronze FWS species: {df_fws_bronze.count():,}")

# Validate earthquake parameters against seismological limits
MAGNITUDE_RANGES = {"ml": (-2, 8), "mb": (0, 9), "mw": (0, 10), "md": (-2, 6)}

df_eq_silver = df_eq_bronze \
    .withColumn("magnitude_valid",
        when(col("magnitude").isNull(), lit(False))
        .when(col("magnitude") < -2, lit(False))
        .when(col("magnitude") > 10, lit(False))
        .otherwise(lit(True))
    ) \
    .withColumn("depth_valid",
        when(col("depth_km") < 0, lit(False))
        .when(col("depth_km") > 700, lit(False))  # Deepest: ~700 km
        .otherwise(lit(True))
    ) \
    .withColumn("coords_valid",
        (col("latitude").between(-90, 90)) & (col("longitude").between(-180, 180))
    ) \
    .withColumn("magnitude_category",
        when(col("magnitude") < 2, lit("Micro (<2)"))
        .when(col("magnitude") < 4, lit("Minor (2-3.9)"))
        .when(col("magnitude") < 5, lit("Light (4-4.9)"))
        .when(col("magnitude") < 6, lit("Moderate (5-5.9)"))
        .when(col("magnitude") < 7, lit("Strong (6-6.9)"))
        .when(col("magnitude") < 8, lit("Major (7-7.9)"))
        .otherwise(lit("Great (8+)"))
    ) \
    .withColumn("depth_category",
        when(col("depth_km") < 70, lit("Shallow (0-70 km)"))
        .when(col("depth_km") < 300, lit("Intermediate (70-300 km)"))
        .otherwise(lit("Deep (300+ km)"))
    ) \
    .withColumn("is_significant",
        (col("magnitude") >= 4.5) | (col("felt_reports") > 100) | (col("alert_level").isin("orange", "red"))
    ) \
    .withColumn("energy_joules",
        pow(lit(10), lit(1.5) * col("magnitude") + lit(4.8))
    )

# Quality score
df_eq_silver = df_eq_silver \
    .withColumn("_dq_score",
        when(col("event_id").isNotNull(), lit(15)).otherwise(lit(0)) +
        when(col("event_time").isNotNull(), lit(15)).otherwise(lit(0)) +
        when(col("magnitude_valid") == True, lit(15)).otherwise(lit(0)) +
        when(col("depth_valid") == True, lit(10)).otherwise(lit(0)) +
        when(col("coords_valid") == True, lit(15)).otherwise(lit(0)) +
        when(col("status") == "reviewed", lit(10)).otherwise(lit(5)) +
        when(col("nst").isNotNull() & (col("nst") >= 10), lit(10)).otherwise(lit(0)) +
        when(col("gap").isNotNull() & (col("gap") < 180), lit(5)).otherwise(lit(0)) +
        when(col("rms").isNotNull() & (col("rms") < 1.0), lit(5)).otherwise(lit(0))
    )

# Deduplication (USGS may update events)
from pyspark.sql.window import Window
w = Window.partitionBy("event_id").orderBy(col("updated_at").desc())
df_eq_deduped = df_eq_silver \
    .withColumn("_row_num", row_number().over(w)) \
    .filter(col("_row_num") == 1) \
    .drop("_row_num")

# Standardize water measurement units
WATER_PARAM_UNITS = {
    "00060": ("Discharge", "cfs"),          # Cubic feet per second
    "00065": ("Gage Height", "ft"),          # Feet
    "00010": ("Water Temperature", "deg C"),
    "00300": ("Dissolved Oxygen", "mg/L"),
    "00400": ("pH", "std units"),
    "00095": ("Specific Conductance", "uS/cm"),
}

param_name_expr = create_map([lit(x) for pair in
    {k: v[0] for k, v in WATER_PARAM_UNITS.items()}.items() for x in pair])
param_unit_expr = create_map([lit(x) for pair in
    {k: v[1] for k, v in WATER_PARAM_UNITS.items()}.items() for x in pair])

df_water_silver = df_water_bronze \
    .withColumn("parameter_name_std",
        coalesce(param_name_expr[col("parameter_code")], col("parameter_name"))
    ) \
    .withColumn("unit_std",
        coalesce(param_unit_expr[col("parameter_code")], col("unit"))
    ) \
    .withColumn("value_valid",
        when(col("value").isNull(), lit(False))
        .when((col("parameter_code") == "00060") & (col("value") < 0), lit(False))
        .when((col("parameter_code") == "00400") & ~col("value").between(0, 14), lit(False))
        .otherwise(lit(True))
    ) \
    .withColumn("state_std", upper(trim(col("state_code")))) \
    .withColumn("is_provisional", col("qualification_code") == "P")

# Deduplication
df_water_deduped = df_water_silver.dropDuplicates(["site_no", "measurement_time", "parameter_code"])

# Clean NPS visitation data
df_nps_silver = df_nps_bronze \
    .withColumn("park_code_std", upper(trim(col("park_code")))) \
    .withColumn("visitors_valid",
        (col("recreation_visitors") >= 0) & (col("total_visitors") >= 0)
    ) \
    .withColumn("visit_date",
        to_date(concat(col("year"), lit("-"), lpad(col("month"), 2, "0"), lit("-01")))
    ) \
    .withColumn("visitors_per_acre",
        when(col("acres") > 0,
            round(col("recreation_visitors") / col("acres"), 2)
        ).otherwise(lit(None))
    ) \
    .withColumn("season",
        when(col("month").isin(12, 1, 2), lit("Winter"))
        .when(col("month").isin(3, 4, 5), lit("Spring"))
        .when(col("month").isin(6, 7, 8), lit("Summer"))
        .otherwise(lit("Fall"))
    )

# Classify and enrich species data
df_species_silver = df_fws_bronze \
    .withColumn("listing_status_std", upper(trim(col("listing_status")))) \
    .withColumn("conservation_priority",
        when(col("listing_status_std") == "ENDANGERED", lit("Critical"))
        .when(col("listing_status_std") == "THREATENED", lit("High"))
        .otherwise(lit("Monitor"))
    ) \
    .withColumn("has_critical_habitat", coalesce(col("critical_habitat"), lit(False))) \
    .withColumn("has_recovery_plan", coalesce(col("recovery_plan"), lit(False))) \
    .withColumn("num_range_states",
        size(split(col("range_states"), ","))
    ) \
    .withColumn("years_listed",
        when(col("listing_date").isNotNull(),
            datediff(current_date(), col("listing_date")) / 365
        ).otherwise(lit(None))
    )

# Write all Silver tables
df_eq_deduped.write.format("delta").mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable("lh_silver.silver_doi_earthquakes")
spark.sql("OPTIMIZE lh_silver.silver_doi_earthquakes ZORDER BY (latitude, longitude, event_time)")

df_water_deduped.write.format("delta").mode("overwrite") \
    .option("overwriteSchema", "true") \
    .partitionBy("state_std") \
    .saveAsTable("lh_silver.silver_doi_water_data")
spark.sql("OPTIMIZE lh_silver.silver_doi_water_data ZORDER BY (site_no, measurement_time)")

df_nps_silver.write.format("delta").mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable("lh_silver.silver_doi_nps_visitation")

df_species_silver.write.format("delta").mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable("lh_silver.silver_doi_fws_species")

print(f"Silver earthquakes: {df_eq_deduped.count():,}")
print(f"Silver water data: {df_water_deduped.count():,}")
print(f"Silver NPS visitation: {df_nps_silver.count():,}")
print(f"Silver FWS species: {df_species_silver.count():,}")

# Eventstream ingestion: USGS Earthquake API -> Eventhouse
import json
import requests
import time
from datetime import datetime, timedelta

USGS_EARTHQUAKE_API = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary"

def fetch_recent_earthquakes(timeframe: str = "hour", min_mag: str = "2.5") -> list:
    """Fetch recent earthquakes from USGS GeoJSON feed.
    timeframe: hour, day, week, month
    min_mag: significant, 4.5, 2.5, 1.0, all
    """
    url = f"{USGS_EARTHQUAKE_API}/{min_mag}_{timeframe}.geojson"
    response = requests.get(url, timeout=15)
    if response.status_code != 200:
        return []

    data = response.json()
    events = []
    for feature in data["features"]:
        props = feature["properties"]
        coords = feature["geometry"]["coordinates"]
        events.append({
            "event_id": feature["id"],
            "event_time": datetime.utcfromtimestamp(props["time"] / 1000).isoformat(),
            "magnitude": props["mag"],
            "magnitude_type": props.get("magType"),
            "depth_km": coords[2],
            "latitude": coords[1],
            "longitude": coords[0],
            "place": props.get("place"),
            "alert_level": props.get("alert"),
            "tsunami": bool(props.get("tsunami", 0)),
            "significance": props.get("sig"),
            "felt_reports": props.get("felt"),
            "status": props.get("status"),
            "ingestion_time": datetime.utcnow().isoformat(),
        })
    return events

# Polling loop for Eventstream
while True:
    earthquakes = fetch_recent_earthquakes("hour", "2.5")
    for eq in earthquakes:
        emit_event(json.dumps(eq))
    print(f"[{datetime.utcnow().isoformat()}] Emitted {len(earthquakes)} earthquake events")
    time.sleep(300)  # 5-minute polling interval

# Seismic risk index by state/region
df_seismic_risk = df_eq_deduped \
    .filter(col("magnitude_valid") == True) \
    .filter(col("coords_valid") == True) \
    .groupBy("state", "region") \
    .agg(
        count("*").alias("total_events"),
        sum(when(col("magnitude") >= 4.0, 1).otherwise(0)).alias("significant_events"),
        sum(when(col("magnitude") >= 6.0, 1).otherwise(0)).alias("major_events"),
        avg("magnitude").alias("avg_magnitude"),
        max("magnitude").alias("max_magnitude"),
        avg("depth_km").alias("avg_depth_km"),
        sum(col("energy_joules")).alias("total_energy_released"),
        sum(when(col("tsunami") == True, 1).otherwise(0)).alias("tsunami_events"),
        sum("felt_reports").alias("total_felt_reports"),
    ) \
    .withColumn("events_per_year",
        round(col("total_events") / 5.0, 1)  # Assuming 5-year window
    )

# Composite risk score (0-100)
from pyspark.sql.window import Window
w_rank = Window.orderBy(col("total_events").desc())
df_seismic_risk = df_seismic_risk \
    .withColumn("freq_rank", percent_rank().over(w_rank)) \
    .withColumn("mag_rank", percent_rank().over(Window.orderBy(col("max_magnitude").desc()))) \
    .withColumn("felt_rank", percent_rank().over(Window.orderBy(col("total_felt_reports").desc()))) \
    .withColumn("seismic_risk_index",
        round((col("freq_rank") * 40 + col("mag_rank") * 35 + col("felt_rank") * 25) * 100, 1)
    )

df_seismic_risk.write.format("delta").mode("overwrite") \
    .saveAsTable("lh_gold.gold_doi_seismic_risk")

# Water resource summary by state and watershed
df_water_resources = df_water_deduped \
    .filter(col("value_valid") == True) \
    .filter(col("parameter_code") == "00060")  # Streamflow discharge
    .groupBy("state_std", "huc_code", "site_type") \
    .agg(
        countDistinct("site_no").alias("monitoring_sites"),
        avg("value").alias("avg_discharge_cfs"),
        min("value").alias("min_discharge_cfs"),
        max("value").alias("max_discharge_cfs"),
        percentile_approx("value", 0.10).alias("p10_discharge"),
        percentile_approx("value", 0.90).alias("p90_discharge"),
        count("*").alias("total_measurements"),
        sum(when(col("is_provisional") == True, 1).otherwise(0)).alias("provisional_count"),
    ) \
    .withColumn("flow_variability",
        when(col("avg_discharge_cfs") > 0,
            round((col("p90_discharge") - col("p10_discharge")) / col("avg_discharge_cfs"), 2)
        ).otherwise(lit(None))
    )

df_water_resources.write.format("delta").mode("overwrite") \
    .saveAsTable("lh_gold.gold_doi_water_resources")

# NPS park performance metrics
df_park_performance = df_nps_silver \
    .groupBy("park_code_std", "park_name", "park_type", "state", "region",
             "latitude", "longitude", "acres") \
    .agg(
        sum("recreation_visitors").alias("total_recreation_visitors"),
        avg("recreation_visitors").alias("avg_monthly_visitors"),
        max("recreation_visitors").alias("peak_monthly_visitors"),
        min("recreation_visitors").alias("min_monthly_visitors"),
        countDistinct("year").alias("years_of_data"),

        # Seasonal patterns
        sum(when(col("season") == "Summer", col("recreation_visitors")).otherwise(0))
            .alias("summer_visitors"),
        sum(when(col("season") == "Winter", col("recreation_visitors")).otherwise(0))
            .alias("winter_visitors"),
    ) \
    .withColumn("seasonality_ratio",
        when(col("winter_visitors") > 0,
            round(col("summer_visitors") / col("winter_visitors"), 2)
        ).otherwise(lit(None))
    ) \
    .withColumn("visitors_per_acre",
        when(col("acres") > 0,
            round(col("total_recreation_visitors") / col("acres"), 0)
        ).otherwise(lit(None))
    ) \
    .withColumn("park_tier",
        when(col("total_recreation_visitors") >= 10000000, lit("Tier 1 - Iconic (10M+)"))
        .when(col("total_recreation_visitors") >= 1000000, lit("Tier 2 - Major (1M-10M)"))
        .when(col("total_recreation_visitors") >= 100000, lit("Tier 3 - Regional (100K-1M)"))
        .otherwise(lit("Tier 4 - Local (<100K)"))
    )

df_park_performance.write.format("delta").mode("overwrite") \
    .saveAsTable("lh_gold.gold_doi_park_performance")

# Biodiversity indicators from FWS species data
df_biodiversity = df_species_silver \
    .groupBy("taxonomic_group", "conservation_priority") \
    .agg(
        count("*").alias("species_count"),
        sum(when(col("has_critical_habitat") == True, 1).otherwise(0)).alias("with_critical_habitat"),
        sum(when(col("has_recovery_plan") == True, 1).otherwise(0)).alias("with_recovery_plan"),
        avg("num_range_states").alias("avg_range_states"),
        avg("years_listed").alias("avg_years_listed"),
    ) \
    .withColumn("recovery_plan_coverage_pct",
        round(col("with_recovery_plan") * 100.0 / col("species_count"), 1)
    ) \
    .withColumn("habitat_designation_pct",
        round(col("with_critical_habitat") * 100.0 / col("species_count"), 1)
    )

df_biodiversity.write.format("delta").mode("overwrite") \
    .saveAsTable("lh_gold.gold_doi_biodiversity")

# Summary
for table in ["gold_doi_seismic_risk", "gold_doi_water_resources",
              "gold_doi_park_performance", "gold_doi_biodiversity"]:
    count = spark.table(f"lh_gold.{table}").count()
    print(f"  {table}: {count:,} rows")

# Data quality checks for DOI natural resources data
from great_expectations.core import ExpectationSuite

doi_expectations = ExpectationSuite(expectation_suite_name="doi_natural_resources_suite")

# Earthquake magnitude range
doi_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "magnitude", "min_value": -2, "max_value": 10, "mostly": 0.999}
})

# Earthquake depth range
doi_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "depth_km", "min_value": 0, "max_value": 700, "mostly": 0.999}
})

# Latitude/longitude validity
doi_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "latitude", "min_value": -90, "max_value": 90}
})

doi_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "longitude", "min_value": -180, "max_value": 180}
})

# Streamflow non-negative
doi_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "value", "min_value": 0, "mostly": 0.99}
})

# Park visitors non-negative
doi_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "recreation_visitors", "min_value": 0}
})

print(f"DOI validation suite: {len(doi_expectations.expectations)} expectations")

# Purview configuration for DOI data
purview_config = {
    "lh_bronze.bronze_doi_earthquakes": {
        "sensitivity_label": "Public - Federal Scientific Data",
        "attribution": "USGS Earthquake Hazards Program",
        "citation": "U.S. Geological Survey, Earthquake Hazards Program (YYYY): ANSS Comprehensive Earthquake Catalog",
        "data_source_url": "https://earthquake.usgs.gov/",
        "update_frequency": "Real-time (seconds) + daily batch review",
    },
    "lh_bronze.bronze_doi_water_data": {
        "sensitivity_label": "Public - Federal Scientific Data",
        "attribution": "USGS National Water Information System (NWIS)",
        "citation": "U.S. Geological Survey, YYYY, National Water Information System data",
        "provisional_data_note": "Recent data may be provisional and subject to revision",
    },
    "lh_bronze.bronze_doi_nps_visitation": {
        "sensitivity_label": "Public - Federal Recreation Data",
        "attribution": "National Park Service, Integrated Resource Management Applications (IRMA)",
        "research_permit_note": "Published research using NPS data may require park superintendent approval",
    },
    "lh_gold.gold_doi_seismic_risk": {
        "sensitivity_label": "Internal - Derived Analytics",
        "disclaimer": "Seismic risk index is a derived composite metric, not an official USGS product",
        "methodology": "Composite of event frequency, maximum magnitude, and felt reports",
    },
}

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