📹 Tutorial 27: Video Security Analytics¶

# Zone crossing detection logic (runs on edge AI node)
def detect_zone_crossing(
    track_id: str,
    current_position: tuple[int, int],
    zone_boundaries: dict[str, list[tuple[int, int]]],
    track_history: dict[str, str]
) -> dict | None:
    """
    Detect when a tracked object crosses from one casino zone to another.

    Args:
        track_id: DeepSORT persistent tracking ID
        current_position: (x, y) pixel coordinates in camera frame
        zone_boundaries: Polygon boundaries for each zone
        track_history: Previous zone assignment per track_id

    Returns:
        Zone crossing event dict, or None if no crossing occurred
    """
    from shapely.geometry import Point, Polygon
    from datetime import datetime, timezone

    point = Point(current_position)
    current_zone = None

    for zone_name, boundary_coords in zone_boundaries.items():
        if Polygon(boundary_coords).contains(point):
            current_zone = zone_name
            break

    if current_zone is None:
        return None

    previous_zone = track_history.get(track_id)
    if previous_zone and previous_zone != current_zone:
        track_history[track_id] = current_zone
        return {
            "event_type": "zone_crossing",
            "track_id": track_id,
            "zone_from": previous_zone,
            "zone_to": current_zone,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "confidence_score": 0.95,
            "alert_level": "INFO"
        }

    track_history[track_id] = current_zone
    return None

# Crowd density calculation (runs in Fabric Eventstream transformation)
def calculate_density_alert(
    zone: str,
    person_count: int,
    zone_capacity: dict[str, int]
) -> str:
    """Determine crowd density alert level based on zone capacity."""
    capacity = zone_capacity.get(zone, 100)
    occupancy_pct = (person_count / capacity) * 100

    if occupancy_pct >= 90:
        return "EMERGENCY"    # Evacuation threshold
    elif occupancy_pct >= 75:
        return "CRITICAL"     # Near capacity
    elif occupancy_pct >= 50:
        return "WARNING"      # Monitor closely
    else:
        return "INFO"         # Normal operations

# Fabric Notebook: Process Video Events from Eventstreams
# ============================================================

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, from_json, current_timestamp, when, lit,
    window, count, avg, max as spark_max, min as spark_min
)
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType,
    IntegerType, TimestampType, MapType
)

spark = SparkSession.builder.getOrCreate()

# Define the video event schema (matches video_event_schema.json)
video_event_schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("camera_id", StringType(), False),
    StructField("camera_location", StringType(), False),
    StructField("event_type", StringType(), False),
    StructField("timestamp", TimestampType(), False),
    StructField("confidence_score", DoubleType(), False),
    StructField("object_class", StringType(), True),
    StructField("object_count", IntegerType(), True),
    StructField("bounding_box", StructType([
        StructField("x", DoubleType()),
        StructField("y", DoubleType()),
        StructField("width", DoubleType()),
        StructField("height", DoubleType())
    ]), True),
    StructField("track_id", StringType(), True),
    StructField("zone_from", StringType(), True),
    StructField("zone_to", StringType(), True),
    StructField("dwell_time_seconds", DoubleType(), True),
    StructField("anomaly_type", StringType(), True),
    StructField("alert_level", StringType(), False),
    StructField("frame_number", IntegerType(), True),
    StructField("video_resolution", StringType(), True),
    StructField("fps", IntegerType(), True),
    StructField("model_name", StringType(), True),
    StructField("model_version", StringType(), True),
    StructField("metadata", MapType(StringType(), StringType()), True),
    StructField("load_time", TimestampType(), False)
])

# Read from Eventstream (streaming mode)
raw_events = (
    spark.readStream
    .format("delta")
    .option("readChangeFeed", "true")
    .table("eventstream_video_security_events")
)

# Parse and validate events
parsed_events = (
    raw_events
    .select(from_json(col("body").cast("string"), video_event_schema).alias("event"))
    .select("event.*")
    .withColumn("fabric_ingestion_time", current_timestamp())
    .filter(col("confidence_score") >= 0.5)  # Drop low-confidence detections
)

