📍 Tutorial 29: Geolocation Analytics¶

# GPS location event from a shuttle tracker
gps_event = {
    "event_id": "evt-gps-20240115-143022-001",
    "device_id": "SHUTTLE-NB-003",
    "device_type": "shuttle_tracker",
    "timestamp": "2024-01-15T14:30:22.000Z",
    "latitude": 36.1147,
    "longitude": -115.1728,
    "altitude_meters": 620.5,
    "accuracy_meters": 3.2,
    "speed_mps": 8.9,
    "heading_degrees": 275.0,
    "source_system": "gps",
    "battery_level": 87.0
}

# Wi-Fi triangulated position from a patron's mobile app
wifi_event = {
    "event_id": "evt-wifi-20240115-143055-042",
    "device_id": "APP-PLY00045891",
    "device_type": "patron_app",
    "timestamp": "2024-01-15T14:30:55.000Z",
    "latitude": 36.1142,
    "longitude": -115.1735,
    "accuracy_meters": 8.0,
    "floor_level": 1,
    "indoor_zone": "SLOT_ZONE_B2",
    "source_system": "wifi_triangulation",
    "battery_level": 62.0
}

# BLE position from an employee badge
ble_event = {
    "event_id": "evt-ble-20240115-143110-088",
    "device_id": "BADGE-EMP-00234",
    "device_type": "employee_badge",
    "timestamp": "2024-01-15T14:31:10.000Z",
    "latitude": 36.1143,
    "longitude": -115.1733,
    "accuracy_meters": 1.5,
    "floor_level": 1,
    "indoor_zone": "TABLE_PIT_3",
    "source_system": "ble_trilateration",
    "battery_level": 94.0
}

# UWB position from a chip tray asset tag
uwb_event = {
    "event_id": "evt-uwb-20240115-143122-005",
    "device_id": "ASSET-CHIPTRAY-0042",
    "device_type": "asset_tag",
    "timestamp": "2024-01-15T14:31:22.000Z",
    "latitude": 36.1143,
    "longitude": -115.1734,
    "accuracy_meters": 0.15,
    "floor_level": 1,
    "indoor_zone": "TABLE_PIT_3",
    "source_system": "uwb",
    "battery_level": 78.0
}

# Hybrid position event -- system selects best available fix
hybrid_event = {
    "event_id": "evt-hyb-20240115-143200-112",
    "device_id": "APP-PLY00045891",
    "device_type": "patron_app",
    "timestamp": "2024-01-15T14:32:00.000Z",
    "latitude": 36.1142,
    "longitude": -115.1735,
    "accuracy_meters": 2.0,
    "floor_level": 1,
    "indoor_zone": "SLOT_ZONE_B2",
    "source_system": "hybrid",
    "battery_level": 61.0
}

# Geofence zone definitions for a Las Vegas casino resort
# Save as: data_generation/config/geofences.py

