Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 68 additions & 17 deletions recon/recon/coverage.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,11 @@
backing, nor is a view into a pit ground; and
- no occupied voxel of the fused reconstruction lies between the camera and the sample, short of
that tolerance.
Cells are half-foot steps clipped to the fitted wall extent, and every sample sits inside it: a
surface beyond the fit's ends can never drive an observation, because the fitted wall line
establishes nothing there (a longer baseline an older scan left behind does not either). On
flat ground a clipped fit yields exactly the cells the old first-to-last grid did.

A cell is observed when every sample row was seen from two camera positions at least 0.25 m
apart, the app's bar for "covered". Wall samples sit on the wall's reconstructed face where that
face is a flat, full-height surface in front of the fitted plane (a pilaster), and on the plane
Expand All @@ -31,7 +36,7 @@
from recon.capture import FEET, UP, Frame
from recon.depth import Depth
from recon.fusion import MAX_RANGE_M, Mesh, Volume
from recon.geometry import WallFrame
from recon.geometry import Ground, WallFrame

CELL_M = 0.1524 # half a foot, the app's cell
WALL_TOP_M = 1.9812 # 6.5 ft, the headroom the export's wall band claims
Expand All @@ -47,6 +52,24 @@
HIDE_ABS_M, HIDE_REL = 0.10, 0.04


def cell_grid(lo: float, hi: float) -> tuple[np.ndarray, np.ndarray]:
"""Coverage cells as (starts, ends) in meters of s: half-foot steps clipped to the fitted
extent [lo, hi]. Interior edges stay on the absolute half-foot grid, so a fit that begins and
ends on the grid gets exactly the cells it always did; an unaligned end clips its boundary
cell instead of letting a sample, and so an observation, fall outside the fit."""
if hi <= lo:
return np.zeros(0), np.zeros(0)
inner = np.arange(np.ceil(lo / CELL_M - 1e-9) * CELL_M, hi, CELL_M)
edges = np.r_[lo, inner[(inner > lo + 1e-9) & (inner < hi - 1e-9)], hi]
return edges[:-1], edges[1:]


def _cell_ends(cells: np.ndarray, ends: np.ndarray | None) -> np.ndarray:
"""The right edge of every cell: `ends` when compute clipped the grid to the fit, else every
cell is a full CELL_M wide, the shape a CellCoverage built by hand carries."""
return cells + CELL_M if ends is None else np.asarray(ends)


def tolerance(distance: np.ndarray) -> np.ndarray:
return np.maximum(HIDE_ABS_M, HIDE_REL * distance)

Expand Down Expand Up @@ -106,14 +129,18 @@ def two_positions(saw: np.ndarray, centres: np.ndarray) -> np.ndarray:
return ((s @ far) * s).sum(axis=-1) > 0


def face_offsets(wall: WallFrame, mesh: Mesh, cells: np.ndarray) -> np.ndarray:
def face_offsets(
wall: WallFrame, mesh: Mesh, cells: np.ndarray, ends: np.ndarray | None = None
) -> np.ndarray:
"""Per cell, how far a flat, full-height face stands in front of the fitted plane (0 if none).
Flat: in at least 75% of the 10 cm height bins from 0.3 to 1.9 m, the front-most surface within
0.6 m lies within 3 cm of their median. A bush or bin is neither, so it stays an occluder."""
ends = _cell_ends(cells, ends)
edges = np.r_[cells, ends[-1]] if len(cells) else cells
loc = wall.local(mesh.vertices.astype(np.float64))
s, h, out = loc[:, 0], loc[:, 1], loc[:, 2]
keep = (out > -0.15) & (out < 0.6) & (h > 0.3) & (h < 1.9)
col = np.floor((s[keep] - cells[0]) / CELL_M).astype(np.int64)
col = np.searchsorted(edges, s[keep], side="right") - 1
row = np.floor((h[keep] - 0.3) / 0.1).astype(np.int64)
ok = (col >= 0) & (col < len(cells))
front = np.full((len(cells), 16), -np.inf)
Expand All @@ -137,6 +164,7 @@ class CellCoverage:
facing_clear: np.ndarray # meters seen empty out from the wall's front
overhead_clearance: np.ndarray # meters up to the first thing overhead, NaN if none seen
overhead_clear: np.ndarray # meters seen empty above the footprint
ends: np.ndarray | None = None # right edge of each cell, meters of s


