Runbook E: PID GTFS-RT (Protobuf adapter)

// Runbook E — PID GTFS-RT · KQL setup
// Idempotentní.

// ──────────────────────────────────────────────────────────────────
// 1. Bronze
// ──────────────────────────────────────────────────────────────────

.create-merge table e_pid_raw (
    EventEnqueuedUtcTime: datetime,
    vehicle_id: string,
    trip_id: string,
    route_id: string,
    route_short_name: string,
    lat: real,
    lon: real,
    bearing: real,
    speed_mps: real,
    timestamp: datetime,
    vehicle_label: string,
    occupancy_status: string,
    current_stop_sequence: int,
    publisher_run_id: string
)

.alter table e_pid_raw policy retention
    @'{"SoftDeletePeriod": "30.00:00:00", "Recoverability": "Disabled"}'

// ──────────────────────────────────────────────────────────────────
// 2. Silver
// ──────────────────────────────────────────────────────────────────

.create-merge table e_pid_clean (
    ingest_ts: datetime,
    timestamp: datetime,
    vehicle_id: string,
    trip_id: string,
    route_id: string,
    route_short_name: string,
    vehicle_type: string,        // tram | bus | metro | train | trolley | other
    lat: real,
    lon: real,
    bearing: real,
    speed_mps: real,
    speed_kmh: real,
    occupancy_status: string,
    occupancy_pct_estimate: int,
    current_stop_sequence: int,
    is_moving: bool
)

.alter table e_pid_clean policy retention
    @'{"SoftDeletePeriod": "180.00:00:00", "Recoverability": "Enabled"}'

// ──────────────────────────────────────────────────────────────────
// 3. Update policy
// ──────────────────────────────────────────────────────────────────

.create-or-alter function fn_e_pid_transform() {
    e_pid_raw
    | extend ingest_ts = EventEnqueuedUtcTime
    | extend speed_kmh = round(speed_mps * 3.6, 1)
    | extend is_moving = (speed_mps > 0.5)
    | extend vehicle_type = case(
        vehicle_id contains "TRAM" or route_id startswith "T", "tram",
        vehicle_id contains "BUS" or route_id startswith "L" or route_id startswith "B", "bus",
        vehicle_id contains "METRO" or route_id startswith "M", "metro",
        vehicle_id contains "TRAIN" or route_id startswith "S", "train",
        vehicle_id contains "TROLLEY", "trolley",
        "other"
      )
    | extend occupancy_pct_estimate = case(
        occupancy_status == "EMPTY", 0,
        occupancy_status == "MANY_SEATS_AVAILABLE", 25,
        occupancy_status == "FEW_SEATS_AVAILABLE", 50,
        occupancy_status == "STANDING_ROOM_ONLY", 75,
        occupancy_status == "CRUSHED_STANDING_ROOM_ONLY", 95,
        occupancy_status == "FULL", 100,
        -1
      )
    | project ingest_ts, timestamp, vehicle_id, trip_id, route_id, route_short_name,
              vehicle_type, lat, lon, bearing, speed_mps, speed_kmh,
              occupancy_status, occupancy_pct_estimate,
              current_stop_sequence, is_moving
}

.alter table e_pid_clean policy update
@'[{"IsEnabled": true, "Source": "e_pid_raw", "Query": "fn_e_pid_transform()", "IsTransactional": true, "PropagateIngestionProperties": true}]'

// ──────────────────────────────────────────────────────────────────
// 4. MV — active vehicles (last position per vehicle, < 5min old)
// ──────────────────────────────────────────────────────────────────

.create-or-alter materialized-view with (backfill=true, lookback=10m)
    mv_e_pid_active_vehicles on table e_pid_clean {
        e_pid_clean
        | summarize arg_max(timestamp, *) by vehicle_id
    }

// ──────────────────────────────────────────────────────────────────
// 5. MV — route summary (5min rolling)
// ──────────────────────────────────────────────────────────────────

