diff --git a/backend/server/placement/scene_bridge.py b/backend/server/placement/scene_bridge.py new file mode 100644 index 000000000..4056fab70 --- /dev/null +++ b/backend/server/placement/scene_bridge.py @@ -0,0 +1,605 @@ +"""Pure bridge from the backend scene contract to the placement wire scene. + +`convert_scene(document: dict) -> dict` takes a `base-scene/v0` document — metres, the +`meter_anchored/v0` frame, validated by `server.contracts.scene.Scene` — and returns the +placement wire scene the unchanged top-level `POST /v1/placements` server accepts +(`server/schemas/scene.schema.json`, feet). This is the pinned consumer interface for the +placement stage's `placement/scene.json` input, which nothing publishes yet +(`backend/server/placement/adapter.py` falls back to a committed reference fixture until +a stage hydrates it through this bridge). + +Purity contract: `convert_scene` reads its argument and nothing else — it loads no +fixture, reads no environment setting, and calls no solver — never mutates its argument, +and returns an equal dict for an equal document on every call. + +Coordinate mapping. Both contracts measure `s` along the wall chain from the meter, +positive to the right seen from outside (scene-and-verdict-format.md §1: "on the meter +wall s = x"), and both order wall baselines left-end-first seen from outside with the +outward normal the baseline direction turned 90° clockwise seen from above. Only the +units differ, so every length converts with exactly 1 ft = 0.3048 m and no s rebasing: +`wire_s_ft = source_s_m / 0.3048`. The s coordinates are nonetheless cross-checked +against the plan geometry (see the refusals below), so a document whose s values do not +match its own baselines is refused, never silently accepted. + +What converts (the initial supported geometry is one straight observed house wall, an +observed meter anchor and level ground, in the declared meter-anchored frame): + +- `meter` — the anchor object (point or wall_rect geometry), with position and error in + feet. A point meter keeps its measured `pos_m`; a wall_rect meter sits on the wall at + its span centre at mid-height, `(bottom_m + top_m) / 2`, which the frame definition + (origin on the ground directly below the meter) and the s=0-centre check below pin to + the same place the source frame does. +- `walls` — the single chain wall: scaled baseline, `height_ft`, `plus_minus_ft`. The + wire `source` field is omitted (absent means tap): the source contract records how the + wall's *points* were supported (`support`), not how the line was found, and guessing a + provenance would be worse than the schema default. `polygon_m`, `s_range_m` + (after validation), `observed_fraction` and `support` have no wire field and are + dropped; `observed_fraction` is redundant with the converted coverage bands. +- `objects` — observed wall_rect hazards with the exact class map (below) and the + observed `source.kind` map (`vlm`→`vlm`, `tap`→`tap`, `tape`→`tape`). The class map + follows the wire enum and its own prose ("Disconnects, sub-panels, EV chargers and + solar equipment are elec_box"; the footprint note names gas regulators and pipe runs): + `window/door/garage_door/gas_meter/vent/downspout/pool/battery` map to themselves, + `ac_unit`→`ac`, `gas_regulator/gas_pipe`→`gas_meter`, + `disconnect/panel/wall_box/meter_can`→`elec_box` (the c1 converter's precedent). + `conf` carries the source confidence unchanged (it is recognition confidence in both + contracts and never a distance error); `plus_minus_ft` converts with the lengths; + `attrs` carries `operable` and `well` when present — a missing attribute stays missing, + which both contracts read as unknown. `ground_level`, `movable` and `text` have no + wire field and are dropped; the object's `source.views`/`model` and `label` likewise. +- `ground` — the plane surface must be exactly level at frame height 0; observed regions + with a wire material (`concrete/gravel/lawn/mulch/deck/drive` map to themselves) become + wire ground patches. The surface's `plus_minus_m`, region `source` and region + `confidence` have no wire field and are dropped. +- `overheads` and `facing` — observed entries on the wall, converted with the lengths. + `object_id` cross-references have no wire field and are dropped. +- `coverage` — the chain ends (`limit`/`unexplored` with their notes) and the explicit + bands from `coverage.bands` and every `coverage.sensitivity[].bands` entry: band name, + `span_ft` and `out_ft` convert with the lengths; `views` and `best_incidence_deg` have + no wire field and are dropped. A coverage without bands converts to ends only, which + the wire parser reads as nothing observed — an absent band stays unobserved. + +Refusals — every refusal raises `SceneBridgeError`, a `ValueError` carrying a `code` and +a JSON-pointer `pointer` string that names the offending input: + +- code `placement_scene_unsupported` (valid scene, outside the bounded scope): more than + one wall (`/walls`); a wall whose role is not `house` (`/walls/0/role`) or whose + evidence is not `observed` (`/walls/0/evidence`) or whose base sits above the ground + (`/walls/0/base_y_m`); any object that is not observed evidence (`/objects//evidence` + — generated or inferred evidence is never promoted to observed); an unmapped class + (`/objects//class`); unsupported geometry (`/objects//geometry/kind` — only the + meter may be a point, only hazards on the wall convert); a wall_rect `standoff_m` + (`/objects//geometry/standoff_m` — a stand-off hazard needs a footprint this bridge + does not synthesise); an unmapped source kind (`/objects//source/kind`); a + heightfield or non-level ground surface (`/ground/surface`); a sloped region + (`/ground/regions//slope_deg`); an unmapped material (`/ground/regions//material`); + non-observed ground regions (`/ground/regions//evidence`); non-observed overheads or + facing (`/overheads//evidence`, `/facing//evidence`); a facing `lower_bound` + (`/facing//lower_bound` — the wire cannot express it); grid-based coverage + (`/coverage/grid` — only explicit bands convert); keyframes (`/keyframes` — the wire + requires per-frame image files the backend contract does not carry) and generated + regions (`/generated_regions`). +- code `placement_scene_invalid` (the source contradicts itself or the schema): anything + `Scene` itself rejects, pointer from the pydantic error location; the cross-reference + failures of `scene_consistency_errors`; a chain that is not exactly the one wall + (`/chain/wall_ids`) or has corners (`/chain/corners`); a meter that projects past its + wall (`/frame/anchor`) or a wall_rect meter whose span is not centred on s=0 + (`/objects//geometry/s_m`); s coordinates that disagree with the plan geometry — a + wall `s_range_m` that is not its baseline span relative to the meter + (`/walls/0/s_range_m`), a chain end not at the wall's end (`/chain/ends//s_m`), a + hazard span running past its wall (`/objects//geometry/s_m`), a reversed span + (`.../s_m`) — compared at `_S_NOISE_M`, an arithmetic float-noise bound, not a + measurement tolerance; non-finite numbers, pointer at the offending field; negative or + inverted wall_rect heights. +""" + +from __future__ import annotations + +import math +from typing import Any, NoReturn + +from pydantic import ValidationError + +from server.contracts.scene import Scene, scene_consistency_errors + +UNSUPPORTED = "placement_scene_unsupported" +INVALID = "placement_scene_invalid" + +# 1 ft = 0.3048 m exactly, so lengths convert with one exact factor in one direction: +# dividing by 0.3048, the correctly rounded conversion, never multiplying by a rounded +# reciprocal. + +# Arithmetic float-noise bound for cross-checking declared s coordinates against plan +# geometry. Several orders below the smallest measurement error either contract +# represents (the tape default is 0.05 ft); it exists so `4.0 + 1e-16` style noise in a +# producer's own arithmetic does not refuse an otherwise consistent scene. It is not a +# measurement tolerance: nothing is rounded, absorbed or guessed at this scale. +_S_NOISE_M = 1e-6 + +_CLASS: dict[str, str] = { + "window": "window", + "door": "door", + "garage_door": "garage_door", + "ac_unit": "ac", + "gas_meter": "gas_meter", + "gas_regulator": "gas_meter", + "gas_pipe": "gas_meter", + "vent": "vent", + "downspout": "downspout", + "pool": "pool", + "battery": "battery", + "disconnect": "elec_box", + "panel": "elec_box", + "wall_box": "elec_box", + "meter_can": "elec_box", +} + +_SOURCE_KIND: dict[str, str] = {"vlm": "vlm", "tap": "tap", "tape": "tape"} + +_MATERIAL: dict[str, str] = { + "concrete": "concrete", + "gravel": "gravel", + "lawn": "lawn", + "mulch": "mulch", + "deck": "deck", + "drive": "drive", +} + + +class SceneBridgeError(ValueError): + """The scene cannot be bridged honestly. `code` is `placement_scene_unsupported` + (valid scene, outside the bounded scope) or `placement_scene_invalid` (the source + contradicts itself); `pointer` names the offending field of the input document.""" + + def __init__(self, message: str, *, code: str, pointer: str) -> None: + super().__init__(f"{pointer}: {message}") + self.code = code + self.pointer = pointer + + +def _unsupported(message: str, pointer: str) -> NoReturn: + raise SceneBridgeError(message, code=UNSUPPORTED, pointer=pointer) + + +def _invalid(message: str, pointer: str) -> NoReturn: + raise SceneBridgeError(message, code=INVALID, pointer=pointer) + + +def _ft(value: Any) -> float: + return float(value) / 0.3048 + + +def _refuse_nonfinite(document: dict) -> None: + """No number in a scene may be NaN or infinite. JSON Schema would accept both.""" + stack: list[tuple[Any, str]] = [(document, "")] + while stack: + value, path = stack.pop() + if isinstance(value, dict): + stack.extend((item, f"{path}/{key}") for key, item in value.items()) + elif isinstance(value, list): + stack.extend((item, f"{path}/{index}") for index, item in enumerate(value)) + elif isinstance(value, int | float) and not isinstance(value, bool): + if not math.isfinite(value): + _invalid(f"{value!r} is not a finite number", path or "/") + + +def _validated(document: dict) -> Scene: + try: + scene = Scene.model_validate(document) + except ValidationError as exc: + error = exc.errors()[0] + pointer = "/" + "/".join(str(part) for part in error["loc"]) + _invalid(error["msg"], pointer) + errors = scene_consistency_errors(document) + if errors: + _invalid(errors[0], "/") + return scene + + +def convert_scene(document: dict) -> dict: + """`base-scene/v0` in metres to the placement wire scene in feet. Pure: see the + module docstring for the mapping, the drops and every refusal.""" + if not isinstance(document, dict): + _invalid("the scene must be a JSON object", "/") + _refuse_nonfinite(document) + return _convert(_validated(document)) + + +def _convert(scene: Scene) -> dict: + if len(scene.walls) != 1: + _unsupported( + f"{len(scene.walls)} walls; the bridge starts with one straight house wall", + "/walls", + ) + if scene.keyframes: + _unsupported( + "keyframes do not convert (the wire format requires a per-frame image file " + "the backend scene does not carry)", + "/keyframes", + ) + if scene.generated_regions: + _unsupported( + "generated regions are unknown space, not observed geometry; the wire format cannot carry them", + "/generated_regions", + ) + if scene.coverage.grid is not None: + _unsupported( + "grid coverage does not convert; only explicit coverage bands do", + "/coverage/grid", + ) + + wall = scene.walls[0] + _wall_scope(wall) + + chain = scene.chain + if chain.wall_ids != [wall.id]: + _invalid( + f"chain {chain.wall_ids!r} must be exactly the one wall {wall.id!r}", + "/chain/wall_ids", + ) + if chain.corners: + _invalid("a one-wall chain has no corners", "/chain/corners") + + meter_index, meter_pos, u_meter = _meter(scene, wall) + lo_s, hi_s = _wall_s_bounds(wall, u_meter) + + span_checks = _SpanChecks(wall.id, lo_s, hi_s) + + meter = scene.objects[meter_index] + out: dict[str, Any] = { + "meter": { + "pos": [_ft(meter_pos[0]), _ft(meter_pos[1]), _ft(meter_pos[2])], + "wall_id": wall.id, + "plus_minus_ft": _ft(meter.plus_minus_m), + }, + "walls": [ + { + "id": wall.id, + "baseline": [ + [_ft(wall.baseline_m[0][0]), _ft(wall.baseline_m[0][1])], + [_ft(wall.baseline_m[1][0]), _ft(wall.baseline_m[1][1])], + ], + "height_ft": _ft(wall.height_m), + "plus_minus_ft": _ft(wall.plus_minus_m), + } + ], + } + + objects_out = _objects(scene, meter_index, span_checks) + if objects_out: + out["objects"] = objects_out + + ground_out = _ground(scene, wall.id) + if ground_out: + out["ground"] = ground_out + + overheads_out = _measured(scene.overheads or [], wall.id, span_checks, "overheads") + if overheads_out: + out["overheads"] = overheads_out + + facing_out = _measured(scene.facing or [], wall.id, span_checks, "facing") + if facing_out: + out["facing"] = facing_out + + out["coverage"] = _coverage(scene, wall.id, span_checks) + return out + + +def _wall_scope(wall) -> None: + """The single wall must be an observed house wall sitting on the ground.""" + if wall.role != "house": + _unsupported(f"wall role {wall.role!r}; only a house wall backs a battery", "/walls/0/role") + if wall.evidence != "observed": + _unsupported( + f"wall evidence {wall.evidence!r}; only an observed wall converts", + "/walls/0/evidence", + ) + if wall.base_y_m is not None and float(wall.base_y_m) != 0.0: + _unsupported( + "the wall base sits above the ground; the bridge starts with level ground", + "/walls/0/base_y_m", + ) + + +def _wall_s_bounds(wall, u_meter: float) -> tuple[float, float]: + """The wall's ends in chain s (metres from the meter), plus its declared range check.""" + length = _baseline_length(wall) + lo_s, hi_s = -u_meter, length - u_meter + if wall.s_range_m is not None: + declared_lo, declared_hi = float(wall.s_range_m[0]), float(wall.s_range_m[1]) + if abs(declared_lo - lo_s) > _S_NOISE_M or abs(declared_hi - hi_s) > _S_NOISE_M or declared_lo > declared_hi: + _invalid( + f"s_range_m [{declared_lo}, {declared_hi}] disagrees with the baseline " + f"and the meter anchor, which put the wall at s in [{lo_s}, {hi_s}]", + "/walls/0/s_range_m", + ) + return lo_s, hi_s + + +def _baseline_length(wall) -> float: + (x0, z0), (x1, z1) = ( + (float(wall.baseline_m[0][0]), float(wall.baseline_m[0][1])), + (float(wall.baseline_m[1][0]), float(wall.baseline_m[1][1])), + ) + length = math.hypot(x1 - x0, z1 - z0) + if length <= 0.0: + _invalid("the baseline repeats its first point", "/walls/0/baseline") + return length + + +def _meter(scene: Scene, wall) -> tuple[int, tuple[float, float, float], float]: + """The anchor meter's index, its 3D position in metres, and the arc distance of its + projection from the wall's left end (u_meter). s = arc - u_meter on the wall.""" + anchor_id = scene.frame.anchor.element_id + meter_index = next((index for index, obj in enumerate(scene.objects) if obj.id == anchor_id), None) + if meter_index is None: + _invalid(f"the anchor {anchor_id!r} names no object", "/frame/anchor") + meter_path = f"/objects/{meter_index}" + meter = scene.objects[meter_index] + if meter.evidence != "observed": + _unsupported( + f"meter anchor evidence {meter.evidence!r}; only an observed meter converts", + f"{meter_path}/evidence", + ) + + (x0, z0) = (float(wall.baseline_m[0][0]), float(wall.baseline_m[0][1])) + (x1, z1) = (float(wall.baseline_m[1][0]), float(wall.baseline_m[1][1])) + length = _baseline_length(wall) + along_x, along_z = (x1 - x0) / length, (z1 - z0) / length + geometry = meter.geometry + + if geometry.kind == "point": + pos_x, pos_y, pos_z = (float(value) for value in geometry.pos_m) + if pos_y < 0.0: + _invalid("the meter sits below the ground", f"{meter_path}/geometry/pos_m") + plan_x, plan_z = pos_x, pos_z + failure_pointer = f"{meter_path}/geometry/pos_m" + elif geometry.kind == "wall_rect": + _wall_rect_on_wall(geometry, meter_path, wall.id) + if geometry.standoff_m is not None and float(geometry.standoff_m) != 0.0: + _unsupported( + "a stand-off meter anchor needs a footprint the bridge does not synthesise", + f"{meter_path}/geometry/standoff_m", + ) + s0, s1 = float(geometry.s_m[0]), float(geometry.s_m[1]) + if s0 > s1: + _invalid("reversed s_m span", f"{meter_path}/geometry/s_m") + if abs((s0 + s1) / 2) > _S_NOISE_M: + _invalid( + f"the meter anchor's span is centred on s={(s0 + s1) / 2}, not s=0 " + "(the frame origin is on the ground directly below the meter)", + f"{meter_path}/geometry/s_m", + ) + if float(geometry.bottom_m) < 0.0 or float(geometry.top_m) < 0.0: + _invalid("a negative height is below the ground", f"{meter_path}/geometry") + if float(geometry.bottom_m) > float(geometry.top_m): + _invalid("bottom_m is above top_m", f"{meter_path}/geometry") + # The frame origin is defined on the ground directly below the meter, so the + # meter's plan position is the origin's projection onto its wall. + plan_x, plan_z = 0.0, 0.0 + pos_y = (float(geometry.bottom_m) + float(geometry.top_m)) / 2 + failure_pointer = "/frame/anchor" + else: + _unsupported( + f"meter geometry {geometry.kind!r}; only point and wall_rect convert", + f"{meter_path}/geometry/kind", + ) + + arc = (plan_x - x0) * along_x + (plan_z - z0) * along_z + if arc < -_S_NOISE_M or arc > length + _S_NOISE_M: + _invalid( + f"the meter anchor projects {arc:.3f} m from the wall's left end, past the " + f"{length:.3f} m wall; the meter must sit on its wall", + failure_pointer, + ) + u_meter = min(max(arc, 0.0), length) + return meter_index, (plan_x, pos_y, plan_z), u_meter + + +def _wall_rect_on_wall(geometry, object_path: str, wall_id: str) -> None: + if geometry.wall_id != wall_id: + _invalid( + f"the object sits on unknown wall {geometry.wall_id!r}", + f"{object_path}/geometry/wall_id", + ) + + +class _SpanChecks: + """Converts and bounds-checks declared s spans against the wall's geometry.""" + + def __init__(self, wall_id: str, lo_s: float, hi_s: float) -> None: + self.wall_id = wall_id + self.lo_s = lo_s + self.hi_s = hi_s + + def span_ft(self, s_m: list[Any], pointer: str, *, bounded: bool) -> list[float]: + s0, s1 = float(s_m[0]), float(s_m[1]) + if s0 > s1: + _invalid(f"reversed span [{s0}, {s1}]", pointer) + if bounded and (s0 < self.lo_s - _S_NOISE_M or s1 > self.hi_s + _S_NOISE_M): + _invalid( + f"span [{s0}, {s1}] runs past its wall at s in [{self.lo_s}, {self.hi_s}]", + pointer, + ) + return [_ft(s0), _ft(s1)] + + +def _objects(scene: Scene, meter_index: int, checks: _SpanChecks) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + for index, obj in enumerate(scene.objects): + if index == meter_index: + continue + path = f"/objects/{index}" + if obj.evidence != "observed": + _unsupported( + f"evidence {obj.evidence!r}; hazards convert only when observed, never as generated or inferred", + f"{path}/evidence", + ) + wire_class = _CLASS.get(obj.class_) + if wire_class is None: + _unsupported(f"class {obj.class_!r} has no wire type", f"{path}/class") + if obj.wall_id is not None and obj.wall_id != checks.wall_id: + _invalid( + f"the object names wall {obj.wall_id!r}, not {checks.wall_id!r}", + f"{path}/wall_id", + ) + geometry = obj.geometry + if geometry.kind != "wall_rect": + _unsupported( + f"geometry {geometry.kind!r}; the bridge starts with wall-relative rectangular hazards", + f"{path}/geometry/kind", + ) + _wall_rect_on_wall(geometry, path, checks.wall_id) + if geometry.standoff_m is not None and float(geometry.standoff_m) != 0.0: + _unsupported( + "a stand-off hazard needs a wire footprint the bridge does not synthesise", + f"{path}/geometry/standoff_m", + ) + if float(geometry.bottom_m) < 0.0 or float(geometry.top_m) < 0.0: + _invalid("a negative height is below the ground", f"{path}/geometry") + if float(geometry.bottom_m) > float(geometry.top_m): + _invalid("bottom_m is above top_m", f"{path}/geometry") + wire_source = _SOURCE_KIND.get(obj.source.kind) + if wire_source is None: + _unsupported( + f"source kind {obj.source.kind!r}; only vlm, tap and tape map", + f"{path}/source/kind", + ) + entry: dict[str, Any] = { + "type": wire_class, + "wall_id": checks.wall_id, + "span_ft": checks.span_ft(geometry.s_m, f"{path}/geometry/s_m", bounded=True), + "bottom_ft": _ft(geometry.bottom_m), + "top_ft": _ft(geometry.top_m), + "source": wire_source, + "conf": float(obj.confidence), + "plus_minus_ft": _ft(obj.plus_minus_m), + } + attrs: dict[str, Any] = {} + if obj.attrs is not None: + if obj.attrs.operable is not None: + attrs["operable"] = obj.attrs.operable + if obj.attrs.well is not None: + attrs["well"] = obj.attrs.well + if attrs: + entry["attrs"] = attrs + out.append(entry) + return out + + +def _ground(scene: Scene, wall_id: str) -> list[dict[str, Any]]: + surface = scene.ground.surface + if surface.kind != "plane": + _unsupported( + f"ground surface {surface.kind!r}; the bridge starts with a plane ground", + "/ground/surface", + ) + normal = [float(value) for value in surface.normal] + if normal != [0.0, 1.0, 0.0]: + _unsupported( + f"ground normal {normal!r}; the bridge starts with level ground", + "/ground/surface", + ) + if float(surface.offset_m) != 0.0: + _unsupported( + f"ground at height {float(surface.offset_m)}; the bridge starts with the ground at frame height 0", + "/ground/surface", + ) + out: list[dict[str, Any]] = [] + for index, region in enumerate(scene.ground.regions or []): + path = f"/ground/regions/{index}" + if region.evidence != "observed": + _unsupported( + f"region evidence {region.evidence!r}; ground converts only when observed", + f"{path}/evidence", + ) + if region.slope_deg is not None and float(region.slope_deg) != 0.0: + _unsupported( + f"sloped region ({float(region.slope_deg)} deg); the bridge starts with level ground", + f"{path}/slope_deg", + ) + wire_type = _MATERIAL.get(region.material) + if wire_type is None: + _unsupported(f"material {region.material!r} has no wire type", f"{path}/material") + entry: dict[str, Any] = { + "type": wire_type, + "polygon": [[_ft(point[0]), _ft(point[1])] for point in region.polygon_m], + } + if region.plus_minus_m is not None: + entry["plus_minus_ft"] = _ft(region.plus_minus_m) + out.append(entry) + return out + + +def _measured(entries, wall_id: str, checks: _SpanChecks, key: str) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + for index, entry in enumerate(entries): + path = f"/{key}/{index}" + if entry.evidence != "observed": + _unsupported( + f"evidence {entry.evidence!r}; the measurement converts only when observed", + f"{path}/evidence", + ) + if entry.wall_id != wall_id: + _invalid(f"the entry names unknown wall {entry.wall_id!r}", f"{path}/wall_id") + record: dict[str, Any] = { + "wall_id": wall_id, + "span_ft": checks.span_ft(entry.s_m, f"{path}/s_m", bounded=True), + } + if key == "overheads": + record["clearance_ft"] = _ft(entry.clearance_m) + else: + if entry.lower_bound: + _unsupported( + "a lower-bound facing depth cannot be written as a wire depth_ft", + f"{path}/lower_bound", + ) + record["depth_ft"] = _ft(entry.depth_m) + record["plus_minus_ft"] = _ft(entry.plus_minus_m) + out.append(record) + return out + + +def _coverage(scene: Scene, wall_id: str, checks: _SpanChecks) -> dict[str, Any]: + def end_out(end, side: str) -> dict[str, Any]: + record: dict[str, Any] = {"kind": end.kind} + if end.s_m is not None: + expected = checks.lo_s if side == "left" else checks.hi_s + if abs(float(end.s_m) - expected) > _S_NOISE_M: + _invalid( + f"{side} chain end s_m {float(end.s_m)} disagrees with the wall's {side} end at s {expected}", + f"/chain/ends/{side}/s_m", + ) + if end.note is not None: + record["note"] = end.note + return record + + out: dict[str, Any] = { + "ends": { + "left": end_out(scene.chain.ends.left, "left"), + "right": end_out(scene.chain.ends.right, "right"), + } + } + + observed: list[dict[str, Any]] = [] + + def band_out(band, pointer: str) -> None: + if band.wall_id is not None and band.wall_id != wall_id: + _invalid(f"the band names unknown wall {band.wall_id!r}", f"{pointer}/wall_id") + record: dict[str, Any] = { + "band": band.band, + "span_ft": checks.span_ft(band.s_m, f"{pointer}/s_m", bounded=False), + } + if band.out_m is not None: + record["out_ft"] = _ft(band.out_m) + observed.append(record) + + for index, band in enumerate(scene.coverage.bands or []): + band_out(band, f"/coverage/bands/{index}") + for sensitivity_index, sensitivity in enumerate(scene.coverage.sensitivity or []): + for band_index, band in enumerate(sensitivity.bands): + band_out(band, f"/coverage/sensitivity/{sensitivity_index}/bands/{band_index}") + + if observed: + out["observed"] = observed + return out diff --git a/backend/server/placement/tests/test_scene_bridge.py b/backend/server/placement/tests/test_scene_bridge.py new file mode 100644 index 000000000..22320b9d3 --- /dev/null +++ b/backend/server/placement/tests/test_scene_bridge.py @@ -0,0 +1,668 @@ +"""Tests for server.placement.scene_bridge. + +Expected values are computed here from first principles — metres divided by 0.3048 +written out in this file, plan geometry with math.hypot — never from the bridge's own +constants, and the converted output is validated by the real root scene parser +(`server/scene.py`, the code behind POST /v1/placements) and `scene.schema.json`. +""" + +from __future__ import annotations + +import copy +import json +import math +from pathlib import Path +from typing import Any + +import pytest +from jsonschema import Draft202012Validator + +from server.placement.adapter import load_solver +from server.placement.scene_bridge import ( + INVALID, + UNSUPPORTED, + SceneBridgeError, + convert_scene, +) + +_REPO_ROOT = Path(__file__).resolve().parents[4] +_WIRE_SCHEMA = Draft202012Validator(json.loads((_REPO_ROOT / "server" / "schemas" / "scene.schema.json").read_text())) + + +def base_scene() -> dict[str, Any]: + """One straight observed house wall from s = -3 m to s = 2 m (the meter projects at + arc 3.0 m, a nonzero wall s origin), an observed meter anchor, level ground, mapped + hazards, ground region, coverage band, overhead and facing.""" + return { + "format": "base-scene/v0", + "scene_id": "bridge-test", + "provenance": { + "data": "synthetic", + "site": "bridge-test", + "producer": {"path": "scene-bridge-test"}, + "scale_source": "native_metric", + "gravity_source": "ground_truth", + }, + "frame": { + "units": "m", + "convention": "meter_anchored/v0", + "anchor": {"element_id": "meter", "plus_minus_m": 0.01}, + }, + "walls": [ + { + "id": "w", + "role": "house", + "baseline_m": [[-3.0, 0.0], [5.0, 0.0]], + "height_m": 3.0, + "plus_minus_m": 0.05, + "evidence": "observed", + } + ], + "chain": { + "wall_ids": ["w"], + "ends": { + "left": {"kind": "unexplored"}, + "right": {"kind": "limit", "note": "fence line"}, + }, + }, + "ground": { + "surface": {"kind": "plane", "normal": [0, 1, 0], "offset_m": 0}, + "plus_minus_m": 0.05, + "regions": [ + { + "id": "pad", + "material": "concrete", + "polygon_m": [[0.2, 0.5], [2.0, 0.5], [2.0, 3.0], [0.2, 3.0]], + "evidence": "observed", + "plus_minus_m": 0.1, + } + ], + }, + "objects": [ + { + "id": "meter", + "class": "meter", + "geometry": {"kind": "point", "pos_m": [0.0, 1.5, 0.02]}, + "confidence": 1, + "plus_minus_m": 0.02, + "source": {"kind": "reconstruction"}, + "evidence": "observed", + }, + { + "id": "win", + "class": "window", + "geometry": { + "kind": "wall_rect", + "wall_id": "w", + "s_m": [1.0, 2.0], + "bottom_m": 1.0, + "top_m": 2.2, + }, + "confidence": 0.9, + "plus_minus_m": 0.08, + "source": {"kind": "vlm"}, + "evidence": "observed", + }, + { + "id": "gas", + "class": "gas_meter", + "geometry": { + "kind": "wall_rect", + "wall_id": "w", + "s_m": [-0.6, -0.3], + "bottom_m": 0.3, + "top_m": 0.9, + }, + "confidence": 0.95, + "plus_minus_m": 0.05, + "source": {"kind": "tap"}, + "evidence": "observed", + "attrs": {"operable": True}, + }, + ], + "coverage": { + "method": "scene-bridge-test", + "sensitivity": [ + { + "surface": "wall", + "max_incidence_deg": 75, + "bands": [{"band": "wall", "s_m": [-2.0, 4.0], "out_m": 2.13}], + } + ], + }, + "overheads": [ + { + "wall_id": "w", + "s_m": [1.0, 2.0], + "clearance_m": 2.1, + "plus_minus_m": 0.1, + "evidence": "observed", + } + ], + "facing": [ + { + "wall_id": "w", + "s_m": [-1.0, 1.0], + "depth_m": 8.0, + "plus_minus_m": 0.3, + "evidence": "observed", + } + ], + } + + +def by_id(document: dict[str, Any], object_id: str) -> dict[str, Any]: + return next(obj for obj in document["objects"] if obj["id"] == object_id) + + +def _parse_with_root_parser(converted: dict[str, Any]) -> Any: + """The real root scene parser (schema + geometry) behind POST /v1/placements.""" + modules = load_solver() + loaded = modules["rules"].load_rules() + return modules["scene"].parse_scene(converted, loaded.rules) + + +def test_happy_path_fields_in_independently_computed_feet() -> None: + converted = convert_scene(base_scene()) + + assert converted["meter"] == { + "pos": [0.0 / 0.3048, 1.5 / 0.3048, 0.02 / 0.3048], + "wall_id": "w", + "plus_minus_ft": 0.02 / 0.3048, + } + wall = converted["walls"][0] + assert wall["id"] == "w" + assert wall["baseline"] == [[-3.0 / 0.3048, 0.0], [5.0 / 0.3048, 0.0]] + assert wall["height_ft"] == 3.0 / 0.3048 + assert wall["plus_minus_ft"] == 0.05 / 0.3048 + assert "source" not in wall + + window, gas = converted["objects"] + assert window["type"] == "window" + assert window["span_ft"] == [1.0 / 0.3048, 2.0 / 0.3048] + assert window["bottom_ft"] == 1.0 / 0.3048 + assert window["top_ft"] == 2.2 / 0.3048 + assert window["source"] == "vlm" + assert window["conf"] == 0.9 + assert window["plus_minus_ft"] == 0.08 / 0.3048 + assert "attrs" not in window # no attrs stays unknown, never false + + assert gas["type"] == "gas_meter" + assert gas["span_ft"] == [-0.6 / 0.3048, -0.3 / 0.3048] + assert gas["source"] == "tap" + assert gas["attrs"] == {"operable": True} + + assert converted["ground"] == [ + { + "type": "concrete", + "polygon": [[x / 0.3048, z / 0.3048] for x, z in [(0.2, 0.5), (2.0, 0.5), (2.0, 3.0), (0.2, 3.0)]], + "plus_minus_ft": 0.1 / 0.3048, + } + ] + assert converted["overheads"] == [ + { + "wall_id": "w", + "span_ft": [1.0 / 0.3048, 2.0 / 0.3048], + "clearance_ft": 2.1 / 0.3048, + "plus_minus_ft": 0.1 / 0.3048, + } + ] + assert converted["facing"] == [ + { + "wall_id": "w", + "span_ft": [-1.0 / 0.3048, 1.0 / 0.3048], + "depth_ft": 8.0 / 0.3048, + "plus_minus_ft": 0.3 / 0.3048, + } + ] + assert converted["coverage"]["ends"] == { + "left": {"kind": "unexplored"}, + "right": {"kind": "limit", "note": "fence line"}, + } + assert converted["coverage"]["observed"] == [ + {"band": "wall", "span_ft": [-2.0 / 0.3048, 4.0 / 0.3048], "out_ft": 2.13 / 0.3048} + ] + + +@pytest.mark.parametrize( + "mutate", + [ + pytest.param(lambda doc: None, id="point-meter"), + pytest.param( + lambda doc: by_id(doc, "meter").update( + { + "geometry": { + "kind": "wall_rect", + "wall_id": "w", + "s_m": [-0.1, 0.1], + "bottom_m": 1.4, + "top_m": 1.8, + } + } + ), + id="wall-rect-meter", + ), + pytest.param( + lambda doc: doc["walls"][0].update({"baseline_m": [[3.0, 1.0], [9.0, 5.0]]}), + id="rotated-wall", + ), + ], +) +def test_output_is_valid_for_the_real_root_scene_parser_and_schema(mutate) -> None: + document = base_scene() + mutate(document) + if document["walls"][0]["baseline_m"] == [[3.0, 1.0], [9.0, 5.0]]: + # The meter stays on the rotated wall line (3, 1) + 2.5 m along it. + length = math.hypot(6.0, 4.0) + by_id(document, "meter")["geometry"] = { + "kind": "point", + "pos_m": [ + 3.0 + (6.0 / length) * 2.5, + 1.5, + 1.0 + (4.0 / length) * 2.5, + ], + } + + converted = convert_scene(document) + + _WIRE_SCHEMA.validate(converted) + parsed = _parse_with_root_parser(converted) + assert json.dumps(converted, sort_keys=True, allow_nan=False) # JSON-compatible + + # The parser's unrolled s=0 is the meter's projection onto the wall, in feet: the + # same plan x as the meter, and within the meter's plan offset of the meter itself + # (0.02 m off-wall for the point meter, on-wall otherwise). + meter_plan_ft = (converted["meter"]["pos"][0], converted["meter"]["pos"][2]) + s0_point = parsed.point_at(0.0) + assert s0_point[0] == pytest.approx(meter_plan_ft[0], abs=1e-6) + assert math.hypot(s0_point[0] - meter_plan_ft[0], s0_point[1] - meter_plan_ft[1]) == pytest.approx( + 0.0, abs=0.02 / 0.3048 + 1e-6 + ) + + +def test_rotated_translated_wall_and_nonzero_s_origin_place_spans_independently() -> None: + """The wall is translated off the origin and rotated; the hazard span is checked + against plan geometry computed here, through the root parser's own unrolling.""" + document = base_scene() + document["walls"][0]["baseline_m"] = [[3.0, 1.0], [9.0, 5.0]] + length = math.hypot(9.0 - 3.0, 5.0 - 1.0) + along = ((9.0 - 3.0) / length, (5.0 - 1.0) / length) + t_meter = 2.5 + meter = by_id(document, "meter") + meter["geometry"] = { + "kind": "point", + "pos_m": [3.0 + along[0] * t_meter, 1.5, 1.0 + along[1] * t_meter], + } + + converted = convert_scene(document) + + # Feet computed independently: the scaled baseline and the span from source s. + assert converted["walls"][0]["baseline"] == [ + [3.0 / 0.3048, 1.0 / 0.3048], + [9.0 / 0.3048, 5.0 / 0.3048], + ] + window = converted["objects"][0] + assert window["span_ft"] == [1.0 / 0.3048, 2.0 / 0.3048] + + # Independent placement: the hazard at source s=1.0 m sits at arc t_meter + 1.0 m + # along the wall, i.e. plan (3, 1) + 3.5 m * along — in feet. The root parser must + # find exactly that point at the converted span start, re-deriving s from geometry. + parsed = _parse_with_root_parser(converted) + expected = ( + (3.0 + along[0] * (t_meter + 1.0)) / 0.3048, + (1.0 + along[1] * (t_meter + 1.0)) / 0.3048, + ) + point = parsed.point_at(window["span_ft"][0]) + assert point[0] == pytest.approx(expected[0], rel=1e-9) + assert point[1] == pytest.approx(expected[1], rel=1e-9) + + +def test_meter_wall_rect_geometry_places_the_meter_at_its_span_centre_mid_height() -> None: + document = base_scene() + by_id(document, "meter")["geometry"] = { + "kind": "wall_rect", + "wall_id": "w", + "s_m": [-0.1, 0.1], + "bottom_m": 1.4, + "top_m": 1.8, + } + + converted = convert_scene(document) + + # Plan: the frame origin (directly below the meter) projected on the wall; + # height: mid-span of the meter's own wall_rect. + assert converted["meter"]["pos"] == [0.0 / 0.3048, (1.4 + 1.8) / 2 / 0.3048, 0.0] + parsed = _parse_with_root_parser(converted) + assert parsed.point_at(0.0)[0] == pytest.approx(0.0, abs=1e-6) + + +def test_absent_coverage_band_stays_unobserved_and_unexplored_ends_are_kept() -> None: + document = base_scene() + document["coverage"] = {"method": "scene-bridge-test"} + + converted = convert_scene(document) + + assert converted["coverage"] == { + "ends": { + "left": {"kind": "unexplored"}, + "right": {"kind": "limit", "note": "fence line"}, + } + } + assert "observed" not in converted["coverage"] + + +def test_unknown_attributes_stay_unknown() -> None: + document = base_scene() + by_id(document, "win")["attrs"] = { + "operable": None, + "well": None, + "ground_level": True, + "movable": False, + "text": "sticker", + } + + converted = convert_scene(document) + + assert "attrs" not in converted["objects"][0] + + +def test_class_and_source_map_exactly() -> None: + mapping = { + "ac_unit": "ac", + "gas_regulator": "gas_meter", + "gas_pipe": "gas_meter", + "disconnect": "elec_box", + "panel": "elec_box", + "wall_box": "elec_box", + "meter_can": "elec_box", + "vent": "vent", + "downspout": "downspout", + "pool": "pool", + "battery": "battery", + "door": "door", + "garage_door": "garage_door", + } + for source_class, wire_type in mapping.items(): + document = base_scene() + by_id(document, "gas")["class"] = source_class + converted = convert_scene(document) + assert converted["objects"][-1]["type"] == wire_type + for kind in ("vlm", "tap", "tape"): + document["objects"][-1]["source"] = {"kind": kind} + assert convert_scene(document)["objects"][-1]["source"] == kind + + +def _expect_bridge_error(document: Any, *, code: str, pointer: str) -> SceneBridgeError: + with pytest.raises(SceneBridgeError) as exc_info: + convert_scene(document) + error = exc_info.value + assert isinstance(error, ValueError) + assert error.code == code + assert error.pointer == pointer + return error + + +def test_unsupported_refusals_carry_code_and_pointer() -> None: + document = base_scene() + + two_walls = copy.deepcopy(document) + two_walls["walls"].append( + { + "id": "w2", + "role": "house", + "baseline_m": [[10.0, 0.0], [12.0, 0.0]], + "height_m": 3.0, + "plus_minus_m": 0.05, + "evidence": "observed", + } + ) + two_walls["chain"]["wall_ids"] = ["w", "w2"] + _expect_bridge_error(two_walls, code=UNSUPPORTED, pointer="/walls") + + cases: list[tuple[Any, str, str]] = [] + for evidence in ("inferred", "generated"): + inferred = copy.deepcopy(document) + by_id(inferred, "win")["evidence"] = evidence + cases.append((inferred, UNSUPPORTED, "/objects/1/evidence")) + inferred_meter = copy.deepcopy(document) + by_id(inferred_meter, "meter")["evidence"] = "inferred" + cases.append((inferred_meter, UNSUPPORTED, "/objects/0/evidence")) + + fence = copy.deepcopy(document) + fence["walls"][0]["role"] = "fence" + cases.append((fence, UNSUPPORTED, "/walls/0/role")) + unobserved_wall = copy.deepcopy(document) + unobserved_wall["walls"][0]["evidence"] = "inferred" + cases.append((unobserved_wall, UNSUPPORTED, "/walls/0/evidence")) + raised_wall = copy.deepcopy(document) + raised_wall["walls"][0]["base_y_m"] = 0.4 + cases.append((raised_wall, UNSUPPORTED, "/walls/0/base_y_m")) + + for unsupported_class in ("tree", "eave", "porch_roof", "stairs", "hose_bib", "other"): + unmapped = copy.deepcopy(document) + by_id(unmapped, "win")["class"] = unsupported_class + cases.append((unmapped, UNSUPPORTED, "/objects/1/class")) + + standoff = copy.deepcopy(document) + by_id(standoff, "win")["geometry"]["standoff_m"] = 0.3 + cases.append((standoff, UNSUPPORTED, "/objects/1/geometry/standoff_m")) + standoff_meter = copy.deepcopy(document) + by_id(standoff_meter, "meter")["geometry"] = { + "kind": "wall_rect", + "wall_id": "w", + "s_m": [-0.1, 0.1], + "bottom_m": 1.4, + "top_m": 1.8, + "standoff_m": 0.05, + } + cases.append((standoff_meter, UNSUPPORTED, "/objects/0/geometry/standoff_m")) + + for source_kind in ("detector", "annotation", "laser_scan", "lidar", "reconstruction", "generated"): + unmapped_source = copy.deepcopy(document) + by_id(unmapped_source, "win")["source"] = {"kind": source_kind} + cases.append((unmapped_source, UNSUPPORTED, "/objects/1/source/kind")) + + heightfield = copy.deepcopy(document) + heightfield["ground"]["surface"] = {"kind": "heightfield", "asset": "h"} + cases.append((heightfield, UNSUPPORTED, "/ground/surface")) + tilted = copy.deepcopy(document) + tilted["ground"]["surface"] = {"kind": "plane", "normal": [0.05, 0.998, 0], "offset_m": 0} + cases.append((tilted, UNSUPPORTED, "/ground/surface")) + lifted = copy.deepcopy(document) + lifted["ground"]["surface"] = {"kind": "plane", "normal": [0, 1, 0], "offset_m": 0.3} + cases.append((lifted, UNSUPPORTED, "/ground/surface")) + sloped = copy.deepcopy(document) + sloped["ground"]["regions"][0]["slope_deg"] = 4.0 + cases.append((sloped, UNSUPPORTED, "/ground/regions/0/slope_deg")) + for material in ("dirt", "pavers", "walkway", "pool", "water", "unknown"): + unmapped_material = copy.deepcopy(document) + unmapped_material["ground"]["regions"][0]["material"] = material + cases.append((unmapped_material, UNSUPPORTED, "/ground/regions/0/material")) + generated_region = copy.deepcopy(document) + generated_region["ground"]["regions"][0]["evidence"] = "generated" + cases.append((generated_region, UNSUPPORTED, "/ground/regions/0/evidence")) + inferred_region = copy.deepcopy(document) + inferred_region["ground"]["regions"][0]["evidence"] = "inferred" + cases.append((inferred_region, UNSUPPORTED, "/ground/regions/0/evidence")) + + inferred_overhead = copy.deepcopy(document) + inferred_overhead["overheads"][0]["evidence"] = "inferred" + cases.append((inferred_overhead, UNSUPPORTED, "/overheads/0/evidence")) + lower_bound_facing = copy.deepcopy(document) + lower_bound_facing["facing"][0]["lower_bound"] = True + cases.append((lower_bound_facing, UNSUPPORTED, "/facing/0/lower_bound")) + + grid = copy.deepcopy(document) + grid["coverage"]["grid"] = { + "asset": "a", + "voxel_m": 0.1, + "origin_m": [0, 0, 0], + "shape": [2, 2, 2], + } + grid["assets"] = [{"id": "a", "kind": "coverage_grid", "path": "coverage.npz"}] + cases.append((grid, UNSUPPORTED, "/coverage/grid")) + + keyframes = copy.deepcopy(document) + keyframes["keyframes"] = [ + { + "id": "k1", + "pose": [1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1], + "intrinsics": [1000.0, 1000.0, 640.0, 480.0], + "w": 1280, + "h": 960, + } + ] + cases.append((keyframes, UNSUPPORTED, "/keyframes")) + + generated_regions = copy.deepcopy(document) + generated_regions["generated_regions"] = [ + { + "id": "gen", + "geometry": {"kind": "asset_mask", "asset": "a"}, + "method": "inpaint", + } + ] + cases.append((generated_regions, UNSUPPORTED, "/generated_regions")) + + for bad_document, code, pointer in cases: + error = _expect_bridge_error(bad_document, code=code, pointer=pointer) + assert str(error).startswith(pointer) + + +def test_invalid_refusals_carry_code_and_pointer() -> None: + document = base_scene() + + cases: list[tuple[Any, str]] = [] + + wrong_chain = copy.deepcopy(document) + wrong_chain["chain"]["wall_ids"] = ["w2"] + # The source contract's consistency check refuses the unknown wall reference before + # the bridge's own chain check runs. + cases.append((wrong_chain, "/")) + corners = copy.deepcopy(document) + corners["chain"]["corners"] = [ + { + "id": "c1", + "left_wall": "w", + "right_wall": "w", + "point_m": [5.0, 0.0], + } + ] + cases.append((corners, "/chain/corners")) + + # A geometry that is not a wall_rect is refused by the source contract itself; the + # pydantic discriminator emits the union tag in the error location. + for kind, field in (("box", "pos_m"), ("footprint", "bottom_m"), ("point", "center_m")): + wrong_geom = copy.deepcopy(document) + by_id(wrong_geom, "win")["geometry"] = { + "kind": kind, + "pos_m": [1.0, 1.5, 0.0], + "center_m": [1.0, 1.5, 0.0], + "size_m": [1.0, 1.0, 0.2], + "polygon_m": [[0.0, 0.0], [1.0, 0.0], [1.0, 1.0]], + } + cases.append((wrong_geom, f"/objects/1/geometry/{kind}/{field}")) + + off_wall = copy.deepcopy(document) + # The meter anchors at arc 3.0 m on an 8 m wall, so the wall ends at s = 5.0 m. + by_id(off_wall, "win")["geometry"]["s_m"] = [4.5, 5.5] + cases.append((off_wall, "/objects/1/geometry/s_m")) + off_wall_left = copy.deepcopy(document) + by_id(off_wall_left, "gas")["geometry"]["s_m"] = [-3.5, -3.2] # wall starts at -3.0 + cases.append((off_wall_left, "/objects/2/geometry/s_m")) + + s_range = copy.deepcopy(document) + s_range["walls"][0]["s_range_m"] = [-2.0, 5.0] # geometry says [-3.0, 2.0] + cases.append((s_range, "/walls/0/s_range_m")) + + end_s = copy.deepcopy(document) + end_s["chain"]["ends"]["right"]["s_m"] = 2.5 + cases.append((end_s, "/chain/ends/right/s_m")) + + off_centre_meter = copy.deepcopy(document) + by_id(off_centre_meter, "meter")["geometry"] = { + "kind": "wall_rect", + "wall_id": "w", + "s_m": [0.2, 0.5], + "bottom_m": 1.4, + "top_m": 1.8, + } + cases.append((off_centre_meter, "/objects/0/geometry/s_m")) + + past_end_meter = copy.deepcopy(document) + by_id(past_end_meter, "meter")["geometry"] = { + "kind": "point", + "pos_m": [6.0, 1.5, 0.0], # past the wall's right end at x = 5.0 + } + cases.append((past_end_meter, "/objects/0/geometry/pos_m")) + + reversed_band = copy.deepcopy(document) + reversed_band["coverage"]["sensitivity"][0]["bands"][0]["s_m"] = [4.0, -2.0] + cases.append((reversed_band, "/coverage/sensitivity/0/bands/0/s_m")) + reversed_hazard = copy.deepcopy(document) + by_id(reversed_hazard, "win")["geometry"]["s_m"] = [2.0, 1.0] + cases.append((reversed_hazard, "/objects/1/geometry/s_m")) + + inverted_heights = copy.deepcopy(document) + by_id(inverted_heights, "win")["geometry"]["bottom_m"] = 2.4 + cases.append((inverted_heights, "/objects/1/geometry")) + negative_height = copy.deepcopy(document) + by_id(negative_height, "win")["geometry"]["bottom_m"] = -0.2 + cases.append((negative_height, "/objects/1/geometry")) + + for value in (float("nan"), float("inf"), float("-inf")): + nonfinite = copy.deepcopy(document) + by_id(nonfinite, "win")["geometry"]["s_m"] = [value, 2.0] + cases.append((nonfinite, "/objects/1/geometry/s_m/0")) + nonfinite_meter = copy.deepcopy(document) + by_id(nonfinite_meter, "meter")["geometry"]["pos_m"] = [0.0, value, 0.02] + cases.append((nonfinite_meter, "/objects/0/geometry/pos_m/1")) + + missing_field = copy.deepcopy(document) + del by_id(missing_field, "win")["evidence"] + cases.append((missing_field, "/objects/1/evidence")) + extra_field = copy.deepcopy(document) + extra_field["walls"][0]["baseline_ft"] = [[-10.0, 0.0], [16.0, 0.0]] + cases.append((extra_field, "/walls/0/baseline_ft")) + not_a_meter_anchor = copy.deepcopy(document) + not_a_meter_anchor["frame"]["anchor"]["element_id"] = "win" + cases.append((not_a_meter_anchor, "/")) + + for bad_document, pointer in cases: + _expect_bridge_error(bad_document, code=INVALID, pointer=pointer) + + +def test_non_object_input_is_invalid() -> None: + for bad in ([], "scene", None, 3): + with pytest.raises(SceneBridgeError) as exc_info: + convert_scene(bad) # type: ignore[arg-type] + assert exc_info.value.code == INVALID + assert exc_info.value.pointer == "/" + + +def test_conversion_is_deterministic_and_never_mutates_its_argument() -> None: + document = base_scene() + frozen = copy.deepcopy(document) + + first = convert_scene(document) + second = convert_scene(document) + + assert document == frozen # no mutation, at any depth + assert first == second + assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True) + assert convert_scene(copy.deepcopy(document)) == first + + +def test_the_committed_orchestrator_stub_scene_converts_and_parses() -> None: + from server.orchestrator.stubs._common import scene_document + + converted = convert_scene(scene_document("stub-bridge")) + + _WIRE_SCHEMA.validate(converted) + parsed = _parse_with_root_parser(converted) + assert converted["meter"]["pos"][1] == pytest.approx((1.4 + 1.8) / 2 / 0.3048) + assert parsed.point_at(0.0)[0] == pytest.approx(0.0, abs=1e-6) + assert converted["coverage"]["ends"] == { + "left": {"kind": "limit"}, + "right": {"kind": "limit"}, + }