CASINO_GEOFENCES = [
    # ----- Gaming Floor -----
    {
        "geofence_id": "GF-MAIN-FLOOR",
        "geofence_name": "Main Casino Floor",
        "type": "polygon",
        "coordinates": [
            [36.1140, -115.1740], [36.1140, -115.1720],
            [36.1150, -115.1720], [36.1150, -115.1740]
        ],
        "floor_level": 1,
        "category": "gaming",
        "triggers": ["enter", "exit", "dwell"],
        "dwell_threshold_seconds": 300,
        "marketing_eligible": True
    },
    {
        "geofence_id": "GF-HIGH-LIMIT",
        "geofence_name": "High Limit Salon",
        "type": "circular",
        "center": [36.1145, -115.1730],
        "radius_meters": 25,
        "floor_level": 1,
        "category": "gaming_vip",
        "triggers": ["enter", "exit", "dwell"],
        "dwell_threshold_seconds": 120,
        "marketing_eligible": True
    },
    {
        "geofence_id": "GF-SPORTSBOOK",
        "geofence_name": "Sportsbook Lounge",
        "type": "polygon",
        "coordinates": [
            [36.1148, -115.1738], [36.1148, -115.1732],
            [36.1152, -115.1732], [36.1152, -115.1738]
        ],
        "floor_level": 1,
        "category": "gaming",
        "triggers": ["enter", "exit", "dwell"],
        "dwell_threshold_seconds": 600,
        "marketing_eligible": True
    },
    # ----- Hospitality -----
    {
        "geofence_id": "GF-BUFFET",
        "geofence_name": "Grand Buffet",
        "type": "circular",
        "center": [36.1143, -115.1725],
        "radius_meters": 30,
        "floor_level": 1,
        "category": "dining",
        "triggers": ["enter", "exit", "dwell"],
        "dwell_threshold_seconds": 900,
        "marketing_eligible": True
    },
    {
        "geofence_id": "GF-STEAKHOUSE",
        "geofence_name": "Prime Steakhouse",
        "type": "circular",
        "center": [36.1147, -115.1722],
        "radius_meters": 15,
        "floor_level": 1,
        "category": "dining",
        "triggers": ["enter", "dwell"],
        "dwell_threshold_seconds": 600,
        "marketing_eligible": True
    },
    # ----- Hotel -----
    {
        "geofence_id": "GF-HOTEL-LOBBY",
        "geofence_name": "Hotel Lobby",
        "type": "polygon",
        "coordinates": [
            [36.1155, -115.1735], [36.1155, -115.1725],
            [36.1160, -115.1725], [36.1160, -115.1735]
        ],
        "floor_level": 1,
        "category": "hotel",
        "triggers": ["enter", "exit"],
        "dwell_threshold_seconds": 180,
        "marketing_eligible": False
    },
    # ----- Outdoor -----
    {
        "geofence_id": "GF-POOL-DECK",
        "geofence_name": "Pool Deck & Cabanas",
        "type": "circular",
        "center": [36.1160, -115.1730],
        "radius_meters": 60,
        "floor_level": 0,
        "category": "amenity",
        "triggers": ["enter", "exit", "dwell"],
        "dwell_threshold_seconds": 1800,
        "marketing_eligible": True
    },
    {
        "geofence_id": "GF-PARKING-A",
        "geofence_name": "Parking Garage A",
        "type": "polygon",
        "coordinates": [
            [36.1130, -115.1745], [36.1130, -115.1735],
            [36.1138, -115.1735], [36.1138, -115.1745]
        ],
        "floor_level": 0,
        "category": "parking",
        "triggers": ["enter", "exit"],
        "dwell_threshold_seconds": 60,
        "marketing_eligible": False
    },
    {
        "geofence_id": "GF-VALET",
        "geofence_name": "Valet Drop-Off",
        "type": "circular",
        "center": [36.1152, -115.1740],
        "radius_meters": 20,
        "floor_level": 0,
        "category": "parking",
        "triggers": ["enter", "exit"],
        "dwell_threshold_seconds": 120,
        "marketing_eligible": False
    }
]

# Geofence event evaluation logic
# This runs in the Silver layer Spark job

from shapely.geometry import Point, Polygon
from shapely.ops import nearest_points

def evaluate_geofence(lat: float, lon: float, geofences: list) -> dict:
    """
    Evaluate which geofence zone a point falls within.
    Returns geofence_id, geofence_name, and event type.
    """
    point = Point(lon, lat)  # Shapely uses (x, y) = (lon, lat)

    for fence in geofences:
        if fence["type"] == "circular":
            center = Point(fence["center"][1], fence["center"][0])
            # Approximate: 1 degree lat ~111,320 meters
            radius_deg = fence["radius_meters"] / 111320.0
            if point.distance(center) <= radius_deg:
                return {
                    "geofence_id": fence["geofence_id"],
                    "geofence_name": fence["geofence_name"],
                    "category": fence["category"]
                }

        elif fence["type"] == "polygon":
            coords = [(c[1], c[0]) for c in fence["coordinates"]]
            polygon = Polygon(coords)
            if polygon.contains(point):
                return {
                    "geofence_id": fence["geofence_id"],
                    "geofence_name": fence["geofence_name"],
                    "category": fence["category"]
                }

    return {"geofence_id": None, "geofence_name": None, "category": None}

