liveboard: auto-build a geo map from model lat/lon (Path B)

When the model has latitude/longitude columns and the board has no geo viz,
synthesize a GEO_BUBBLE so the layout places it full-width under the KPIs
("geo on almost anything, when the data supports it").

golden_layout.py:
- detect_geo_columns(): find lat/lon (word-boundary match, so "Platform"
isn't a false hit) + a geo attribute + a bubble measure from model columns
- _build_geo_viz(): synthesize a GEO_BUBBLE answer (clone a sibling's model
ref; search_query [Lat] [Lon] sum [Measure] [GeoDim]; minimal mapview
client_state that auto-fits)
- _maybe_build_geo_viz(): only fires when lat/lon exist and no geo viz present
- build_dashboard_plan gains model_columns; apply_golden_two_tab_layout adds
the synthesized viz so it lands in the Geographic View group

liveboard_creator.py: _fetch_model_columns() resolves the board's model
(export w/ fqn -> model TML) and passes its columns to the plan; best-effort
(returns [] on failure so liveboard creation never breaks).

Verified live: stripped Tixr's geo viz, fed the model columns, and the engine
rebuilt a GEO_BUBBLE that renders as a venue map.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

demoprep_app/liveboards/golden_layout.py

@@ -23,6 +23,7 @@ def build_dashboard_plan(
     logo_url: str | None = None,
     company_url: str | None = None,
     keep_overview_note: bool = False,
+    model_columns: List[Dict[str, Any]] | None = None,
 ) -> Dict[str, Any]:
     """Return the first-pass dashboard architecture for lab-created liveboards.
 
@@ -38,6 +39,7 @@ def build_dashboard_plan(
         "logo_url": logo_url,
         "company_url": company_url,
         "keep_overview_note": keep_overview_note,
+        "model_columns": model_columns,
         "executive_tab": executive_tab,
         "manager_tab": manager_tab,
         "target_viz_count": target_viz_count,
@@ -155,6 +157,15 @@ def apply_golden_two_tab_layout(
         visualizations = [viz for viz in visualizations if "note_tile" not in viz]
     liveboard["visualizations"] = visualizations
 
+    # Auto-build a geo map when the model has coordinates and the board has none.
+    geo_viz = _maybe_build_geo_viz(data_visualizations, plan.get("model_columns"))
+    geo_change = ""
+    if geo_viz is not None:
+        data_visualizations.append(geo_viz)
+        visualizations.append(geo_viz)
+        liveboard["visualizations"] = visualizations
+        geo_change = f"Built geo map: {geo_viz['answer']['name']}"
+
     kpi_changes = _convert_periodic_metric_charts(data_visualizations)
     chart_changes = _reduce_heatmaps_and_apply_palettes(data_visualizations)
     mode_changes = _force_chart_mode(data_visualizations)
@@ -173,7 +184,7 @@ def apply_golden_two_tab_layout(
         f"Applied golden two-tab layout ({len(groups)} groups)",
         f"Tabs: {_executive_tab_name(plan)}, {_manager_tab_name(plan)}",
     ]
-    for change in (kpi_changes, chart_changes, mode_changes, color_changes, title_changes):
+    for change in (geo_change, kpi_changes, chart_changes, mode_changes, color_changes, title_changes):
         if change:
             changes.append(change)
     return updated, changes
@@ -581,6 +592,104 @@ def _clean_viz_titles(visualizations: List[Dict[str, Any]]) -> str:
     return f"Cleaned {cleaned} viz title(s)" if cleaned else ""
 
 
+_LAT_RE = re.compile(r"\blat(itude)?\b", re.IGNORECASE)
+_LON_RE = re.compile(r"\b(lon(gitude)?|lng)\b", re.IGNORECASE)
+_GEO_DIM_HINTS = (
+    "city", "venue", "location", "region", "country", "state", "province",
+    "market", "branch", "store", "site", "territory", "metro", "county",
+)
+
+
+def detect_geo_columns(model_columns: List[Dict[str, Any]]) -> Dict[str, str] | None:
+    """From a model's columns [{name, type}], find lat/lon + a geo attribute + a
+    measure. Returns {lat, lon, geodim, measure} or None if not geographic."""
+    names = [(c.get("name", ""), str(c.get("type", "")).upper()) for c in (model_columns or []) if c.get("name")]
+    lat = next((n for n, _ in names if _LAT_RE.search(n)), None)
+    lon = next((n for n, _ in names if _LON_RE.search(n)), None)
+    if not (lat and lon):
+        return None
+    attrs = [n for n, t in names if t == "ATTRIBUTE"]
+    lat_prefix = _LAT_RE.sub("", lat).strip()  # e.g. "Venue Lat" -> "Venue"
+    geodim = None
+    if lat_prefix:
+        geodim = next((n for n in attrs if n.lower().startswith(lat_prefix.lower())), None)
+    if not geodim:
+        geodim = next((n for n in attrs if any(h in n.lower() for h in _GEO_DIM_HINTS)), None)
+    if not geodim:
+        geodim = attrs[0] if attrs else None
+    measures = [n for n, t in names if t == "MEASURE" and not _LAT_RE.search(n) and not _LON_RE.search(n)]
+    measure = next(
+        (n for n in measures if any(k in n.lower() for k in ("revenue", "sales", "amount", "value", "gmv", "spend"))),
+        measures[0] if measures else None,
+    )
+    if not (geodim and measure):
+        return None
+    return {"lat": lat, "lon": lon, "geodim": geodim, "measure": measure}
+
+
+def _geo_client_state(lat_col: str, lon_col: str, geodim: str, measure_col: str) -> str:
+    state = {
+        "version": "V4DOT2",
+        "chartProperties": {
+            "gridLines": {},
+            "responsiveLayoutPreference": "USER_PREFERRED_ON",
+            "chartSpecific": {"dataFieldArea": "column"},
+        },
+        "columnProperties": [{"columnId": c, "columnProperty": {}} for c in (geodim, lat_col, lon_col, measure_col)],
+        "axisProperties": [
+            {"id": "geo-y", "properties": {"axisType": "Y", "linkedColumns": [lat_col], "isOpposite": False}},
+            {"id": "geo-x", "properties": {"axisType": "X", "linkedColumns": [geodim]}},
+        ],
+        "systemMultiColorSeriesColors": [
+            {"serieName": lat_col, "colorMap": [{"serieName": geodim,
+             "color": ["#ffffb2", "#fddd87", "#fba35d", "#f75534", "#f9140a", "#d70315", "#b10026"]}]}
+        ],
+    }
+    return json_dumps_compact(state)
+
+
+def _build_geo_viz(sibling_answer: Dict[str, Any], geo: Dict[str, str], viz_id: str = "Viz_Geo") -> Dict[str, Any]:
+    lat, lon, geodim, measure = geo["lat"], geo["lon"], geo["geodim"], geo["measure"]
+    t_lat, t_lon, t_meas = f"Total {lat}", f"Total {lon}", f"Total {measure}"
+    return {
+        "id": viz_id,
+        "answer": {
+            "name": f"{measure} by {geodim}",
+            "tables": deepcopy(sibling_answer.get("tables", [])),
+            "search_query": f"[{lat}] [{lon}] sum [{measure}] [{geodim}]",
+            "answer_columns": [{"name": t_meas}, {"name": t_lat}, {"name": t_lon}, {"name": geodim}],
+            "chart": {
+                "type": "GEO_BUBBLE",
+                "chart_columns": [{"column_id": t_meas}, {"column_id": t_lat}, {"column_id": t_lon}, {"column_id": geodim}],
+                "axis_configs": [{"x": [geodim], "y": [t_lat]}],
+                "client_state": "",
+                "client_state_v2": _geo_client_state(t_lat, t_lon, geodim, t_meas),
+            },
+            "display_mode": "CHART_MODE",
+        },
+    }
+
+
+def _maybe_build_geo_viz(
+    visualizations: List[Dict[str, Any]],
+    model_columns: List[Dict[str, Any]] | None,
+) -> Dict[str, Any] | None:
+    """Synthesize a GEO_BUBBLE when the model has lat/lon and the board has none."""
+    if not model_columns:
+        return None
+    if any(str(v.get("answer", {}).get("chart", {}).get("type", "")).startswith("GEO")
+           for v in visualizations if "note_tile" not in v):
+        return None
+    geo = detect_geo_columns(model_columns)
+    if not geo:
+        return None
+    sibling = next((v.get("answer", {}) for v in visualizations
+                    if "note_tile" not in v and v.get("answer", {}).get("tables")), None)
+    if not sibling:
+        return None
+    return _build_geo_viz(sibling, geo)
+
+
 def _apply_color_by_dimension(visualizations: List[Dict[str, Any]]) -> str:
     """Color ADVANCED_* charts by an attribute via custom_chart_config's
     'slice-with-color' slot — the new chart engine's real color mechanism

liveboard_creator.py

@@ -4080,6 +4080,50 @@ def create_liveboard_from_model_mcp(
         return {'success': False, 'error': err}
 
 
+def _fetch_model_columns(ts_client, liveboard_guid: str) -> List[Dict]:
+    """Export the board (with fqn) -> resolve its model -> return the model's
+    columns as [{name, type}]. Best-effort: returns [] on any failure, so the
+    geo auto-build simply doesn't fire rather than breaking liveboard creation.
+    """
+    try:
+        yaml.SafeLoader.add_constructor(
+            "tag:yaml.org,2002:value", lambda loader, node: loader.construct_scalar(node)
+        )
+        export = ts_client.session.post(
+            f"{ts_client.base_url}/api/rest/2.0/metadata/tml/export",
+            json={"metadata": [{"identifier": liveboard_guid}], "export_fqn": True, "export_associated": False},
+            timeout=60,
+        )
+        if export.status_code != 200:
+            return []
+        board = yaml.safe_load(export.json()[0].get("edoc", "")) or {}
+        model_fqn = None
+        for viz in board.get("liveboard", {}).get("visualizations", []):
+            for tbl in (viz.get("answer", {}).get("tables") or []):
+                if tbl.get("fqn"):
+                    model_fqn = tbl["fqn"]
+                    break
+            if model_fqn:
+                break
+        if not model_fqn:
+            return []
+        m_export = ts_client.session.post(
+            f"{ts_client.base_url}/api/rest/2.0/metadata/tml/export",
+            json={"metadata": [{"identifier": model_fqn}], "export_associated": False},
+            timeout=60,
+        )
+        if m_export.status_code != 200:
+            return []
+        model_root = (yaml.safe_load(m_export.json()[0].get("edoc", "")) or {}).get("model") or {}
+        return [
+            {"name": c.get("name"), "type": (c.get("properties") or {}).get("column_type")}
+            for c in model_root.get("columns", []) if c.get("name")
+        ]
+    except Exception as exc:
+        print(f"  [geo] model-column fetch failed; skipping geo auto-build: {exc}", flush=True)
+        return []
+
+
 def enhance_mcp_liveboard(
     liveboard_guid: str,
     company_data: Dict,
@@ -4550,12 +4594,14 @@ def enhance_mcp_liveboard(
                     build_dashboard_plan,
                 )
 
+                model_columns = _fetch_model_columns(ts_client, liveboard_guid)
                 plan = build_dashboard_plan(
                     company_name=company_data.get('name', 'Company'),
                     use_case=company_data.get('use_case') or company_data.get('use_case_name') or 'Analytics',
                     target_viz_count=len(visualizations),
                     logo_url=company_data.get('logo_url'),
                     company_url=company_data.get('url') or company_data.get('company_url') or company_data.get('website'),
+                    model_columns=model_columns,
                 )
                 liveboard_tml, golden_changes = apply_golden_two_tab_layout(liveboard_tml, plan)
                 enhancements_applied.extend(golden_changes)