Runbook D: Camera ALPR simulátor

// Runbook D — Camera ALPR Simulator · KQL setup
// Idempotentní.

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

.create-merge table d_alpr_raw (
    EventEnqueuedUtcTime: datetime,
    event_id: string,
    camera_id: string,
    camera_lat: real,
    camera_lon: real,
    timestamp: datetime,
    license_plate: string,
    vehicle_class: string,
    speed_kmh: int,
    speed_limit_kmh: int,
    direction: string,
    confidence: real,
    color: string,
    publisher_run_id: string
)

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

// ──────────────────────────────────────────────────────────────────
// 2. Silver — adds derived columns
// ──────────────────────────────────────────────────────────────────

.create-merge table d_alpr_clean (
    ingest_ts: datetime,
    timestamp: datetime,
    event_id: string,
    camera_id: string,
    highway: string,
    km_marker: int,
    camera_lat: real,
    camera_lon: real,
    license_plate: string,
    plate_country: string,
    vehicle_class: string,
    speed_kmh: int,
    speed_limit_kmh: int,
    speed_excess_kmh: int,
    is_speeder: bool,
    is_severe_speeder: bool,
    direction: string,
    confidence: real,
    color: string
)

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

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

.create-or-alter function fn_d_alpr_transform() {
    d_alpr_raw
    | extend ingest_ts = EventEnqueuedUtcTime
    | extend highway = extract(@"^CAM-(D\d+)", 1, camera_id)
    | extend km_marker = toint(extract(@"KM-(\d+)", 1, camera_id))
    | extend plate_country = case(
        license_plate matches regex @"^[0-9][A-Z]{2}-[0-9]{4}$", "CZ",
        license_plate matches regex @"^[A-Z]{2}\s?[0-9]{2}\s?[A-Z]{3}$", "GB",
        license_plate matches regex @"^[A-Z]{1,3}-[A-Z]{1,2}\s?[0-9]{1,4}[A-Z]?$", "DE",
        "OTHER"
      )
    | extend speed_excess_kmh = max_of(0, speed_kmh - speed_limit_kmh)
    | extend is_speeder = (speed_excess_kmh > 0)
    | extend is_severe_speeder = (speed_excess_kmh >= 20)
    | project ingest_ts, timestamp, event_id, camera_id, highway, km_marker,
              camera_lat, camera_lon, license_plate, plate_country,
              vehicle_class, speed_kmh, speed_limit_kmh, speed_excess_kmh,
              is_speeder, is_severe_speeder, direction, confidence, color
}

.alter table d_alpr_clean policy update
@'[{"IsEnabled": true, "Source": "d_alpr_raw", "Query": "fn_d_alpr_transform()", "IsTransactional": true, "PropagateIngestionProperties": true}]'

// ──────────────────────────────────────────────────────────────────
// 4. MV — count per camera, 5min windows
// ──────────────────────────────────────────────────────────────────

.create-or-alter materialized-view with (backfill=true, lookback=2h)
    mv_d_alpr_count_per_camera_5min on table d_alpr_clean {
        d_alpr_clean
        | summarize n_events = count(),
                    avg_speed = avg(speed_kmh),
                    n_speeders = countif(is_speeder),
                    n_severe = countif(is_severe_speeder)
                    by bin(timestamp, 5m), camera_id, highway
        | extend window_start = timestamp
        | project-away timestamp
    }

// ──────────────────────────────────────────────────────────────────
// 5. MV — speeders rolling table
// ──────────────────────────────────────────────────────────────────

.create-or-alter materialized-view with (backfill=true, lookback=1d)
    mv_d_alpr_speeders on table d_alpr_clean {
        d_alpr_clean
        | where is_speeder
    }

// ──────────────────────────────────────────────────────────────────
// 6. Watchlist function
// ──────────────────────────────────────────────────────────────────

.create-or-alter function fn_d_alpr_watchlist_match(plates:dynamic, window:timespan = 1h) {
    d_alpr_clean
    | where timestamp > ago(window)
    | where license_plate in (plates)
    | project timestamp, license_plate, camera_id, highway, speed_kmh, vehicle_class
    | order by timestamp desc
}

// Usage example:
// fn_d_alpr_watchlist_match(dynamic(["1AB-2345", "STO-9999"]), 24h)

"""
Notebook D — Camera ALPR Simulator → Eventstream Custom Endpoint

Simuluje fleet 50 ALPR kamer rozmístěných na CZ dálnicích D1, D5, D8, D11, D35.
Realistic license plate generator, speed distribution, vehicle class mix.

Throughput: ~100 events/s sustained (50 cameras × 2 events/s).
Tunable přes EVENTS_PER_SEC_PER_CAMERA.
"""

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

import requests

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

