"""W20A2: FP geometry readback bridge — pure deterministic builder. Consumes a ``base_location_dossier`` (W20A) and emits: - ``readback`` — a ``GeometryReadback`` shape with marker cells. status ∈ {ok, synthetic_fixture, failed}. **W20A2 produces synthetic_fixture ONLY**; the ``ok`` status path is reserved for a future wave that wires a real VLM provider (see the provider-slot docstring below). - ``geometry`` — geometry candidates derived from the readback: per-marker cells, per-unit camera/look-at candidate cells, direction-vector records, view-cone records keyed on a generic lens enum, visible-unit / visible-opening superset maps, and wall/door invalidation diagnostics. W20 boundaries enforced in this module: - exact-ID only — every join uses marker number / fp_id exact equality. - no semantic reference policy, no camera pick. Code emits **candidate** sets and diagnostics; the consumer (W20B) makes the LLM-owned decisions. - no scenario-specific static literals (the only string keys here are generic enum / shape-kind strings). - no LLM / image / VLM / DB / ImageAsset imports. - the synthetic fixture path is intentionally a placeholder. It exists so downstream code (W20B planner contract, HTML overlay emitter, tests) can be exercised before the real VLM is wired. The geometry decisions made off synthetic cells must not be promoted to a production smoke without first being recomputed against a real VLM readback. Real VLM provider slot ====================== A ``call_vlm_grid_readback`` callable can be passed into ``compute_readback(...)`` to swap the synthetic_fixture path for a real provider. The callable signature is:: Callable[[BaseLocationDossier, fp_image_path], GeometryReadback] The W20A2 wave **does not** wire any real provider. ``compute_readback`` with ``vlm_provider=None`` (default) always returns the synthetic_fixture readback. A future wave behind explicit approval will wire a litellm vision call to populate the ``ok`` path; see ``backend/scripts/experiment_floor_plan_vlm_readback_slice.py`` (W17C) for the historical reference pattern. """ from __future__ import annotations from typing import Any, Callable, Dict, List, Optional, Tuple # Aligned with W20A / W19B-1 partition enum (frozenset reuse would create # a reverse dependency; re-declare to keep the module self-contained). BASE_STRUCTURAL_UNIT = "base_structural_unit" BASE_OPENING = "base_opening" BASE_PERSISTENT_FIXTURE = "base_persistent_fixture" BASE_PERSISTENT_FURNITURE = "base_persistent_furniture" BASE_KINDS: Tuple[str, ...] = ( BASE_STRUCTURAL_UNIT, BASE_OPENING, BASE_PERSISTENT_FIXTURE, BASE_PERSISTENT_FURNITURE, ) # Mirrors base_location_dossier.VIEW_CONE_LENS_ENUM_MAP. Generic — no # scenario tokens. VIEW_CONE_LENS_ENUM_MAP: Dict[str, int] = { "wide": 90, "normal": 50, "telephoto": 25, } DEFAULT_GRID_SIZE: Tuple[int, int] = (10, 10) class GeometryReadbackError(Exception): """Fail-closed signal for the readback / geometry builder.""" # Type aliases for readability — JSON-shape dicts. ReadbackEntry = Dict[str, Any] GeometryReadback = Dict[str, Any] GeometryCandidates = Dict[str, Any] # ───────────────────────────────────────────────────────────────────── # Synthetic readback fixture # ───────────────────────────────────────────────────────────────────── def _spread_marker_cells( *, base_inventory: List[Dict[str, Any]], grid_size: Tuple[int, int], ) -> Tuple[List[ReadbackEntry], List[str]]: """Deterministically spread base markers across the grid. Marker order is ``BASE_KINDS`` outer × ``number ascending`` inner. Each marker takes the next unused cell in column-major order. This is **not** a realistic floor-plan layout — it exists so geometry computation has something to operate on before a real VLM readback is wired up. Raises ``GeometryReadbackError`` when the grid cannot fit every base marker (``rows * cols < |markers|``). This is intentional: a silent overwrite would let synthetic geometry collide cells and mask bugs downstream. """ rows, cols = grid_size diagnostics: List[str] = [] by_kind: Dict[str, List[Dict[str, Any]]] = {k: [] for k in BASE_KINDS} for entry in base_inventory: decision = entry.get("base_layer_decision") if decision in by_kind: by_kind[decision].append(entry) observed: List[ReadbackEntry] = [] used: set = set() cell_idx = 0 total_to_place = sum(len(v) for v in by_kind.values()) grid_capacity = rows * cols if total_to_place > grid_capacity: raise GeometryReadbackError( f"synthetic_fixture cannot fit {total_to_place} base markers " f"in a {rows}x{cols} grid (capacity {grid_capacity})" ) for kind in BASE_KINDS: for entry in sorted(by_kind[kind], key=lambda e: int(e["number"])): while True: # Column-major spread keeps successive markers visually # far apart in a typical 10×10 layout. r = cell_idx % rows c = cell_idx // rows cell_idx += 1 if c >= cols: # Should not be reached because the capacity guard # above already rejects oversized inputs. raise GeometryReadbackError( "synthetic_fixture column overflow — guard " "should have prevented this" ) if (r, c) in used: continue used.add((r, c)) observed.append( { "number": int(entry["number"]), "row": r, "col": c, "kind": kind, } ) break diagnostics.append( "synthetic_fixture: marker cells are deterministic placeholders, " "not real VLM observations. Geometry derived from this readback " "must not be promoted to a production smoke." ) return observed, diagnostics def compute_synthetic_readback_fixture( *, dossier: Dict[str, Any], grid_size: Tuple[int, int] = DEFAULT_GRID_SIZE, ) -> GeometryReadback: """Return a synthetic_fixture-status readback for the given dossier. The readback shape mirrors what a real VLM provider would return, so the geometry / HTML layer below treats both code paths uniformly. """ fp_id = dossier.get("fp_id") if not isinstance(fp_id, str) or not fp_id: raise GeometryReadbackError( f"dossier.fp_id missing or non-string (got {fp_id!r})" ) base_inv = dossier.get("base_marker_inventory") or [] if not base_inv: raise GeometryReadbackError( f"dossier fp_id={fp_id!r} has empty base_marker_inventory" ) observed, diagnostics = _spread_marker_cells( base_inventory=base_inv, grid_size=grid_size, ) inv_numbers = {int(e["number"]) for e in base_inv} seen_numbers = {e["number"] for e in observed} missing = sorted(inv_numbers - seen_numbers) extra = sorted(seen_numbers - inv_numbers) return { "status": "synthetic_fixture", "fp_id": fp_id, "grid_size": list(grid_size), "observed_markers": observed, "missing_markers": missing, "extra_markers": extra, "confidence": None, "diagnostics": diagnostics, } def compute_readback( *, dossier: Dict[str, Any], fp_image_path: Optional[str] = None, grid_size: Tuple[int, int] = DEFAULT_GRID_SIZE, vlm_provider: Optional[ Callable[..., GeometryReadback] ] = None, ) -> GeometryReadback: """Return a readback for the dossier's fp_id. When ``vlm_provider`` is ``None`` (W20A2 default), the synthetic_fixture path is used and ``fp_image_path`` is ignored. When a provider callable is passed, it is invoked as ``vlm_provider(dossier=..., fp_image_path=..., grid_size=...)`` and its return value must be a readback dict with ``status='ok'``, matching the GeometryReadback shape. Any non-conforming return raises ``GeometryReadbackError``. """ if vlm_provider is None: return compute_synthetic_readback_fixture( dossier=dossier, grid_size=grid_size ) out = vlm_provider( dossier=dossier, fp_image_path=fp_image_path, grid_size=grid_size, ) if not isinstance(out, dict): raise GeometryReadbackError( f"vlm_provider returned non-dict ({type(out).__name__})" ) if out.get("status") != "ok": raise GeometryReadbackError( f"vlm_provider returned status={out.get('status')!r}; only " f"'ok' is accepted on the real-provider path" ) if out.get("fp_id") != dossier.get("fp_id"): raise GeometryReadbackError( f"vlm_provider fp_id mismatch: {out.get('fp_id')!r} != " f"dossier {dossier.get('fp_id')!r}" ) return out # ───────────────────────────────────────────────────────────────────── # Geometry candidates # ───────────────────────────────────────────────────────────────────── def _validate_readback_cells( *, readback: GeometryReadback, grid_size: Tuple[int, int] ) -> List[str]: """Validate readback cell shape and return cell-collision diagnostics. W20E6-A: distinct marker numbers may share a single 10x10 coarse cell. The hard fail on duplicate ``(row, col)`` has been lowered to a returned diagnostic so the geometry layer can soft-degrade instead of dropping the whole readback. All other invariants stay fail-closed: - duplicate marker NUMBER -> raise - non-int row/col -> raise - out-of-grid cell -> raise - unknown kind -> raise Returns a list of per-pair collision messages (empty when no collisions). The caller folds these into the geometry diagnostics. """ rows, cols = grid_size seen_numbers: set = set() cell_owner: Dict[Tuple[int, int], int] = {} cell_collision_diagnostics: List[str] = [] for entry in readback.get("observed_markers") or []: num = entry.get("number") r = entry.get("row") c = entry.get("col") kind = entry.get("kind") if not isinstance(num, int) or isinstance(num, bool): raise GeometryReadbackError( f"observed_marker.number must be int (got {num!r})" ) if num in seen_numbers: raise GeometryReadbackError( f"observed_marker.number={num} duplicated in readback" ) seen_numbers.add(num) if not isinstance(r, int) or not isinstance(c, int): raise GeometryReadbackError( f"observed_marker #{num} row/col must be int " f"(got row={r!r}, col={c!r})" ) if not (0 <= r < rows and 0 <= c < cols): raise GeometryReadbackError( f"observed_marker #{num} cell ({r},{c}) out of grid " f"({rows}x{cols})" ) prior = cell_owner.get((r, c)) if prior is not None: cell_collision_diagnostics.append( f"observed_marker #{num} shares 10x10 cell ({r},{c}) " f"with marker #{prior} (W20E6-A: surfaced as diagnostic, " f"not a hard failure)" ) else: cell_owner[(r, c)] = num if kind not in BASE_KINDS: raise GeometryReadbackError( f"observed_marker #{num} kind={kind!r} not in " f"{list(BASE_KINDS)}" ) return cell_collision_diagnostics def compute_geometry_candidates( *, dossier: Dict[str, Any], readback: GeometryReadback, ) -> GeometryCandidates: """Compute geometry candidates from the dossier + readback. The output carries only **candidate** sets and diagnostics — no LLM-owned decisions. W20 boundaries: code never picks a camera, never picks a reference image, and never asserts that two candidates depict the same physical space. Raises ``GeometryReadbackError`` on: - readback.status not in {ok, synthetic_fixture} - readback cell out of grid, duplicate cell, unknown kind - readback fp_id mismatch with dossier """ if readback.get("status") not in {"ok", "synthetic_fixture"}: raise GeometryReadbackError( f"readback.status={readback.get('status')!r} not in " f"{{ok, synthetic_fixture}}" ) if readback.get("fp_id") != dossier.get("fp_id"): raise GeometryReadbackError( f"readback.fp_id={readback.get('fp_id')!r} != dossier " f"fp_id={dossier.get('fp_id')!r}" ) grid = tuple(readback.get("grid_size") or DEFAULT_GRID_SIZE) if len(grid) != 2 or not all( isinstance(v, int) and v > 0 for v in grid ): raise GeometryReadbackError( f"readback.grid_size invalid: {readback.get('grid_size')!r}" ) grid_size: Tuple[int, int] = (int(grid[0]), int(grid[1])) cell_collision_diagnostics: List[str] = _validate_readback_cells( readback=readback, grid_size=grid_size ) marker_cells: Dict[int, List[int]] = {} marker_kinds: Dict[int, str] = {} for entry in readback["observed_markers"]: marker_cells[entry["number"]] = [entry["row"], entry["col"]] marker_kinds[entry["number"]] = entry["kind"] def _by_kind(kind: str) -> Dict[int, List[int]]: return { n: list(cell) for n, cell in marker_cells.items() if marker_kinds[n] == kind } unit_cells = _by_kind(BASE_STRUCTURAL_UNIT) opening_cells = _by_kind(BASE_OPENING) fixture_cells = _by_kind(BASE_PERSISTENT_FIXTURE) furniture_cells = _by_kind(BASE_PERSISTENT_FURNITURE) units_sorted = sorted(unit_cells) openings_sorted = sorted(opening_cells) # Camera cell candidates: W20A2 simple — the unit's anchor cell is # the only candidate. A future wave with VLM-observed unit rects # can expand this to every interior cell of the rect. camera_cell_candidates_per_unit: Dict[int, List[List[int]]] = { u: [list(unit_cells[u])] for u in units_sorted } # Look-at cell candidates per unit: every opening, fixture, and # furniture cell on the floor plan. We cannot derive per-unit # membership of openings/fixtures from cell positions alone (would # require either VLM-observed unit rects or wall polylines). A # diagnostic surfaces that caveat. Cells are de-duped (W20E6-A: # two markers may share a 10x10 coarse cell, so a naive # concatenation would produce duplicate candidates). _look_at_cell_set: set = set() for source in ( opening_cells.values(), fixture_cells.values(), furniture_cells.values(), ): for cell in source: _look_at_cell_set.add(tuple(cell)) shared_look_at_cells: List[List[int]] = sorted( list(c) for c in _look_at_cell_set ) look_at_cell_candidates_per_unit: Dict[int, List[List[int]]] = { u: [list(c) for c in shared_look_at_cells] for u in units_sorted } # Direction vector records: every (camera_unit, look_at_unit) pair. direction_vector_records: List[Dict[str, Any]] = [] for cam_u in units_sorted: cam_cell = unit_cells[cam_u] for tgt_u in units_sorted: if tgt_u == cam_u: continue tgt_cell = unit_cells[tgt_u] dy = tgt_cell[0] - cam_cell[0] dx = tgt_cell[1] - cam_cell[1] direction_vector_records.append( { "camera_unit": cam_u, "camera_cell": list(cam_cell), "look_at_unit": tgt_u, "look_at_cell": list(tgt_cell), "dx": dx, "dy": dy, "length_l1": abs(dx) + abs(dy), } ) # View cone records: per (direction_vector, lens_enum). Each cone # is bound to a concrete camera→look-at candidate pair (not just # a camera unit) so the HTML layer can rotate the cone polygon # along the actual candidate direction instead of a fixed axis. # Code does NOT pick a final camera or final look-at — the cone # is a candidate, one per (direction_vector × lens) cross-product. # When only one (or zero) unit is observed, direction_vector_records # is empty and so is this list (with a diagnostic surfaced below). view_cone_records: List[Dict[str, Any]] = [] for dv in direction_vector_records: for lens, fov in VIEW_CONE_LENS_ENUM_MAP.items(): view_cone_records.append( { "camera_unit": dv["camera_unit"], "camera_cell": list(dv["camera_cell"]), "look_at_unit": dv["look_at_unit"], "look_at_cell": list(dv["look_at_cell"]), "dx": dv["dx"], "dy": dv["dy"], "lens_enum": lens, "fov_deg": fov, } ) # Visible-unit / visible-opening candidates: supersets — every # other unit (resp. every opening) is surfaced as a candidate, and # the wall/door invalidation diagnostics call out that the actual # adjacency relations cannot be derived from cells alone. visible_units_candidates: Dict[int, List[int]] = { cam_u: [u for u in units_sorted if u != cam_u] for cam_u in units_sorted } visible_openings_candidates: Dict[int, List[int]] = { cam_u: list(openings_sorted) for cam_u in units_sorted } wall_door_diagnostics: List[str] = [ "readback does not encode wall segment polylines; wall/door " "line-of-sight invalidation cannot be performed at this layer.", "opening-to-unit adjacency cannot be derived from cell " "positions alone; consumer must treat visible_units_candidates " "and visible_openings_candidates as supersets, not committed sets.", ] diagnostics: List[str] = [] # W20E6-A: surface per-pair collision diagnostics first so the # consumer (HTML overlay + LLM planner prompt) can flag the # readback as imperfect without losing the cells themselves. diagnostics.extend(cell_collision_diagnostics) if readback["status"] == "synthetic_fixture": diagnostics.append( "geometry candidates computed from a synthetic_fixture " "readback; placeholder cells, not production geometry." ) if not units_sorted: diagnostics.append( "zero base_structural_unit cells in readback; geometry " "candidates are empty." ) if not direction_vector_records: diagnostics.append( "fewer than two base_structural_unit cells in readback; " "direction-vector candidates are empty and view-cone " "records were skipped (a cone needs a direction)." ) return { "fp_id": dossier["fp_id"], "grid_size": list(grid_size), "readback_status": readback["status"], "marker_cells": {str(n): list(c) for n, c in marker_cells.items()}, "unit_marker_cells": {str(n): list(c) for n, c in unit_cells.items()}, "opening_marker_cells": { str(n): list(c) for n, c in opening_cells.items() }, "fixture_marker_cells": { str(n): list(c) for n, c in fixture_cells.items() }, "furniture_marker_cells": { str(n): list(c) for n, c in furniture_cells.items() }, "camera_cell_candidates_per_unit": { str(u): [list(c) for c in cells] for u, cells in camera_cell_candidates_per_unit.items() }, "look_at_cell_candidates_per_unit": { str(u): [list(c) for c in cells] for u, cells in look_at_cell_candidates_per_unit.items() }, "direction_vector_records": direction_vector_records, "view_cone_records": view_cone_records, "visible_units_candidates": { str(u): list(v) for u, v in visible_units_candidates.items() }, "visible_openings_candidates": { str(u): list(v) for u, v in visible_openings_candidates.items() }, "wall_door_invalidation_diagnostics": wall_door_diagnostics, "cell_collision_diagnostics": list(cell_collision_diagnostics), "diagnostics": diagnostics, }