🌿 Tutorial 35: EPA Environmental Analytics¶

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

from generators.federal.epa_generator import EPAGenerator

generator = EPAGenerator(
    output_path="/lakehouse/default/Files/raw/epa/",
    num_air_quality_records=100000,    # 100K air quality observations
    num_tri_records=10000,             # 10K toxic release records
    num_water_records=20000,           # 20K water compliance records
    num_ghg_records=5000,              # 5K GHG emission records
    date_range=("2018-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 EPA data sources
import requests

# 1. AQS API - Air Quality Data
AQS_EMAIL = "your.email@example.com"
AQS_KEY = "your_aqs_key"

def fetch_aqs_data(state_code: str, county_code: str, param: str,
                   begin_date: str, end_date: str) -> dict:
    """Fetch air quality data from EPA AQS API."""
    url = "https://aqs.epa.gov/data/api/dailyData/byCounty"
    params = {
        "email": AQS_EMAIL, "key": AQS_KEY,
        "param": param,          # 88101=PM2.5, 44201=Ozone
        "bdate": begin_date, "edate": end_date,
        "state": state_code, "county": county_code,
    }
    response = requests.get(url, params=params)
    return response.json()

# 2. TRI Bulk Download
TRI_URL = "https://enviro.epa.gov/triexplorer/release_fac?triession=TRIQQM5"
# Also: https://www.epa.gov/toxics-release-inventory-tri-program/tri-basic-data-files-calendar-years-1987-present

# 3. ECHO API - Compliance Data
def fetch_echo_facilities(state: str, program: str = "ALL") -> dict:
    """Fetch facility compliance data from ECHO."""
    url = "https://echo.epa.gov/cgi-bin/get_facilities"
    params = {
        "output": "JSON",
        "p_st": state,
        "p_med": program,
    }
    response = requests.get(url, params=params)
    return response.json()

# Example: Fetch PM2.5 data for Los Angeles County
aqs_data = fetch_aqs_data("06", "037", "88101", "20240101", "20240131")
print(f"AQS records: {len(aqs_data.get('Data', []))}")

# Schema for EPA air quality monitoring data (matches epa_air_quality_schema.json)
from pyspark.sql.types import *

air_quality_schema = StructType([
    StructField("monitor_id", StringType(), False),            # AQS site ID (state-county-site)
    StructField("site_name", StringType(), True),
    StructField("measurement_date", DateType(), False),
    StructField("measurement_time", StringType(), True),       # HH:MM local time
    StructField("parameter_code", StringType(), True),         # EPA AQS parameter code
    StructField("parameter_name", StringType(), True),         # PM2.5, Ozone, NO2, SO2, CO
    StructField("sample_duration", StringType(), True),        # 1 HOUR, 24 HOUR, etc.
    StructField("concentration", DoubleType(), True),          # Measured value
    StructField("units_of_measure", StringType(), True),       # ug/m3, ppb, ppm
    StructField("aqi_value", IntegerType(), True),             # AQI (0-500)
    StructField("aqi_category", StringType(), True),           # Good, Moderate, Unhealthy, etc.

    # Geography
    StructField("state_code", StringType(), True),
    StructField("county_code", StringType(), True),
    StructField("county_name", StringType(), True),
    StructField("latitude", DoubleType(), True),
    StructField("longitude", DoubleType(), True),
    StructField("cbsa_code", StringType(), True),              # Core-Based Statistical Area
    StructField("cbsa_name", StringType(), True),

    # Quality
    StructField("method_code", StringType(), True),            # Measurement method
    StructField("data_validity", StringType(), True),          # Valid, Suspect, Excluded
    StructField("exceptional_event", StringType(), True),      # Wildfire, dust storm, etc.
])

print(f"Air quality schema: {len(air_quality_schema.fields)} fields")

# Schema for EPA TRI (Toxic Release Inventory)
tri_release_schema = StructType([
    StructField("tri_facility_id", StringType(), False),       # EPA TRI facility ID
    StructField("facility_name", StringType(), True),
    StructField("reporting_year", IntegerType(), True),
    StructField("chemical_name", StringType(), True),
    StructField("cas_number", StringType(), True),             # CAS Registry Number
    StructField("clean_air_act_chemical", BooleanType(), True),
    StructField("carcinogen", BooleanType(), True),

    # Release quantities (pounds)
    StructField("total_releases_lbs", DoubleType(), True),
    StructField("fugitive_air_lbs", DoubleType(), True),       # Air - non-point
    StructField("stack_air_lbs", DoubleType(), True),          # Air - stack/point
    StructField("water_releases_lbs", DoubleType(), True),
    StructField("land_releases_lbs", DoubleType(), True),
    StructField("underground_injection_lbs", DoubleType(), True),
    StructField("offsite_transfers_lbs", DoubleType(), True),

    # Facility details
    StructField("industry_sector", StringType(), True),
    StructField("naics_code", StringType(), True),
    StructField("street_address", StringType(), True),
    StructField("city", StringType(), True),
    StructField("state", StringType(), True),
    StructField("zip_code", StringType(), True),
    StructField("county", StringType(), True),
    StructField("latitude", DoubleType(), True),
    StructField("longitude", DoubleType(), True),
    StructField("federal_facility", BooleanType(), True),
])

print(f"TRI schema: {len(tri_release_schema.fields)} fields")

# Schema for EPA water compliance (ECHO/SDWIS)
water_compliance_schema = StructType([
    StructField("pwsid", StringType(), False),                 # Public Water System ID
    StructField("pws_name", StringType(), True),
    StructField("violation_id", StringType(), True),
    StructField("violation_type", StringType(), True),         # MCL, TT, Monitoring, etc.
    StructField("contaminant_code", StringType(), True),
    StructField("contaminant_name", StringType(), True),
    StructField("compliance_period_begin", DateType(), True),
    StructField("compliance_period_end", DateType(), True),
    StructField("is_health_based", BooleanType(), True),       # Health-based violation
    StructField("enforcement_action", StringType(), True),

    # Water system details
    StructField("pws_type", StringType(), True),               # Community, NTNC, TNC
    StructField("population_served", IntegerType(), True),
    StructField("source_type", StringType(), True),            # Ground Water, Surface Water
    StructField("state_code", StringType(), True),
    StructField("county", StringType(), True),
    StructField("city_served", StringType(), True),
    StructField("epa_region", StringType(), True),
])

print(f"Water compliance schema: {len(water_compliance_schema.fields)} fields")

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

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

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

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

# Air quality
df_air = spark.read.schema(air_quality_schema).parquet(f"{source_path}air_quality/")
df_air_bronze = add_bronze_metadata(df_air)
df_air_bronze.write.format("delta").mode("append") \
    .partitionBy("state_code", "_load_date") \
    .saveAsTable("lh_bronze.bronze_epa_air_quality")

# TRI
df_tri = spark.read.schema(tri_release_schema).parquet(f"{source_path}tri/")
df_tri_bronze = add_bronze_metadata(df_tri)
df_tri_bronze.write.format("delta").mode("append") \
    .partitionBy("reporting_year") \
    .saveAsTable("lh_bronze.bronze_epa_tri_releases")

# Water compliance
df_water = spark.read.schema(water_compliance_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") \
    .saveAsTable("lh_bronze.bronze_epa_water_compliance")

print(f"Bronze air quality: {df_air_bronze.count():,}")
print(f"Bronze TRI releases: {df_tri_bronze.count():,}")
print(f"Bronze water compliance: {df_water_bronze.count():,}")

# EPA AQI breakpoint calculation
# Reference: https://www.airnow.gov/aqi/aqi-basics/

AQI_BREAKPOINTS = {
    "PM2.5": [  # 24-hour average (ug/m3)
        (0.0, 12.0, 0, 50),          # Good
        (12.1, 35.4, 51, 100),       # Moderate
        (35.5, 55.4, 101, 150),      # Unhealthy for Sensitive Groups
        (55.5, 150.4, 151, 200),     # Unhealthy
        (150.5, 250.4, 201, 300),    # Very Unhealthy
        (250.5, 500.4, 301, 500),    # Hazardous
    ],
    "OZONE": [  # 8-hour average (ppm)
        (0.000, 0.054, 0, 50),
        (0.055, 0.070, 51, 100),
        (0.071, 0.085, 101, 150),
        (0.086, 0.105, 151, 200),
        (0.106, 0.200, 201, 300),
    ],
    "NO2": [  # 1-hour average (ppb)
        (0, 53, 0, 50),
        (54, 100, 51, 100),
        (101, 360, 101, 150),
        (361, 649, 151, 200),
        (650, 1249, 201, 300),
        (1250, 2049, 301, 500),
    ],
}

def calculate_aqi(concentration, breakpoints):
    """Calculate AQI from concentration using EPA breakpoint formula."""
    for c_lo, c_hi, i_lo, i_hi in breakpoints:
        if c_lo <= concentration <= c_hi:
            aqi = ((i_hi - i_lo) / (c_hi - c_lo)) * (concentration - c_lo) + i_lo
            return round(aqi)
    return None  # Out of range

# Apply AQI calculation as UDF
from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType

@udf(IntegerType())
def aqi_udf(concentration, parameter):
    if concentration is None or parameter is None:
        return None
    bp = AQI_BREAKPOINTS.get(parameter.upper())
    if bp is None:
        return None
    return calculate_aqi(concentration, bp)

df_air_silver = df_air_bronze \
    .withColumn("aqi_calculated", aqi_udf(col("concentration"), col("parameter_name"))) \
    .withColumn("aqi_category_calculated",
        when(col("aqi_calculated") <= 50, lit("Good"))
        .when(col("aqi_calculated") <= 100, lit("Moderate"))
        .when(col("aqi_calculated") <= 150, lit("Unhealthy for Sensitive Groups"))
        .when(col("aqi_calculated") <= 200, lit("Unhealthy"))
        .when(col("aqi_calculated") <= 300, lit("Very Unhealthy"))
        .when(col("aqi_calculated") > 300, lit("Hazardous"))
        .otherwise(lit("Unknown"))
    )

# Normalize TRI release quantities and classify chemicals
df_tri_silver = df_tri_bronze \
    .withColumn("total_air_releases_lbs",
        coalesce(col("fugitive_air_lbs"), lit(0)) + coalesce(col("stack_air_lbs"), lit(0))
    ) \
    .withColumn("total_onsite_releases_lbs",
        coalesce(col("total_air_releases_lbs"), lit(0)) +
        coalesce(col("water_releases_lbs"), lit(0)) +
        coalesce(col("land_releases_lbs"), lit(0))
    ) \
    .withColumn("total_releases_tons", round(col("total_releases_lbs") / 2000, 2)) \
    .withColumn("state_std", upper(trim(col("state")))) \
    .withColumn("release_tier",
        when(col("total_releases_lbs") >= 1000000, lit("Major (>1M lbs)"))
        .when(col("total_releases_lbs") >= 100000, lit("Large (100K-1M lbs)"))
        .when(col("total_releases_lbs") >= 10000, lit("Medium (10K-100K lbs)"))
        .when(col("total_releases_lbs") >= 1000, lit("Small (1K-10K lbs)"))
        .otherwise(lit("Minor (<1K lbs)"))
    ) \
    .withColumn("chemical_category",
        when(col("carcinogen") == True, lit("Carcinogen"))
        .when(col("clean_air_act_chemical") == True, lit("Clean Air Act"))
        .otherwise(lit("Other Toxic"))
    )

# Deduplication
df_tri_deduped = df_tri_silver.dropDuplicates(["tri_facility_id", "reporting_year", "chemical_name"])

# Validate and enrich water compliance records
df_water_silver = df_water_bronze \
    .withColumn("state_std", upper(trim(col("state_code")))) \
    .withColumn("violation_severity",
        when(col("is_health_based") == True, lit("Health-Based"))
        .when(col("violation_type").isin("MCL", "TT"), lit("Regulatory"))
        .when(col("violation_type") == "Monitoring", lit("Monitoring"))
        .otherwise(lit("Administrative"))
    ) \
    .withColumn("population_impact_tier",
        when(col("population_served") >= 100000, lit("Large System (100K+)"))
        .when(col("population_served") >= 10000, lit("Medium System (10K-100K)"))
        .when(col("population_served") >= 1000, lit("Small System (1K-10K)"))
        .otherwise(lit("Very Small (<1K)"))
    ) \
    .withColumn("violation_duration_days",
        when(col("compliance_period_end").isNotNull(),
            datediff(col("compliance_period_end"), col("compliance_period_begin"))
        ).otherwise(lit(None))
    )

# Deduplication
df_water_deduped = df_water_silver.dropDuplicates(["violation_id"])

# Air quality DQ scoring
df_air_silver = df_air_silver \
    .withColumn("_dq_score",
        when(col("monitor_id").isNotNull(), lit(15)).otherwise(lit(0)) +
        when(col("measurement_date").isNotNull(), lit(15)).otherwise(lit(0)) +
        when(col("concentration").isNotNull() & (col("concentration") >= 0), lit(15)).otherwise(lit(0)) +
        when(col("aqi_calculated").isNotNull(), lit(15)).otherwise(lit(0)) +
        when(col("parameter_name").isNotNull(), lit(10)).otherwise(lit(0)) +
        when(col("data_validity") == "Valid", lit(10)).otherwise(lit(0)) +
        when(col("latitude").isNotNull() & col("longitude").isNotNull(), lit(10)).otherwise(lit(0)) +
        when(col("state_code").isNotNull(), lit(5)).otherwise(lit(0)) +
        when(col("method_code").isNotNull(), lit(5)).otherwise(lit(0))
    ) \
    .withColumn("_silver_timestamp", current_timestamp())

# Write Silver tables
df_air_silver.write.format("delta").mode("overwrite") \
    .option("overwriteSchema", "true") \
    .partitionBy("state_code") \
    .saveAsTable("lh_silver.silver_epa_air_quality")

df_tri_deduped.write.format("delta").mode("overwrite") \
    .option("overwriteSchema", "true") \
    .partitionBy("reporting_year") \
    .saveAsTable("lh_silver.silver_epa_tri_releases")

df_water_deduped.write.format("delta").mode("overwrite") \
    .option("overwriteSchema", "true") \
    .partitionBy("state_std") \
    .saveAsTable("lh_silver.silver_epa_water_compliance")

spark.sql("OPTIMIZE lh_silver.silver_epa_air_quality ZORDER BY (monitor_id, measurement_date)")
spark.sql("OPTIMIZE lh_silver.silver_epa_tri_releases ZORDER BY (state_std, chemical_name)")
spark.sql("OPTIMIZE lh_silver.silver_epa_water_compliance ZORDER BY (state_std, violation_severity)")

print(f"Silver air quality: {df_air_silver.count():,}")
print(f"Silver TRI releases: {df_tri_deduped.count():,}")
print(f"Silver water compliance: {df_water_deduped.count():,}")

# Eventstream ingestion: AirNow API -> Eventhouse for real-time AQI
import json
import requests
import time
from datetime import datetime

AIRNOW_API_KEY = os.environ.get("AIRNOW_API_KEY", "YOUR_KEY_HERE")

# Monitor AQI for major metro areas
MONITORED_ZIPS = [
    "10001", "90001", "60601", "77001", "85001",
    "30301", "80201", "94101", "33101", "98101",
]

def fetch_current_aqi(zip_code: str) -> list:
    """Fetch current AQI from AirNow for a ZIP code."""
    url = "https://www.airnowapi.org/aq/observation/zipCode/current/"
    params = {
        "format": "application/json",
        "zipCode": zip_code,
        "distance": 25,
        "API_KEY": AIRNOW_API_KEY,
    }
    response = requests.get(url, params=params, timeout=10)
    return response.json() if response.status_code == 200 else []

# Polling loop for Eventstream
while True:
    for zip_code in MONITORED_ZIPS:
        observations = fetch_current_aqi(zip_code)
        for obs in observations:
            event = {
                "zip_code": zip_code,
                "parameter": obs.get("ParameterName"),
                "aqi": obs.get("AQI"),
                "category": obs.get("Category", {}).get("Name"),
                "reporting_area": obs.get("ReportingArea"),
                "state": obs.get("StateCode"),
                "latitude": obs.get("Latitude"),
                "longitude": obs.get("Longitude"),
                "event_time": datetime.utcnow().isoformat(),
            }
            emit_event(json.dumps(event))
    time.sleep(3600)  # Hourly polling (AirNow updates hourly)

# AQI summary by state and monitoring site
df_aqi_index = df_air_silver \
    .filter(col("data_validity") == "Valid") \
    .groupBy("state_code", "county_name", "cbsa_name", "parameter_name") \
    .agg(
        avg("aqi_calculated").alias("avg_aqi"),
        max("aqi_calculated").alias("max_aqi"),
        min("aqi_calculated").alias("min_aqi"),
        percentile_approx("aqi_calculated", 0.95).alias("p95_aqi"),
        count("*").alias("observation_count"),

        # Days in each AQI category
        sum(when(col("aqi_category_calculated") == "Good", 1).otherwise(0)).alias("days_good"),
        sum(when(col("aqi_category_calculated") == "Moderate", 1).otherwise(0)).alias("days_moderate"),
        sum(when(col("aqi_category_calculated").contains("Unhealthy"), 1).otherwise(0)).alias("days_unhealthy"),
        sum(when(col("aqi_category_calculated") == "Hazardous", 1).otherwise(0)).alias("days_hazardous"),

        # Exceptional events
        sum(when(col("exceptional_event").isNotNull(), 1).otherwise(0)).alias("exceptional_event_days"),
    ) \
    .withColumn("unhealthy_pct",
        round((col("days_unhealthy") + col("days_hazardous")) * 100.0 / col("observation_count"), 2)
    )

df_aqi_index.write.format("delta").mode("overwrite") \
    .saveAsTable("lh_gold.gold_epa_aqi_index")

# TRI facility risk profiles
df_facility_risk = df_tri_deduped \
    .groupBy("tri_facility_id", "facility_name", "state_std", "county",
             "industry_sector", "latitude", "longitude") \
    .agg(
        count("*").alias("reporting_years"),
        sum("total_releases_lbs").alias("lifetime_releases_lbs"),
        sum("total_air_releases_lbs").alias("lifetime_air_releases_lbs"),
        sum("water_releases_lbs").alias("lifetime_water_releases_lbs"),
        avg("total_releases_lbs").alias("avg_annual_releases_lbs"),
        countDistinct("chemical_name").alias("distinct_chemicals"),
        sum(when(col("carcinogen") == True, col("total_releases_lbs")).otherwise(0))
            .alias("carcinogen_releases_lbs"),
        max("reporting_year").alias("most_recent_report"),
    ) \
    .withColumn("risk_tier",
        when(col("carcinogen_releases_lbs") > 100000, lit("Critical"))
        .when(col("lifetime_releases_lbs") > 1000000, lit("High"))
        .when(col("lifetime_releases_lbs") > 100000, lit("Elevated"))
        .when(col("lifetime_releases_lbs") > 10000, lit("Moderate"))
        .otherwise(lit("Low"))
    )

df_facility_risk.write.format("delta").mode("overwrite") \
    .saveAsTable("lh_gold.gold_epa_facility_risk")

spark.sql("OPTIMIZE lh_gold.gold_epa_facility_risk ZORDER BY (state_std, risk_tier)")

# Water system compliance scorecards
df_water_scorecard = df_water_deduped \
    .groupBy("state_std", "population_impact_tier") \
    .agg(
        countDistinct("pwsid").alias("total_systems"),
        count("*").alias("total_violations"),
        sum(when(col("is_health_based") == True, 1).otherwise(0)).alias("health_violations"),
        sum(when(col("violation_type") == "MCL", 1).otherwise(0)).alias("mcl_violations"),
        sum(when(col("violation_type") == "Monitoring", 1).otherwise(0)).alias("monitoring_violations"),
        sum(when(col("enforcement_action").isNotNull(), 1).otherwise(0)).alias("enforcement_actions"),
        sum("population_served").alias("total_population_affected"),
        avg("violation_duration_days").alias("avg_violation_duration_days"),
    ) \
    .withColumn("violations_per_system",
        round(col("total_violations") * 1.0 / col("total_systems"), 2)
    ) \
    .withColumn("health_violation_rate",
        round(col("health_violations") * 100.0 / col("total_violations"), 2)
    )

df_water_scorecard.write.format("delta").mode("overwrite") \
    .saveAsTable("lh_gold.gold_epa_water_scorecard")

# Environmental justice: correlate pollution burden with demographics
# Uses EPA EJScreen methodology conceptually (actual EJScreen data would be joined here)

df_ej = df_tri_deduped \
    .groupBy("state_std", "county") \
    .agg(
        count("*").alias("tri_facilities"),
        sum("total_releases_lbs").alias("total_releases_lbs"),
        sum("carcinogen_releases_lbs").alias("carcinogen_releases_lbs"),
    )

# Join with AQI data for multi-domain environmental burden
df_aqi_county = df_air_silver \
    .groupBy("state_code", "county_name") \
    .agg(
        avg("aqi_calculated").alias("avg_aqi"),
        sum(when(col("aqi_category_calculated").contains("Unhealthy"), 1).otherwise(0))
            .alias("unhealthy_air_days"),
    )

df_water_county = df_water_deduped \
    .groupBy("state_std", "county") \
    .agg(
        sum(when(col("is_health_based") == True, 1).otherwise(0)).alias("health_water_violations"),
        sum("population_served").alias("affected_population"),
    )

# Composite environmental burden score
df_env_justice = df_ej \
    .join(df_aqi_county, (df_ej.state_std == df_aqi_county.state_code) &
          (df_ej.county == df_aqi_county.county_name), "left") \
    .join(df_water_county, ["state_std", "county"], "left") \
    .fillna(0) \
    .withColumn("env_burden_index",
        round(
            (col("tri_facilities") * 20 +
             coalesce(col("unhealthy_air_days"), lit(0)) * 30 +
             coalesce(col("health_water_violations"), lit(0)) * 25 +
             when(col("carcinogen_releases_lbs") > 0, lit(25)).otherwise(lit(0))),
            1
        )
    ) \
    .withColumn("env_burden_percentile",
        percent_rank().over(Window.orderBy(col("env_burden_index")))
    ) \
    .withColumn("ej_priority",
        when(col("env_burden_percentile") >= 0.80, lit("High Priority"))
        .when(col("env_burden_percentile") >= 0.50, lit("Moderate"))
        .otherwise(lit("Lower Burden"))
    )

df_env_justice.write.format("delta").mode("overwrite") \
    .saveAsTable("lh_gold.gold_epa_env_justice")

# Summary
for table in ["gold_epa_aqi_index", "gold_epa_facility_risk",
              "gold_epa_water_scorecard", "gold_epa_env_justice"]:
    count = spark.table(f"lh_gold.{table}").count()
    print(f"  {table}: {count:,} rows")

# Data quality checks specific to EPA environmental data
from great_expectations.core import ExpectationSuite

epa_expectations = ExpectationSuite(expectation_suite_name="epa_env_suite")

# AQI range (0-500)
epa_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "aqi_calculated", "min_value": 0, "max_value": 500}
})

