scripts/audit_pipeline.py

"""Step-by-step pipeline audit → one shareable HTML page for an address.

Runs the REAL pipeline with the capture hook and renders every stage in order:
imagery → parcel → to-curb geometry → color-threshold veg → per-class segmentation
(EoMT + incumbent, WITH class legends) → lawn-area mask → LiDAR ground points →
final lawn/removed classification. Lets the owner audit exactly what each step does.

Runs prod-parity by default (SAM_RESTRICT, ROW_TO_CURB, GREEN_RECLAIM, LAWN_CASCADE);
override via env like any pipeline run. Output is a self-contained .html (data-URI
images) — open it or share the file.

    python scripts/audit_pipeline.py --address "7863 N 144th Ave, Bennington, NE 68007" \
        --out data/outputs/audit_bennington.html
"""
from __future__ import annotations

import argparse
import base64
import io
import os

import numpy as np
from dotenv import load_dotenv
from PIL import Image, ImageDraw

load_dotenv()
os.environ.setdefault("SAM_RESTRICT", "1")
os.environ.setdefault("ROW_TO_CURB", "1")
os.environ.setdefault("GREEN_RECLAIM", "1")
os.environ.setdefault("LAWN_CASCADE", "1")

from lawn_estimator.pipeline import run  # noqa: E402
from lawn_estimator.segmentation import (  # noqa: E402
    CASCADE_PRIMARY_MODEL_ID,
    LAWN_MODEL_ID,
    _predict_classes,
)

# ADE20K names for the EoMT cascade primary (outdoor-relevant subset).
ADE = {0: "wall", 1: "building", 2: "sky", 4: "tree", 6: "road", 9: "grass",
       11: "sidewalk", 13: "earth", 17: "plant", 21: "water", 20: "car", 25: "?",
       29: "field", 46: "sand", 52: "path", 94: "land"}
# Empirical meanings for the incumbent's generic labels.
M2F = {0: "background", 1: "open (grass+dirt)", 2: "street-edge band", 3: "pavement",
       4: "canopy", 6: '"water" (fires on flat turf)', 7: "roofs (as cropland)"}
PALETTE = [(80, 200, 60), (0, 110, 40), (220, 60, 60), (150, 110, 70), (235, 220, 120),
           (170, 120, 40), (60, 130, 235), (120, 120, 130), (200, 60, 200), (230, 130, 30),
           (90, 200, 200), (200, 120, 200), (255, 180, 40)]


def b64(img: Image.Image, max_w: int = 900) -> str:
    if img.width > max_w:
        img = img.resize((max_w, round(img.height * max_w / img.width)), Image.LANCZOS)
    buf = io.BytesIO()
    img.convert("RGB").save(buf, "JPEG", quality=82)
    return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode()


def overlay(base: Image.Image, mask: np.ndarray, color, alpha=0.55) -> Image.Image:
    arr = np.asarray(base.convert("RGB")).astype(np.float32)
    arr[mask] = arr[mask] * (1 - alpha) + np.array(color, np.float32) * alpha
    return Image.fromarray(arr.astype(np.uint8))


def draw_outlines(base: Image.Image, outlines, color, width=4) -> Image.Image:
    img = base.convert("RGB").copy()
    d = ImageDraw.Draw(img)
    for xs, ys in outlines:  # each ring is (xs, ys) — separate coord lists
        pts = [(float(x), float(y)) for x, y in zip(xs, ys, strict=False)]
        if len(pts) > 1:
            d.line(pts + [pts[0]], fill=color, width=width)
    return img


def class_panel(base: Image.Image, pred: np.ndarray, names: dict) -> tuple[str, list]:
    """Colored per-class overlay + legend chips (name, %, color) for present classes."""
    arr = np.asarray(base.convert("RGB")).astype(np.float32) * 0.45
    legend = []
    codes = [c for c in np.unique(pred) if (pred == c).sum() / pred.size >= 0.003]
    for i, c in enumerate(sorted(codes, key=lambda c: -(pred == c).sum())):
        col = PALETTE[i % len(PALETTE)]
        arr[pred == c] += np.array(col, np.float32) * 0.55
        legend.append({"name": names.get(int(c), f"class {c}"),
                       "color": "#{:02x}{:02x}{:02x}".format(*col),
                       "pct": round(float((pred == c).sum() / pred.size * 100), 1)})
    return b64(Image.fromarray(np.clip(arr, 0, 255).astype(np.uint8))), legend


