RTI Library: architektura

DP600_Demo_RTI                    # main workspace pro celou library

es_a_bicycles                     # demo A
es_b_chmi                         # demo B
es_c_ndic                         # demo C
es_d_alpr                         # demo D
es_e_pid                          # demo E

a_bicycles_raw                    # demo A (pokud chceš sink)
b_chmi_raw         b_chmi_clean   # demo B
c_ndic_raw         c_ndic_clean
d_alpr_raw         d_alpr_clean
e_pid_raw          e_pid_clean

mv_b_chmi_last_per_station        # latest reading per station
mv_c_ndic_avg_5min_per_loop       # 5-min avg counts per traffic loop
mv_d_alpr_count_per_camera_5min   # 5-min count per camera
mv_e_pid_active_vehicles          # currently active vehicles

fn_kraj_from_idreg(idReg:int)     # lookup helper
fn_speed_class(kmh:int)           # bucketize

nb_b_chmi_publisher               # publisher pro demo B
nb_c_ndic_publisher
nb_d_alpr_simulator
nb_e_pid_adapter

rtd_b_chmi_live
rtd_d_alpr_live

sm_rti_combined                   # composite model nad celou KQL DB
rpt_b_chmi_realtime
rpt_d_alpr_realtime

import requests
import json
import time
from datetime import datetime, timezone

EVENTSTREAM_URL = "https://<your-region>.eventstream.fabric.microsoft.com/..."
EVENTSTREAM_KEY = mssparkutils.credentials.getSecret("kv-rti", "es-key-b")  # uložené v KeyVaultu

def publish_event(event: dict):
    """Send single event to Eventstream Custom Endpoint."""
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {EVENTSTREAM_KEY}",
    }
    r = requests.post(EVENTSTREAM_URL, headers=headers, data=json.dumps(event), timeout=10)
    r.raise_for_status()

def publish_batch(events: list[dict]):
    """Send batch (newline-delimited JSON)."""
    headers = {
        "Content-Type": "application/x-ndjson",
        "Authorization": f"Bearer {EVENTSTREAM_KEY}",
    }
    body = "\n".join(json.dumps(e) for e in events)
    r = requests.post(EVENTSTREAM_URL, headers=headers, data=body, timeout=30)
    r.raise_for_status()

// Step 1: Bronze table (Eventstream destination)
.create-merge table b_chmi_raw (
    EventEnqueuedUtcTime: datetime,
    idRegistration: int,
    startTime: datetime,
    idValueType: string,
    value: real
)

// Step 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
)

// Step 3: Update policy function (transforms raw -> clean)
.create-or-alter function fn_b_chmi_transform() {
    b_chmi_raw
    | extend ingest_ts = EventEnqueuedUtcTime
    | extend measurement_ts = todatetime(startTime)
    // dim_pollutant = bezna KQL tabulka (lookup nahrany .set-or-replace, viz runbook B);
    // update policy nesmi cist externaldata ani external tables
    | join kind=leftouter dim_pollutant on $left.idValueType == $right.pollutant_code
    | extend over_limit = (value > eu_limit)
    | project ingest_ts, measurement_ts, idRegistration, pollutant_code, pollutant_name, value, unit, eu_limit, over_limit
}

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

// Step 5: Materialized view nad Silver
.create-or-alter materialized-view with (backfill=true)
    mv_b_chmi_last_per_station on table b_chmi_clean {
        b_chmi_clean
        | summarize arg_max(measurement_ts, *) by idRegistration, pollutant_code
    }

// 1. summarize bin() — equivalent SQL GROUP BY date_trunc
b_chmi_clean
| where measurement_ts > ago(24h)
| summarize avg(value) by bin(measurement_ts, 1h), pollutant_code

// 2. make-series — vyrovnaná time series s gaps
b_chmi_clean
| where pollutant_code == "PM10"
| make-series avg(value) on measurement_ts step 1h by idRegistration

// 3. arg_max / arg_min — latest / earliest per group
b_chmi_clean
| summarize arg_max(measurement_ts, *) by idRegistration

// 4. percentiles
b_chmi_clean
| summarize percentiles(value, 50, 95, 99) by pollutant_code

// 5. series_decompose / series_detect_anomalies
b_chmi_clean
| make-series val=avg(value) on measurement_ts step 1h
| extend (anomalies, score, _) = series_decompose_anomalies(val, 1.5)