Transportation Safety Analytics¶

# Databricks notebook source
# MAGIC %md
# MAGIC # Aviation Incident Trend Analysis
# MAGIC Analyze ASRS incident reports by category, severity, and temporal patterns.

# COMMAND ----------

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Load ASRS incident data from bronze layer
df_asrs = spark.read.format("delta").load(
    "Tables/bronze_dot_asrs_incidents"
)

# Parse report date and extract temporal features
df_asrs_enriched = (
    df_asrs
    .withColumn("report_date", F.to_date("report_date_str", "yyyy-MM-dd"))
    .withColumn("report_year", F.year("report_date"))
    .withColumn("report_month", F.month("report_date"))
    .withColumn("report_quarter", F.quarter("report_date"))
    .filter(F.col("report_year") >= 2015)
)

# Aggregate incident counts by category and year
df_incident_trends = (
    df_asrs_enriched
    .groupBy("report_year", "event_type", "contributing_factor_primary")
    .agg(
        F.count("*").alias("incident_count"),
        F.countDistinct("report_id").alias("unique_reports"),
        F.countDistinct("aircraft_type").alias("aircraft_types_involved")
    )
    .orderBy("report_year", F.desc("incident_count"))
)

# Calculate year-over-year change using window functions
window_yoy = Window.partitionBy("event_type").orderBy("report_year")
df_incident_trends = df_incident_trends.withColumn(
    "yoy_change_pct",
    F.round(
        (F.col("incident_count") - F.lag("incident_count").over(window_yoy))
        / F.lag("incident_count").over(window_yoy) * 100, 2
    )
)

# Identify top emerging risk categories (fastest-growing incident types)
df_emerging_risks = (
    df_incident_trends
    .filter(F.col("report_year") == 2024)
    .filter(F.col("yoy_change_pct") > 10)
    .orderBy(F.desc("yoy_change_pct"))
    .limit(10)
)

display(df_emerging_risks)

# Write to silver layer
df_incident_trends.write.format("delta").mode("overwrite").saveAsTable(
    "lh_silver.silver_dot_asrs_incident_trends"
)

# Databricks notebook source
# MAGIC %md
# MAGIC # Airline Delay Pattern Analysis
# MAGIC Analyze BTS on-time performance by carrier, airport, and delay cause.

# COMMAND ----------

from pyspark.sql import functions as F

# Load BTS on-time performance data
df_flights = spark.read.format("delta").load(
    "Tables/bronze_dot_bts_ontime_performance"
)

# Calculate delay metrics per carrier per month
df_carrier_performance = (
    df_flights
    .filter(F.col("year") >= 2020)
    .groupBy("year", "month", "reporting_airline", "reporting_airline_name")
    .agg(
        F.count("*").alias("total_flights"),
        F.sum(F.when(F.col("dep_del15") == 1, 1).otherwise(0)).alias("delayed_flights"),
        F.sum(F.when(F.col("cancelled") == 1, 1).otherwise(0)).alias("cancelled_flights"),
        F.sum(F.when(F.col("diverted") == 1, 1).otherwise(0)).alias("diverted_flights"),
        F.avg("arr_delay_minutes").alias("avg_arrival_delay_min"),
        F.sum("carrier_delay").alias("total_carrier_delay_min"),
        F.sum("weather_delay").alias("total_weather_delay_min"),
        F.sum("nas_delay").alias("total_nas_delay_min"),
        F.sum("security_delay").alias("total_security_delay_min"),
        F.sum("late_aircraft_delay").alias("total_late_aircraft_delay_min")
    )
)

# Compute on-time performance rate and reliability score
df_carrier_performance = (
    df_carrier_performance
    .withColumn(
        "on_time_pct",
        F.round((F.col("total_flights") - F.col("delayed_flights") - F.col("cancelled_flights"))
                / F.col("total_flights") * 100, 2)
    )
    .withColumn(
        "cancellation_rate_pct",
        F.round(F.col("cancelled_flights") / F.col("total_flights") * 100, 2)
    )
    .withColumn(
        "reliability_score",
        F.round(
            F.col("on_time_pct") * 0.6
            + (100 - F.col("cancellation_rate_pct")) * 0.3
            + F.least(F.lit(100), F.lit(100) - F.col("avg_arrival_delay_min")) * 0.1,
            2
        )
    )
)

# Airport congestion analysis — top 30 busiest airports
df_airport_delays = (
    df_flights
    .filter(F.col("year") == 2024)
    .groupBy("origin", "origin_city_name")
    .agg(
        F.count("*").alias("departures"),
        F.avg("dep_delay_minutes").alias("avg_dep_delay"),
        F.percentile_approx("dep_delay_minutes", 0.95).alias("p95_dep_delay"),
        F.sum(F.when(F.col("cancelled") == 1, 1).otherwise(0)).alias("cancellations")
    )
    .orderBy(F.desc("departures"))
    .limit(30)
)

display(df_airport_delays)