def lidar_panel(base: Image.Image, viz: dict) -> str:
    """Ground points colored: green = counted lawn, red = removed."""
    img = base.convert("RGB").copy()
    d = ImageDraw.Draw(img)
    gpx, gpy = viz.get("ground_px"), viz.get("ground_py")
    if gpx is None:
        return b64(img)
    is_lawn = viz.get("lawn_mask")
    for k in range(len(gpx)):
        x, y = float(gpx[k]), float(gpy[k])
        col = (60, 220, 60) if (is_lawn is not None and is_lawn[k]) else (235, 60, 60)
        d.ellipse([x - 2, y - 2, x + 2, y + 2], fill=col)
    return b64(img)


STEP = """<section class="step">
  <div class="hd"><span class="n">{n}</span><h2>{title}</h2><span class="sq">{sqft}</span></div>
  <p class="dsc">{desc}</p>
  <img src="{img}" alt="{title}">
  {legend}
</section>"""


def legend_html(legend) -> str:
    if not legend:
        return ""
    chips = "".join(
        f'<span class="chip"><span class="sw" style="background:{e["color"]}"></span>'
        f'{e["name"]} <b>{e["pct"]}%</b></span>' for e in legend)
    return f'<div class="legend">{chips}</div>'


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--address", required=True)
    ap.add_argument("--out", required=True)
    ap.add_argument("--imagery", default="google")
    args = ap.parse_args()

    cap: dict = {}
    result = run(args.address, imagery=args.imagery, capture=cap)
    img = cap["image"]

    steps = []
    steps.append(("1", "Analysis imagery", f"{cap['imagery']} tile, {img.size[0]}×{img.size[1]} px — "
                  "the single image every measurement is computed on.", "", b64(img), None))
    steps.append(("2", "Legal parcel", "County parcel boundary (point-in-polygon lookup), yellow outline. "
                  f"{cap['parcel_area_sqft']:,.0f} sqft legal lot.", "",
                  b64(draw_outlines(img, cap["legal_outlines"], (255, 220, 0))), None))
    ext = "; ".join(f"{e.get('street') or '?'} +{e['area_sqft']:,.0f}" for e in cap["extensions"])
    steps.append(("3", "Estimation geometry (to the curb)",
                  f"Parcel extended to street-facing curbs (cyan), neighbors subtracted. "
                  f"{cap['estimation_area_sqft']:,.0f} sqft. Extensions: {ext or 'none'}.", "",
                  b64(draw_outlines(draw_outlines(img, cap["legal_outlines"], (255, 220, 0)),
                                    cap["estimation_outlines"], (0, 220, 220))), None))
    steps.append(("4", "Color-threshold vegetation (baseline)",
                  "A pure HSV/green rule (no neural net) — the sanity baseline, not the estimate.",
                  f"{cap['rgb_veg_sqft']:,.0f} sqft", b64(overlay(img, cap["veg_in_parcel"], (60, 220, 60))), None))

    n = 5
    cascade = os.getenv("LAWN_CASCADE", "").lower() in ("1", "true", "yes", "on")
    if cascade:
        eomt = _predict_classes(CASCADE_PRIMARY_MODEL_ID, img)
        im, leg = class_panel(img, eomt, ADE)
        steps.append((str(n), "Segmentation — EoMT-DINOv3 (cascade primary)",
                      "The primary read, real class names. Note where it mislabels green lawn "
                      "(e.g. 'building' on shaded turf) — the cascade + green-reclaim recover that.",
                      "", im, leg))
        n += 1
    m2f = _predict_classes(LAWN_MODEL_ID, img)
    im, leg = class_panel(img, m2f, M2F)
    steps.append((str(n), "Segmentation — mask2former (incumbent"
                  + ("/arbiter)" if cascade else ")"),
                  "Empirical class meanings (published labels are wrong). "
                  + ("Arbitrates the cascade's sidewalk/earth." if cascade else "Lawn = open + canopy."),
                  "", im, leg))
    n += 1
    steps.append((str(n), "Lawn-area mask (candidate lawn)",
                  "Union of the segmentation decision (+ green reclaim): which pixels are eligible "
                  "lawn. This gates which LiDAR points count.", "",
                  b64(overlay(img, cap["lawn_area_in_parcel"], (60, 220, 60))), None))
    n += 1
    steps.append((str(n), "LiDAR ground points → classified",
                  "Ground returns inside the estimation area, colored by the mask: green = counted "
                  "as lawn, red = removed (hardscape/roof, incl. SAM restrict). The sqft is "
                  "green_points / all_ground × ground-sampled area.",
                  f"{result['lawn_sqft']:,.0f} sqft", lidar_panel(img, result_est_viz(result, cap)), None))

    cards = "\n".join(STEP.format(n=s[0], title=s[1], desc=s[2], sqft=s[3], img=s[4],
                                  legend=legend_html(s[5])) for s in steps)
    html = PAGE.replace("<!--STEPS-->", cards).replace("{addr}", args.address).replace(
        "{final}", f"{result['lawn_sqft']:,.0f}").replace("{method}", result["method"])
    with open(args.out, "w", encoding="utf-8") as f:
        f.write(html)
    print(f"audit page: {args.out} ({os.path.getsize(args.out)/1e6:.1f} MB) — {len(steps)} steps")


