"""Ablation study → one shareable HTML page. Runs the REAL pipeline in several configurations over a set of addresses to isolate what each signal contributes — especially the LiDAR-vs-national-scale question (how much does dropping LiDAR cost?) and how much the neural segmenter adds over a dumb color rule. Emits a matrix table (config × address, sqft + % vs the reference) and a per-address strip of each config's measured-lawn overlay. python scripts/run_ablation.py --csv data/evals/addresses/cascade_vs_precascade.csv \ --out data/outputs/ablation.html Uses the byte-safe ablation flags (LAWN_FORCE_RGB_ONLY, LAWN_RGB_MODEL_OFF) — all off in normal runs, so prod is unaffected. """ from __future__ import annotations import argparse import base64 import csv import io import os import numpy as np from dotenv import load_dotenv from PIL import Image, ImageDraw load_dotenv() from lawn_estimator.pipeline import run # noqa: E402 # All flags this study toggles — cleared before each config so runs don't leak. ALL_FLAGS = ["SAM_RESTRICT", "ROW_TO_CURB", "GREEN_RECLAIM", "LAWN_CASCADE", "LAWN_FORCE_RGB_ONLY", "LAWN_RGB_MODEL_OFF", "LAWN_MODEL"] # The ablation matrix: segmentation {cascade, pre-cascade, color-only} × LiDAR {on, off}. CONFIGS = [ ("Cascade + LiDAR", "reference — current prod pipeline", {"SAM_RESTRICT": "1", "ROW_TO_CURB": "1", "GREEN_RECLAIM": "1", "LAWN_CASCADE": "1"}), ("Cascade · NO LiDAR", "imagery-only — the national-scale question", {"SAM_RESTRICT": "1", "ROW_TO_CURB": "1", "GREEN_RECLAIM": "1", "LAWN_CASCADE": "1", "LAWN_FORCE_RGB_ONLY": "1"}), ("Pre-cascade + LiDAR", "incumbent + green reclaim (v0.2)", {"SAM_RESTRICT": "1", "ROW_TO_CURB": "1", "GREEN_RECLAIM": "1"}), ("Color baseline + LiDAR", "no neural model — HSV green rule × LiDAR", {"SAM_RESTRICT": "1", "ROW_TO_CURB": "1", "LAWN_RGB_MODEL_OFF": "1"}), ("Color baseline · NO LiDAR", "the floor: green pixels × pixel area", {"ROW_TO_CURB": "1", "LAWN_RGB_MODEL_OFF": "1", "LAWN_FORCE_RGB_ONLY": "1"}), ] def b64(img: Image.Image, max_w: int = 460) -> 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=78) return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode() def cell_viz(cap: dict, result: dict) -> str: """Compact overlay of the measured lawn for this config.""" img = cap["image"].convert("RGB") est = cap.get("est") viz = getattr(est, "viz", {}) if est else {} gpx, gpy = viz.get("ground_px"), viz.get("ground_py") if gpx is not None: # LiDAR path — dot the ground points d = ImageDraw.Draw(img) is_lawn = viz.get("lawn_mask") for k in range(0, len(gpx), 2): # subsample for the thumbnail col = (50, 220, 50) if (is_lawn is not None and is_lawn[k]) else (235, 60, 60) d.ellipse([gpx[k] - 2, gpy[k] - 2, gpx[k] + 2, gpy[k] + 2], fill=col) else: # RGB-only — fill the lawn mask fill = viz.get("lawn_fill", cap.get("lawn_area_in_parcel")) if fill is not None: arr = np.asarray(img).astype(np.float32) arr[fill] = arr[fill] * 0.45 + np.array([50, 220, 50], np.float32) * 0.55 img = Image.fromarray(arr.astype(np.uint8)) return b64(img) def apply_env(flags: dict) -> None: for f in ALL_FLAGS: os.environ.pop(f, None) for k, v in flags.items(): os.environ[k] = v def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--csv", required=True) ap.add_argument("--out", required=True) ap.add_argument("--imagery", default="google") args = ap.parse_args() with open(args.csv, newline="", encoding="utf-8") as f: addresses = [r[next(iter(r))].strip() for r in csv.DictReader(f) if r[next(iter(r))].strip()] # results[config_name][address] = (sqft, viz_b64) results: dict[str, dict[str, tuple]] = {} for cname, _desc, flags in CONFIGS: results[cname] = {} for addr in addresses: apply_env(flags) try: cap: dict = {} res = run(addr, imagery=args.imagery, capture=cap) results[cname][addr] = (res["lawn_sqft"], cell_viz(cap, res)) print(f"[{cname}] {addr}: {res['lawn_sqft']:,.0f}", flush=True) except Exception as exc: results[cname][addr] = (None, "") print(f"[{cname}] {addr}: FAILED {exc}", flush=True) ref = CONFIGS[0][0] short = [a.split(",")[0] for a in addresses] # matrix table head = "".join(f"{s}" for s in short) rows = "" for cname, desc, _ in CONFIGS: cells = "" for addr in addresses: sqft, _ = results[cname][addr] refv = results[ref][addr][0] if sqft is None: cells += '—' else: pct = (sqft - refv) / refv * 100 if refv else 0 cls = "" if cname == ref else ("hi" if pct > 4 else "lo" if pct < -4 else "ok") delta = "" if cname == ref else f'{pct:+.0f}%' cells += f'{sqft:,.0f}{delta}' rows += f'{cname}{desc}{cells}' # per-address viz strips strips = "" for addr, s in zip(addresses, short, strict=False): cardset = "" for cname, _d, _f in CONFIGS: sqft, viz = results[cname][addr] lbl = f"{sqft:,.0f}" if sqft is not None else "—" cardset += (f'
{cname}' f'
{cname}{lbl}
') strips += f'

{s}

{cardset}
' html = PAGE.replace("", head).replace("", rows).replace( "", strips).replace("{ref}", ref).replace("{n}", str(len(addresses))) with open(args.out, "w", encoding="utf-8") as f: f.write(html) print(f"\nablation page: {args.out} ({os.path.getsize(args.out)/1e6:.1f} MB)") PAGE = """Ablation study — what each signal contributes

Ablation study

Each configuration is the REAL pipeline with one signal changed, over {n} addresses. Reference = {ref}; the % next to each number is the change vs that. Green dots (below) = LiDAR points counted as lawn, red = removed; solid green = an imagery-only lawn fill (no LiDAR).

configuration

Read the "· NO LiDAR" rows against the reference: that gap is what dropping LiDAR costs (the national-scale question). The "Color baseline" rows show what the neural segmenter adds over a plain green rule.

""" if __name__ == "__main__": main()