print(f"Schema validated. Streaming video events into Bronze layer...")

# Fabric Notebook: 01_bronze_video_security_events.py
# ============================================================
# Bronze Layer — Raw Video Analytics Event Ingestion
# ============================================================

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, current_timestamp, input_file_name, lit,
    year, month, dayofmonth, hour
)
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType,
    IntegerType, TimestampType, MapType
)

spark = SparkSession.builder.getOrCreate()

# ------------------------------------------------------------
# Schema: matches data_generation/schemas/analytics/video_event_schema.json
# ------------------------------------------------------------
BRONZE_SCHEMA = StructType([
    StructField("event_id", StringType(), nullable=False),
    StructField("camera_id", StringType(), nullable=False),
    StructField("camera_location", StringType(), nullable=False),
    StructField("event_type", StringType(), nullable=False),
    StructField("timestamp", StringType(), nullable=False),
    StructField("confidence_score", DoubleType(), nullable=False),
    StructField("object_class", StringType(), nullable=True),
    StructField("object_count", IntegerType(), nullable=True),
    StructField("bounding_box", StructType([
        StructField("x", DoubleType()),
        StructField("y", DoubleType()),
        StructField("width", DoubleType()),
        StructField("height", DoubleType())
    ]), nullable=True),
    StructField("track_id", StringType(), nullable=True),
    StructField("zone_from", StringType(), nullable=True),
    StructField("zone_to", StringType(), nullable=True),
    StructField("dwell_time_seconds", DoubleType(), nullable=True),
    StructField("anomaly_type", StringType(), nullable=True),
    StructField("alert_level", StringType(), nullable=False),
    StructField("frame_number", IntegerType(), nullable=True),
    StructField("video_resolution", StringType(), nullable=True),
    StructField("fps", IntegerType(), nullable=True),
    StructField("model_name", StringType(), nullable=True),
    StructField("model_version", StringType(), nullable=True),
    StructField("metadata", MapType(StringType(), StringType()), nullable=True),
    StructField("load_time", StringType(), nullable=False)
])

# ------------------------------------------------------------
# Read raw events from Eventstream landing zone
# ------------------------------------------------------------
raw_video_events = (
    spark.read
    .format("json")
    .schema(BRONZE_SCHEMA)
    .load("Files/landing/video_security_events/")
)

# ------------------------------------------------------------
# Add Bronze layer metadata columns
# ------------------------------------------------------------
bronze_video_events = (
    raw_video_events
    .withColumn("_bronze_ingestion_time", current_timestamp())
    .withColumn("_bronze_source_file", input_file_name())
    .withColumn("_bronze_batch_id", lit("batch_001"))
    # Partition columns for query performance
    .withColumn("_year", year(col("timestamp").cast("timestamp")))
    .withColumn("_month", month(col("timestamp").cast("timestamp")))
    .withColumn("_day", dayofmonth(col("timestamp").cast("timestamp")))
    .withColumn("_hour", hour(col("timestamp").cast("timestamp")))
)

# ------------------------------------------------------------
# Write to Bronze Delta table (append-only, partitioned)
# ------------------------------------------------------------
(
    bronze_video_events.write
    .format("delta")
    .mode("append")
    .partitionBy("_year", "_month", "_day", "_hour")
    .saveAsTable("lh_bronze.bronze_video_security_events")
)

row_count = spark.table("lh_bronze.bronze_video_security_events").count()
print(f"Bronze video security events loaded: {row_count:,} records")
print(f"Partitioned by: _year / _month / _day / _hour")

# Quick validation after Bronze load
# ============================================================

bronze_df = spark.table("lh_bronze.bronze_video_security_events")

# Basic counts
print(f"Total records: {bronze_df.count():,}")
print(f"Distinct cameras: {bronze_df.select('camera_id').distinct().count()}")
print(f"Distinct zones: {bronze_df.select('camera_location').distinct().count()}")