# Notebook: nb_geolocation_h3_indexing
# Compute H3 spatial indices for location events

# Install h3-pyspark in your Fabric environment
# %pip install h3-pyspark h3

import h3
from pyspark.sql.functions import udf, col, lit, struct
from pyspark.sql.types import StringType, ArrayType, DoubleType

# --- UDF: Convert lat/lon to H3 index ---
@udf(returnType=StringType())
def lat_lon_to_h3(lat, lon, resolution):
    """Convert latitude/longitude to H3 hexagonal index."""
    if lat is None or lon is None:
        return None
    try:
        return h3.latlng_to_cell(lat, lon, resolution)
    except Exception:
        return None

# --- UDF: Get H3 hex center coordinates ---
@udf(returnType=ArrayType(DoubleType()))
def h3_to_center(h3_index):
    """Get the center lat/lon of an H3 hex cell."""
    if h3_index is None:
        return None
    try:
        lat, lon = h3.cell_to_latlng(h3_index)
        return [lat, lon]
    except Exception:
        return None

# --- UDF: Get H3 hex boundary for visualization ---
@udf(returnType=ArrayType(ArrayType(DoubleType())))
def h3_to_boundary(h3_index):
    """Get the boundary vertices of an H3 hex cell."""
    if h3_index is None:
        return None
    try:
        boundary = h3.cell_to_boundary(h3_index)
        return [[lat, lon] for lat, lon in boundary]
    except Exception:
        return None

# --- Apply H3 indexing to location events ---
df_locations = spark.table("bronze_geolocation_events")

df_with_h3 = (
    df_locations
    .withColumn("h3_res10", lat_lon_to_h3(col("latitude"), col("longitude"), lit(10)))
    .withColumn("h3_res11", lat_lon_to_h3(col("latitude"), col("longitude"), lit(11)))
    .withColumn("h3_res12", lat_lon_to_h3(col("latitude"), col("longitude"), lit(12)))
)

display(df_with_h3.select(
    "event_id", "latitude", "longitude", "h3_res10", "h3_res11", "h3_res12"
).limit(10))

# Aggregate patron density by H3 hex cell
# This drives the heat-map visualization in Power BI

from pyspark.sql.functions import count, countDistinct, avg, sum as spark_sum

patron_density = (
    df_with_h3
    .filter(col("device_type") == "patron_app")
    .filter(col("timestamp") >= "2024-01-15T00:00:00Z")
    .groupBy("h3_res10")
    .agg(
        count("event_id").alias("event_count"),
        countDistinct("device_id").alias("unique_patrons"),
        avg("accuracy_meters").alias("avg_accuracy"),
        avg("speed_mps").alias("avg_speed_mps")
    )
    .withColumn("hex_center", h3_to_center(col("h3_res10")))
    .withColumn("hex_boundary", h3_to_boundary(col("h3_res10")))
)

display(patron_density.orderBy(col("unique_patrons").desc()).limit(20))

# Notebook: nb_geolocation_stream_ingest
# Structured Streaming from IoT Hub via Event Hub-compatible endpoint

from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType,
    IntegerType, TimestampType
)
from pyspark.sql.functions import from_json, col, current_timestamp

