scripts/run_eval.py

"""Labeled before/after accuracy evaluation runner.

Runs the real pipeline twice over an address list — a BEFORE leg (base env flags
only) and an AFTER leg (base + toggled flags) — and saves everything under a
labeled, dated directory so results are reusable and comparable across sessions:

    data/evals/<label>/<YYYY-MM-DD_HHMM>/
        manifest.json       label, git sha, imagery, env of both legs, model pins
        results_before.csv  same columns as the regression CSVs
        results_after.csv
        delta.csv           per-address before/after lawn_sqft and % change
        viz/<slug>__before.png / __after.png   4-panel renders (gitignored)

CSVs + manifest are small and git-trackable; the PNGs fall under the global
*.png ignore. Accuracy evals run on GOOGLE imagery (what prod uses) — the NAIP
byte-identical gate is a separate regression check, not an accuracy reference.

Usage (prod-parity Exp 5 example):
    python scripts/run_eval.py --label exp5-green-reclaim \\
        --csv data/evals/addresses/exp5_green_reclaim.csv --imagery google \\
        --base SAM_RESTRICT=1 --base ROW_TO_CURB=1 --toggle GREEN_RECLAIM=1
"""

from __future__ import annotations

import argparse
import csv
import json
import os
import re
import shutil
import subprocess
import sys
from datetime import datetime
from pathlib import Path

from dotenv import load_dotenv

load_dotenv()

from lawn_estimator.config import DATA_DIR  # noqa: E402
from lawn_estimator.pipeline import run  # noqa: E402
from lawn_estimator.segmentation import MODEL_REVISIONS  # noqa: E402

FIELDNAMES = [
    "Address", "Status", "RGB_Vegetation_sqft", "LiDAR_Lawn_sqft",
    "Parcel_Area_sqft", "Estimation_Area_sqft", "Ground_Sampled_sqft",
    "LiDAR_Pct_of_Estimation", "Region", "Method", "Confidence",
    "Warning", "Error", "Timestamp",
]


def _parse_env_pairs(pairs: list[str]) -> dict[str, str]:
    env = {}
    for pair in pairs:
        if "=" not in pair:
            sys.exit(f"ERROR: expected NAME=value, got {pair!r}")
        name, value = pair.split("=", 1)
        env[name.strip()] = value.strip()
    return env


def _read_addresses(csv_path: Path) -> list[str]:
    with open(csv_path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        col = next((c for c in (reader.fieldnames or []) if c.strip().lower() == "address"), None)
        if col is None:
            sys.exit(f"ERROR: {csv_path} has no 'Address' column (found {reader.fieldnames}).")
        return [row[col].strip() for row in reader if row[col].strip()]


def _git(*args: str) -> str:
    try:
        return subprocess.run(
            ["git", *args], capture_output=True, text=True, check=True
        ).stdout.strip()
    except Exception:
        return ""


def _slug(address: str) -> str:
    return re.sub(r"[^\w\s-]", "", address).strip().replace(" ", "_")[:60]


def _apply_leg_env(base: dict[str, str], toggles: dict[str, str], leg: str) -> None:
    """Config() reads os.environ at construction inside pipeline.run, so setting
    the process env between legs is what switches the behavior under test."""
    for name, value in base.items():
        os.environ[name] = value
    for name, value in toggles.items():
        if leg == "after":
            os.environ[name] = value
        else:
            os.environ.pop(name, None)


def _run_leg(leg: str, addresses: list[str], imagery: str, out_dir: Path) -> list[dict]:
    viz_dir = out_dir / "viz"
    viz_dir.mkdir(parents=True, exist_ok=True)
    rows = []
    for i, address in enumerate(addresses, 1):
        print(f"\n=== [{leg} {i}/{len(addresses)}] {address} ===")
        row = dict.fromkeys(FIELDNAMES, "")
        row.update(Address=address, Timestamp=datetime.now().isoformat(timespec="seconds"))
        try:
            result = run(address, imagery=imagery)
            row.update(
                Status="ok",
                RGB_Vegetation_sqft=round(result["rgb_veg_sqft"], 1),
                LiDAR_Lawn_sqft=round(result["lidar_lawn_sqft"], 1),
                Parcel_Area_sqft=round(result["parcel_area_sqft"], 1),
                Estimation_Area_sqft=round(result["estimation_area_sqft"], 1),
                Ground_Sampled_sqft=round(result["ground_sampled_sqft"], 1),
                LiDAR_Pct_of_Estimation=round(
                    result["lidar_lawn_sqft"] / result["estimation_area_sqft"] * 100, 1
                ),
                Region=result.get("region", ""),
                Method=result.get("method", ""),
                Confidence=result.get("confidence", ""),
                Warning=result.get("warning") or "",
            )
            viz = result.get("visualization_path")
            if viz and Path(viz).exists():
                shutil.copy(viz, viz_dir / f"{_slug(address)}__{leg}.png")
        except Exception as exc:  # a bad address shouldn't kill the whole eval
            row.update(Status="error", Error=f"{type(exc).__name__}: {exc}")
            print(f"  FAILED: {row['Error']}")
        rows.append(row)
    return rows


def _write_csv(path: Path, rows: list[dict], fieldnames: list[str]) -> None:
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)