# Event type distribution
bronze_df.groupBy("event_type").count().orderBy("count", ascending=False).show()

# Alert level distribution
bronze_df.groupBy("alert_level").count().orderBy("count", ascending=False).show()

# Null check on required fields
from pyspark.sql.functions import sum as spark_sum, when, col

null_checks = bronze_df.select(
    spark_sum(when(col("event_id").isNull(), 1).otherwise(0)).alias("null_event_id"),
    spark_sum(when(col("camera_id").isNull(), 1).otherwise(0)).alias("null_camera_id"),
    spark_sum(when(col("event_type").isNull(), 1).otherwise(0)).alias("null_event_type"),
    spark_sum(when(col("alert_level").isNull(), 1).otherwise(0)).alias("null_alert_level"),
)
null_checks.show()

# Fabric Notebook: 02_silver_video_security_events.py
# ============================================================
# Silver Layer — Enriched Video Security Analytics
# ============================================================

from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import (
    col, to_timestamp, current_timestamp, lit, when,
    row_number, count, avg, max as spark_max, min as spark_min,
    sum as spark_sum, datediff, expr, concat_ws,
    first, last, collect_list, size, round as spark_round
)

spark = SparkSession.builder.getOrCreate()

# ------------------------------------------------------------
# Read from Bronze
# ------------------------------------------------------------
bronze_df = spark.table("lh_bronze.bronze_video_security_events")

# Cast timestamp string to proper TimestampType
bronze_df = bronze_df.withColumn(
    "event_timestamp", to_timestamp(col("timestamp"))
)

# ------------------------------------------------------------
# 5.1 Deduplication: same track_id within 5-second window
# Keep highest confidence detection per track per window
# ------------------------------------------------------------
dedup_window = Window.partitionBy(
    "track_id",
    "camera_location",
    # 5-second tumbling window
    (col("event_timestamp").cast("long") / 5).cast("long")
).orderBy(col("confidence_score").desc())

deduped_df = (
    bronze_df
    .filter(col("track_id").isNotNull())
    .withColumn("_dedup_rank", row_number().over(dedup_window))
    .filter(col("_dedup_rank") == 1)
    .drop("_dedup_rank")
)

# Events without track_id pass through without dedup
no_track_df = bronze_df.filter(col("track_id").isNull())

silver_base = deduped_df.unionByName(no_track_df, allowMissingColumns=True)

print(f"Bronze records: {bronze_df.count():,}")
print(f"After dedup: {silver_base.count():,}")
print(f"Duplicates removed: {bronze_df.count() - silver_base.count():,}")

# ------------------------------------------------------------
# 5.2 Enrichment: zone names, severity scores, camera metadata
# ------------------------------------------------------------

# Zone display name mapping
zone_display_names = {
    "slot_floor_a": "Slot Floor - Section A",
    "slot_floor_b": "Slot Floor - Section B",
    "table_games": "Table Games Pit",
    "cage_area": "Cashier Cage",
    "entrance_main": "Main Entrance",
    "entrance_valet": "Valet Entrance",
    "parking_garage": "Parking Structure",
    "elevator_lobby": "Elevator Lobby",
    "restaurant": "Restaurant & Dining",
    "hotel_lobby": "Hotel Lobby",
    "pool_area": "Pool & Cabana",
    "convention_hall": "Convention Center",
    "back_of_house": "Back of House",
    "surveillance_room": "Surveillance Operations"
}

# Alert severity numeric score for sorting and aggregation
alert_severity_scores = {
    "INFO": 1,
    "WARNING": 2,
    "CRITICAL": 3,
    "EMERGENCY": 4
}

# Apply enrichments
from pyspark.sql.functions import create_map, coalesce

zone_map_expr = create_map(
    *[item for pair in zone_display_names.items() for item in (lit(pair[0]), lit(pair[1]))]
)

severity_map_expr = create_map(
    *[item for pair in alert_severity_scores.items() for item in (lit(pair[0]), lit(pair[1]))]
)