.create-or-alter materialized-view with (backfill=true, lookback=2h)
    mv_e_pid_route_summary on table e_pid_clean {
        e_pid_clean
        | summarize n_vehicles = dcount(vehicle_id),
                    avg_speed_kmh = avg(speed_kmh),
                    avg_occupancy = avg(occupancy_pct_estimate)
                    by bin(timestamp, 5m), route_id, route_short_name, vehicle_type
        | extend window_start = timestamp
        | project-away timestamp
    }

// ──────────────────────────────────────────────────────────────────
// 6. Helper: vehicles in geographic bbox
// ──────────────────────────────────────────────────────────────────

.create-or-alter function fn_e_pid_in_bbox(
    lat_min:real, lat_max:real, lon_min:real, lon_max:real,
    age:timespan = 2m
) {
    mv_e_pid_active_vehicles
    | where timestamp > ago(age)
    | where lat between (lat_min .. lat_max) and lon between (lon_min .. lon_max)
    | project vehicle_id, route_short_name, vehicle_type, lat, lon, speed_kmh, occupancy_status
}

// Praha centrum:
//   fn_e_pid_in_bbox(50.080, 50.095, 14.410, 14.450, 2m)

"""
Notebook E — PID GTFS-RT Protobuf Adapter

Decoder pro Pražská integrovaná doprava real-time vehicle positions.
GTFS-RT je binary Protocol Buffers format — Fabric Eventstream out-of-the-box neumí.
Tenhle Notebook je adapter: poll → decode → publish JSON.

DOCS:
  https://api.golemio.cz/v2/pid/docs/openapi/
  https://gtfs.org/realtime/reference/
"""

import json
import time
import uuid
from datetime import datetime, timezone

import requests
from google.transit import gtfs_realtime_pb2

PID_URL = "https://api.golemio.cz/v2/vehiclepositions/gtfsrt"
GOLEMIO_KEY = "<your-golemio-api-key>"

ES_URL = "<your-eventstream-custom-endpoint-url>"
ES_KEY = "<your-eventstream-access-key>"

POLL_INTERVAL_SEC = 20         # GTFS-RT refreshuje ~20s na zdroji
BATCH_SIZE = 500
MAX_RETRIES = 3
RETRY_BACKOFF_SEC = 3

def fetch_gtfs_rt() -> bytes:
    """Download GTFS-RT FeedMessage protobuf bytes."""
    headers = {
        "X-Access-Token": GOLEMIO_KEY,
        "Accept": "application/x-protobuf",
    }
    resp = requests.get(PID_URL, headers=headers, timeout=15)
    resp.raise_for_status()
    return resp.content

def decode_feed(pb_bytes: bytes) -> tuple[gtfs_realtime_pb2.FeedMessage, datetime]:
    """Decode protobuf bytes → FeedMessage + feed timestamp."""
    feed = gtfs_realtime_pb2.FeedMessage()
    feed.ParseFromString(pb_bytes)
    feed_ts = datetime.fromtimestamp(feed.header.timestamp, tz=timezone.utc) \
              if feed.header.timestamp else datetime.now(timezone.utc)
    return feed, feed_ts

_OCCUPANCY_NAMES = {
    0: "EMPTY",
    1: "MANY_SEATS_AVAILABLE",
    2: "FEW_SEATS_AVAILABLE",
    3: "STANDING_ROOM_ONLY",
    4: "CRUSHED_STANDING_ROOM_ONLY",
    5: "FULL",
    6: "NOT_ACCEPTING_PASSENGERS",
}

