Runbook B: ČHMÚ HTTP polling

// Runbook B — ČHMÚ Polling · KQL setup
// Idempotentní, lze spustit opakovaně.
// Předpoklad:
//   - KQL DB `kql_rti_demo` v Eventhouse `eh_rti_demo`
//   - Lakehouse `lh_rti_lookups` s dim_kraj.csv, dim_pollutant.csv, dim_station.csv (Files/lookups/)
//   - Eventstream `es_b_chmi` napojený na destination = `b_chmi_raw`

// ──────────────────────────────────────────────────────────────────
// 1. Bronze table — Eventstream destination
// ──────────────────────────────────────────────────────────────────

.create-merge table b_chmi_raw (
    EventEnqueuedUtcTime: datetime,    // auto-added by Eventstream
    idRegistration: int,
    startTime: datetime,
    idValueType: string,
    value: real
)

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

// ──────────────────────────────────────────────────────────────────
// 2. Silver table — typed, joined, business model
// ──────────────────────────────────────────────────────────────────

.create-merge table b_chmi_clean (
    ingest_ts: datetime,
    measurement_ts: datetime,
    idRegistration: int,
    pollutant_code: string,
    pollutant_name: string,
    value: real,
    unit: string,
    eu_limit: real,
    over_limit: bool,
    station_nazev: string,
    kraj_kod: string,
    kraj_nazev: string,
    lat: real,
    lon: real
)

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

// ──────────────────────────────────────────────────────────────────
// 3. Lookup external tables (point at Lakehouse Files/lookups/)
// ──────────────────────────────────────────────────────────────────
// Replace <abfss URL> with actual OneLake path. Get URL via:
//   Lakehouse → right-click file → Properties → ABFSS path

.create-or-alter external table dim_pollutant_ext (
    pollutant_code: string,
    pollutant_name: string,
    unit: string,
    eu_limit: real
)
kind = storage
dataformat = csv
(
    'abfss://<workspace-id>@onelake.dfs.fabric.microsoft.com/lh_rti_lookups.Lakehouse/Files/lookups/dim_pollutant.csv'
)
with (includeHeaders = 'all')

.create-or-alter external table dim_station_ext (
    loc_code: string,
    id_registration: int,
    station_nazev: string,
    kraj_kod: string,
    kraj_nazev: string,
    lat: real,
    lon: real
)
kind = storage
dataformat = csv
(
    'abfss://<workspace-id>@onelake.dfs.fabric.microsoft.com/lh_rti_lookups.Lakehouse/Files/lookups/dim_station.csv'
)
with (includeHeaders = 'all')

// ──────────────────────────────────────────────────────────────────
// 3b. Materializace lookupu do beznych tabulek — update policy nesmi cist external tables
// ────────────────────────────────────────────────────────────

.set-or-replace dim_pollutant <| external_table('dim_pollutant_ext')
.set-or-replace dim_station   <| external_table('dim_station_ext')

// Pokud .alter policy update nize odmitne join kvuli streaming ingestion na b_chmi_raw:
// .alter table b_chmi_raw policy streamingingestion disable

// ────────────────────────────────────────────────────────────
// 4. Update policy function (transforms raw → clean)
// ──────────────────────────────────────────────────────────────────

.create-or-alter function fn_b_chmi_transform() {
    b_chmi_raw
    | extend ingest_ts = EventEnqueuedUtcTime
    | extend measurement_ts = startTime
    | join kind=leftouter dim_pollutant on $left.idValueType == $right.pollutant_code
    | join kind=leftouter dim_station on $left.idRegistration == $right.id_registration
    | extend over_limit = iif(isnotnull(eu_limit), value > eu_limit, false)
    | extend station_nazev = iif(isempty(station_nazev), strcat("Unmapped (id=", tostring(idRegistration), ")"), station_nazev)
    | extend kraj_kod = iif(isempty(kraj_kod), "UNK", kraj_kod)
    | extend kraj_nazev = iif(isempty(kraj_nazev), "Neznámý", kraj_nazev)
    | project ingest_ts, measurement_ts, idRegistration,
              pollutant_code = idValueType,
              pollutant_name = coalesce(pollutant_name, idValueType),
              value,
              unit = coalesce(unit, ""),
              eu_limit = coalesce(eu_limit, real(null)),
              over_limit,
              station_nazev, kraj_kod, kraj_nazev,
              lat, lon
}

// ──────────────────────────────────────────────────────────────────
// 5. Wire update policy
// ──────────────────────────────────────────────────────────────────

.alter table b_chmi_clean policy update
@'[{"IsEnabled": true, "Source": "b_chmi_raw", "Query": "fn_b_chmi_transform()", "IsTransactional": true, "PropagateIngestionProperties": true}]'

// ──────────────────────────────────────────────────────────────────
// 6. Materialized view — last reading per station + pollutant
// ──────────────────────────────────────────────────────────────────

.create-or-alter materialized-view with (backfill=true, lookback=1d)
    mv_b_chmi_last_per_station on table b_chmi_clean {
        b_chmi_clean
        | summarize arg_max(measurement_ts, *) by idRegistration, pollutant_code
    }

// ──────────────────────────────────────────────────────────────────
// 7. Helper function — kraj summary
// ──────────────────────────────────────────────────────────────────