silver_enriched = (
    silver_base
    .withColumn("zone_display_name",
        coalesce(zone_map_expr[col("camera_location")], col("camera_location")))
    .withColumn("alert_severity_score",
        coalesce(severity_map_expr[col("alert_level")], lit(0)))
    .withColumn("is_high_priority",
        when(col("alert_level").isin("CRITICAL", "EMERGENCY"), True).otherwise(False))
    .withColumn("detection_hour", hour(col("event_timestamp")))
    .withColumn("detection_date", col("event_timestamp").cast("date"))
    .withColumn("_silver_processed_time", current_timestamp())
)

print(f"Enriched records: {silver_enriched.count():,}")

# ------------------------------------------------------------
# 5.3 Zone Activity Summary (pre-aggregated for dashboards)
# ------------------------------------------------------------

zone_activity = (
    silver_enriched
    .groupBy(
        "camera_location",
        "zone_display_name",
        "detection_date",
        "detection_hour"
    )
    .agg(
        count("*").alias("total_events"),
        spark_sum(when(col("event_type") == "object_detection", 1).otherwise(0))
            .alias("detection_count"),
        spark_sum(when(col("event_type") == "zone_crossing", 1).otherwise(0))
            .alias("zone_crossing_count"),
        spark_sum(when(col("event_type") == "anomaly", 1).otherwise(0))
            .alias("anomaly_count"),
        spark_sum(when(col("event_type") == "loitering", 1).otherwise(0))
            .alias("loitering_count"),
        spark_sum(when(col("event_type") == "crowd_density", 1).otherwise(0))
            .alias("crowd_density_count"),
        spark_sum(when(col("is_high_priority"), 1).otherwise(0))
            .alias("high_priority_count"),
        spark_round(avg("confidence_score"), 4).alias("avg_confidence"),
        spark_round(avg("dwell_time_seconds"), 2).alias("avg_dwell_time_sec"),
        spark_max("alert_severity_score").alias("max_severity_score")
    )
)

# ------------------------------------------------------------
# Write Silver tables
# ------------------------------------------------------------
(
    silver_enriched.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("lh_silver.silver_video_security_events")
)

(
    zone_activity.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("lh_silver.silver_video_zone_activity")
)

print(f"Silver video events: {silver_enriched.count():,}")
print(f"Silver zone activity: {zone_activity.count():,}")

# Fabric Notebook: 03_gold_video_security_kpis.py
# ============================================================
# Gold Layer — Security Analytics KPIs
# ============================================================

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, current_timestamp, count, sum as spark_sum, avg,
    max as spark_max, min as spark_min, when, lit, round as spark_round,
    datediff, current_date, countDistinct, expr, percent_rank
)
from pyspark.sql import Window

spark = SparkSession.builder.getOrCreate()

silver_events = spark.table("lh_silver.silver_video_security_events")

# ------------------------------------------------------------
# 6.1 Daily Security KPI Summary
# ------------------------------------------------------------
daily_kpis = (
    silver_events
    .groupBy("detection_date")
    .agg(
        count("*").alias("total_events"),
        countDistinct("camera_id").alias("active_cameras"),
        countDistinct("track_id").alias("unique_tracks"),

        # Event type breakdowns
        spark_sum(when(col("event_type") == "object_detection", 1).otherwise(0))
            .alias("detections"),
        spark_sum(when(col("event_type") == "zone_crossing", 1).otherwise(0))
            .alias("zone_crossings"),
        spark_sum(when(col("event_type") == "anomaly", 1).otherwise(0))
            .alias("anomalies"),
        spark_sum(when(col("event_type") == "loitering", 1).otherwise(0))
            .alias("loitering_events"),
        spark_sum(when(col("event_type") == "tailgating", 1).otherwise(0))
            .alias("tailgating_events"),
        spark_sum(when(col("event_type") == "abandoned_object", 1).otherwise(0))
            .alias("abandoned_objects"),
        spark_sum(when(col("event_type") == "crowd_density", 1).otherwise(0))
            .alias("crowd_density_alerts"),
        spark_sum(when(col("event_type") == "face_match", 1).otherwise(0))
            .alias("face_matches"),

        # Severity breakdowns
        spark_sum(when(col("alert_level") == "EMERGENCY", 1).otherwise(0))
            .alias("emergency_alerts"),
        spark_sum(when(col("alert_level") == "CRITICAL", 1).otherwise(0))
            .alias("critical_alerts"),
        spark_sum(when(col("alert_level") == "WARNING", 1).otherwise(0))
            .alias("warning_alerts"),

        # Quality metrics
        spark_round(avg("confidence_score"), 4).alias("avg_confidence"),
        spark_round(avg("dwell_time_seconds"), 2).alias("avg_dwell_time_sec"),
    )
    .withColumn("_gold_processed_time", current_timestamp())
)

