"""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 = """
{n}

{title}

{sqft}

{desc}

{title} {legend}
""" def legend_html(legend) -> str: if not legend: return "" chips = "".join( f'' f'{e["name"]} {e["pct"]}%' for e in legend) return f'
{chips}
' 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("", 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 = """Pipeline audit — {addr}

Pipeline audit

{addr}

Final measured lawn: {final} sqft  ·  method: {method}. Every step below feeds this number — scroll to audit each stage.
""" if __name__ == "__main__": main()