# Schema matching geolocation_schema.json
location_schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("device_id", StringType(), False),
    StructField("device_type", StringType(), False),
    StructField("timestamp", StringType(), False),
    StructField("latitude", DoubleType(), False),
    StructField("longitude", DoubleType(), False),
    StructField("altitude_meters", DoubleType(), True),
    StructField("accuracy_meters", DoubleType(), True),
    StructField("speed_mps", DoubleType(), True),
    StructField("heading_degrees", DoubleType(), True),
    StructField("h3_index", StringType(), True),
    StructField("geofence_id", StringType(), True),
    StructField("geofence_name", StringType(), True),
    StructField("geofence_event", StringType(), True),
    StructField("geofence_dwell_seconds", DoubleType(), True),
    StructField("poi_name", StringType(), True),
    StructField("poi_distance_meters", DoubleType(), True),
    StructField("floor_level", IntegerType(), True),
    StructField("indoor_zone", StringType(), True),
    StructField("proximity_trigger", StringType(), True),
    StructField("source_system", StringType(), False),
    StructField("battery_level", DoubleType(), True),
    StructField("load_time", StringType(), True)
])

# Read from Eventstream (Event Hub-compatible endpoint)
raw_stream = (
    spark.readStream
    .format("eventhubs")
    .options(**eh_conf)
    .load()
    .select(from_json(col("body").cast("string"), location_schema).alias("data"))
    .select("data.*")
    .withColumn("load_time", current_timestamp())
)

# Write to Bronze Delta table
(
    raw_stream.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "Files/checkpoints/geolocation_bronze")
    .toTable("bronze_geolocation_events")
)

# Notebook: nb_bronze_geolocation
# Ingest raw geolocation events into Bronze Delta table

from pyspark.sql.functions import (
    col, lit, current_timestamp, to_timestamp,
    year, month, dayofmonth, hour
)

# Read raw events (batch mode for initial/backfill load)
raw_df = spark.read.json("Files/raw/geolocation_events/*.json")

# Minimal transformations: type casting + partition columns
bronze_df = (
    raw_df
    .withColumn("event_timestamp", to_timestamp(col("timestamp")))
    .withColumn("load_timestamp", current_timestamp())
    .withColumn("year", year(col("event_timestamp")))
    .withColumn("month", month(col("event_timestamp")))
    .withColumn("day", dayofmonth(col("event_timestamp")))
    .withColumn("hour", hour(col("event_timestamp")))
)

# Write to Bronze Delta table (partitioned by date for query performance)
(
    bronze_df.write
    .format("delta")
    .mode("append")
    .partitionBy("year", "month", "day")
    .saveAsTable("bronze_geolocation_events")
)

record_count = spark.table("bronze_geolocation_events").count()
print(f"Bronze geolocation events: {record_count:,} records")

# Quick validation of Bronze data quality
bronze = spark.table("bronze_geolocation_events")

# Coordinate range check
invalid_coords = bronze.filter(
    (col("latitude") < -90) | (col("latitude") > 90) |
    (col("longitude") < -180) | (col("longitude") > 180)
).count()

print(f"Invalid coordinates: {invalid_coords}")
assert invalid_coords == 0, "Found events with out-of-range coordinates"

# Device type distribution
bronze.groupBy("device_type").count().orderBy("count", ascending=False).show()

# Source system distribution
bronze.groupBy("source_system").count().orderBy("count", ascending=False).show()

# Notebook: nb_silver_geolocation
# Silver layer: spatial enrichment pipeline

import h3
from pyspark.sql.functions import (
    udf, col, lag, lead, lit, when, coalesce,
    unix_timestamp, sqrt, pow as spark_pow, atan2, degrees,
    broadcast
)
from pyspark.sql.types import StringType, DoubleType
from pyspark.sql.window import Window

# ---- H3 Index UDFs ----
@udf(returnType=StringType())
def compute_h3(lat, lon, res):
    if lat is None or lon is None:
        return None
    try:
        return h3.latlng_to_cell(lat, lon, res)
    except Exception:
        return None

# Read Bronze data
bronze = spark.table("bronze_geolocation_events")

# Step 1: Compute H3 indices at multiple resolutions
silver_h3 = (
    bronze
    .withColumn("h3_res9", compute_h3(col("latitude"), col("longitude"), lit(9)))
    .withColumn("h3_res10", compute_h3(col("latitude"), col("longitude"), lit(10)))
    .withColumn("h3_res11", compute_h3(col("latitude"), col("longitude"), lit(11)))
)

