Spaces:
Running
Running
| """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() | |