def result_est_viz(result, cap):
    est = cap.get("est")
    return getattr(est, "viz", {}) if est is not None else {}


PAGE = """<title>Pipeline audit — {addr}</title>
<meta name="viewport" content="width=device-width, initial-scale=1">
<style>
:root{--bg:#0f130d;--card:#181d14;--ink:#e6ebe0;--soft:#9fb090;--line:#2c3626;--acc:#5db85d}
*{box-sizing:border-box}
body{margin:0;background:var(--bg);color:var(--ink);font:16px/1.55 "Avenir Next",system-ui,sans-serif}
main{max-width:940px;margin:0 auto;padding:24px 16px 64px}
h1{font-size:clamp(22px,4vw,30px);margin:.2em 0}
.lead{color:var(--soft);margin:0 0 8px}
.final{background:var(--card);border:1px solid var(--line);border-left:4px solid var(--acc);
  border-radius:10px;padding:12px 16px;margin:14px 0 6px;font-size:18px}
.final b{color:var(--acc);font-size:22px}
.step{background:var(--card);border:1px solid var(--line);border-radius:12px;padding:14px;margin:18px 0}
.hd{display:flex;align-items:center;gap:10px}
.hd .n{background:var(--acc);color:#06210a;font-weight:700;width:28px;height:28px;border-radius:50%;
  display:flex;align-items:center;justify-content:center;flex:none}
.hd h2{font-size:17px;margin:0;flex:1}
.hd .sq{color:var(--acc);font-weight:600;font-variant-numeric:tabular-nums}
.dsc{color:var(--soft);font-size:14px;margin:8px 2px 10px}
.step img{width:100%;border-radius:8px;display:block}
.legend{display:flex;flex-wrap:wrap;gap:6px;margin-top:10px}
.chip{display:flex;align-items:center;gap:6px;background:#0f130d;border:1px solid var(--line);
  border-radius:7px;padding:4px 9px;font-size:12.5px}
.chip .sw{width:13px;height:13px;border-radius:3px;border:1px solid #0006}
</style>
<main>
<h1>Pipeline audit</h1>
<p class="lead">{addr}</p>
<div class="final">Final measured lawn: <b>{final} sqft</b> &nbsp;·&nbsp; method: {method}. Every
step below feeds this number — scroll to audit each stage.</div>
<!--STEPS-->
</main>"""


if __name__ == "__main__":
    main()