print("H3 indexing complete")

# Step 2: Evaluate geofence membership using broadcast join
# Load geofence definitions as a Spark DataFrame

from shapely.geometry import Point, Polygon

geofence_data = [
    ("GF-MAIN-FLOOR", "Main Casino Floor", "polygon",
     [[36.1140,-115.1740],[36.1140,-115.1720],[36.1150,-115.1720],[36.1150,-115.1740]],
     None, None, "gaming"),
    ("GF-HIGH-LIMIT", "High Limit Salon", "circular",
     None, [36.1145, -115.1730], 25.0, "gaming_vip"),
    ("GF-SPORTSBOOK", "Sportsbook Lounge", "polygon",
     [[36.1148,-115.1738],[36.1148,-115.1732],[36.1152,-115.1732],[36.1152,-115.1738]],
     None, None, "gaming"),
    ("GF-BUFFET", "Grand Buffet", "circular",
     None, [36.1143, -115.1725], 30.0, "dining"),
    ("GF-POOL-DECK", "Pool Deck & Cabanas", "circular",
     None, [36.1160, -115.1730], 60.0, "amenity"),
    ("GF-PARKING-A", "Parking Garage A", "polygon",
     [[36.1130,-115.1745],[36.1130,-115.1735],[36.1138,-115.1735],[36.1138,-115.1745]],
     None, None, "parking"),
]

@udf(returnType=StringType())
def match_geofence_id(lat, lon):
    """Match a coordinate to the first containing geofence."""
    if lat is None or lon is None:
        return None
    point = Point(lon, lat)
    for gf in geofence_data:
        gf_id, _, gf_type, coords, center, radius, _ = gf
        if gf_type == "circular" and center and radius:
            center_pt = Point(center[1], center[0])
            if point.distance(center_pt) <= radius / 111320.0:
                return gf_id
        elif gf_type == "polygon" and coords:
            poly = Polygon([(c[1], c[0]) for c in coords])
            if poly.contains(point):
                return gf_id
    return None

@udf(returnType=StringType())
def match_geofence_name(lat, lon):
    """Return the name of the matched geofence."""
    if lat is None or lon is None:
        return None
    point = Point(lon, lat)
    for gf in geofence_data:
        _, gf_name, gf_type, coords, center, radius, _ = gf
        if gf_type == "circular" and center and radius:
            center_pt = Point(center[1], center[0])
            if point.distance(center_pt) <= radius / 111320.0:
                return gf_name
        elif gf_type == "polygon" and coords:
            poly = Polygon([(c[1], c[0]) for c in coords])
            if poly.contains(point):
                return gf_name
    return None

# Apply geofence matching
silver_geo = (
    silver_h3
    .withColumn("matched_geofence_id", match_geofence_id(col("latitude"), col("longitude")))
    .withColumn("matched_geofence_name", match_geofence_name(col("latitude"), col("longitude")))
)

print("Geofence matching complete")

# Step 3: Calculate distance to nearest POI

POI_LIST = [
    ("POI-SLOTS-A", "Slot Zone A", 36.1142, -115.1733),
    ("POI-SLOTS-B", "Slot Zone B", 36.1144, -115.1728),
    ("POI-TABLE-PIT-1", "Table Pit 1", 36.1146, -115.1731),
    ("POI-TABLE-PIT-2", "Table Pit 2", 36.1148, -115.1729),
    ("POI-CASHIER", "Main Cashier", 36.1141, -115.1726),
    ("POI-BAR-CENTER", "Center Bar", 36.1145, -115.1730),
    ("POI-BUFFET-ENTRANCE", "Buffet Entrance", 36.1143, -115.1724),
    ("POI-STEAKHOUSE", "Prime Steakhouse", 36.1147, -115.1722),
    ("POI-VIP-LOUNGE", "VIP Lounge", 36.1149, -115.1735),
    ("POI-SPORTSBOOK", "Sportsbook Window", 36.1150, -115.1736),
]

