"""The tuning cockpit — a local Gradio app. Three working tabs (plus export buttons): Label brush-paint the mowable area on each gold lot -> sqft + mask Tune configure any recipe, Run -> per-step image gallery + measurement + error/IoU Compare run saved recipes over the gold set -> ranked metrics table Run it locally: python -m tuning.app (or `make tune`) Nothing here touches src/lawn_estimator — it drives the pipeline read-only via the harness. """ from __future__ import annotations import json from dataclasses import asdict, fields import gradio as gr from tuning import cache as tcache from tuning import export, gold, label_ui, metrics, render from tuning.harness import ( LIDAR_MODES, RESTRICTS, SEGMENTERS, Recipe, try_run_recipe, ) from tuning.label_ui import addr_choices as _addr_choices from tuning.label_ui import addr_from_label as _addr_from_label RECIPES_PATH = tcache.TUNING_DIR / "recipes.json" EXPORT_DIR = tcache.TUNING_DIR / "exports" FRAME = label_ui.FRAME # the canonical gold frame (google, z20, 640, x2) # --------------------------------------------------------------------------- # Saved-recipe registry (JSON-persisted; seeded with the two presets) # --------------------------------------------------------------------------- def _recipe_fields() -> list[str]: return [f.name for f in fields(Recipe)] def load_recipes() -> dict[str, Recipe]: presets = {"baseline": Recipe(), "prod": Recipe.prod()} if RECIPES_PATH.exists(): for name, d in json.loads(RECIPES_PATH.read_text(encoding="utf-8")).items(): presets[name] = Recipe(**{k: v for k, v in d.items() if k in _recipe_fields()}) return presets def save_recipe(recipe: Recipe) -> None: RECIPES_PATH.parent.mkdir(parents=True, exist_ok=True) saved = {} if RECIPES_PATH.exists(): saved = json.loads(RECIPES_PATH.read_text(encoding="utf-8")) saved[recipe.name] = asdict(recipe) RECIPES_PATH.write_text(json.dumps(saved, indent=2), encoding="utf-8") # --------------------------------------------------------------------------- # Small helpers # --------------------------------------------------------------------------- def _ids(text: str) -> list[int] | None: text = (text or "").strip() if not text: return None return [int(x) for x in text.replace(" ", "").split(",") if x] def _prompts(text: str, default: list[str]) -> list[str]: parts = [p.strip() for p in (text or "").split(",") if p.strip()] return parts or default def _same_grid(frame: dict, recipe) -> bool: """The gold mask registers to the projection grid (zoom/size/scale), independent of which imagery painted it — so a leaf-off (county) run can still be scored by IoU against a leaf-on gold mask as long as the grid matches.""" return all(frame.get(k) == getattr(recipe, k) for k in ("zoom", "image_size", "image_scale")) # --------------------------------------------------------------------------- # TUNE tab # --------------------------------------------------------------------------- TUNE_INPUTS: list[str] = [ "name", "address", "imagery", "zoom", "image_size", "image_scale", "row_buffer_ft", "row_to_curb", "road_half_width_ft", "row_curb_cap_ft", "street_max_dist_m", "segmenter", "lawn_model", "lawn_class_ids", "cascade_arbitrated_class_ids", "canopy_class_id", "green_reclaim", "sam3_lawn_prompts", "sam3_canopy_prompts", "sam3_score_threshold", "sam3_mask_threshold", "restrict", "seed_step", "min_piece", "roof_min_bldg", "roof_area_frac", "concrete_gray", "concrete_green", "sam3_restrict_prompts", "lidar", "min_ground_points", "fusion_base_segmenter", "fusion_hardscape_detector", "fusion_undercanopy_only", "fusion_gray_sat", "fusion_sam3_prompts", ] def build_recipe(vals: dict) -> Recipe: def num(v, cast): return None if v in (None, "") else cast(v) return Recipe( name=vals["name"] or "custom", imagery=vals["imagery"], zoom=int(vals["zoom"]), image_size=int(vals["image_size"]), image_scale=int(vals["image_scale"]), row_buffer_ft=float(vals["row_buffer_ft"]), row_to_curb=bool(vals["row_to_curb"]), road_half_width_ft=float(vals["road_half_width_ft"]), row_curb_cap_ft=float(vals["row_curb_cap_ft"]), street_max_dist_m=float(vals["street_max_dist_m"]), segmenter=vals["segmenter"], lawn_model=vals["lawn_model"].strip(), lawn_class_ids=_ids(vals["lawn_class_ids"]), cascade_arbitrated_class_ids=_ids(vals["cascade_arbitrated_class_ids"]), canopy_class_id=num(vals["canopy_class_id"], int), green_reclaim=bool(vals["green_reclaim"]), sam3_lawn_prompts=_prompts(vals["sam3_lawn_prompts"], ["grass lawn"]), sam3_canopy_prompts=_prompts(vals["sam3_canopy_prompts"], ["tree canopy"]), sam3_score_threshold=float(vals["sam3_score_threshold"]), sam3_mask_threshold=float(vals["sam3_mask_threshold"]), restrict=vals["restrict"], seed_step=num(vals["seed_step"], int), min_piece=num(vals["min_piece"], int), roof_min_bldg=num(vals["roof_min_bldg"], int), roof_area_frac=num(vals["roof_area_frac"], float), concrete_gray=num(vals["concrete_gray"], float), concrete_green=num(vals["concrete_green"], float), sam3_restrict_prompts=_prompts(vals["sam3_restrict_prompts"], ["house roof", "driveway", "sidewalk"]), lidar=vals["lidar"], min_ground_points=num(vals["min_ground_points"], int), fusion_base_segmenter=vals["fusion_base_segmenter"], fusion_hardscape_detector=vals["fusion_hardscape_detector"], fusion_undercanopy_only=bool(vals["fusion_undercanopy_only"]), fusion_gray_sat=float(vals["fusion_gray_sat"]), fusion_sam3_prompts=_prompts(vals["fusion_sam3_prompts"], ["driveway", "sidewalk", "pavement", "concrete", "patio"]), ) def _step_gallery(result, recipe): """(PIL image, caption) pairs for the per-step gallery — the interactive audit.""" cap = result.capture img = cap["image"] out = [(img, "1 · analysis imagery")] out.append((render.draw_outlines(img, cap["legal_outlines"], (255, 220, 0)), f"2 · legal parcel {cap['parcel_area_sqft']:,.0f} sqft")) geo_img = render.draw_outlines(render.draw_outlines(img, cap["legal_outlines"], (255, 220, 0)), cap["estimation_outlines"], (0, 220, 220)) out.append((geo_img, f"3 · to-curb geometry {cap['estimation_area_sqft']:,.0f} sqft")) out.append((render.overlay(img, cap["veg_in_parcel"], (60, 220, 60)), f"4 · color veg {cap['rgb_veg_sqft']:,.0f} sqft")) if recipe.segmenter in ("mask2former", "eomt-cascade", "eomt-solo", "trained"): from lawn_estimator.segmentation import CASCADE_PRIMARY_MODEL_ID, LAWN_MODEL_ID, _predict_classes from tuning.harness import _segmenter_model_id if recipe.segmenter == "eomt-cascade": models = [(CASCADE_PRIMARY_MODEL_ID, render.ADE_NAMES), (LAWN_MODEL_ID, render.M2F_NAMES)] else: mid = _segmenter_model_id(recipe) names = render.ADE_NAMES if ("eomt" in mid or "ade" in mid) else render.M2F_NAMES models = [(mid, names)] for mid, names in models: panel, _leg = render.class_panel(img, _predict_classes(mid, img), names) out.append((panel, f"seg · {mid.split('/')[-1]}")) if cap.get("leaf_off_image") is not None: # leaf-on/leaf-off fusion audit loff = cap["leaf_off_image"] out.append((loff, "leaf-off (DOGIS) — same frame")) out.append((render.overlay(loff, cap["fusion_carve"], (235, 60, 60)), f"carved hardscape ({recipe.fusion_hardscape_detector}, under canopy)")) out.append((render.overlay(img, cap["fusion_lawn_before"] & cap["parcel_mask"], (60, 220, 60)), "lawn BEFORE carve (leaf-on)")) out.append((render.overlay(img, cap["lawn_area_in_parcel"], (60, 220, 60)), f"lawn mask → {result.lawn_sqft:,.0f} sqft")) if cap.get("not_lawn_mask") is not None: out.append((render.overlay(img, cap["not_lawn_mask"], (235, 60, 60)), f"restrict ({recipe.restrict}) — not-lawn removed")) if cap["est"].viz.get("ground_px") is not None: out.append((render.lidar_panel(img, cap["est"].viz), f"LiDAR classified · {result.lawn_sqft:,.0f} sqft ({result.method})")) return out def tune_run(*args): vals = dict(zip(TUNE_INPUTS, args, strict=True)) addr = _addr_from_label(vals["address"]) or vals["address"] if not addr: return [], "Enter an address.", None recipe = build_recipe(vals) result, err = try_run_recipe(addr, recipe, preview_restrict=True) if err: return [], f"**Run failed:** {err}", None gallery = _step_gallery(result, recipe) lines = [f"**{result.lawn_sqft:,.0f} sqft** · {result.method} ({result.confidence}) · " f"{result.region} · est. area {result.estimation_area_sqft:,.0f} sqft"] gold_mask, gold_meta = gold.load_mask(addr), gold.load_meta(addr) if gold_mask is not None and gold_meta and _same_grid(gold_meta.get("frame", {}), recipe) \ and gold_mask.shape == result.capture["lawn_area_in_parcel"].shape: row = metrics.lot_error(result.lawn_sqft, gold_meta["mask_sqft"]) sc = metrics.mask_scores(result.capture["lawn_area_in_parcel"], gold_mask) gallery.append((render.iou_overlay(result.capture["image"], result.capture["lawn_area_in_parcel"], gold_mask), f"vs gold · IoU {sc['iou']}")) lines.append(f"**vs gold:** true {row['true_sqft']:,.0f} sqft · error {row['signed_err']:+,.0f} " f"({row['ape_pct']:.1f}%) · IoU {sc['iou']} · precision {sc['precision']} · recall {sc['recall']}") # stash for export _LAST["result"], _LAST["recipe"] = result, recipe _LAST["gold_mask"], _LAST["gold_meta"] = gold_mask, gold_meta return gallery, "\n\n".join(lines), None _LAST: dict = {} def tune_export(): if "result" not in _LAST: return None html = export.audit_html(_LAST["result"], _LAST["recipe"], _LAST.get("gold_mask"), _LAST.get("gold_meta")) EXPORT_DIR.mkdir(parents=True, exist_ok=True) path = EXPORT_DIR / f"audit_{gold.slug(_LAST['result'].address)}.html" path.write_text(html, encoding="utf-8") return str(path) def tune_save_recipe(*args): vals = dict(zip(TUNE_INPUTS, args, strict=True)) recipe = build_recipe(vals) save_recipe(recipe) return f"Saved recipe **{recipe.name}**.", gr.update(choices=list(load_recipes())) # --------------------------------------------------------------------------- # COMPARE tab # --------------------------------------------------------------------------- def compare_run(recipe_names: list[str], progress=gr.Progress()): recipes = load_recipes() chosen = [recipes[n] for n in recipe_names if n in recipes] labeled = gold.labeled_records() if not chosen or not labeled: return [], [], None entries, table = [], [] tile_caches: dict[str, tcache.CachedImagery] = {} for rc in progress.tqdm(chosen, desc="recipes"): rows = [] for addr, meta in labeled.items(): tc = tile_caches.get(rc.imagery) result, err = try_run_recipe(addr, rc, tile_cache=tc) if err: continue row = metrics.lot_error(result.lawn_sqft, meta["mask_sqft"]) gm = gold.load_mask(addr) if gm is not None and _same_grid(meta.get("frame", {}), rc) \ and gm.shape == result.capture["lawn_area_in_parcel"].shape: row["iou"] = metrics.mask_scores(result.capture["lawn_area_in_parcel"], gm)["iou"] rows.append(row) summary = metrics.aggregate(rows) entries.append({"name": rc.name, "summary": summary}) table.append([rc.name, summary.get("n", 0), summary.get("mae"), summary.get("mape_pct"), summary.get("bias_pct"), summary.get("r2"), summary.get("mean_iou")]) table.sort(key=lambda r: (r[2] is None, r[2] if r[2] is not None else 0)) _LAST["leaderboard"] = entries return table, entries, None def compare_export(): if "leaderboard" not in _LAST: return None html = export.leaderboard_html(_LAST["leaderboard"]) EXPORT_DIR.mkdir(parents=True, exist_ok=True) path = EXPORT_DIR / "leaderboard.html" path.write_text(html, encoding="utf-8") return str(path) # --------------------------------------------------------------------------- # UI # --------------------------------------------------------------------------- def build_ui() -> gr.Blocks: d = Recipe() # defaults with gr.Blocks(title="Lawn Estimator — tuning suite") as demo: gr.Markdown("# Lawn Estimator — pipeline tuning suite\n" "Drives the production pipeline read-only. Label gold lots, tune any recipe, " "compare recipes against the gold set.") label_ui.build_label_tab() # shared with the standalone labeler (tuning.label_app) with gr.Tab("Tune"): with gr.Row(): t_name = gr.Textbox("custom", label="Recipe name", scale=1) t_addr = gr.Dropdown(_addr_choices(), label="Address (or type any)", allow_custom_value=True, scale=3) with gr.Accordion("Imagery + to-curb geometry", open=False): with gr.Row(): t_imagery = gr.Dropdown(["google", "county"], value=d.imagery, label="imagery (google = leaf-on · county = DOGIS leaf-off)") t_zoom = gr.Number(d.zoom, label="zoom", precision=0) t_size = gr.Number(d.image_size, label="image_size", precision=0) t_scale = gr.Number(d.image_scale, label="image_scale", precision=0) with gr.Row(): t_rowbuf = gr.Number(d.row_buffer_ft, label="row_buffer_ft") t_curb = gr.Checkbox(d.row_to_curb, label="row_to_curb") t_halfroad = gr.Number(d.road_half_width_ft, label="road_half_width_ft") t_curbcap = gr.Number(d.row_curb_cap_ft, label="row_curb_cap_ft") t_streetmax = gr.Number(d.street_max_dist_m, label="street_max_dist_m") with gr.Accordion("Segmenter", open=True): with gr.Row(): t_seg = gr.Dropdown(list(SEGMENTERS), value=d.segmenter, label="segmenter") t_model = gr.Textbox(d.lawn_model, label="lawn_model (mask2former/trained id)") t_green = gr.Checkbox(d.green_reclaim, label="green_reclaim") with gr.Row(): t_lawnids = gr.Textbox("", label="lawn_class_ids (e.g. 9,4)") t_arbids = gr.Textbox("", label="cascade_arbitrated_class_ids (e.g. 11,13)") t_canopy = gr.Textbox("", label="canopy_class_id") with gr.Row(): t_s3lawn = gr.Textbox(", ".join(d.sam3_lawn_prompts), label="sam3_lawn_prompts") t_s3can = gr.Textbox(", ".join(d.sam3_canopy_prompts), label="sam3_canopy_prompts") t_s3score = gr.Number(d.sam3_score_threshold, label="sam3_score_threshold") t_s3mask = gr.Number(d.sam3_mask_threshold, label="sam3_mask_threshold") with gr.Accordion("Restrict (not-lawn) + LiDAR", open=False): with gr.Row(): t_restrict = gr.Dropdown(list(RESTRICTS), value=d.restrict, label="restrict") t_lidar = gr.Dropdown(list(LIDAR_MODES), value=d.lidar, label="lidar") t_minground = gr.Textbox("", label="min_ground_points") with gr.Row(): t_seed = gr.Textbox("", label="seed_step") t_minpiece = gr.Textbox("", label="min_piece") t_roofbldg = gr.Textbox("", label="roof_min_bldg") t_roofarea = gr.Textbox("", label="roof_area_frac") t_cgray = gr.Textbox("", label="concrete_gray") t_cgreen = gr.Textbox("", label="concrete_green") t_s3restrict = gr.Textbox(", ".join(d.sam3_restrict_prompts), label="sam3_restrict_prompts") with gr.Accordion("Leaf-on/leaf-off fusion (segmenter='leafon_leafoff_fusion', Douglas only)", open=False): with gr.Row(): t_fbase = gr.Dropdown(["eomt-cascade", "mask2former", "eomt-solo"], value=d.fusion_base_segmenter, label="fusion_base_segmenter (leaf-on)") t_fdet = gr.Dropdown(["gray", "sam3"], value=d.fusion_hardscape_detector, label="fusion_hardscape_detector (leaf-off)") t_funder = gr.Checkbox(d.fusion_undercanopy_only, label="fusion_undercanopy_only") t_fgray = gr.Number(d.fusion_gray_sat, label="fusion_gray_sat") t_fprompts = gr.Textbox(", ".join(d.fusion_sam3_prompts), label="fusion_sam3_prompts") with gr.Row(): t_run = gr.Button("Run recipe", variant="primary") t_saverec = gr.Button("Save recipe") t_export = gr.Button("Export audit HTML") t_result = gr.Markdown() t_gallery = gr.Gallery(label="Pipeline steps", columns=3, height=560) t_file = gr.File(label="Audit HTML") tune_inputs = [t_name, t_addr, t_imagery, t_zoom, t_size, t_scale, t_rowbuf, t_curb, t_halfroad, t_curbcap, t_streetmax, t_seg, t_model, t_lawnids, t_arbids, t_canopy, t_green, t_s3lawn, t_s3can, t_s3score, t_s3mask, t_restrict, t_seed, t_minpiece, t_roofbldg, t_roofarea, t_cgray, t_cgreen, t_s3restrict, t_lidar, t_minground, t_fbase, t_fdet, t_funder, t_fgray, t_fprompts] t_run.click(tune_run, tune_inputs, [t_gallery, t_result, t_file]) t_export.click(tune_export, None, t_file) with gr.Tab("Compare"): gr.Markdown("Run saved recipes over every labeled gold lot; ranked by MAE.") cmp_pick = gr.CheckboxGroup(list(load_recipes()), label="Recipes", value=["baseline", "prod"]) with gr.Row(): cmp_run = gr.Button("Run over gold set", variant="primary") cmp_export = gr.Button("Export leaderboard HTML") cmp_table = gr.Dataframe(headers=["recipe", "n", "MAE", "MAPE%", "bias%", "R²", "meanIoU"], label="Leaderboard", interactive=False) cmp_file = gr.File(label="Leaderboard HTML") cmp_state = gr.State() cmp_run.click(compare_run, [cmp_pick], [cmp_table, cmp_state, cmp_file]) cmp_export.click(compare_export, None, cmp_file) t_saverec.click(tune_save_recipe, tune_inputs, [t_result, cmp_pick]) return demo def main() -> None: # Gradio 6 moved `theme` from the Blocks constructor to launch(). build_ui().launch(inbrowser=True, theme=gr.themes.Soft()) if __name__ == "__main__": main()