def wall_and_ground(
Expand All @@ -146,10 +174,14 @@ def wall_and_ground(
vol: Volume,
mesh: Mesh,
cells: np.ndarray,
ends: np.ndarray | None = None,
ground: Ground | None = None,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
ends = _cell_ends(cells, ends)
centres = np.array([f.center for f in frames])
faces = face_offsets(wall, mesh, cells)
alongs = np.stack([cells + 0.25 * CELL_M, cells + 0.75 * CELL_M], axis=1) # (C, 2)
faces = face_offsets(wall, mesh, cells, ends)
widths = ends - cells
alongs = np.stack([cells + 0.25 * widths, cells + 0.75 * widths], axis=1) # (C, 2)
heights = np.linspace(0.05, WALL_TOP_M, WALL_ROWS)
S = np.broadcast_to(alongs[:, :, None], (len(cells), 2, WALL_ROWS))
Hh = np.broadcast_to(heights[None, None, :], S.shape)
Expand All @@ -163,7 +195,16 @@ def wall_and_ground(
outs = np.append(np.arange(0.05, GROUND_MAX_M - 1e-6, CELL_M), GROUND_MAX_M)
S = np.broadcast_to(alongs[:, :, None], (len(cells), 2, len(outs)))
Oo = np.maximum(np.broadcast_to(outs[None, None, :], S.shape), faces[:, None, None] + 0.05)
gpts = wall.world(S, np.zeros_like(S), Oo).reshape(-1, 3)
if ground is None:
h = np.zeros_like(S)
else:
# The fitted ground, not the horizontal plane through the meter's foot: at (s, out) the
# surface's height above the meter's ground is height_at there minus its height at the
# meter, so on a slope the samples follow the ground instead of hanging above or below
# it, where nothing was built and the depth test would read visible ground unobserved.
flat = wall.world(S, np.zeros_like(S), Oo)
h = ground.height_at(flat[..., 0], flat[..., 2]) - wall.ground_y
gpts = wall.world(S, h, Oo).reshape(-1, 3)
gseen = two_positions(seen_by(gpts, UP, frames, depths, vol), centres).reshape(
len(cells), 2, len(outs)
)
Expand Down Expand Up @@ -202,7 +243,7 @@ def wall_front(wall: WallFrame, vol: Volume, across: np.ndarray) -> float:
return front + vol.voxel


def free_space(wall: WallFrame, vol: Volume, cells: np.ndarray):
def free_space(wall: WallFrame, vol: Volume, cells: np.ndarray, ends: np.ndarray | None = None):
"""Per cell: (facing gap or NaN, facing clear, overhead clearance or NaN, overhead clear).
The facing gap and clear distance are measured out from the wall's reconstructed front, not
from the fitted plane: in front of a pilaster 0.3 m proud, something 1.3 m out of the plane
Expand All @@ -211,11 +252,12 @@ def free_space(wall: WallFrame, vol: Volume, cells: np.ndarray):
clearance is a height above the ground over the footprint in front of that face."""
step = vol.voxel
n = len(cells)
widths = _cell_ends(cells, ends) - cells
facing_gap, facing_clear = np.full(n, np.nan), np.zeros(n)
over_gap, over_clear = np.full(n, np.nan), np.zeros(n)
fh = np.arange(FACING_BAND_M[0], FACING_BAND_M[1] + 1e-9, step)
for c in range(n):
across = cells[c] + CELL_M * np.array([0.25, 0.5, 0.75])
across = cells[c] + widths[c] * np.array([0.25, 0.5, 0.75])
front = wall_front(wall, vol, across)
hits = 0
for out in np.arange(front + step, front + FACING_MAX_M, step):
Expand Down Expand Up @@ -244,13 +286,20 @@ def free_space(wall: WallFrame, vol: Volume, cells: np.ndarray):


def compute(
wall: WallFrame, frames: list[Frame], depths: dict[str, Depth], vol: Volume, mesh: Mesh
wall: WallFrame,
frames: list[Frame],
depths: dict[str, Depth],
vol: Volume,
mesh: Mesh,
ground: Ground | None = None,
) -> CellCoverage:
lo, hi = wall.s_range
cells = np.arange(np.floor(lo / CELL_M) * CELL_M, hi - 1e-9, CELL_M)
wall_ok, ground_out, _ = wall_and_ground(wall, frames, depths, vol, mesh, cells)
fg, fc, og, oc = free_space(wall, vol, cells)
return CellCoverage(cells, wall_ok, ground_out, fg, fc, og, oc)
"""`ground`, the fit the wall's ground height came from, lets the ground samples follow the
surface off the horizontal plane through the meter's foot; the cells stop at the fitted wall
extent, so no exported observation runs past what the fit established."""
cells, ends = cell_grid(*wall.s_range)
wall_ok, ground_out, _ = wall_and_ground(wall, frames, depths, vol, mesh, cells, ends, ground)
fg, fc, og, oc = free_space(wall, vol, cells, ends)
return CellCoverage(cells, wall_ok, ground_out, fg, fc, og, oc, ends)


def _runs(mask: np.ndarray) -> list[tuple[int, int]]:
Expand All @@ -273,8 +322,10 @@ def observed(cov: CellCoverage, ground_levels_ft=(2.0, 4.0, 6.0, 8.0, 10.0)) ->
"""The coverage as scene.json `coverage.observed` entries (feet, s from the meter).

Ground is reported at a few depths so every clearance's request can be met by one entry;
`out_ft` is always a distance the samples cleared, rounded down."""
span = lambda a, b: [_ft(cov.cells[a]), _ft(cov.cells[b - 1] + CELL_M)] # noqa: E731
`out_ft` is always a distance the samples cleared, rounded down. A span stops at the fitted
extent: its end cells may be partial ones, clipped to where the fit established the wall."""
ends = _cell_ends(cov.cells, cov.ends)
span = lambda a, b: [_ft(cov.cells[a]), _ft(ends[b - 1])] # noqa: E731
entries = [{"band": "wall", "span_ft": span(a, b)} for a, b in _runs(cov.wall)]
for level in ground_levels_ft:
for a, b in _runs(cov.ground_out >= level * FEET - 1e-9):
Expand Down Expand Up @@ -317,7 +368,7 @@ def runs_of(values: np.ndarray) -> list[tuple[int, int]]:
def entry(a, b, key, values):
return {
"wall_id": wall_id,
"span_ft": [_ft(cov.cells[a]), _ft(cov.cells[b - 1] + CELL_M)],
"span_ft": [_ft(cov.cells[a]), _ft(_cell_ends(cov.cells, cov.ends)[b - 1])],
key: _ft(float(values[a:b].min())),
"plus_minus_ft": plus_minus_ft,
}
Expand Down
19 changes: 13 additions & 6 deletions recon/recon/pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,10 @@ def reconstruct(
# Only frames with a depth map can see anything (depth.depth_maps: LiDAR frames saved
# without depth get none).
seeing = [f for f in capture.frames if f.id in depths]
cov = coverage.compute(wall, seeing, depths, vol, surface)
# The fitted ground goes in so ground samples sit on the surface it fit, not on the horizontal
# plane through the meter's foot: on sloping ground that plane hangs above or below the real
# surface, and ground the scan saw read as unobserved.
cov = coverage.compute(wall, seeing, depths, vol, surface, ground)
_log(f"coverage over {len(cov.cells)} cells ({time.perf_counter() - t0:.0f} s)")
return {
"depths": depths,
Expand Down Expand Up @@ -150,9 +153,11 @@ def run(

def _coverage_doc(cov: coverage.CellCoverage) -> dict:
ft = lambda a: [None if not np.isfinite(x) else round(float(x) / FEET, 3) for x in a] # noqa: E731
ends = cov.cells + coverage.CELL_M if cov.ends is None else cov.ends
return {
"cell_ft": round(coverage.CELL_M / FEET, 3),
"cells_s_ft": ft(cov.cells),
"cells_end_ft": ft(ends), # a cell clipped to the fit's end is narrower than cell_ft
"wall_observed": cov.wall.tolist(),
"ground_out_ft": ft(cov.ground_out),
"facing_gap_ft": ft(cov.facing_gap),
Expand All @@ -164,7 +169,6 @@ def _coverage_doc(cov: coverage.CellCoverage) -> dict:

def _report(capture: Capture, r: dict, geo: dict, doc: dict, result: dict | None) -> str:
cov, dr = r["coverage"], r["depth_report"]
cell = coverage.CELL_M / FEET
wall, ground = geo["walls"][0], geo["ground"]
lo, hi = wall["s_range_ft"]
lines = [
Expand All @@ -179,17 +183,20 @@ def _report(capture: Capture, r: dict, geo: dict, doc: dict, result: dict | None
f"- Scale fitted for {dr['fitted']} of {dr['frames']} frames: MoGe-2 x "
f"{dr['median_scale']:.3f} median ({a:.3f} to {b:.3f}); the rest take the median."
)
facing = np.isfinite(cov.facing_gap).sum() * cell
over = np.isfinite(cov.overhead_clearance).sum() * cell
widths_ft = (cov.cells + coverage.CELL_M if cov.ends is None else cov.ends) - cov.cells
facing = np.isfinite(cov.facing_gap) @ widths_ft / FEET
over = np.isfinite(cov.overhead_clearance) @ widths_ft / FEET
obs = doc["coverage"]["observed"]
lines += [
f"- Model: {len(r['mesh'].vertices)} vertices at {r['volume'].voxel * 100:.1f} cm voxels.",
f"- Wall: {lo:.1f} to {hi:.1f} ft of s, plane fit RMS {wall['fit_rms_in']:.1f} in; "
f"{len(geo['other_walls'])} other wall stretches.",
f"- Ground: {ground['height_ft']:+.2f} ft, tilt {ground['tilt_deg']:.1f} deg, "
f"fit RMS {ground['fit_rms_in']:.1f} in.",
f"- Wall band observed: {cov.wall.sum() * cell:.1f} of {len(cov.cells) * cell:.1f} ft.",
f"- Ground seen 4 ft out or more: {(cov.ground_out >= 4 * FEET).sum() * cell:.1f} ft.",
f"- Wall band observed: {widths_ft[cov.wall].sum() / FEET:.1f} of "
f"{widths_ft.sum() / FEET:.1f} ft.",
f"- Ground seen 4 ft out or more: "
f"{widths_ft[cov.ground_out >= 4 * FEET].sum() / FEET:.1f} ft.",
f"- Facing gap measured over {facing:.1f} ft; overhead clearance over {over:.1f} ft.",
f"- scene.json: {len(obs)} observed entries, {len(doc['facing'])} facing, "
f"{len(doc['overheads'])} overheads.",
Expand Down
81 changes: 81 additions & 0 deletions recon/results/coverage_extent_witnesses.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
# Coverage-extent witnesses: baseline, pipeline witnesses, schema receipts

The record of the witness pass on top of ea38351 ("Coverage stops at the fitted extent and
samples the fitted ground"), which fixed the two coverage defects the handoff files as issues
#95 and #96. Everything here ran offline on synthetic bundles: no real home capture, no private
rule, no depth model, no server call.


## Baseline (head ea38351, the three-file repair untouched)

Run in `recon/`, 2026-10-10:

| Command | Result |
| --- | --- |
| `uv sync --locked` | Resolved 17 packages; environment up to date |
| `uv run ruff check .` | All checks passed |
| `uv run ruff format --check .` | 20 files already formatted |
| `uv run pytest -q` | 90 passed, 50 warnings, 11.7 s |
| `uv run pytest -q tests/test_coverage_fit.py` | 9 passed, 1.8 s |


## Pipeline witnesses (tests/test_coverage_pipeline.py + tests/witness_bundle.py)

Eight witnesses, each through `pipeline.run` (the caller a real bundle takes) on a synthetic
bundle rendered from analytic planes in `witness_bundle`: LiDAR depth only, no model, no
network, no real home capture. The oracle is the pipeline's own written scene.json,
coverage.json and geometry.json, plus refusals it raises:

| Witness | Pins |
| --- | --- |
| observed stops at the fitted extent when the baseline runs past it | every exported span within `[lo, hi]` of geometry.json while the phone's baseline claims 1.9/0.6 m more wall; grid starts at `lo`, boundary cells clipped |
| a hole in the wall stays unobserved through the pipeline | fit bridges a 0.8 m opening, cells across it stay unobserved, neighbours observed, exported spans agree |
| ground observed on a falling fitted plane (0.35 m/m) | 2 ft ground entries exist - impossible for the old horizontal-plane sampler, whose samples sit 0.21 m below the surface at 2 ft out |
| a short wall track (0.6 m) refuses | `RuntimeError: no straight vertical wall` |
| an empty wall track (floor only) refuses | same refusal |
| two frames 5 cm apart observe no wall or ground | under the 0.25 m two-position bar nothing exports (free space keeps its own weaker bar - known defect, commented in the test) |
| a bundle with no keyframes refuses | `ValueError: ... has no keyframes` |
| one LiDAR frame cannot drive coverage | `ValueError: ... only 1 of 1 keyframes carry LiDAR depth` |

`uv run pytest -q tests/test_coverage_pipeline.py` - 8 passed.

## Owned CLI smoke (exact command, offline)

uv run python tests/witness_bundle.py /tmp/witness-smoke/bundle
uv run python -m recon /tmp/witness-smoke/bundle --out /tmp/witness-smoke/out \
--work /tmp/witness-smoke/work --depth lidar --no-server

Output sha256 (head obv/basescanning-004 + these tests):

30a7ea0… scene.json
3e0cf8f… coverage.json
ac5e607… geometry.json
b28eb3b… report.md
438d3af… model.glb

## Which coverage changed (same bundle, pre-change code at c813182 vs head)

| Field | c813182 | head ea38351 + witnesses |
| --- | --- | --- |
| coverage.json cells_s_ft[0] | -3.5 (grid line beyond the fit) | -3.035 (the fitted extent) |
| coverage.json cells_end_ft | absent | present; last cell 5.0→5.332, clipped to the fit |
| scene.json wall span_ft | [-3.0, 5.0] (under-covers the fit) | [-3.0, 5.332] |
| scene.json overhead span_ft | [-3.5, 5.5] (claims past the fit) | [-3.035, 5.332] |
| wall_observed last cell | false (phantom grid cell) | true (cell ends at the real fit end) |
| ground_out_ft | unchanged on flat ground (0.664 tail) | unchanged (slope case is the pytest witness above) |

Full suite with the witnesses: `uv run pytest -q` - 98 passed, 56 warnings, 25.8 s; ruff
check and format clean (23 files).

## Scene-schema compatibility receipt

uv run --with jsonschema python tests/scene_schema_check.py /tmp/witness-smoke/out

/tmp/witness-smoke/out: 0 strict schema errors, 5 keyframe-depth passthrough items
(input extension, not defined by the schema)

The exported scene.json validates against `server/schemas/scene.schema.json` with no strict
errors. The five reported items are the per-keyframe `depth` blocks the input bundle carried
(one per input keyframe), which `scene.build` passes through unchanged; the schema does not
define that input extension, so the receipt reports it separately rather than folding it into
a pass. Full suite rerun after the schema script: 98 passed, ruff check and format clean.
55 changes: 55 additions & 0 deletions recon/tests/scene_schema_check.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
"""Scene-schema compatibility receipts for the witness outputs.

Validates each written scene.json against the committed server schema
(`app/assets/schemas/scene.schema.json`). The input bundle's per-keyframe `depth` block is an
app-side input that the schema does not define and scene.build passes through to its keyframes;
when that is the only failure it is reported on its own line, not folded into a pass.

uv run --with jsonschema python tests/scene_schema_check.py OUT1 [OUT2 ...]
"""

from __future__ import annotations

import argparse
import json
from pathlib import Path

from jsonschema import Draft202012Validator

SCHEMA = Path(__file__).resolve().parents[2] / "server" / "schemas" / "scene.schema.json"


def split_errors(doc: dict) -> tuple[list, list]:
"""Top-level schema errors, and keyframe-`depth` extension errors kept apart."""
validator = Draft202012Validator(json.loads(SCHEMA.read_text()))
strict, depth_ext = [], []
for err in sorted(validator.iter_errors(doc), key=lambda e: list(e.absolute_path)):
if "Additional properties" in err.message and "'depth'" in err.message:
depth_ext.append(err)
else:
strict.append(err)
return strict, depth_ext


def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("outputs", nargs="+", type=Path, help="run output folders with a scene.json")
args = ap.parse_args()
failures = 0
for out in args.outputs:
doc = json.loads((out / "scene.json").read_text())
strict, depth_ext = split_errors(doc)
print(
f"{out}: {len(strict)} strict schema errors, {len(depth_ext)} keyframe-depth "
"passthrough items (input extension, not defined by the schema)"
)
for e in strict:
failures += 1
print(f" STRICT {list(e.absolute_path)}: {e.message[:200]}")
for e in depth_ext[:3]:
print(f" depth passthrough at {list(e.absolute_path)}")
return 1 if failures else 0


if __name__ == "__main__":
raise SystemExit(main())
Loading