@udf(returnType=StringType())
def nearest_poi_name(lat, lon):
    """Find the name of the nearest POI."""
    if lat is None or lon is None:
        return None
    min_dist = float("inf")
    nearest = None
    for _, name, plat, plon in POI_LIST:
        dist = ((lat - plat)**2 + (lon - plon)**2) ** 0.5
        if dist < min_dist:
            min_dist = dist
            nearest = name
    return nearest

@udf(returnType=DoubleType())
def nearest_poi_distance(lat, lon):
    """Calculate distance in meters to nearest POI."""
    if lat is None or lon is None:
        return None
    min_dist = float("inf")
    for _, _, plat, plon in POI_LIST:
        dlat = (lat - plat) * 111320
        dlon = (lon - plon) * 111320 * 0.833  # cos(36 deg) approx
        dist = (dlat**2 + dlon**2) ** 0.5
        if dist < min_dist:
            min_dist = dist
    return round(min_dist, 2)

silver_poi = (
    silver_geo
    .withColumn("nearest_poi", nearest_poi_name(col("latitude"), col("longitude")))
    .withColumn("poi_dist_meters", nearest_poi_distance(col("latitude"), col("longitude")))
)

# Step 4: Derive speed and heading from consecutive location fixes

device_window = Window.partitionBy("device_id").orderBy("event_timestamp")

silver_motion = (
    silver_poi
    .withColumn("prev_lat", lag("latitude", 1).over(device_window))
    .withColumn("prev_lon", lag("longitude", 1).over(device_window))
    .withColumn("prev_ts", lag("event_timestamp", 1).over(device_window))
    .withColumn(
        "time_delta_sec",
        unix_timestamp(col("event_timestamp")) - unix_timestamp(col("prev_ts"))
    )
    .withColumn(
        "distance_meters",
        when(
            col("prev_lat").isNotNull(),
            sqrt(
                spark_pow((col("latitude") - col("prev_lat")) * 111320, 2) +
                spark_pow((col("longitude") - col("prev_lon")) * 111320 * 0.833, 2)
            )
        )
    )
    .withColumn(
        "derived_speed_mps",
        when(
            (col("time_delta_sec") > 0) & (col("distance_meters").isNotNull()),
            col("distance_meters") / col("time_delta_sec")
        )
    )
    .withColumn(
        "derived_heading_deg",
        when(
            col("prev_lat").isNotNull(),
            degrees(atan2(
                col("longitude") - col("prev_lon"),
                col("latitude") - col("prev_lat")
            ))
        )
    )
    .drop("prev_lat", "prev_lon", "prev_ts")
)

# Optional: Apache Sedona for advanced spatial joins
# %pip install apache-sedona

from sedona.spark import SedonaContext

sedona = SedonaContext.create(spark)

# Register location data as spatial DataFrame
sedona.sql("""
    CREATE OR REPLACE TEMP VIEW location_points AS
    SELECT *,
           ST_Point(CAST(longitude AS DECIMAL(10,6)),
                    CAST(latitude AS DECIMAL(10,6))) AS geom
    FROM silver_geolocation
""")

# Register geofence polygons
sedona.sql("""
    CREATE OR REPLACE TEMP VIEW geofence_polygons AS
    SELECT geofence_id, geofence_name, category,
           ST_GeomFromWKT(wkt_geometry) AS geom
    FROM geofence_definitions
""")

# Spatial join: find all points within each geofence
spatial_join_result = sedona.sql("""
    SELECT l.event_id, l.device_id, l.timestamp,
           g.geofence_id, g.geofence_name, g.category
    FROM location_points l
    JOIN geofence_polygons g
    ON ST_Contains(g.geom, l.geom)
""")

display(spatial_join_result.limit(20))