# Write to silver layer
df_carrier_performance.write.format("delta").mode("overwrite").saveAsTable(
    "lh_silver.silver_dot_carrier_performance"
)

# Databricks notebook source
# MAGIC %md
# MAGIC # NHTSA FARS Crash Fatality Analytics
# MAGIC Analyze fatal crash patterns by geography, time, and contributing factors.

# COMMAND ----------

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Load FARS accident-level data
df_crashes = spark.read.format("delta").load(
    "Tables/bronze_dot_fars_accidents"
)

# Load FARS person-level data for impairment and restraint analysis
df_persons = spark.read.format("delta").load(
    "Tables/bronze_dot_fars_persons"
)

# Crash severity and temporal analysis
df_crash_analysis = (
    df_crashes
    .filter(F.col("year") >= 2018)
    .withColumn("crash_hour", F.col("hour").cast("int"))
    .withColumn("is_weekend", F.when(
        F.dayofweek("crash_date").isin(1, 7), True
    ).otherwise(False))
    .withColumn("is_nighttime", F.when(
        (F.col("crash_hour") >= 20) | (F.col("crash_hour") <= 5), True
    ).otherwise(False))
    .groupBy("year", "state_name", "is_weekend", "is_nighttime")
    .agg(
        F.count("*").alias("fatal_crashes"),
        F.sum("fatals").alias("total_fatalities"),
        F.sum(F.when(F.col("drunk_dr") > 0, 1).otherwise(0)).alias("alcohol_involved_crashes"),
        F.avg("fatals").alias("avg_fatalities_per_crash")
    )
)

# Impairment analysis from person-level data
df_impairment = (
    df_persons
    .filter(F.col("year") >= 2018)
    .filter(F.col("per_typ") == 1)  # Drivers only
    .groupBy("year", "state_name")
    .agg(
        F.count("*").alias("total_drivers_in_fatal_crashes"),
        F.sum(F.when(F.col("drinking") == 1, 1).otherwise(0)).alias("drinking_drivers"),
        F.sum(F.when(F.col("drugs") == 1, 1).otherwise(0)).alias("drug_positive_drivers"),
        F.sum(F.when(F.col("rest_use").isin(0, 7, 8), 1).otherwise(0)).alias("unrestrained_drivers")
    )
    .withColumn(
        "impaired_driving_pct",
        F.round(F.col("drinking_drivers") / F.col("total_drivers_in_fatal_crashes") * 100, 2)
    )
)

# Fatality rate per 100M VMT by state (requires FHWA VMT join)
df_fatality_rates = (
    df_crash_analysis
    .groupBy("year", "state_name")
    .agg(F.sum("total_fatalities").alias("annual_fatalities"))
)

display(df_crash_analysis.orderBy("year", F.desc("total_fatalities")))

# Write to gold layer
df_crash_analysis.write.format("delta").mode("overwrite").saveAsTable(
    "lh_gold.gold_dot_crash_fatality_analysis"
)
df_impairment.write.format("delta").mode("overwrite").saveAsTable(
    "lh_gold.gold_dot_driver_impairment_analysis"
)

# Cross-domain: DOT × NOAA weather impact correlation
df_weather_events = spark.read.format("delta").load("Tables/bronze_noaa_storm_events")
df_flight_delays = spark.read.format("delta").load("Tables/silver_dot_carrier_performance")
df_fatal_crashes = spark.read.format("delta").load("Tables/gold_dot_crash_fatality_analysis")

# Join on date and state to correlate weather severity with safety outcomes
df_weather_safety = (
    df_weather_events
    .filter(F.col("event_type").isin("Winter Storm", "Thunderstorm", "Fog", "Ice Storm"))
    .groupBy("year", "month", "state")
    .agg(
        F.count("*").alias("severe_weather_events"),
        F.sum("deaths_direct").alias("weather_direct_deaths")
    )
    .join(
        df_fatal_crashes.groupBy("year", "state_name")
        .agg(F.sum("total_fatalities").alias("highway_fatalities")),
        (F.col("year") == F.col("year")) & (F.col("state") == F.col("state_name")),
        "inner"
    )
)

# Calculate correlation coefficient
correlation = df_weather_safety.stat.corr("severe_weather_events", "highway_fatalities")
print(f"Weather-Highway Fatality Correlation: {correlation:.4f}")

# Cross-domain: DOT × EPA airport emissions impact
df_airport_ops = spark.read.format("delta").load("Tables/bronze_dot_faa_opsnet")
df_air_quality = spark.read.format("delta").load("Tables/bronze_epa_air_quality")

# Join airport operations with nearby air quality monitors
df_airport_emissions = (
    df_airport_ops
    .groupBy("year", "facility_id", "facility_name")
    .agg(F.sum("total_operations").alias("annual_operations"))
    .join(
        df_air_quality
        .filter(F.col("parameter_name") == "PM2.5 - Local Conditions")
        .groupBy("year", "county_name", "state_name")
        .agg(F.avg("arithmetic_mean").alias("avg_pm25")),
        on="year",
        how="inner"
    )
)