# /// script
# requires-python = ">=3.11,<3.14"
# dependencies = [
#   "altair==6.0.0",
#   "marimo==0.25.0",
#   "pandas==3.0.2",
# ]
# ///

import marimo

__generated_with = "0.25.0"
app = marimo.App(width="medium")


@app.cell
def _():
    import altair as alt
    import marimo as mo
    import pandas as pd

    return alt, mo, pd


@app.cell
def _(mo):
    mo.md(
        """
        # Autonomous-driving long-tail audit

        A compact, reproducible marimo demo for slicing perception results by
        model, weather, lighting, object class, and distance. Change the controls
        below: every metric, chart, and failure row updates reactively.

        **Evidence boundary:** all 72 rows are deterministic synthetic fixtures.
        This demonstrates an audit workflow, not road safety, real-world model
        performance, or deployment readiness.
        """
    )
    return


@app.cell
def _(pd):
    weather_cycle = ["clear", "rain", "fog"]
    lighting_cycle = ["day", "night"]
    class_cycle = ["vehicle", "pedestrian", "cyclist"]
    distance_cycle = ["near", "mid", "far"]

    rows = []
    for frame_id in range(36):
        _weather = weather_cycle[frame_id % 3]
        _lighting = lighting_cycle[(frame_id // 3) % 2]
        _object_class = class_cycle[(frame_id // 2) % 3]
        _distance = distance_cycle[(frame_id // 4) % 3]

        for _model in ["baseline", "candidate"]:
            difficulty = (
                (0.10 if _weather == "rain" else 0.18 if _weather == "fog" else 0)
                + (0.12 if _lighting == "night" else 0)
                + (0.15 if _distance == "far" else 0.05 if _distance == "mid" else 0)
                + (0.05 if _object_class != "vehicle" else 0)
            )
            model_gain = 0.09 if _model == "candidate" else 0
            jitter = ((frame_id * 17) % 13 - 6) / 100
            confidence = max(0.12, min(0.97, 0.88 - difficulty + model_gain + jitter))
            detected = confidence >= (0.53 if _model == "candidate" else 0.56)
            iou = max(0.0, min(0.92, confidence - 0.14 + ((frame_id % 5) - 2) / 100)) if detected else 0.0
            latency_ms = 39 + (7 if _model == "candidate" else 0) + (frame_id * 7) % 19
            rows.append(
                {
                    "frame_id": f"scene-{frame_id:03d}",
                    "model": _model,
                    "weather": _weather,
                    "lighting": _lighting,
                    "object_class": _object_class,
                    "distance": _distance,
                    "confidence": round(confidence, 2),
                    "detected": detected,
                    "iou": round(iou, 2),
                    "latency_ms": latency_ms,
                }
            )

    results = pd.DataFrame(rows)
    return (results,)


@app.cell
def _(mo):
    model = mo.ui.dropdown(["all", "baseline", "candidate"], value="all", label="Model")
    weather = mo.ui.dropdown(["all", "clear", "rain", "fog"], value="all", label="Weather")
    lighting = mo.ui.dropdown(["all", "day", "night"], value="all", label="Lighting")
    object_class = mo.ui.dropdown(["all", "vehicle", "pedestrian", "cyclist"], value="all", label="Object class")
    min_confidence = mo.ui.slider(0.0, 0.9, step=0.05, value=0.0, label="Minimum confidence")
    mo.hstack([model, weather, lighting, object_class, min_confidence], widths="equal", wrap=True)
    return lighting, min_confidence, model, object_class, weather


@app.cell
def _(lighting, min_confidence, model, object_class, results, weather):
    filtered = results.copy()
    filters = {
        "model": model.value,
        "weather": weather.value,
        "lighting": lighting.value,
        "object_class": object_class.value,
    }
    for column, value in filters.items():
        if value != "all":
            filtered = filtered[filtered[column] == value]
    filtered = filtered[filtered["confidence"] >= min_confidence.value]
    return (filtered,)


@app.cell
def _(filtered, mo):
    count = len(filtered)
    recall = filtered["detected"].mean() if count else 0
    mean_iou = filtered.loc[filtered["detected"], "iou"].mean() if filtered["detected"].any() else 0
    latency = filtered["latency_ms"].mean() if count else 0
    mo.hstack(
        [
            mo.stat(value=f"{count}", label="Rows in slice"),
            mo.stat(value=f"{recall:.1%}", label="Detection rate"),
            mo.stat(value=f"{mean_iou:.2f}", label="Mean IoU (detected)"),
            mo.stat(value=f"{latency:.1f} ms", label="Mean latency"),
        ],
        widths="equal",
        wrap=True,
    )
    return


@app.cell
def _(alt, filtered, mo):
    summary = (
        filtered.groupby(["model", "weather"], as_index=False)
        .agg(detection_rate=("detected", "mean"), samples=("detected", "size"))
    )
    chart = (
        alt.Chart(summary)
        .mark_bar()
        .encode(
            x=alt.X("weather:N", title="Weather"),
            y=alt.Y("detection_rate:Q", title="Detection rate", scale=alt.Scale(domain=[0, 1])),
            color=alt.Color("model:N", title="Model"),
            xOffset="model:N",
            tooltip=["model", "weather", alt.Tooltip("detection_rate:Q", format=".1%"), "samples"],
        )
        .properties(height=280, title="Detection rate by weather")
    )
    mo.ui.altair_chart(chart)
    return


@app.cell
def _(filtered, mo):
    failures = filtered.loc[
        ~filtered["detected"],
        ["frame_id", "model", "weather", "lighting", "object_class", "distance", "confidence", "latency_ms"],
    ].sort_values(["confidence", "frame_id"])
    mo.vstack(
        [
            mo.md(f"## Failure queue\n\n**{len(failures)}** missed detections in the current slice."),
            mo.ui.table(failures, pagination=True, page_size=10, selection=None),
        ]
    )
    return


@app.cell
def _(mo):
    mo.callout(
        mo.md(
            """
            **How to productionize this:** replace the fixture cell with a versioned
            evaluation artifact, add dataset/model commit hashes, preserve per-scene
            evidence, and pre-register release gates. Aggregate metrics alone do not
            establish safety.
            """
        ),
        kind="warn",
    )
    return


if __name__ == "__main__":
    app.run()