(
    daily_kpis.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("lh_gold.gold_video_security_daily_kpis")
)

print(f"Daily KPI records: {daily_kpis.count():,}")
daily_kpis.orderBy("detection_date", ascending=False).show(5, truncate=False)

# ------------------------------------------------------------
# 6.2 Hot Zone Analysis — zones with abnormal activity levels
# ------------------------------------------------------------

zone_baseline = (
    silver_events
    .groupBy("camera_location", "zone_display_name", "detection_date")
    .agg(
        count("*").alias("daily_events"),
        spark_sum(when(col("is_high_priority"), 1).otherwise(0))
            .alias("daily_high_priority"),
        countDistinct("track_id").alias("daily_unique_tracks"),
        spark_round(avg("dwell_time_seconds"), 2).alias("daily_avg_dwell"),
    )
)

# Calculate zone-level statistics for baseline comparison
zone_stats_window = Window.partitionBy("camera_location")

hot_zone_analysis = (
    zone_baseline
    .withColumn("zone_avg_events", avg("daily_events").over(zone_stats_window))
    .withColumn("zone_std_events", expr(
        "stddev(daily_events) OVER (PARTITION BY camera_location)"
    ))
    .withColumn("z_score",
        (col("daily_events") - col("zone_avg_events")) / col("zone_std_events"))
    .withColumn("is_hot_zone",
        when(col("z_score") > 2.0, True).otherwise(False))
    .withColumn("heat_level",
        when(col("z_score") > 3.0, "EXTREME")
        .when(col("z_score") > 2.0, "HIGH")
        .when(col("z_score") > 1.0, "ELEVATED")
        .when(col("z_score") > 0.0, "NORMAL")
        .otherwise("LOW"))
    .withColumn("_gold_processed_time", current_timestamp())
)

(
    hot_zone_analysis.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("lh_gold.gold_video_hot_zone_analysis")
)

print(f"Hot zone records: {hot_zone_analysis.count():,}")
hot_zone_analysis.filter(col("is_hot_zone")).show(10, truncate=False)

# ------------------------------------------------------------
# 6.3 Incident Pattern Detection
# ------------------------------------------------------------

# Cameras with recurring anomalies (3+ in 24 hours)
incident_patterns = (
    silver_events
    .filter(col("alert_level").isin("CRITICAL", "EMERGENCY"))
    .groupBy("camera_id", "camera_location", "zone_display_name", "detection_date")
    .agg(
        count("*").alias("incident_count"),
        collect_list("event_type").alias("event_types"),
        collect_list("anomaly_type").alias("anomaly_types"),
        spark_min("event_timestamp").alias("first_incident"),
        spark_max("event_timestamp").alias("last_incident"),
        spark_round(avg("confidence_score"), 4).alias("avg_confidence"),
    )
    .filter(col("incident_count") >= 3)
    .withColumn("incident_severity",
        when(col("incident_count") >= 10, "PATTERN_CRITICAL")
        .when(col("incident_count") >= 5, "PATTERN_HIGH")
        .otherwise("PATTERN_MODERATE"))
    .withColumn("_gold_processed_time", current_timestamp())
)

(
    incident_patterns.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("lh_gold.gold_video_incident_patterns")
)

print(f"Incident patterns detected: {incident_patterns.count():,}")