.create-or-alter function fn_b_kraj_summary(window:timespan = 1h) {
    b_chmi_clean
    | where measurement_ts > ago(window)
    | summarize avg_value = avg(value),
                max_value = max(value),
                stations = dcount(idRegistration)
                by kraj_kod, kraj_nazev, pollutant_code
}

// ──────────────────────────────────────────────────────────────────
// 8. Sanity / smoke test queries (run after publisher beží)
// ──────────────────────────────────────────────────────────────────

// Counts
// b_chmi_raw | count
// b_chmi_clean | count
// mv_b_chmi_last_per_station | count

// Latest 10
// b_chmi_clean | top 10 by ingest_ts desc

// Per-hour PM10 trend
// b_chmi_clean
// | where measurement_ts > ago(24h) and pollutant_code == "PM10"
// | summarize avg_pm10 = avg(value) by bin(measurement_ts, 1h)
// | order by measurement_ts asc

// Stations over EU limit (last 6h)
// b_chmi_clean
// | where measurement_ts > ago(6h) and over_limit
// | summarize n = count() by station_nazev, pollutant_code
// | top 10 by n desc

// Per-kraj summary
// fn_b_kraj_summary(6h) | order by avg_value desc

"""
Notebook B — ČHMÚ recent CSV → Eventstream Custom Endpoint publisher

Run mode: Fabric Pure Python notebook (NE PySpark — Pure Python má lepší lifecycle pro continuous runs).
Schedule: every 5 minutes (Fabric Data Pipeline → Notebook activity).
Ad-hoc: run cells manually for live demo.

Cell 1: imports + config
Cell 2: fetch + parse
Cell 3: publish batch
Cell 4: orchestration loop (volitelné, pro continuous run)
"""

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

import requests

CHMI_URL = "https://opendata.chmi.cz/air_quality/recent/data/airquality_1h_avg_CZ.csv"

ES_URL = "https://<region>.eventstream.fabric.microsoft.com/<endpoint-id>/events"
ES_KEY = "<your-eventstream-access-key>"

POLL_INTERVAL_SEC = 300        # 5 min
BATCH_SIZE = 200                # events per HTTP POST
MAX_RETRIES = 3
RETRY_BACKOFF_SEC = 5

def fetch_chmi_snapshot() -> list[dict]:
    """
    Stáhne current CHMU snapshot, parse CSV, vrátí list of event dicts.

    Schema (per row):
      idRegistration: int
      startTime: datetime ISO 8601 string
      idValueType: string (např. PM10, NO2, O3, ...)
      value: float
    """
    resp = requests.get(CHMI_URL, timeout=30)
    resp.raise_for_status()

    # CHMU CSV má leading whitespace v cells → skipinitialspace
    reader = csv.DictReader(
        io.StringIO(resp.text),
        skipinitialspace=True
    )

    events = []
    for row in reader:
        try:
            event = {
                "idRegistration": int(row["idRegistration"]),
                "startTime": row["startTime"].strip(),  # už je v ISO 8601 z CHMU
                "idValueType": row["idValueType"].strip(),
                "value": float(row["value"]),
                # client-side enrichment (volitelné — můžeme i v KQL update policy)
                "publisher_run_id": str(uuid.uuid4()),
                "publisher_ts": datetime.now(timezone.utc).isoformat(),
            }
            events.append(event)
        except (KeyError, ValueError) as e:
            # log a skip — corrupted row
            print(f"Skipping row: {e} | row={row}")
            continue

    return events

def publish_batch(events: list[dict]) -> dict:
    """
    POST batch as NDJSON (newline-delimited JSON).
    Eventstream Custom Endpoint accepts both JSON array and NDJSON.
    NDJSON je preferovaný pro batch (single HTTP roundtrip per batch).

    Returns: dict with stats {sent, failed}.
    """
    if not events:
        return {"sent": 0, "failed": 0}

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

    sent = 0
    failed = 0

    # Chunk do BATCH_SIZE
    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 (HTTP {r.status_code})")
                    time.sleep(wait)
                    continue
                print(f"FAIL chunk po {attempt} retries: {e}")
                failed += len(chunk)
                break
            except requests.RequestException as e:
                if attempt < MAX_RETRIES:
                    wait = RETRY_BACKOFF_SEC * attempt
                    print(f"Network retry {attempt}/{MAX_RETRIES} po {wait}s: {e}")
                    time.sleep(wait)
                    continue
                print(f"FAIL chunk po {attempt} retries: {e}")
                failed += len(chunk)
                break

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

def run_once() -> dict:
    """Single fetch + publish cycle."""
    t0 = time.time()
    print(f"[{datetime.now(timezone.utc).isoformat()}] Fetching CHMU snapshot...")
    events = fetch_chmi_snapshot()
    t_fetch = time.time() - t0
    print(f"  Got {len(events)} events in {t_fetch:.1f}s")

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

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

def run_continuous(max_iterations: int = 0):
    """
    Continuous poll loop.

    max_iterations = 0 → běží navždy (manuálně stop)
    max_iterations > 0 → po N cyklech skončí (testing)
    """
    iteration = 0
    while True:
        iteration += 1
        try:
            stats = run_once()
            print(f"  → iteration {iteration} done, sleeping {POLL_INTERVAL_SEC}s")
        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__":
    # Default: single run
    print(run_once())