"""Compare route aggregates to a historical baseline, not scientific ground truth.""" from __future__ import annotations import csv import math from collections import Counter from pathlib import Path from typing import Any # Absolute and relative limits; allow rounding and undocumented legacy smoothing. TOLERANCES = { "points": (0.0, 0.0), "distance_km": (0.1, 0.01), "ascent_m_smooth5": (20.0, 0.02), "descent_m_smooth5": (20.0, 0.02), "max_elev_m": (1.0, 0.0), } def read_summary(path: Path) -> dict[str, dict[str, Any]]: """Validate required comparison metrics and unique route identities.""" result = {} with path.open(encoding="utf-8", newline="") as stream: for row in csv.DictReader(stream): route_id = row["route_id"] if not route_id or route_id in result: raise ValueError(f"{path}: empty or duplicate route_id {route_id!r}") group = row["route_group"].lower() if group not in {"tmb", "reference"}: raise ValueError(f"{path}: invalid group for {route_id}") result[route_id] = {"route_group": group} for metric in TOLERANCES: value = float(row[metric]) if row[metric] else None if value is not None and not math.isfinite(value): raise ValueError(f"{path}: nonfinite {metric} for {route_id}") if value is None and metric in {"points", "distance_km"}: raise ValueError(f"{path}: missing {metric} for {route_id}") if value is not None and metric != "max_elev_m" and value < 0: raise ValueError(f"{path}: negative {metric} for {route_id}") if metric == "points" and (value < 1 or not value.is_integer()): raise ValueError(f"{path}: invalid point count for {route_id}") result[route_id][metric] = value if not result: raise ValueError(f"{path}: empty route summary") return result def metric_difference( generated: float | None, preserved: float | None, metric: str ) -> dict[str, Any]: """Compare with max(absolute tolerance, relative tolerance × baseline).""" absolute, relative = TOLERANCES[metric] tolerance = max(absolute, relative * abs(preserved)) if preserved is not None else absolute delta = generated - preserved if generated is not None and preserved is not None else None return { "generated": generated, "preserved": preserved, "delta": delta, "tolerance": tolerance, "within_tolerance": delta is not None and abs(delta) <= tolerance, } def totals(rows: dict[str, dict[str, Any]]) -> dict[str, Any]: """Return complete totals; unknown elevations propagate to relevant totals.""" tmb = [row for row in rows.values() if row["route_group"] == "tmb"] result = { "routes": len(rows), "points": sum(r["points"] for r in rows.values()), "tmb_routes": len(tmb), } for metric in ("distance_km", "ascent_m_smooth5", "descent_m_smooth5"): result["tmb_" + metric] = ( sum(r[metric] for r in tmb) if all(r[metric] is not None for r in tmb) else None ) maximums = [r["max_elev_m"] for r in rows.values()] result["max_elev_m"] = max(maximums) if all(v is not None for v in maximums) else None return result def point_counts(path: Path) -> Counter: """Count actual master CSV rows by route, independently of summary counts.""" with path.open(encoding="utf-8", newline="") as stream: return Counter(row["route_id"] for row in csv.DictReader(stream)) def compare_datasets(generated: Path, preserved: Path) -> dict[str, Any]: """Report missing routes, per-route metrics, totals and count consistency.""" new, old = ( read_summary(generated / "routes_summary.csv"), read_summary(preserved / "routes_summary.csv"), ) route_reports = {} for route_id in sorted(new.keys() & old.keys()): route_reports[route_id] = { metric: metric_difference(new[route_id][metric], old[route_id][metric], metric) for metric in TOLERANCES } missing, extra = sorted(old.keys() - new.keys()), sorted(new.keys() - old.keys()) group_mismatches = sorted( r for r in new.keys() & old.keys() if new[r]["route_group"] != old[r]["route_group"] ) checks = {} for label, directory, rows in [("generated", generated, new), ("preserved", preserved, old)]: counts = point_counts(directory / "trail_points_master.csv") checks[label + "_master_point_counts_match_summary"] = dict(counts) == { key: row["points"] for key, row in rows.items() } # A no-TMB registry legitimately produces a header-only TMB summary. with (directory / "tmb_routes_summary.csv").open(encoding="utf-8", newline="") as stream: subset = list(csv.DictReader(stream)) expected = {key: row for key, row in rows.items() if row["route_group"] == "tmb"} checks[label + "_tmb_summary_matches_routes"] = ( len(subset) == len(expected) and {r["route_id"] for r in subset} == set(expected) and all( r["route_group"].lower() == "tmb" and all( (float(r[m]) if r[m] else None) == expected[r["route_id"]][m] for m in TOLERANCES ) for r in subset ) ) return { "generated_totals": totals(new), "preserved_totals": totals(old), "missing_routes": missing, "extra_routes": extra, "group_mismatches": group_mismatches, "checks": checks, "per_route": route_reports, "within_tolerance": not missing and not extra and not group_mismatches and all(checks.values()) and all(m["within_tolerance"] for r in route_reports.values() for m in r.values()), } def format_report(report: dict[str, Any]) -> str: """Render a compact human-readable comparison, including every route metric.""" lines = ["Mountain Twin generated vs preserved (comparison, not scientific validation)"] for metric, value in report["generated_totals"].items(): lines.append(f"{metric}: generated={value}, preserved={report['preserved_totals'][metric]}") for key in ("missing_routes", "extra_routes", "group_mismatches", "checks"): lines.append(f"{key}: {report[key]}") for route_id, metrics in report["per_route"].items(): lines.append(route_id + ":") for name, result in metrics.items(): status = "OK" if result["within_tolerance"] else "REVIEW" lines.append( f" {name}: generated={result['generated']}, preserved={result['preserved']}, delta={result['delta']}, tolerance={result['tolerance']} [{status}]" ) lines.append( "PASS: within comparison tolerances" if report["within_tolerance"] else "REVIEW: missing data, inconsistency, or differences exceed tolerances" ) return "\n".join(lines)