Evaluating map-matching accuracy against ground truth

import numpy as np
import pandas as pd


def route_mismatch(
    truth_links: list,
    matched_links: list,
    link_lengths: dict,
) -> dict:
    """
    Length-weighted route error between a true and a matched link sequence.

    Parameters
    ----------
    truth_links, matched_links : list
        Ordered link ids. Order is used for the topology check; the length
        metric itself is set-based, because a route that visits the right
        links in the wrong order is a separate failure worth naming.
    link_lengths : dict
        link id -> length in METRES.

    Returns
    -------
    dict
        mismatch_frac, missed_m, extra_m, and a topology flag.

    Raises
    ------
    ValueError
        On an empty truth route or a link with no recorded length.
    """
    if not truth_links:
        raise ValueError("Empty ground-truth route.")
    unknown = (set(truth_links) | set(matched_links)) - set(link_lengths)
    if unknown:
        raise ValueError(f"{len(unknown)} link(s) have no length: {list(unknown)[:3]}")

    t, m = set(truth_links), set(matched_links)
    truth_m = sum(link_lengths[l] for l in t)
    missed = sum(link_lengths[l] for l in t - m)     # true links not matched
    extra = sum(link_lengths[l] for l in m - t)      # matched links not driven

    # Symmetric difference over the true length: the standard route-mismatch
    # fraction. It can exceed 1.0 when the matcher invents a long detour,
    # which is correct — that is a worse outcome than matching nothing.
    return {
        "mismatch_frac": (missed + extra) / truth_m,
        "missed_m": missed,
        "extra_m": extra,
        "truth_m": truth_m,
        "order_differs": [l for l in matched_links if l in t] != [l for l in truth_links if l in m],
    }


def evaluate_matcher(
    runs: pd.DataFrame,
    link_lengths: dict,
    road_class: dict | None = None,
) -> pd.DataFrame:
    """
    Score every labelled run and break the result down by road class.

    runs : DataFrame with columns run_id, truth_links, matched_links (lists).
    An aggregate score hides the class where the matcher fails, and that
    class is almost always the one with the shortest links.
    """
    required = {"run_id", "truth_links", "matched_links"}
    missing = required - set(runs.columns)
    if missing:
        raise ValueError(f"Missing required columns: {missing}")
    if runs.empty:
        raise ValueError("No labelled runs to evaluate.")

    rows = []
    for _, r in runs.iterrows():
        res = route_mismatch(r["truth_links"], r["matched_links"], link_lengths)
        res["run_id"] = r["run_id"]
        rows.append(res)
    per_run = pd.DataFrame(rows)

    if road_class is None:
        return per_run

    # Class-level breakdown, weighted by true length in that class.
    cls_rows = []
    for _, r in runs.iterrows():
        t, m = set(r["truth_links"]), set(r["matched_links"])
        for cls in set(road_class.get(l, "unknown") for l in t):
            t_cls = {l for l in t if road_class.get(l, "unknown") == cls}
            m_cls = {l for l in m if road_class.get(l, "unknown") == cls}
            length = sum(link_lengths[l] for l in t_cls)
            wrong = sum(link_lengths[l] for l in (t_cls - m_cls) | (m_cls - t_cls))
            cls_rows.append({"road_class": cls, "truth_m": length, "wrong_m": wrong})
    by_class = (pd.DataFrame(cls_rows).groupby("road_class", as_index=False)
                  .sum(numeric_only=True))
    by_class["mismatch_frac"] = by_class["wrong_m"] / by_class["truth_m"]
    return by_class.sort_values("mismatch_frac", ascending=False)

def validate_evaluation(per_run: pd.DataFrame, by_class: pd.DataFrame,
                        target_frac: float = 0.05) -> dict:
    """Check the evaluation itself before believing its verdict."""
    # 1. Enough runs to say anything. Below ~20 the confidence interval on a
    #    5% mismatch rate is wider than the difference between two matchers.
    assert len(per_run) >= 20, (
        f"only {len(per_run)} labelled runs — the interval on the score is "
        "wider than any difference you are trying to measure"
    )

    # 2. The aggregate is length-weighted, not a mean of per-run fractions.
    #    A mean of fractions over-weights short runs.
    weighted = per_run["missed_m"].sum() + per_run["extra_m"].sum()
    agg = weighted / per_run["truth_m"].sum()
    naive = per_run["mismatch_frac"].mean()
    if abs(agg - naive) > 0.02:
        print(f"note: length-weighted {agg:.1%} vs mean-of-fractions "
              f"{naive:.1%} — report the first")

    # 3. Worst class must be reported, not buried.
    worst = by_class.iloc[0]
    assert worst["mismatch_frac"] < target_frac * 4, (
        f"{worst['road_class']} is at {worst['mismatch_frac']:.1%} against an "
        f"overall target of {target_frac:.0%} — the matcher fails on this class"
    )
    return {"aggregate_mismatch": float(agg), "worst_class": worst["road_class"],
            "worst_frac": float(worst["mismatch_frac"]), "runs": len(per_run)}