EVENTS_PER_SEC_PER_CAMERA = 2.0      # ~100 events/s celkem
BATCH_SIZE = 100                      # events per HTTP POST
PUBLISH_INTERVAL_SEC = 1.0            # publish every 1s
SEED = 42                             # deterministic playback (None = random)

if SEED is not None:
    random.seed(SEED)

CAMERAS = [
    # D1 Praha-Brno
    ("CAM-D1-KM-005", 50.0234, 14.5512, 130),
    ("CAM-D1-KM-022", 50.0532, 14.6745, 130),
    ("CAM-D1-KM-045", 49.9878, 14.8923, 130),
    ("CAM-D1-KM-085", 49.6234, 15.5421, 130),
    ("CAM-D1-KM-120", 49.4567, 16.0234, 130),
    ("CAM-D1-KM-150", 49.3456, 16.4523, 130),
    ("CAM-D1-KM-178", 49.2345, 16.5734, 130),
    ("CAM-D1-KM-195", 49.2134, 16.6234, 130),
    # D5 Praha-Plzeň-Rozvadov
    ("CAM-D5-KM-008", 49.9923, 14.2345, 130),
    ("CAM-D5-KM-035", 49.8234, 13.9456, 130),
    ("CAM-D5-KM-068", 49.7423, 13.3712, 130),
    ("CAM-D5-KM-095", 49.6712, 12.9234, 130),
    ("CAM-D5-KM-130", 49.6334, 12.5645, 130),
    # D8 Praha-Ústí
    ("CAM-D8-KM-012", 50.1845, 14.4234, 130),
    ("CAM-D8-KM-035", 50.3712, 14.2823, 130),
    ("CAM-D8-KM-058", 50.5234, 14.0934, 130),
    ("CAM-D8-KM-080", 50.6634, 13.9234, 130),
    # D11 Praha-Hradec
    ("CAM-D11-KM-008", 50.1234, 14.6712, 130),
    ("CAM-D11-KM-035", 50.1923, 15.0823, 130),
    ("CAM-D11-KM-068", 50.2334, 15.5234, 130),
    ("CAM-D11-KM-088", 50.2812, 15.7912, 130),
    # D35 Hradec-Olomouc-Lipník
    ("CAM-D35-KM-012", 50.1934, 15.9823, 130),
    ("CAM-D35-KM-045", 50.0345, 16.5234, 130),
    ("CAM-D35-KM-080", 49.8923, 17.0512, 130),
    ("CAM-D35-KM-120", 49.6234, 17.2823, 130),
    # D52 Brno-Mikulov
    ("CAM-D52-KM-005", 49.1834, 16.5934, 130),
    ("CAM-D52-KM-025", 48.9923, 16.6234, 130),
    ("CAM-D52-KM-045", 48.8023, 16.6512, 130),
    # D6 Praha-Karlovy Vary
    ("CAM-D6-KM-015", 50.0712, 14.2334, 110),
    ("CAM-D6-KM-040", 50.1023, 13.9534, 110),
    ("CAM-D6-KM-070", 50.1834, 13.5212, 110),
    # D7 Praha-Slaný-Chomutov
    ("CAM-D7-KM-010", 50.1934, 14.3023, 130),
    ("CAM-D7-KM-035", 50.3923, 14.1234, 130),
    ("CAM-D7-KM-060", 50.5234, 13.8923, 130),
    # D10 Praha-Mladá Boleslav-Liberec
    ("CAM-D10-KM-012", 50.1923, 14.6534, 130),
    ("CAM-D10-KM-040", 50.4023, 14.9034, 130),
    ("CAM-D10-KM-068", 50.6234, 15.0312, 130),
    # D46 Vyškov-Olomouc
    ("CAM-D46-KM-008", 49.2745, 17.0512, 130),
    ("CAM-D46-KM-030", 49.4923, 17.1834, 130),
    # D55 Hulín-Břeclav
    ("CAM-D55-KM-015", 49.2734, 17.4523, 130),
    ("CAM-D55-KM-050", 48.9534, 17.3934, 130),
    # D48 Bělotín-Český Těšín
    ("CAM-D48-KM-010", 49.5934, 17.7234, 130),
    ("CAM-D48-KM-035", 49.6712, 18.0823, 130),
    # Pomalejší úseky (D-roads, omezení 110)
    ("CAM-D49-KM-008", 49.2334, 17.6712, 110),
    ("CAM-D49-KM-025", 49.1023, 17.7923, 110),
    # Ulice ve městech (omezení 50)
    ("CAM-MUNI-PHA-01", 50.0823, 14.4234, 50),
    ("CAM-MUNI-PHA-02", 50.0712, 14.4523, 50),
    ("CAM-MUNI-BRN-01", 49.1934, 16.6112, 50),
    ("CAM-MUNI-BRN-02", 49.2034, 16.6034, 50),
    ("CAM-MUNI-OLO-01", 49.5934, 17.2512, 50),
]