# Write enriched data to Silver layer
final_silver = (
    silver_motion
    .select(
        "event_id", "device_id", "device_type", "event_timestamp",
        "latitude", "longitude", "accuracy_meters",
        "h3_res9", "h3_res10", "h3_res11",
        "matched_geofence_id", "matched_geofence_name",
        "nearest_poi", "poi_dist_meters",
        "speed_mps", "derived_speed_mps", "heading_degrees", "derived_heading_deg",
        "floor_level", "indoor_zone", "source_system",
        "battery_level", "load_timestamp",
        "distance_meters", "time_delta_sec",
        "year", "month", "day"
    )
)

(
    final_silver.write
    .format("delta")
    .mode("overwrite")
    .partitionBy("year", "month")
    .saveAsTable("silver_geolocation_events")
)

print(f"Silver geolocation events: {final_silver.count():,} records")

# Notebook: nb_gold_geolocation
# Build patron journey sequences from Silver location events

from pyspark.sql.functions import (
    collect_list, struct, col, count, countDistinct,
    min as spark_min, max as spark_max, sum as spark_sum,
    avg, round as spark_round, datediff, when
)
from pyspark.sql.window import Window

silver = spark.table("silver_geolocation_events")

# Build ordered zone sequence per patron per day
journey_window = Window.partitionBy("device_id", "year", "month", "day").orderBy("event_timestamp")

patron_journeys = (
    silver
    .filter(col("device_type") == "patron_app")
    .filter(col("matched_geofence_name").isNotNull())
    .withColumn("zone_sequence_num", F.row_number().over(journey_window))
    .groupBy("device_id", "year", "month", "day")
    .agg(
        collect_list(
            struct("event_timestamp", "matched_geofence_name", "nearest_poi")
        ).alias("journey_steps"),
        countDistinct("matched_geofence_id").alias("zones_visited"),
        spark_min("event_timestamp").alias("journey_start"),
        spark_max("event_timestamp").alias("journey_end"),
        count("event_id").alias("total_location_pings")
    )
    .withColumn(
        "journey_duration_min",
        spark_round(
            (unix_timestamp(col("journey_end")) - unix_timestamp(col("journey_start"))) / 60, 1
        )
    )
)

(
    patron_journeys.write
    .format("delta")
    .mode("overwrite")
    .saveAsTable("gold_patron_journeys")
)

display(patron_journeys.orderBy(col("zones_visited").desc()).limit(10))

# Aggregate dwell time by geofence zone
# A "dwell session" is a continuous period where a device stays in one zone

from pyspark.sql import functions as F

zone_dwell = (
    silver
    .filter(col("device_type") == "patron_app")
    .filter(col("matched_geofence_name").isNotNull())
    .groupBy("matched_geofence_id", "matched_geofence_name", "year", "month", "day")
    .agg(
        countDistinct("device_id").alias("unique_patrons"),
        count("event_id").alias("total_events"),
        spark_round(avg("time_delta_sec"), 1).alias("avg_ping_interval_sec"),
        # Approximate total dwell = sum of time deltas where device is in zone
        spark_round(spark_sum("time_delta_sec") / 60, 1).alias("total_dwell_minutes"),
        spark_round(avg("poi_dist_meters"), 1).alias("avg_poi_distance_m")
    )
    .withColumn(
        "avg_dwell_per_patron_min",
        spark_round(col("total_dwell_minutes") / col("unique_patrons"), 1)
    )
)

(
    zone_dwell.write
    .format("delta")
    .mode("overwrite")
    .saveAsTable("gold_zone_dwell_time")
)

display(zone_dwell.orderBy(col("unique_patrons").desc()))

# Identify the most common zone-to-zone transitions
from pyspark.sql.functions import lead

route_window = Window.partitionBy("device_id", "year", "month", "day").orderBy("event_timestamp")

zone_transitions = (
    silver
    .filter(col("device_type") == "patron_app")
    .filter(col("matched_geofence_name").isNotNull())
    .withColumn("next_zone", lead("matched_geofence_name", 1).over(route_window))
    .filter(col("next_zone").isNotNull())
    .filter(col("matched_geofence_name") != col("next_zone"))  # Only transitions
    .groupBy("matched_geofence_name", "next_zone")
    .agg(
        count("event_id").alias("transition_count"),
        countDistinct("device_id").alias("unique_patrons")
    )
    .withColumnRenamed("matched_geofence_name", "from_zone")
    .withColumnRenamed("next_zone", "to_zone")
    .orderBy(col("transition_count").desc())
)