# PM2.5 concentration non-negative
epa_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "concentration", "min_value": 0, "mostly": 0.999}
})

# TRI releases non-negative
epa_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "total_releases_lbs", "min_value": 0, "mostly": 0.999}
})

# State code format
epa_expectations.add_expectation({
    "expectation_type": "expect_column_value_lengths_to_equal",
    "kwargs": {"column": "state_code", "value": 2}
})

# CAS number format (XXXXXX-XX-X pattern)
epa_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_match_regex",
    "kwargs": {"column": "cas_number", "regex": r"^\d{2,7}-\d{2}-\d$", "mostly": 0.95}
})

print(f"EPA validation suite: {len(epa_expectations.expectations)} expectations")

# Purview configuration for EPA data
purview_config = {
    "lh_bronze.bronze_epa_air_quality": {
        "sensitivity_label": "Public - Federal Environmental Data",
        "attribution": "U.S. Environmental Protection Agency - Air Quality System (AQS)",
        "regulatory_context": "40 CFR Part 58 - Ambient Air Quality Surveillance",
        "update_frequency": "Hourly (real-time) + daily batch",
    },
    "lh_bronze.bronze_epa_tri_releases": {
        "sensitivity_label": "Public - Federal Environmental Data",
        "attribution": "U.S. EPA - Toxic Release Inventory (TRI)",
        "regulatory_context": "EPCRA Section 313 - Toxic Chemical Release Reporting",
        "note": "Self-reported by facilities; EPA-verified but not guaranteed",
    },
    "lh_gold.gold_epa_env_justice": {
        "sensitivity_label": "Internal - Derived Analytics",
        "disclaimer": "Environmental justice index is a derived composite metric, not an official EPA product",
        "methodology": "Composite of TRI proximity, AQI, water violations (EPA EJScreen-inspired)",
    },
}

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