VEHICLE_CLASSES = ["passenger", "passenger", "passenger", "passenger",
                   "suv", "suv", "truck", "truck", "bus", "motorcycle"]

COLORS = ["white", "white", "black", "black", "silver", "grey", "blue",
          "red", "green", "yellow"]

DIRECTIONS = ["N", "S", "E", "W"]

def gen_cz_plate() -> str:
    """CZ plate format: 1AB-2345 (region digit + 2 letters + 4 digits)"""
    region = random.choice(string.digits)
    letters = "".join(random.choices(string.ascii_uppercase, k=2))
    digits = "".join(random.choices(string.digits, k=4))
    return f"{region}{letters}-{digits}"

def gen_de_plate() -> str:
    """DE plate format: M-AB 1234"""
    region = random.choice(["M", "B", "K", "F", "S", "DD", "L"])
    letters = "".join(random.choices(string.ascii_uppercase, k=2))
    digits = "".join(random.choices(string.digits, k=random.randint(2, 4)))
    return f"{region}-{letters} {digits}"

def gen_plate() -> str:
    """90% CZ, 10% foreign (DE)."""
    if random.random() < 0.1:
        return gen_de_plate()
    return gen_cz_plate()

def gen_speed(limit_kmh: int) -> int:
    """
    Speed distribution:
      80% within limit (limit-30 to limit)
      15% slight speeder (limit to limit+20)
       5% severe speeder (limit+20 to limit+50)
    """
    r = random.random()
    if r < 0.80:
        return random.randint(max(20, limit_kmh - 30), limit_kmh)
    elif r < 0.95:
        return random.randint(limit_kmh, limit_kmh + 20)
    else:
        return random.randint(limit_kmh + 20, limit_kmh + 50)

def gen_event(camera_id, lat, lon, speed_limit, run_id) -> dict:
    return {
        "event_id": str(uuid.uuid4()),
        "camera_id": camera_id,
        "camera_lat": lat,
        "camera_lon": lon,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "license_plate": gen_plate(),
        "vehicle_class": random.choice(VEHICLE_CLASSES),
        "speed_kmh": gen_speed(speed_limit),
        "speed_limit_kmh": speed_limit,
        "direction": random.choice(DIRECTIONS),
        "confidence": round(random.uniform(0.85, 0.99), 2),
        "color": random.choice(COLORS),
        "publisher_run_id": run_id,
    }

def gen_batch(run_id: str) -> list[dict]:
    """Generate one tick worth of events across all cameras."""
    events = []
    n_per_camera = max(1, int(EVENTS_PER_SEC_PER_CAMERA * PUBLISH_INTERVAL_SEC))
    for camera_id, lat, lon, speed_limit in CAMERAS:
        # Stochastic — not every camera fires every tick
        if random.random() > 0.7:  # ~70% activity rate
            for _ in range(n_per_camera):
                events.append(gen_event(camera_id, lat, lon, speed_limit, run_id))
    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}",
    }
    body = "\n".join(json.dumps(e) for e in events)
    try:
        r = requests.post(ES_URL, headers=headers, data=body, timeout=10)
        r.raise_for_status()
        return {"sent": len(events), "failed": 0}
    except Exception as e:
        print(f"Publish failed: {e}")
        return {"sent": 0, "failed": len(events)}

def run_simulator(duration_sec: int = 600, verbose: bool = True):
    """
    Run simulator pro `duration_sec` sekund.

    duration_sec=0 → běží navždy (manuálně stop)
    """
    run_id = str(uuid.uuid4())
    t_start = time.time()
    total_sent = 0
    iteration = 0

    print(f"[ALPR sim start] run_id={run_id} duration={duration_sec}s "
          f"cameras={len(CAMERAS)} target_eps={EVENTS_PER_SEC_PER_CAMERA * len(CAMERAS):.0f}")

    while True:
        iteration += 1
        events = gen_batch(run_id)
        result = publish_batch(events)
        total_sent += result["sent"]

        if verbose and iteration % 10 == 0:
            elapsed = time.time() - t_start
            eps = total_sent / max(1, elapsed)
            print(f"  [{iteration:4d}] elapsed={elapsed:.0f}s sent={total_sent} eps={eps:.1f}")

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

        time.sleep(PUBLISH_INTERVAL_SEC)

    elapsed = time.time() - t_start
    print(f"[ALPR sim done] sent={total_sent} elapsed={elapsed:.0f}s avg_eps={total_sent/elapsed:.1f}")

if __name__ == "__main__":
    run_simulator(duration_sec=300)