(
    zone_transitions.write
    .format("delta")
    .mode("overwrite")
    .saveAsTable("gold_zone_transitions")
)

display(zone_transitions.limit(20))

# Track proximity-triggered marketing events and conversion rates
proximity_marketing = (
    silver
    .filter(col("device_type") == "patron_app")
    .filter(col("nearest_poi").isNotNull())
    .filter(col("poi_dist_meters") <= 15.0)  # Within 15 meters of POI
    .groupBy("nearest_poi", "year", "month", "day")
    .agg(
        countDistinct("device_id").alias("patrons_in_proximity"),
        count("event_id").alias("proximity_events"),
        spark_round(avg("poi_dist_meters"), 1).alias("avg_distance_m")
    )
)

(
    proximity_marketing.write
    .format("delta")
    .mode("overwrite")
    .saveAsTable("gold_proximity_marketing")
)

# Join location data with gaming revenue for zone-level revenue attribution
# Requires gold_slot_performance or similar table from Tutorial 03

gaming_revenue = spark.table("gold_slot_performance")

zone_revenue = (
    silver
    .filter(col("device_type") == "patron_app")
    .filter(col("matched_geofence_name").isNotNull())
    .select("device_id", "matched_geofence_name", "event_timestamp", "year", "month", "day")
    .join(
        gaming_revenue.select("player_id", "total_coin_in", "net_win", "session_date"),
        (col("device_id") == col("player_id")) &
        (col("day") == col("session_date")),
        "inner"
    )
    .groupBy("matched_geofence_name", "year", "month")
    .agg(
        spark_sum("total_coin_in").alias("attributed_coin_in"),
        spark_sum("net_win").alias("attributed_net_win"),
        countDistinct("device_id").alias("attributed_patrons")
    )
)

(
    zone_revenue.write
    .format("delta")
    .mode("overwrite")
    .saveAsTable("gold_zone_revenue_attribution")
)

# Generate GeoJSON for H3 hex grid visualization
import json
import h3

def h3_hex_to_geojson(h3_density_df):
    """Convert H3 density DataFrame to GeoJSON FeatureCollection."""
    features = []
    for row in h3_density_df.collect():
        h3_idx = row["h3_res10"]
        boundary = h3.cell_to_boundary(h3_idx)
        # GeoJSON uses [lon, lat] order and requires closed ring
        coords = [[lon, lat] for lat, lon in boundary]
        coords.append(coords[0])  # Close the polygon

        feature = {
            "type": "Feature",
            "geometry": {
                "type": "Polygon",
                "coordinates": [coords]
            },
            "properties": {
                "h3_index": h3_idx,
                "event_count": row["event_count"],
                "unique_patrons": row["unique_patrons"],
                "avg_speed": row["avg_speed_mps"]
            }
        }
        features.append(feature)

    return {"type": "FeatureCollection", "features": features}

# Create and save GeoJSON
density_df = spark.table("gold_zone_dwell_time")
# (Use the patron_density DataFrame from Step 3.3 for H3-level data)
# geojson = h3_hex_to_geojson(patron_density)
# Save to Files for Power BI Shape Map or custom visual

# Bridge to ArcGIS: Export Silver data for ArcGIS GeoAnalytics
from arcgis.features import GeoAccessor

silver_pdf = spark.table("silver_geolocation_events").limit(100000).toPandas()
silver_sdf = GeoAccessor.from_xy(silver_pdf, "longitude", "latitude")

# Publish as hosted Feature Layer for ArcGIS analysis
layer = silver_sdf.spatial.to_featurelayer(
    title="Casino Patron Locations - Fabric Silver",
    gis=gis,
    folder="Fabric Geolocation"
)
print(f"Published to ArcGIS: {layer.url}")