Runbook C: NDIC dopravní detektory

// Runbook C — NDIC Traffic Counters · KQL setup
// Idempotentní.

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

.create-merge table c_ndic_raw (
    EventEnqueuedUtcTime: datetime,
    site_id: string,
    site_name: string,
    lat: real,
    lon: real,
    measurement_ts: datetime,
    vehicle_flow_rate: int,
    avg_speed_kmh: real,
    publisher_run_id: string,
    publisher_ts: datetime
)

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

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

.create-merge table c_ndic_clean (
    ingest_ts: datetime,
    measurement_ts: datetime,
    site_id: string,
    site_name: string,
    highway: string,
    km_marker: real,
    lat: real,
    lon: real,
    vehicle_flow_rate: int,
    avg_speed_kmh: real,
    speed_class: string,
    is_congested: bool
)

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

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

.create-or-alter function fn_c_ndic_transform() {
    c_ndic_raw
    | extend ingest_ts = EventEnqueuedUtcTime
    | extend highway = extract(@"^MS-(D\d+)", 1, site_id)
    | extend km_marker = todouble(extract(@"KM-(\d+)", 1, site_id))
    | extend speed_class = case(
        avg_speed_kmh < 30, "stopped",
        avg_speed_kmh < 60, "slow",
        avg_speed_kmh < 90, "normal",
        avg_speed_kmh < 120, "fast",
        "very_fast"
      )
    | extend is_congested = (avg_speed_kmh < 50 and vehicle_flow_rate > 800)
    | project ingest_ts, measurement_ts, site_id, site_name, highway, km_marker,
              lat, lon, vehicle_flow_rate, avg_speed_kmh, speed_class, is_congested
}

.alter table c_ndic_clean policy update
@'[{"IsEnabled": true, "Source": "c_ndic_raw", "Query": "fn_c_ndic_transform()", "IsTransactional": true, "PropagateIngestionProperties": true}]'

// ──────────────────────────────────────────────────────────────────
// 4. MV — last per site
// ──────────────────────────────────────────────────────────────────

.create-or-alter materialized-view with (backfill=true, lookback=1d)
    mv_c_ndic_last_per_site on table c_ndic_clean {
        c_ndic_clean
        | summarize arg_max(measurement_ts, *) by site_id
    }

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

.create-or-alter materialized-view with (backfill=true, lookback=2h)
    mv_c_ndic_congestion_5min on table c_ndic_clean {
        c_ndic_clean
        | where is_congested
        | summarize n_congested_sites = dcount(site_id),
                    avg_speed = avg(avg_speed_kmh)
                    by bin(measurement_ts, 5m), highway
    }

// ──────────────────────────────────────────────────────────────────
// 6. Helper: incident detection
// ──────────────────────────────────────────────────────────────────

.create-or-alter function fn_c_ndic_incidents(window:timespan = 15m, drop_pct:real = 30.0) {
    let current = mv_c_ndic_last_per_site
                  | where measurement_ts > ago(window);
    let baseline = c_ndic_clean
                   | where measurement_ts between (ago(7d) .. ago(1d))
                   | summarize baseline_speed = avg(avg_speed_kmh) by site_id;
    current
    | join kind=inner baseline on site_id
    | extend speed_drop_pct = round((baseline_speed - avg_speed_kmh) / baseline_speed * 100, 1)
    | where speed_drop_pct > drop_pct
    | project site_id, site_name, highway, km_marker,
              current_speed = avg_speed_kmh, baseline_speed, speed_drop_pct,
              lat, lon
    | order by speed_drop_pct desc
}

"""
Notebook C — NDIC DATEX II → Eventstream Custom Endpoint publisher

DATEX II XML parsing + auth API + retry pattern.

DOCS: https://api.dopravniinfo.cz/docs/openapi/
"""

import json
import time
import uuid
import xml.etree.ElementTree as ET
from datetime import datetime, timezone

import requests