def main() -> None:
    parser = argparse.ArgumentParser(description="Labeled before/after accuracy eval.")
    parser.add_argument("--label", required=True, help="Eval name, e.g. exp5-green-reclaim")
    parser.add_argument("--csv", required=True, help="CSV with an 'Address' column")
    parser.add_argument("--imagery", default="google", choices=["auto", "google", "naip", "county"])
    parser.add_argument("--base", action="append", default=[],
                        help="NAME=value env set on BOTH legs (repeatable), e.g. SAM_RESTRICT=1")
    parser.add_argument("--toggle", action="append", default=[],
                        help="NAME=value env set ONLY on the after leg (repeatable)")
    parser.add_argument("--notes", default="", help="Free-text note stored in the manifest")
    args = parser.parse_args()

    base = _parse_env_pairs(args.base)
    toggles = _parse_env_pairs(args.toggle)
    if not toggles:
        sys.exit("ERROR: --toggle is required — an eval with no toggled flag has no 'after'.")
    addresses = _read_addresses(Path(args.csv))

    out_dir = DATA_DIR / "evals" / args.label / datetime.now().strftime("%Y-%m-%d_%H%M")
    out_dir.mkdir(parents=True, exist_ok=True)
    print(f"Eval '{args.label}': {len(addresses)} addresses, imagery={args.imagery}")
    print(f"Results -> {out_dir}")

    manifest = {
        "label": args.label,
        "created": datetime.now().isoformat(timespec="seconds"),
        "git_sha": _git("rev-parse", "HEAD"),
        "git_branch": _git("rev-parse", "--abbrev-ref", "HEAD"),
        "git_dirty": bool(_git("status", "--porcelain")),
        "imagery": args.imagery,
        "base_env": base,
        "toggle_env": toggles,
        "model_revisions": MODEL_REVISIONS,
        "addresses": addresses,
        "notes": args.notes,
    }
    (out_dir / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")

    legs = {}
    for leg in ("before", "after"):
        _apply_leg_env(base, toggles, leg)
        legs[leg] = _run_leg(leg, addresses, args.imagery, out_dir)
        _write_csv(out_dir / f"results_{leg}.csv", legs[leg], FIELDNAMES)

    deltas = []
    for b, a in zip(legs["before"], legs["after"], strict=True):
        ok = b["Status"] == "ok" and a["Status"] == "ok"
        before_sqft = float(b["LiDAR_Lawn_sqft"]) if ok else None
        after_sqft = float(a["LiDAR_Lawn_sqft"]) if ok else None
        deltas.append({
            "Address": b["Address"],
            "Before_Lawn_sqft": b["LiDAR_Lawn_sqft"],
            "After_Lawn_sqft": a["LiDAR_Lawn_sqft"],
            "Delta_sqft": round(after_sqft - before_sqft, 1) if ok else "",
            "Delta_pct": round((after_sqft - before_sqft) / before_sqft * 100, 1)
            if ok and before_sqft else "",
            "Method_before": b["Method"],
            "Method_after": a["Method"],
            "Status": "ok" if ok else "error",
        })
    _write_csv(out_dir / "delta.csv", deltas, list(deltas[0].keys()))

    print(f"\n{'='*60}")
    print(f"{'Address':<45} {'before':>9} {'after':>9} {'Δ%':>7}")
    for d in deltas:
        print(f"{d['Address']:<45} {d['Before_Lawn_sqft']:>9} {d['After_Lawn_sqft']:>9} "
              f"{str(d['Delta_pct']):>7}")
    print(f"\nSaved to {out_dir}")


if __name__ == "__main__":
    main()