def feed_to_events(feed: gtfs_realtime_pb2.FeedMessage,
                   feed_ts: datetime,
                   run_id: str) -> list[dict]:
    """Iterate FeedMessage entities → list of vehicle event dicts."""
    events = []

    for entity in feed.entity:
        if not entity.HasField("vehicle"):
            continue

        v = entity.vehicle
        pos = v.position if v.HasField("position") else None
        trip = v.trip if v.HasField("trip") else None
        veh = v.vehicle if v.HasField("vehicle") else None

        if not pos:
            continue

        ts_unix = v.timestamp if v.timestamp else feed_ts.timestamp()
        ts_iso = datetime.fromtimestamp(ts_unix, tz=timezone.utc).isoformat()

        event = {
            "vehicle_id": veh.id if veh else entity.id,
            "trip_id": trip.trip_id if trip else "",
            "route_id": trip.route_id if trip else "",
            "route_short_name": "",  # GTFS-RT pure neobsahuje, doplnit lookupem z routes.txt (volitelné)
            "lat": pos.latitude,
            "lon": pos.longitude,
            "bearing": pos.bearing if pos.HasField("bearing") else 0.0,
            "speed_mps": pos.speed if pos.HasField("speed") else 0.0,
            "timestamp": ts_iso,
            "vehicle_label": veh.label if veh and veh.HasField("label") else "",
            "occupancy_status": _OCCUPANCY_NAMES.get(v.occupancy_status, "UNKNOWN"),
            "current_stop_sequence": v.current_stop_sequence if v.HasField("current_stop_sequence") else -1,
            "publisher_run_id": run_id,
        }
        events.append(event)

    return events

def publish_batch(events: list[dict]) -> dict:
    if not events:
        return {"sent": 0, "failed": 0}

    headers = {
        "Content-Type": "application/x-ndjson",
        "Authorization": f"Bearer {ES_KEY}",
    }

    sent = 0
    failed = 0

    for i in range(0, len(events), BATCH_SIZE):
        chunk = events[i:i + BATCH_SIZE]
        body = "\n".join(json.dumps(e) for e in chunk)

        for attempt in range(1, MAX_RETRIES + 1):
            try:
                r = requests.post(ES_URL, headers=headers, data=body, timeout=30)
                r.raise_for_status()
                sent += len(chunk)
                break
            except requests.HTTPError as e:
                if r.status_code in (429, 502, 503, 504) and attempt < MAX_RETRIES:
                    time.sleep(RETRY_BACKOFF_SEC * attempt)
                    continue
                failed += len(chunk)
                break
            except requests.RequestException:
                if attempt < MAX_RETRIES:
                    time.sleep(RETRY_BACKOFF_SEC * attempt)
                    continue
                failed += len(chunk)
                break

    return {"sent": sent, "failed": failed}

def run_once() -> dict:
    run_id = str(uuid.uuid4())[:8]
    t0 = time.time()

    pb_bytes = fetch_gtfs_rt()
    t_fetch = time.time() - t0

    feed, feed_ts = decode_feed(pb_bytes)
    t_decode = time.time() - t0 - t_fetch

    events = feed_to_events(feed, feed_ts, run_id)
    t_transform = time.time() - t0 - t_fetch - t_decode

    stats = publish_batch(events)
    t_publish = time.time() - t0 - t_fetch - t_decode - t_transform

    print(f"[{run_id}] feed_ts={feed_ts.isoformat()} "
          f"events={len(events)} sent={stats['sent']} failed={stats['failed']} "
          f"fetch={t_fetch:.1f}s decode={t_decode:.2f}s "
          f"publish={t_publish:.1f}s")

    return {
        "events": len(events),
        "sent": stats["sent"],
        "failed": stats["failed"],
        "fetch_sec": round(t_fetch, 2),
        "decode_sec": round(t_decode, 3),
        "publish_sec": round(t_publish, 2),
    }

def run_continuous(duration_sec: int = 600):
    t_start = time.time()
    iteration = 0
    while True:
        iteration += 1
        try:
            run_once()
        except Exception as e:
            print(f"  ✗ iteration {iteration} failed: {e}")

        elapsed = time.time() - t_start
        if duration_sec and elapsed >= duration_sec:
            break

        time.sleep(POLL_INTERVAL_SEC)

if __name__ == "__main__":
    run_continuous(duration_sec=300)  # 5 min