NDIC_URL = "https://api.dopravniinfo.cz/v1/datex2/measurement/recent"
NDIC_API_KEY = "<your-ndic-api-key>"

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

POLL_INTERVAL_SEC = 60
BATCH_SIZE = 200
MAX_RETRIES = 3
RETRY_BACKOFF_SEC = 5

NS = {"d2": "http://datex2.eu/schema/2/2_0"}

def fetch_ndic_measurement_xml() -> bytes:
    """Download recent measurement XML from NDIC."""
    headers = {
        "Authorization": f"ApiKey {NDIC_API_KEY}",
        "Accept": "application/xml",
    }
    resp = requests.get(NDIC_URL, headers=headers, timeout=30)
    resp.raise_for_status()
    return resp.content

def parse_datex2_measurements(xml_bytes: bytes) -> list[dict]:
    """
    Parse DATEX II measurement XML → list of event dicts.

    Note: DATEX II je generic standard, exact element structure
    se může lišit podle endpoint verze. Tahle implementace je
    pragmatic best-effort + tolerantní k missing fields.
    """
    root = ET.fromstring(xml_bytes)

    publisher_run_id = str(uuid.uuid4())
    publisher_ts = datetime.now(timezone.utc).isoformat()

    events = []

    # Iterate all measurementSiteRecord elements
    # Use .// to find them anywhere in tree
    for site in root.findall(".//d2:measurementSiteRecord", NS):
        try:
            site_id = site.get("id") or ""

            name_el = site.find(".//d2:measurementSiteName", NS)
            site_name = name_el.text if name_el is not None else ""

            lat_el = site.find(".//d2:latitude", NS)
            lon_el = site.find(".//d2:longitude", NS)
            lat = float(lat_el.text) if lat_el is not None else None
            lon = float(lon_el.text) if lon_el is not None else None

            mts_el = site.find(".//d2:measurementTimeDefault", NS)
            measurement_ts = mts_el.text if mts_el is not None else publisher_ts

            flow_el = site.find(".//d2:vehicleFlowRate", NS)
            speed_el = site.find(".//d2:averageVehicleSpeed", NS)
            flow = int(flow_el.text) if flow_el is not None else 0
            speed = float(speed_el.text) if speed_el is not None else 0.0

            event = {
                "site_id": site_id,
                "site_name": site_name,
                "lat": lat,
                "lon": lon,
                "measurement_ts": measurement_ts,
                "vehicle_flow_rate": flow,
                "avg_speed_kmh": speed,
                "publisher_run_id": publisher_run_id,
                "publisher_ts": publisher_ts,
            }
            events.append(event)
        except Exception as e:
            print(f"Skipping site {site.get('id')}: {e}")
            continue

    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:
                    wait = RETRY_BACKOFF_SEC * attempt
                    print(f"Retry {attempt}/{MAX_RETRIES} po {wait}s")
                    time.sleep(wait)
                    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:
    t0 = time.time()
    print(f"[{datetime.now(timezone.utc).isoformat()}] Fetching NDIC...")
    xml = fetch_ndic_measurement_xml()
    events = parse_datex2_measurements(xml)
    t_fetch = time.time() - t0
    print(f"  Parsed {len(events)} events in {t_fetch:.1f}s")

    t1 = time.time()
    stats = publish_batch(events)
    t_pub = time.time() - t1
    print(f"  Published: sent={stats['sent']}, failed={stats['failed']} in {t_pub:.1f}s")

    return {**stats, "fetched": len(events), "fetch_sec": round(t_fetch, 2),
            "publish_sec": round(t_pub, 2)}

def run_continuous(max_iterations: int = 0):
    iteration = 0
    while True:
        iteration += 1
        try:
            run_once()
        except Exception as e:
            print(f"  ✗ iteration {iteration} failed: {e}")
        if max_iterations and iteration >= max_iterations:
            break
        time.sleep(POLL_INTERVAL_SEC)

if __name__ == "__main__":
    print(run_once())