#!/usr/bin/env python3 """Build solidified accessibility meshes and turntable GIFs from completed depth.""" from __future__ import annotations import argparse import hashlib import html import json import math import os import sys from collections import Counter, defaultdict from pathlib import Path from typing import Any import cv2 import numpy as np from PIL import Image, ImageDraw, ImageOps PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT)) from accessibilityamodal.category_overrides import ( # noqa: E402 load_reviewed_category_overrides, resolve_geometry_category_for_modeling, ) from accessibilityamodal.reconstruct import camera_intrinsics, pixels_to_points CONTINUOUS_SURFACE_CATEGORIES = { "walkway", "ramp", "curb_cut", "raised_curb", "tactile_paving", } RAISED_CURB_CATEGORIES = {"raised_curb"} _FILE_SHA256_CACHE: dict[tuple[str, int, int], str] = {} def read_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def read_jsonl(path: Path) -> list[dict[str, Any]]: text = path.read_text(encoding="utf-8") stripped = text.lstrip() if stripped.startswith("["): data = json.loads(text) if not isinstance(data, list): raise ValueError(f"Expected a JSON list or JSONL rows in {path}") return [row for row in data if isinstance(row, dict)] rows = [] for line in text.splitlines(): if line.strip(): rows.append(json.loads(line)) return rows def sha256_file(path: Path) -> str: stat = path.stat() key = (str(path.resolve()), int(stat.st_size), int(stat.st_mtime_ns)) cached = _FILE_SHA256_CACHE.get(key) if cached is not None: return cached digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) value = digest.hexdigest() _FILE_SHA256_CACHE[key] = value return value def read_mask(path: Path, shape: tuple[int, int]) -> np.ndarray: mask = np.asarray(ImageOps.exif_transpose(Image.open(path)).convert("L")) if mask.shape[:2] != shape: raise ValueError( f"Mask/geometry raster mismatch for {path}: mask={mask.shape[:2]}, geometry={shape}. " "Refusing to resize a derived mask because it can hide an alignment error." ) return mask > 127 def read_mask_nearest(path: Path, shape: tuple[int, int]) -> np.ndarray: """Read an authoritative source mask with explicit nearest-neighbor mapping.""" mask = np.asarray(ImageOps.exif_transpose(Image.open(path)).convert("L")) if mask.shape[:2] != shape: if not aspect_ratio_matches(mask.shape[:2], shape): raise ValueError( f"Source mask/geometry aspect mismatch for {path}: " f"mask={mask.shape[:2]}, geometry={shape}." ) h, w = shape mask = np.asarray( Image.fromarray(mask, mode="L").resize( (w, h), Image.Resampling.NEAREST, ) ) return mask > 127 def aspect_ratio_matches(source_shape: tuple[int, int], target_shape: tuple[int, int]) -> bool: source_h, source_w = source_shape target_h, target_w = target_shape return abs(source_w * target_h - target_w * source_h) <= max(source_w, target_w) def largest_component(mask: np.ndarray) -> np.ndarray: count, labels, stats, _ = cv2.connectedComponentsWithStats(mask.astype(np.uint8), connectivity=8) if count <= 1: return mask largest = 1 + int(np.argmax(stats[1:, cv2.CC_STAT_AREA])) return labels == largest def clean_target_mask(mask: np.ndarray, close_kernel: int, keep_largest: bool) -> np.ndarray: cleaned = mask.astype(np.uint8) if close_kernel > 1: kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (close_kernel, close_kernel)) cleaned = cv2.morphologyEx(cleaned, cv2.MORPH_CLOSE, kernel) cleaned_bool = cleaned > 0 if keep_largest and np.any(cleaned_bool): cleaned_bool = largest_component(cleaned_bool) return cleaned_bool def fill_enclosed_mask_holes(mask: np.ndarray) -> np.ndarray: """Fill background components that cannot reach the image boundary. This is used only to make a render surface topologically continuous. It never replaces or writes the reviewed amodal annotation. """ source = mask.astype(bool) if not np.any(source): return source.copy() inverse = (~source).astype(np.uint8) component_count, labels = cv2.connectedComponents(inverse, connectivity=8) if component_count <= 1: return source.copy() border_labels = np.unique( np.concatenate( (labels[0], labels[-1], labels[:, 0], labels[:, -1]) ) ) enclosed = inverse.astype(bool) & ~np.isin(labels, border_labels) return source | enclosed def mask_component_count(mask: np.ndarray, *, connectivity: int = 8) -> int: count, _ = cv2.connectedComponents( mask.astype(np.uint8), connectivity=connectivity, ) return max(int(count) - 1, 0) def largest_axis_aligned_rectangle(mask: np.ndarray) -> tuple[int, int, int, int]: """Return the largest all-true rectangle as inclusive x/y bounds.""" source = mask.astype(bool) if source.ndim != 2 or not np.any(source): raise ValueError("Largest rectangle requires a non-empty 2D mask") height, width = source.shape histogram = np.zeros(width, dtype=np.int32) best_area = 0 best = (0, 0, 0, 0) for y in range(height): histogram = np.where(source[y], histogram + 1, 0) stack: list[int] = [] for x in range(width + 1): current_height = int(histogram[x]) if x < width else 0 while stack and int(histogram[stack[-1]]) > current_height: column = stack.pop() rectangle_height = int(histogram[column]) left = stack[-1] + 1 if stack else 0 rectangle_width = x - left area = rectangle_height * rectangle_width if area > best_area: best_area = area best = ( left, x - 1, y - rectangle_height + 1, y, ) stack.append(x) if best_area <= 0: raise ValueError("Largest rectangle search produced no supported cells") return best def build_continuous_surface_support_mask( target_mask: np.ndarray, *, stride: int, close_kernel: int = 0, visible_mask: np.ndarray | None = None, hidden_mask: np.ndarray | None = None, obstacle_mask: np.ndarray | None = None, automatic_close_added_ratio: float = 0.05, maximum_close_added_ratio: float = 0.10, maximum_large_hole_ratio: float = 0.05, obstacle_hole_overlap_ratio: float = 0.25, maximum_bridge_added_ratio: float = 0.005, maximum_opening_removed_ratio: float = 0.03, minimum_opening_hidden_retention: float = 0.95, ) -> tuple[np.ndarray, dict[str, Any]]: """Create a conservative, render-only support mask for a road surface. Sub-grid breaks are closed, fully enclosed holes are filled, and only very short nearby components may be bridged inside a tightly bounded evidence envelope. The operation does not use a row-span/convex-hull expansion and therefore does not turn unrelated background into target annotation. """ if stride < 1: raise ValueError("continuous surface stride must be positive") source = target_mask.astype(bool) for name, auxiliary in ( ("visible_mask", visible_mask), ("hidden_mask", hidden_mask), ("obstacle_mask", obstacle_mask), ): if auxiliary is not None and auxiliary.shape != source.shape: raise ValueError(f"{name} must match the continuous target raster") authoritative = source.copy() if visible_mask is not None: authoritative |= visible_mask.astype(bool) if hidden_mask is not None: authoritative |= hidden_mask.astype(bool) if not np.any(authoritative): raise ValueError("No target pixels for continuous surface support") kernel_size = int(close_kernel) if close_kernel > 0 else 4 * int(stride) + 1 kernel_size = max(kernel_size, 3) if kernel_size % 2 == 0: kernel_size += 1 kernel = cv2.getStructuringElement( cv2.MORPH_ELLIPSE, (kernel_size, kernel_size), ) obstacle_envelope = None if obstacle_mask is not None and np.any(obstacle_mask): full_obstacle = fill_enclosed_mask_holes(obstacle_mask.astype(bool)) obstacle_kernel_size = 4 * int(stride) + 1 obstacle_kernel = cv2.getStructuringElement( cv2.MORPH_ELLIPSE, (obstacle_kernel_size, obstacle_kernel_size), ) obstacle_envelope = cv2.dilate( full_obstacle.astype(np.uint8), obstacle_kernel, ).astype(bool) closed_candidate = cv2.morphologyEx( authoritative.astype(np.uint8), cv2.MORPH_CLOSE, kernel, ).astype(bool) close_added_candidate = closed_candidate & ~authoritative automatic_close_added_limit = max( 1, int(math.floor(float(authoritative.sum()) * automatic_close_added_ratio)), ) close_added_limit = max( 1, int(math.floor(float(authoritative.sum()) * maximum_close_added_ratio)), ) close_added_obstacle_overlap = ( float(np.count_nonzero(close_added_candidate & obstacle_envelope)) / max(int(close_added_candidate.sum()), 1) if obstacle_envelope is not None else 0.0 ) close_gate_passed = ( int(close_added_candidate.sum()) <= automatic_close_added_limit or ( int(close_added_candidate.sum()) <= close_added_limit and close_added_obstacle_overlap >= obstacle_hole_overlap_ratio ) ) closed = closed_candidate if close_gate_passed else authoritative.copy() inverse = (~closed).astype(np.uint8) component_count, labels, stats, _ = cv2.connectedComponentsWithStats( inverse, connectivity=8, ) border_labels = set( int(value) for value in np.unique( np.concatenate((labels[0], labels[-1], labels[:, 0], labels[:, -1])) ) ) source_pixels = int(authoritative.sum()) small_hole_limit = max( 4 * int(stride) * int(stride), int(math.floor(0.01 * source_pixels)), ) large_hole_limit = max( small_hole_limit, int(math.floor(maximum_large_hole_ratio * source_pixels)), ) supported = closed.copy() small_hole_pixels = 0 obstacle_gated_hole_pixels = 0 large_render_only_hole_pixels = 0 for label in range(1, component_count): if label in border_labels: continue area = int(stats[label, cv2.CC_STAT_AREA]) hole = labels == label if area <= small_hole_limit: supported[hole] = True small_hole_pixels += area continue overlap_ratio = ( float(np.count_nonzero(hole & obstacle_envelope)) / max(area, 1) if obstacle_envelope is not None else 0.0 ) supported[hole] = True if area <= large_hole_limit and overlap_ratio >= obstacle_hole_overlap_ratio: obstacle_gated_hole_pixels += area else: large_render_only_hole_pixels += area enclosed_hole_added_pixels = int((supported & ~closed).sum()) # Connect only short, significant surface fragments. This is intentionally # much stricter than a hull/row-span envelope: the bridge must stay next to # the reviewed surface or inside the authoritative obstacle neighbourhood, # and the total added area is capped at 0.5% by default. bridge_added_limit = max( 1, int(math.floor(maximum_bridge_added_ratio * source_pixels)), ) bridge_added_pixels = 0 bridge_count = 0 bridge_rejected_count = 0 bridge_distances_pixels: list[float] = [] bridge_maximum_distance = 3.0 * float(stride) * math.sqrt(2.0) bridge_width = 2 * int(stride) + 1 component_count_before_bridge = mask_component_count(supported) if component_count_before_bridge > 1: count, labels, stats, _ = cv2.connectedComponentsWithStats( supported.astype(np.uint8), connectivity=8, ) primary_label = 1 + int(np.argmax(stats[1:, cv2.CC_STAT_AREA])) anchor = labels == primary_label bridge_allowed_kernel = cv2.getStructuringElement( cv2.MORPH_ELLIPSE, (6 * int(stride) + 1, 6 * int(stride) + 1), ) bridge_allowed = cv2.dilate( authoritative.astype(np.uint8), bridge_allowed_kernel, ).astype(bool) if obstacle_envelope is not None: bridge_allowed |= obstacle_envelope minimum_component_pixels = max( 4 * int(stride) * int(stride), int(math.ceil(0.001 * source_pixels)), ) component_labels = sorted( range(1, count), key=lambda label: int(stats[label, cv2.CC_STAT_AREA]), reverse=True, ) for label in component_labels: if label == primary_label: continue component = labels == label carries_hidden = bool( hidden_mask is not None and np.any(component & hidden_mask.astype(bool)) ) if ( int(stats[label, cv2.CC_STAT_AREA]) < minimum_component_pixels and not carries_hidden ): continue distance, nearest_label_map = cv2.distanceTransformWithLabels( (~anchor).astype(np.uint8), cv2.DIST_L2, 5, labelType=cv2.DIST_LABEL_PIXEL, ) component_y, component_x = np.nonzero(component) nearest_index = int(np.argmin(distance[component_y, component_x])) source_y = int(component_y[nearest_index]) source_x = int(component_x[nearest_index]) distance_pixels = float(distance[source_y, source_x]) nearest_label = int(nearest_label_map[source_y, source_x]) anchor_y, anchor_x = np.nonzero(anchor) anchor_labels = nearest_label_map[anchor_y, anchor_x] matches = np.flatnonzero(anchor_labels == nearest_label) if matches.size == 0: bridge_rejected_count += 1 continue target_index = int(matches[0]) target_y = int(anchor_y[target_index]) target_x = int(anchor_x[target_index]) candidate = np.zeros_like(supported, dtype=np.uint8) cv2.line( candidate, (source_x, source_y), (target_x, target_y), color=1, thickness=bridge_width, lineType=cv2.LINE_8, ) candidate = candidate.astype(bool) added_candidate = candidate & ~supported evidence_fraction = float( np.count_nonzero(candidate & bridge_allowed) ) / max(int(candidate.sum()), 1) if ( distance_pixels > bridge_maximum_distance or evidence_fraction < 0.90 or bridge_added_pixels + int(added_candidate.sum()) > bridge_added_limit ): bridge_rejected_count += 1 continue supported |= candidate anchor |= component | candidate added_pixels = int(added_candidate.sum()) bridge_added_pixels += added_pixels bridge_count += 1 bridge_distances_pixels.append(distance_pixels) # A short accepted bridge can enclose a final sub-grid pocket. before_post_bridge_hole_fill = supported.copy() supported = fill_enclosed_mask_holes(supported) post_bridge_hole_added_pixels = int( np.count_nonzero(supported & ~before_post_bridge_hole_fill) ) # Remove only thin render-footprint hairs. The same 13 px scale that closes # obstacle cuts removes sub-resolution protrusions without changing a broad # Y junction into a hull or trapezoid. This is accepted only when at least # 95% of hidden evidence and 97% of authoritative support survive. opening_candidate = cv2.morphologyEx( supported.astype(np.uint8), cv2.MORPH_OPEN, kernel, ).astype(bool) opening_removed = supported & ~opening_candidate opening_removed_ratio = float( np.count_nonzero(opening_removed & authoritative) ) / max(source_pixels, 1) opening_hidden_retention = ( float(np.count_nonzero(opening_candidate & hidden_mask.astype(bool))) / max(int(hidden_mask.sum()), 1) if hidden_mask is not None and np.any(hidden_mask) else 1.0 ) opening_gate_passed = bool( opening_removed_ratio <= maximum_opening_removed_ratio and opening_hidden_retention >= minimum_opening_hidden_retention and np.any(opening_candidate) ) if opening_gate_passed: # Opening is a render-footprint cleanup, never an annotation erosion. # Restore every reviewed visible/hidden pixel before the final hole # fill so thin synthetic hairs can be removed without reopening an # occluder cut or trimming authoritative amodal evidence. supported = fill_enclosed_mask_holes( opening_candidate | authoritative ) added = supported & ~authoritative hidden_pixels = int(hidden_mask.sum()) if hidden_mask is not None else 0 retained_hidden = ( int(np.count_nonzero(supported & hidden_mask.astype(bool))) if hidden_mask is not None else 0 ) return supported, { "role": "render_only_geometry_support_not_annotation", "policy": ( "authoritative_components_plus_bounded_close_all_enclosed_holes_" "and_short_evidence_bridges" ), "row_span_or_convex_hull_used": False, "generation_mask_used": False, "annotation_mask_modified": False, "stride": int(stride), "close_kernel": kernel_size, "source_pixels": int(source.sum()), "authoritative_pixels": source_pixels, "visible_pixels": int(visible_mask.sum()) if visible_mask is not None else None, "hidden_pixels": hidden_pixels if hidden_mask is not None else None, "hidden_retained_pixels": retained_hidden if hidden_mask is not None else None, "hidden_retention_fraction": ( float(retained_hidden / hidden_pixels) if hidden_pixels > 0 else None ), "close_added_candidate_pixels": int(close_added_candidate.sum()), "automatic_close_added_limit_pixels": automatic_close_added_limit, "close_added_limit_pixels": close_added_limit, "close_added_obstacle_overlap_fraction": close_added_obstacle_overlap, "close_gate_passed": close_gate_passed, "close_added_pixels": int((closed & ~authoritative).sum()), "small_hole_limit_pixels": small_hole_limit, "large_hole_limit_pixels": large_hole_limit, "small_hole_added_pixels": small_hole_pixels, "obstacle_gated_hole_added_pixels": obstacle_gated_hole_pixels, "large_render_only_hole_added_pixels": large_render_only_hole_pixels, "rejected_large_hole_pixels": 0, "all_enclosed_holes_filled": True, "obstacle_hole_overlap_ratio": float(obstacle_hole_overlap_ratio), "full_obstacle_used_for_hole_gate": obstacle_envelope is not None, "enclosed_hole_added_pixels": enclosed_hole_added_pixels, "post_bridge_hole_added_pixels": post_bridge_hole_added_pixels, "support_pixels": int(supported.sum()), "support_added_pixels": int(added.sum()), "source_components": mask_component_count(authoritative), "support_components": mask_component_count(supported), "components_before_short_bridge": component_count_before_bridge, "short_bridge_count": bridge_count, "short_bridge_rejected_count": bridge_rejected_count, "short_bridge_added_pixels": bridge_added_pixels, "short_bridge_added_limit_pixels": bridge_added_limit, "short_bridge_maximum_distance_pixels": bridge_maximum_distance, "short_bridge_width_pixels": bridge_width, "short_bridge_distances_pixels": bridge_distances_pixels, "footprint_opening_kernel": kernel_size, "footprint_opening_candidate_removed_pixels": int( opening_removed.sum() ), "footprint_opening_removed_authoritative_pixels": int( np.count_nonzero(opening_removed & authoritative) ), "footprint_opening_removed_authoritative_ratio": opening_removed_ratio, "footprint_opening_hidden_retention_fraction": opening_hidden_retention, "footprint_opening_gate_passed": opening_gate_passed, "footprint_opening_policy": ( "thin_hair_removal_only_without_hull_row_span_or_trapezoid" ), "detached_hidden_bridge_policy": ( "only_short_bounded_bridges_near_reviewed_surface_or_obstacle_evidence" ), } def complete_continuous_surface_texture( colors: np.ndarray, geometry_support: np.ndarray, obstacle_mask: np.ndarray | None, *, stride: int, ) -> tuple[np.ndarray, dict[str, Any]]: """Replace obstacle appearance with sharp same-surface exemplar texels. This is a 3D-render-only texture adapter. It never writes back to the frozen 2D completion. Every changed byte is copied from a non-obstacle texel on the same connected surface; no diffusion, mean fill, Telea, or semantic color tint is used, which avoids the foggy appearance of the old renderer while removing poles/people/cars painted flat onto the road. """ if colors.ndim != 3 or colors.shape[2] != 3: raise ValueError("Continuous surface texture must be an HxWx3 RGB image") if geometry_support.shape != colors.shape[:2]: raise ValueError("Geometry support and surface texture must share a raster") if stride < 1: raise ValueError("continuous surface texture stride must be positive") if obstacle_mask is None or not np.any(obstacle_mask): return colors.copy(), { "role": "render_only_3d_texture_not_2d_completion", "policy": "no_obstacle_texture_repair_needed", "obstacle_texture_pixels": 0, "repaired_pixels": 0, "source_file_modified": False, "diffusion_or_mean_fill_used": False, } if obstacle_mask.shape != colors.shape[:2]: raise ValueError("Obstacle mask and surface texture must share a raster") support = geometry_support.astype(bool) obstacle = obstacle_mask.astype(bool) repair = support & obstacle if not np.any(repair): return colors.copy(), { "role": "render_only_3d_texture_not_2d_completion", "policy": "obstacle_does_not_overlap_surface", "obstacle_texture_pixels": 0, "repaired_pixels": 0, "source_file_modified": False, "diffusion_or_mean_fill_used": False, } kernel_size = 2 * int(stride) + 1 kernel = cv2.getStructuringElement( cv2.MORPH_ELLIPSE, (kernel_size, kernel_size), ) expanded_obstacle = cv2.dilate( obstacle.astype(np.uint8), kernel, ).astype(bool) output = colors.copy() geometry_component_count, geometry_component_labels = cv2.connectedComponents( support.astype(np.uint8), connectivity=8, ) repair_component_count, repair_component_labels, repair_stats, _ = ( cv2.connectedComponentsWithStats( repair.astype(np.uint8), connectivity=8, ) ) repaired_pixels = 0 skipped_pixels = 0 repaired_components = 0 skipped_components = 0 mirrored_pixels = 0 translated_pixels = 0 translated_components = 0 translation_shifts: list[dict[str, Any]] = [] donor_distances: list[np.ndarray] = [] donor_linear_indices: list[np.ndarray] = [] h, w = support.shape for repair_label in range(1, repair_component_count): component_repair = repair_component_labels == repair_label repair_y, repair_x = np.nonzero(component_repair) if repair_x.size == 0: continue geometry_values = geometry_component_labels[component_repair] geometry_values = geometry_values[geometry_values > 0] if geometry_values.size == 0: skipped_components += 1 skipped_pixels += int(repair_x.size) continue geometry_label = int( np.argmax(np.bincount(geometry_values.astype(np.int64))) ) component = geometry_component_labels == geometry_label donor = component & ~expanded_obstacle if int(donor.sum()) < max(16, int(0.01 * component.sum())): donor = component & ~obstacle if not np.any(donor): skipped_components += 1 skipped_pixels += int(repair_x.size) continue original_repair_y = repair_y.copy() original_repair_x = repair_x.copy() filled = np.zeros(repair_x.shape[0], dtype=bool) sample_y_all = np.zeros(repair_x.shape[0], dtype=np.int32) sample_x_all = np.zeros(repair_x.shape[0], dtype=np.int32) x0 = int(repair_stats[repair_label, cv2.CC_STAT_LEFT]) y0 = int(repair_stats[repair_label, cv2.CC_STAT_TOP]) box_width = int(repair_stats[repair_label, cv2.CC_STAT_WIDTH]) box_height = int(repair_stats[repair_label, cv2.CC_STAT_HEIGHT]) template = component_repair[ y0 : y0 + box_height, x0 : x0 + box_width, ].astype(np.float32) translation_overlap = 0.0 chosen_dx = 0 chosen_dy = 0 if ( box_height <= h and box_width <= w and template.size > 0 and np.any(template) ): response = cv2.matchTemplate( donor.astype(np.float32), template, cv2.TM_CCORR, ) response = response / max(float(template.sum()), 1.0) candidate_count = min(128, int(response.size)) flat = response.reshape(-1) if candidate_count > 0: candidate_indices = np.argpartition( flat, -candidate_count, )[-candidate_count:] maximum_overlap = float(flat[candidate_indices].max()) minimum_candidate_overlap = max(0.70, maximum_overlap - 0.02) ring = ( cv2.dilate(component_repair.astype(np.uint8), kernel).astype(bool) & donor ) ring_values = colors[ring].astype(np.float32) ring_mean = ( ring_values.mean(axis=0) if ring_values.size else None ) best_score = float("inf") best: tuple[int, int, float] | None = None response_width = response.shape[1] for flat_index in candidate_indices: overlap = float(flat[int(flat_index)]) if overlap < minimum_candidate_overlap: continue source_top = int(flat_index) // response_width source_left = int(flat_index) % response_width dy = source_top - y0 dx = source_left - x0 candidate_y = original_repair_y + dy candidate_x = original_repair_x + dx candidate_valid = donor[candidate_y, candidate_x] if not np.any(candidate_valid): continue source_values = colors[ candidate_y[candidate_valid], candidate_x[candidate_valid], ].astype(np.float32) color_cost = ( float(np.mean(np.abs(source_values.mean(axis=0) - ring_mean))) / 255.0 if ring_mean is not None else 0.0 ) shift_cost = ( abs(float(dx)) / max(float(w), 1.0) + 3.0 * abs(float(dy)) / max(float(h), 1.0) ) score = 20.0 * (1.0 - overlap) + shift_cost + color_cost if score < best_score: best_score = score best = (dx, dy, overlap) if best is not None: chosen_dx, chosen_dy, translation_overlap = best translated_y = original_repair_y + chosen_dy translated_x = original_repair_x + chosen_dx translated_valid = donor[translated_y, translated_x] output[ original_repair_y[translated_valid], original_repair_x[translated_valid], ] = colors[ translated_y[translated_valid], translated_x[translated_valid], ] filled[translated_valid] = True sample_y_all[translated_valid] = translated_y[translated_valid] sample_x_all[translated_valid] = translated_x[translated_valid] translated_count = int(np.count_nonzero(translated_valid)) translated_pixels += translated_count if translated_count > 0: translated_components += 1 translation_shifts.append( { "component": int(repair_label), "dx": int(chosen_dx), "dy": int(chosen_dy), "support_overlap_fraction": translation_overlap, "translated_pixels": translated_count, "component_pixels": int(original_repair_x.size), } ) remaining = ~filled repair_y = original_repair_y[remaining] repair_x = original_repair_x[remaining] if repair_x.size == 0: repaired_components += 1 repaired_pixels += int(original_repair_x.size) donor_linear_indices.append( (sample_y_all * w + sample_x_all).astype(np.int64) ) continue distance, nearest_labels = cv2.distanceTransformWithLabels( (~donor).astype(np.uint8), cv2.DIST_L2, 5, labelType=cv2.DIST_LABEL_PIXEL, ) maximum_label = int(nearest_labels.max()) nearest_y = np.zeros(maximum_label + 1, dtype=np.int32) nearest_x = np.zeros(maximum_label + 1, dtype=np.int32) donor_y, donor_x = np.nonzero(donor) donor_labels = nearest_labels[donor_y, donor_x] nearest_y[donor_labels] = donor_y nearest_x[donor_labels] = donor_x repair_labels = nearest_labels[repair_y, repair_x] valid = (repair_labels > 0) & (repair_labels <= maximum_label) if not np.all(valid): skipped_pixels += int(np.count_nonzero(~valid)) repair_y = repair_y[valid] repair_x = repair_x[valid] repair_labels = repair_labels[valid] if repair_x.size == 0: skipped_components += 1 continue source_y = nearest_y[repair_labels] source_x = nearest_x[repair_labels] reflected_y = np.clip(2 * source_y - repair_y, 0, h - 1) reflected_x = np.clip(2 * source_x - repair_x, 0, w - 1) reflected_valid = donor[reflected_y, reflected_x] sample_y = np.where(reflected_valid, reflected_y, source_y) sample_x = np.where(reflected_valid, reflected_x, source_x) output[repair_y, repair_x] = colors[sample_y, sample_x] remaining_indices = np.flatnonzero(remaining)[valid] sample_y_all[remaining_indices] = sample_y sample_x_all[remaining_indices] = sample_x repaired_components += 1 repaired_pixels += int(original_repair_x.size) mirrored_pixels += int(np.count_nonzero(reflected_valid)) donor_distances.append(distance[repair_y, repair_x].astype(np.float32)) donor_linear_indices.append( (sample_y_all * w + sample_x_all).astype(np.int64) ) distances = ( np.concatenate(donor_distances) if donor_distances else np.empty((0,), dtype=np.float32) ) linear_indices = ( np.concatenate(donor_linear_indices) if donor_linear_indices else np.empty((0,), dtype=np.int64) ) reuse_counts = ( np.unique(linear_indices, return_counts=True)[1] if linear_indices.size else np.empty((0,), dtype=np.int64) ) unchanged_outside_repair = bool( np.array_equal(output[~repair], colors[~repair]) ) copied_texel_provenance_fraction = ( float(repaired_pixels / max(int(repair.sum()), 1)) ) return output, { "role": "render_only_3d_texture_not_2d_completion", "policy": ( "same_connected_surface_coherent_translated_exemplar_then_" "mirrored_nearest_fallback_without_diffusion_or_bbox_warp" ), "source_file_modified": False, "frozen_2d_consumed_read_only": True, "diffusion_or_mean_fill_used": False, "telea_inpaint_used": False, "obstacle_texture_pixels": int(repair.sum()), "repaired_pixels": repaired_pixels, "skipped_pixels": skipped_pixels, "repaired_components": repaired_components, "skipped_components": skipped_components, "donor_exclusion_kernel": kernel_size, "mirrored_donor_fraction": ( float(mirrored_pixels / max(repaired_pixels, 1)) ), "translated_exemplar_pixels": translated_pixels, "translated_exemplar_fraction": ( float(translated_pixels / max(repaired_pixels, 1)) ), "translated_exemplar_components": translated_components, "translation_shifts": translation_shifts, "copied_texel_provenance_fraction": copied_texel_provenance_fraction, "outside_repair_byte_identical": unchanged_outside_repair, "donor_distance_mean_pixels": ( float(np.mean(distances)) if distances.size else 0.0 ), "donor_distance_p95_pixels": ( float(np.percentile(distances, 95)) if distances.size else 0.0 ), "donor_distance_max_pixels": ( float(np.max(distances)) if distances.size else 0.0 ), "unique_donor_fraction": ( float(np.unique(linear_indices).size / max(linear_indices.size, 1)) ), "donor_reuse_p95": ( float(np.percentile(reuse_counts, 95)) if reuse_counts.size else 0.0 ), "donor_reuse_max": ( int(np.max(reuse_counts)) if reuse_counts.size else 0 ), } def resolve_source_image(sample_dir: Path) -> Path | None: manifest_path = sample_dir / "pipeline_manifest.json" if not manifest_path.is_file(): return None manifest = read_json(manifest_path) image = manifest.get("image") if image and Path(image).is_file(): return Path(image) source_manifest = manifest.get("source_pipeline_manifest") if source_manifest and Path(source_manifest).is_file(): source = read_json(Path(source_manifest)) image = source.get("image") if image and Path(image).is_file(): return Path(image) return None def prepare_texture( sample_dir: Path, geometry_dir: Path, shape: tuple[int, int], *, hidden_tint: float, texture_override: Path | None = None, ) -> tuple[np.ndarray, str]: h, w = shape if texture_override is not None: if not texture_override.is_file(): raise FileNotFoundError( "The explicit frozen 2D texture is missing; refusing to fall " f"back to a lower-quality or stale texture: {texture_override}" ) if hidden_tint > 0: raise ValueError( "A frozen 2D texture cannot be color-tinted in the 3D path. " "Set --hidden-tint 0 to preserve its appearance." ) completed_candidates = [ texture_override, sample_dir / "03_2d_completion" / "completed_rgb_selected.png", sample_dir / "amodal2d" / "completed_rgb_selected.png", geometry_dir / "completed_rgb_inpaint.png", ] completed_rgb = next( ( candidate for candidate in completed_candidates if candidate is not None and candidate.is_file() ), None, ) if completed_rgb is not None: texture = np.asarray(ImageOps.exif_transpose(Image.open(completed_rgb)).convert("RGB")) source = str(completed_rgb) already_completed = True else: source_image = resolve_source_image(sample_dir) already_completed = False if completed_rgb is None and source_image is not None: texture = np.asarray(ImageOps.exif_transpose(Image.open(source_image)).convert("RGB")) source = str(source_image) elif completed_rgb is None: texture = np.asarray(Image.open(geometry_dir / "completed_region_overlay.png").convert("RGB")) source = str(geometry_dir / "completed_region_overlay.png") if texture.shape[:2] != (h, w): if already_completed: if not aspect_ratio_matches(texture.shape[:2], (h, w)): raise ValueError( f"Completed RGB/geometry aspect mismatch: rgb={texture.shape[:2]}, geometry={(h, w)}." ) source_shape = texture.shape[:2] texture = np.asarray( Image.fromarray(texture).resize((w, h), Image.Resampling.LANCZOS) ) source += ( f" + read_only_texture_resample_{source_shape[1]}x{source_shape[0]}" f"_to_{w}x{h}" ) else: if not aspect_ratio_matches(texture.shape[:2], (h, w)): raise ValueError( f"Fallback source RGB/geometry aspect mismatch: rgb={texture.shape[:2]}, geometry={(h, w)}." ) texture = np.asarray( Image.fromarray(texture).resize((w, h), Image.Resampling.BILINEAR) ) hidden_path = geometry_dir / "hidden_completion_mask.png" if hidden_path.is_file(): hidden = read_mask(hidden_path, (h, w)) if np.any(hidden): if not already_completed: inpaint_mask = (hidden.astype(np.uint8) * 255) texture = cv2.inpaint(texture, inpaint_mask, 5, cv2.INPAINT_TELEA) if hidden_tint > 0: tint = np.array([255, 95, 25], dtype=np.float32) texture = texture.astype(np.float32) texture[hidden] = (1.0 - hidden_tint) * texture[hidden] + hidden_tint * tint texture = np.clip(texture, 0, 255).astype(np.uint8) source += " + clean_hidden_texture" if already_completed else " + hidden_texture_inpaint" if hidden_tint > 0: source += f" + hidden_tint_{hidden_tint:.3f}" return texture.astype(np.uint8), source def fit_plane(points: np.ndarray) -> tuple[np.ndarray, float] | None: if points.shape[0] < 50: return None if points.shape[0] > 50000: points = points[np.linspace(0, points.shape[0] - 1, 50000).astype(np.int64)] sample = points for _ in range(4): centroid = sample.mean(axis=0) centered = sample - centroid try: _, _, vh = np.linalg.svd(centered, full_matrices=False) except np.linalg.LinAlgError: return None normal = vh[-1] norm = float(np.linalg.norm(normal)) if norm < 1e-6: return None normal = normal / norm d = -float(np.dot(normal, centroid)) residual = np.abs(sample @ normal + d) threshold = max(float(np.percentile(residual, 75)), float(np.median(residual)) * 2.5, 1e-6) retained = sample[residual <= threshold] if retained.shape[0] < 50 or retained.shape[0] >= sample.shape[0] * 0.98: return normal.astype(np.float32), d sample = retained return normal.astype(np.float32), d def plane_depth_for_pixels( shape: tuple[int, int], plane: tuple[np.ndarray, float], fx: float, fy: float, cx: float, cy: float, ) -> np.ndarray: h, w = shape normal, d = plane xs, ys = np.meshgrid(np.arange(w, dtype=np.float32), np.arange(h, dtype=np.float32)) rays = np.stack([(xs - cx) / fx, (ys - cy) / fy, np.ones((h, w), dtype=np.float32)], axis=-1) denom = rays @ normal with np.errstate(divide="ignore", invalid="ignore"): depth = -d / denom depth[~np.isfinite(depth)] = 0 depth[depth <= 0] = 0 return depth.astype(np.float32) def robust_extent_from_points(points: np.ndarray) -> dict[str, Any]: if points.shape[0] == 0: return { "span_x": 1.0, "span_y": 0.2, "span_z": 1.0, "footprint_diag": 1.0, "median_z": 1.0, "percentile_low": [0.0, 0.0, 0.0], "percentile_high": [1.0, 0.2, 1.0], } sample = points if sample.shape[0] > 80000: sample = sample[np.linspace(0, sample.shape[0] - 1, 80000).astype(np.int64)] low = np.percentile(sample, 3, axis=0) high = np.percentile(sample, 97, axis=0) span = np.maximum(high - low, 1e-4) return { "span_x": float(span[0]), "span_y": float(span[1]), "span_z": float(span[2]), "footprint_diag": float(np.hypot(span[0], span[2])), "median_z": float(np.median(sample[:, 2])), "percentile_low": [float(value) for value in low], "percentile_high": [float(value) for value in high], } def depth_scale_for_mask(depth: np.ndarray, target_mask: np.ndarray) -> dict[str, Any]: h, w = depth.shape fx, fy, cx, cy = camera_intrinsics(w, h, None, None, None, None) valid = target_mask & np.isfinite(depth) & (depth > 0) points = pixels_to_points(depth, fx, fy, cx, cy)[valid] scale = robust_extent_from_points(points) scale["valid_points"] = int(points.shape[0]) return scale def continuous_depth_for_category( depth: np.ndarray, target_mask: np.ndarray, visible_mask: np.ndarray | None, category: str, ) -> tuple[np.ndarray, dict[str, Any]]: if category not in CONTINUOUS_SURFACE_CATEGORIES: return depth, {"surface_model": "depth_surface"} h, w = depth.shape fx, fy, cx, cy = camera_intrinsics(w, h, None, None, None, None) points = pixels_to_points(depth, fx, fy, cx, cy) fit_mask = target_mask if visible_mask is None else (visible_mask & target_mask) fit_mask &= np.isfinite(depth) & (depth > 0) plane = fit_plane(points[fit_mask]) if plane is None: return depth, {"surface_model": "depth_surface_plane_fit_failed", "fit_points": int(fit_mask.sum())} plane_depth = plane_depth_for_pixels((h, w), plane, fx, fy, cx, cy) observed = depth[target_mask & np.isfinite(depth) & (depth > 0)] if observed.size == 0: return depth, {"surface_model": "depth_surface_no_observed_values", "fit_points": int(fit_mask.sum())} low, high = np.percentile(observed, [1, 99]) plausible = target_mask & np.isfinite(plane_depth) & (plane_depth > 0.5 * low) & (plane_depth < 1.5 * high) continuous = depth.copy() continuous[plausible] = plane_depth[plausible] surface_model = ( "raised_curb_planar_cap" if category in RAISED_CURB_CATEGORIES else "robust_single_plane" ) return continuous, { "surface_model": surface_model, "fit_points": int(fit_mask.sum()), "plane_normal": [float(v) for v in plane[0]], "plane_d": float(plane[1]), "plane_applied_pixels": int(plausible.sum()), } def rotation_matrix(yaw_degrees: float, pitch_degrees: float, roll_degrees: float = 0.0) -> np.ndarray: yaw = math.radians(yaw_degrees) pitch = math.radians(pitch_degrees) roll = math.radians(roll_degrees) cy, sy = math.cos(yaw), math.sin(yaw) cp, sp = math.cos(pitch), math.sin(pitch) cr, sr = math.cos(roll), math.sin(roll) ry = np.array([[cy, 0.0, sy], [0.0, 1.0, 0.0], [-sy, 0.0, cy]], dtype=np.float32) rx = np.array([[1.0, 0.0, 0.0], [0.0, cp, -sp], [0.0, sp, cp]], dtype=np.float32) rz = np.array([[cr, -sr, 0.0], [sr, cr, 0.0], [0.0, 0.0, 1.0]], dtype=np.float32) return rz @ rx @ ry def quality_review_ids(path: Path | None) -> set[str]: if path is None or not path.is_file(): return set() data = read_json(path) return {row["sample_id"] for row in data.get("review_candidates", [])} def load_taxonomy(path: Path | None) -> dict[str, Any]: if path is None or not path.is_file(): return {} return read_json(path).get("categories", {}) def select_rows( rows: list[dict[str, Any]], review_ids: set[str], limit: int, per_category_limit: int, include_review: bool, sample_ids: list[str] | None = None, ) -> list[dict[str, Any]]: if sample_ids: by_id = {row["sample_id"]: row for row in rows} missing = [sample_id for sample_id in sample_ids if sample_id not in by_id] if missing: raise ValueError(f"sample_ids not found in manifest: {', '.join(missing)}") return [by_id[sample_id] for sample_id in sample_ids] selected = [] counts: Counter[str] = Counter() for row in rows: sample_id = row["sample_id"] if not include_review and sample_id in review_ids: continue category = str(row.get("category") or "unknown") if per_category_limit > 0 and counts[category] >= per_category_limit: continue selected.append(row) counts[category] += 1 if limit > 0 and len(selected) >= limit: break return selected def line_x_bounds_on_mask( y_center: float, slope: float, mask: np.ndarray, ) -> tuple[int, int] | None: h, w = mask.shape x_center = (w - 1) / 2.0 xs = np.arange(w, dtype=np.float32) ys = np.rint(y_center + slope * (xs - x_center)).astype(np.int32) in_frame = (ys >= 0) & (ys < h) if not np.any(in_frame): return None valid_xs = xs[in_frame].astype(np.int32) valid_ys = ys[in_frame] on_mask = mask[valid_ys, valid_xs] if np.any(on_mask): selected_xs = valid_xs[on_mask] return int(selected_xs.min()), int(selected_xs.max()) mask_ys, mask_xs = np.nonzero(mask) if mask_xs.size == 0: return None return int(mask_xs.min()), int(mask_xs.max()) def sample_nearest_depth( depth: np.ndarray, valid: np.ndarray, x: float, y: float, fallback: float, ) -> float: h, w = depth.shape cx = int(np.clip(round(x), 0, w - 1)) cy = int(np.clip(round(y), 0, h - 1)) if valid[cy, cx]: return float(depth[cy, cx]) for radius in (2, 4, 8, 14, 22): y0 = max(0, cy - radius) y1 = min(h, cy + radius + 1) x0 = max(0, cx - radius) x1 = min(w, cx + radius + 1) patch_valid = valid[y0:y1, x0:x1] if np.any(patch_valid): return float(np.median(depth[y0:y1, x0:x1][patch_valid])) return fallback def sample_nearest_color(colors: np.ndarray, x: float, y: float) -> np.ndarray: h, w = colors.shape[:2] cx = int(np.clip(round(x), 0, w - 1)) cy = int(np.clip(round(y), 0, h - 1)) return colors[cy, cx].astype(np.uint8) def regularize_depth_row(depth_values: np.ndarray) -> np.ndarray: values = depth_values.astype(np.float32).copy() finite = np.isfinite(values) & (values > 0) if finite.sum() == 0: return values fallback = float(np.median(values[finite])) values[~finite] = fallback if len(values) < 4: return values xs = np.linspace(-1.0, 1.0, len(values), dtype=np.float32) residual_center = float(np.median(values)) residual_scale = float(np.median(np.abs(values - residual_center))) + 1e-6 inliers = np.abs(values - residual_center) <= 3.0 * residual_scale if inliers.sum() >= 2: degree = 1 coefficients = np.polyfit(xs[inliers], values[inliers], degree) fitted = np.polyval(coefficients, xs).astype(np.float32) values = 0.35 * values + 0.65 * fitted kernel = np.array([0.20, 0.60, 0.20], dtype=np.float32) padded = np.pad(values, (1, 1), mode="edge") return np.convolve(padded, kernel, mode="valid").astype(np.float32) def polygon_signed_area(points: np.ndarray) -> float: """Return the signed area of an Nx2 polygon without repeating its first point.""" xs = points[:, 0].astype(np.float64) ys = points[:, 1].astype(np.float64) return 0.5 * float(np.sum(xs * np.roll(ys, -1) - np.roll(xs, -1) * ys)) def triangulate_simple_polygon(points: np.ndarray) -> list[tuple[int, int, int]]: """Triangulate a simple image-space polygon with deterministic ear clipping.""" polygon = np.asarray(points, dtype=np.float64).reshape((-1, 2)) if len(polygon) < 3: raise ValueError("A polygon needs at least three vertices") orientation = 1.0 if polygon_signed_area(polygon) > 0 else -1.0 remaining = list(range(len(polygon))) triangles: list[tuple[int, int, int]] = [] def cross(a: np.ndarray, b: np.ndarray, c: np.ndarray) -> float: ab = b - a ac = c - a return float(ab[0] * ac[1] - ab[1] * ac[0]) def inside_triangle(point: np.ndarray, a: np.ndarray, b: np.ndarray, c: np.ndarray) -> bool: c1 = orientation * cross(a, b, point) c2 = orientation * cross(b, c, point) c3 = orientation * cross(c, a, point) return c1 >= -1e-8 and c2 >= -1e-8 and c3 >= -1e-8 guard = 0 while len(remaining) > 3: ear_found = False for offset, current in enumerate(remaining): previous = remaining[offset - 1] following = remaining[(offset + 1) % len(remaining)] a, b, c = polygon[[previous, current, following]] if orientation * cross(a, b, c) <= 1e-8: continue if any( inside_triangle(polygon[candidate], a, b, c) for candidate in remaining if candidate not in {previous, current, following} ): continue triangles.append((previous, current, following)) del remaining[offset] ear_found = True break guard += 1 if not ear_found or guard > len(polygon) * len(polygon): raise ValueError("Could not triangulate simplified curb contour") triangles.append(tuple(remaining)) return triangles def simplified_largest_contour( mask: np.ndarray, epsilon_ratio: float, ) -> tuple[np.ndarray, dict[str, Any]]: """Extract a simplified continuous outline, avoiding raster-staircase side walls.""" contours, _ = cv2.findContours( mask.astype(np.uint8), cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE, ) if not contours: raise ValueError("No target contour for raised curb") contour = max(contours, key=cv2.contourArea) perimeter = float(cv2.arcLength(contour, True)) epsilon = max(1.0, float(epsilon_ratio) * perimeter) simplified = cv2.approxPolyDP(contour, epsilon, True).reshape((-1, 2)).astype(np.float32) if len(simplified) < 3: raise ValueError("Raised-curb contour collapsed during simplification") return simplified, { "raw_contour_vertices": int(len(contour)), "contour_vertices": int(len(simplified)), "contour_perimeter_pixels": perimeter, "contour_epsilon_pixels": epsilon, "contour_epsilon_ratio": float(epsilon_ratio), } def build_raised_curb_prism_mesh( depth: np.ndarray, colors: np.ndarray, target_mask: np.ndarray, *, height: float | None, height_ratio: float, contour_epsilon_ratio: float, stride: int, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, dict[str, Any]]: """Build one smooth, vertically extruded curb body from a simplified mask contour.""" h, w = depth.shape valid = target_mask & np.isfinite(depth) & (depth > 0) if not np.any(valid): raise ValueError("No valid target depth values for raised curb") contour, contour_metadata = simplified_largest_contour( target_mask, contour_epsilon_ratio, ) triangles = triangulate_simple_polygon(contour) fx, fy, cx, cy = camera_intrinsics(w, h, None, None, None, None) points = pixels_to_points(depth, fx, fy, cx, cy) fallback_depth = float(np.median(depth[valid])) top_vertices: list[np.ndarray] = [] for x, y in contour: sampled_depth = sample_nearest_depth(depth, valid, float(x), float(y), fallback_depth) ray = np.array([(float(x) - cx) / fx, (float(y) - cy) / fy, 1.0], dtype=np.float32) top_vertices.append(ray * sampled_depth) scale = robust_extent_from_points(points[valid]) median_depth = float(np.median(depth[valid])) if height is not None: curb_height = float(height) height_source = "explicit_relative_height" else: # Curb height is a cross-section property, so it must not grow with # total curb length. Use relative depth and bounded vertical evidence # instead of the footprint diagonal, which is dominated by long curbs. automatic_height_cap = 0.35 * median_depth curb_height = min( max( float(height_ratio) * median_depth, 0.065 * median_depth, 1.15 * max(float(scale["span_y"]), 1e-3), ), automatic_height_cap, ) height_source = "median_depth_ratio_with_vertical_evidence_cap" top_array = np.asarray(top_vertices, dtype=np.float32) bottom_array = top_array.copy() bottom_array[:, 1] += curb_height target_colors = colors[valid].astype(np.float32) base_color = np.median(target_colors, axis=0) base_color = 0.84 * base_color + 0.16 * np.array([172.0, 172.0, 166.0], dtype=np.float32) top_color_array = np.repeat( np.clip(base_color, 0, 255).astype(np.uint8)[None, :], len(top_array), axis=0, ) bottom_color_array = np.clip(top_color_array.astype(np.float32) * 0.62, 0, 255).astype(np.uint8) vertices = np.concatenate([top_array, bottom_array], axis=0) vertex_colors = np.concatenate([top_color_array, bottom_color_array], axis=0) count = len(top_array) top_faces = list(triangles) bottom_faces = [(c + count, b + count, a + count) for a, b, c in triangles] side_faces: list[tuple[int, int, int]] = [] for a in range(count): b = (a + 1) % count # Reverse the shared top/bottom boundary directions relative to the # caps so the closed prism has consistent winding and stable normals. side_faces.append((b, a, a + count)) side_faces.append((b, a + count, b + count)) faces = np.asarray(top_faces + bottom_faces + side_faces, dtype=np.int64) metadata = { "front_faces": len(top_faces), "back_faces": len(bottom_faces), "side_faces": len(side_faces), "boundary_side_faces": len(side_faces), "wedge_support_faces": 0, "stair_sidewall_faces": 0, "stair_sidewall_bottom_faces": 0, "solid_thickness": curb_height, "max_depth_jump": None, "stride": stride, "support_mode": "raised_curb_prism", "stair_surface_mode": "not_applicable", "stair_rows": [], "stair_cross_samples": None, "stair_edge_slope": None, "stair_step_count": None, "surface_model": "raised_curb_contour_prism", "curb_height": curb_height, "curb_height_ratio": float(height_ratio), "curb_height_source": height_source, **contour_metadata, "depth_scaled_dimensions": { "height": curb_height, "source_span_x": float(scale["span_x"]), "source_span_y": float(scale["span_y"]), "source_span_z": float(scale["span_z"]), "footprint_diag": float(scale["footprint_diag"]), "median_depth": median_depth, "automatic_height_cap": 0.35 * median_depth, "depth_units": "relative_pseudo_depth", }, } return vertices, vertex_colors, faces, metadata def build_continuous_stair_mesh( depth: np.ndarray, colors: np.ndarray, target_mask: np.ndarray, *, stride: int, thickness: float | None, thickness_ratio: float, support_mode: str, wedge_base_drop_ratio: float, stair_edges_y: list[int] | None, stair_edge_slope: float, cross_samples: int, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, dict[str, Any]]: h, w = depth.shape valid = target_mask & np.isfinite(depth) & (depth > 0) depth_values = depth[valid] if depth_values.size == 0: raise ValueError("No valid target depth values") median_depth = float(np.median(depth_values)) depth_span = float(np.percentile(depth_values, 95) - np.percentile(depth_values, 5)) solid_thickness = float(thickness) if thickness is not None else max( 0.04 * median_depth, thickness_ratio * max(depth_span, 1e-3), ) mask_ys, mask_xs = np.nonzero(target_mask) if mask_ys.size == 0: raise ValueError("No target mask pixels") top_y = int(mask_ys.min()) bottom_y = int(mask_ys.max()) min_gap = max(3, int(round(stride * 0.75))) requested_rows = [top_y] requested_rows.extend(int(edge) for edge in (stair_edges_y or []) if top_y < int(edge) < bottom_y) requested_rows.append(bottom_y) requested_rows = sorted(requested_rows) row_centers: list[int] = [] for row in requested_rows: if not row_centers or abs(row - row_centers[-1]) >= min_gap: row_centers.append(int(row)) else: row_centers[-1] = int(round((row_centers[-1] + row) * 0.5)) if len(row_centers) < 2: row_centers = [top_y, bottom_y] bounds = [line_x_bounds_on_mask(float(row), stair_edge_slope, target_mask) for row in row_centers] if any(item is None for item in bounds): raise ValueError("Could not derive stair line bounds from target mask") typed_bounds = [item for item in bounds if item is not None] widest = max(right - left for left, right in typed_bounds) if cross_samples <= 0: cross_samples = int(np.clip(math.ceil(widest / max(stride, 1)) + 1, 12, 96)) cross_samples = max(4, int(cross_samples)) fx, fy, cx, cy = camera_intrinsics(w, h, None, None, None, None) image_x_center = (w - 1) / 2.0 vertices: list[np.ndarray] = [] vertex_colors: list[np.ndarray] = [] front_id = np.full((len(row_centers), cross_samples), -1, dtype=np.int64) back_id = np.full_like(front_id, -1) row_depths: list[np.ndarray] = [] row_pixels: list[list[tuple[float, float]]] = [] global_fallback = float(np.median(depth_values)) for row, (left, right) in zip(row_centers, typed_bounds): xs = np.linspace(float(left), float(right), cross_samples, dtype=np.float32) pixels = [(float(x), float(row + stair_edge_slope * (float(x) - image_x_center))) for x in xs] sampled = np.array( [sample_nearest_depth(depth, valid, x, y, global_fallback) for x, y in pixels], dtype=np.float32, ) row_depths.append(regularize_depth_row(sampled)) row_pixels.append(pixels) for row_index, (pixels, depths) in enumerate(zip(row_pixels, row_depths)): for col_index, ((x, y), z) in enumerate(zip(pixels, depths)): z = float(max(z, 1e-4)) point = np.array( [ (float(x) - cx) * z / fx, (float(y) - cy) * z / fy, z, ], dtype=np.float32, ) front_id[row_index, col_index] = len(vertices) vertices.append(point) vertex_colors.append(sample_nearest_color(colors, x, y)) back_depth = z + solid_thickness back_point = point * (back_depth / max(z, 1e-6)) back_id[row_index, col_index] = len(vertices) vertices.append(back_point.astype(np.float32)) vertex_colors.append(np.clip(vertex_colors[-1].astype(np.float32) * 0.54, 0, 255).astype(np.uint8)) front_faces: list[tuple[int, int, int]] = [] back_faces: list[tuple[int, int, int]] = [] for row_index in range(len(row_centers) - 1): for col_index in range(cross_samples - 1): a = int(front_id[row_index, col_index]) b = int(front_id[row_index, col_index + 1]) c = int(front_id[row_index + 1, col_index]) d = int(front_id[row_index + 1, col_index + 1]) ba = int(back_id[row_index, col_index]) bb = int(back_id[row_index, col_index + 1]) bc = int(back_id[row_index + 1, col_index]) bd = int(back_id[row_index + 1, col_index + 1]) front_faces.append((a, c, b)) front_faces.append((b, c, d)) back_faces.append((bb, bc, ba)) back_faces.append((bd, bc, bb)) boundary_side_faces: list[tuple[int, int, int]] = [] def append_backed_strip(front_a: int, front_b: int, back_a: int, back_b: int) -> None: boundary_side_faces.append((front_a, front_b, back_a)) boundary_side_faces.append((front_b, back_b, back_a)) last_row = len(row_centers) - 1 last_col = cross_samples - 1 for col_index in range(cross_samples - 1): append_backed_strip( int(front_id[0, col_index]), int(front_id[0, col_index + 1]), int(back_id[0, col_index]), int(back_id[0, col_index + 1]), ) append_backed_strip( int(front_id[last_row, col_index + 1]), int(front_id[last_row, col_index]), int(back_id[last_row, col_index + 1]), int(back_id[last_row, col_index]), ) for row_index in range(len(row_centers) - 1): append_backed_strip( int(front_id[row_index + 1, 0]), int(front_id[row_index, 0]), int(back_id[row_index + 1, 0]), int(back_id[row_index, 0]), ) append_backed_strip( int(front_id[row_index, last_col]), int(front_id[row_index + 1, last_col]), int(back_id[row_index, last_col]), int(back_id[row_index + 1, last_col]), ) stair_sidewall_faces: list[tuple[int, int, int]] = [] stair_sidewall_bottom_faces: list[tuple[int, int, int]] = [] if support_mode == "stair_sidewall": front_points = np.asarray([vertices[int(front_id[row_index, col])] for row_index in range(len(row_centers)) for col in (0, last_col)], dtype=np.float32) base_y = float(front_points[:, 1].max()) + max( solid_thickness * 1.5, wedge_base_drop_ratio * max(median_depth, 1e-3), 1e-3, ) def make_bottom(top_id: int) -> int: bottom = vertices[top_id].astype(np.float32).copy() bottom[1] = base_y bottom_id = len(vertices) vertices.append(bottom) vertex_colors.append(np.clip(vertex_colors[top_id].astype(np.float32) * 0.70, 0, 255).astype(np.uint8)) return bottom_id left_ids = [int(front_id[row_index, 0]) for row_index in range(len(row_centers))] right_ids = [int(front_id[row_index, last_col]) for row_index in range(len(row_centers))] left_bottom = [make_bottom(vertex_id) for vertex_id in left_ids] right_bottom = [make_bottom(vertex_id) for vertex_id in right_ids] for top_a, top_b, bottom_a, bottom_b in zip(left_ids[:-1], left_ids[1:], left_bottom[:-1], left_bottom[1:]): stair_sidewall_faces.append((top_a, top_b, bottom_a)) stair_sidewall_faces.append((top_b, bottom_b, bottom_a)) for top_a, top_b, bottom_a, bottom_b in zip(right_ids[:-1], right_ids[1:], right_bottom[:-1], right_bottom[1:]): stair_sidewall_faces.append((top_a, bottom_a, top_b)) stair_sidewall_faces.append((top_b, bottom_a, bottom_b)) for left_a, left_b, right_a, right_b in zip(left_bottom[:-1], left_bottom[1:], right_bottom[:-1], right_bottom[1:]): stair_sidewall_bottom_faces.append((left_a, right_a, left_b)) stair_sidewall_bottom_faces.append((left_b, right_a, right_b)) side_faces = boundary_side_faces + stair_sidewall_faces + stair_sidewall_bottom_faces face_rows = front_faces + back_faces + side_faces faces = ( np.asarray(face_rows, dtype=np.int64).reshape((-1, 3)) if face_rows else np.empty((0, 3), dtype=np.int64) ) metadata = { "front_faces": len(front_faces), "back_faces": len(back_faces), "side_faces": len(side_faces), "boundary_side_faces": len(boundary_side_faces), "wedge_support_faces": 0, "stair_sidewall_faces": len(stair_sidewall_faces), "stair_sidewall_bottom_faces": len(stair_sidewall_bottom_faces), "solid_thickness": solid_thickness, "max_depth_jump": None, "stride": stride, "support_mode": support_mode, "stair_surface_mode": "continuous_step_strips", "stair_rows": row_centers, "stair_edge_slope": float(stair_edge_slope), "stair_cross_samples": int(cross_samples), } return np.asarray(vertices, dtype=np.float32), np.asarray(vertex_colors, dtype=np.uint8), faces, metadata def infer_stair_edge_rows( depth: np.ndarray, colors: np.ndarray, target_mask: np.ndarray, stair_edges_y: list[int] | None, ) -> tuple[list[int], dict[str, Any]]: """Filter duplicate/noisy tread candidates without inventing extra stairs. The upstream detector intentionally has high recall. A presentation mesh must not interpret every weak or repeated line as a physical riser. This filter combines target support, RGB horizontal-edge evidence, and depth discontinuity evidence, then merges the characteristic clusters produced when one physical edge is detected several times. """ mask_ys, _ = np.nonzero(target_mask) if mask_ys.size == 0: return [], { "policy": "empty_target", "candidate_rows": [], "retained_rows": [], "confidence": 0.0, } top_y = int(mask_ys.min()) bottom_y = int(mask_ys.max()) candidates = sorted( { int(row) for row in (stair_edges_y or []) if top_y <= int(row) <= bottom_y } ) if not candidates: return [], { "policy": "no_supported_detector_rows", "candidate_rows": [], "retained_rows": [], "confidence": 0.0, } gray = cv2.cvtColor(colors, cv2.COLOR_RGB2GRAY).astype(np.float32) / 255.0 rgb_gradient = np.abs(cv2.Sobel(gray, cv2.CV_32F, 0, 1, ksize=3)) depth_f = depth.astype(np.float32).copy() finite_depth = np.isfinite(depth_f) & (depth_f > 0) replacement = ( float(np.median(depth_f[finite_depth])) if np.any(finite_depth) else 1.0 ) depth_f[~finite_depth] = replacement depth_gradient = np.abs(cv2.Sobel(depth_f, cv2.CV_32F, 0, 1, ksize=3)) rgb_reference = max(float(np.percentile(rgb_gradient[target_mask], 75)), 1e-4) depth_reference = max( float(np.percentile(depth_gradient[target_mask], 75)), 1e-4, ) band_radius = max(2, int(round(target_mask.shape[0] / 512.0))) evidence: list[dict[str, Any]] = [] for row in candidates: row_slice = slice( max(0, row - band_radius), min(target_mask.shape[0], row + band_radius + 1), ) support = target_mask[row_slice] rgb_values = rgb_gradient[row_slice][support] depth_values = depth_gradient[row_slice][support] evidence.append( { "row": row, "support_pixels": int(support.sum()), "rgb_strength": ( float(np.percentile(rgb_values, 75)) / rgb_reference if rgb_values.size else 0.0 ), "depth_strength": ( float(np.percentile(depth_values, 75)) / depth_reference if depth_values.size else 0.0 ), } ) maximum_support = max(item["support_pixels"] for item in evidence) for item in evidence: support_ratio = item["support_pixels"] / max(maximum_support, 1) item["support_ratio"] = float(support_ratio) item["score"] = float( 0.50 * min(item["rgb_strength"], 3.0) / 3.0 + 0.30 * min(item["depth_strength"], 3.0) / 3.0 + 0.20 * support_ratio ) supported = [ item for item in evidence if ( item["support_ratio"] >= 0.12 and item["score"] >= 0.16 and max(item["rgb_strength"], item["depth_strength"]) >= 0.25 ) ] if not supported: return [], { "policy": "insufficient_rgb_depth_edge_evidence", "candidate_rows": candidates, "evidence": evidence, "retained_rows": [], "confidence": 0.0, "step_count_policy": ( "no_physical_step_count_inferred_from_unsupported_rows" ), } supported.sort(key=lambda item: item["row"]) # A detector often returns two or three parallel lines for one stair nose. # When the row gaps are clearly bimodal, small-gap runs are one physical # edge and large gaps separate adjacent physical steps. rows = [int(item["row"]) for item in supported] gap_policy = "non_maximum_suppression" if len(rows) >= 5: gaps = np.diff(rows).astype(np.float32) sorted_gaps = np.sort(gaps) ratios = sorted_gaps[1:] / np.maximum(sorted_gaps[:-1], 1.0) split_index = int(np.argmax(ratios)) if ratios.size else -1 split_ratio = float(ratios[split_index]) if split_index >= 0 else 1.0 threshold = ( 0.5 * float(sorted_gaps[split_index] + sorted_gaps[split_index + 1]) if split_index >= 0 else 0.0 ) large_gap_count = int(np.sum(gaps > threshold)) if threshold > 0 else 0 small_gap_count = int(np.sum(gaps <= threshold)) if threshold > 0 else 0 if split_ratio >= 1.8 and large_gap_count >= 2 and small_gap_count >= 2: groups: list[list[dict[str, Any]]] = [[supported[0]]] for gap, item in zip(gaps, supported[1:]): if float(gap) > threshold: groups.append([item]) else: groups[-1].append(item) merged: list[dict[str, Any]] = [] for group in groups: weights = np.asarray( [max(float(item["score"]), 1e-4) for item in group], dtype=np.float64, ) merged_row = int( round( np.average( [float(item["row"]) for item in group], weights=weights, ) ) ) merged.append( { **max(group, key=lambda item: item["score"]), "row": merged_row, "merged_rows": [int(item["row"]) for item in group], } ) supported = merged gap_policy = "bimodal_gap_cluster_merge" if gap_policy == "non_maximum_suppression": minimum_separation = max(4, int(round(0.055 * max(bottom_y - top_y, 1)))) retained: list[dict[str, Any]] = [] for item in sorted(supported, key=lambda value: value["score"], reverse=True): if all( abs(int(item["row"]) - int(other["row"])) >= minimum_separation for other in retained ): retained.append(item) supported = sorted(retained, key=lambda item: item["row"]) retained_rows = [int(item["row"]) for item in supported] return retained_rows, { "policy": gap_policy, "candidate_rows": candidates, "evidence": evidence, "retained_rows": retained_rows, "confidence": ( float(np.mean([item["score"] for item in supported])) if supported else 0.0 ), "step_count_policy": "one_retained_physical_riser_row_equals_one_step", } def trusted_stair_edges_from_manifest( geometry_manifest: dict[str, Any], *, minimum_confidence: float = 0.35, ) -> tuple[list[int], dict[str, Any]]: """Reject detector rows whose provenance cannot support a physical count.""" raw_rows = [ int(value) for value in geometry_manifest.get("stair_edges_y", []) ] used_regular_fallback = bool( geometry_manifest.get("used_regular_fallback_stair_edges", False) ) confidence_value = geometry_manifest.get("stair_edge_confidence") confidence = ( float(confidence_value) if confidence_value is not None else None ) source = ( geometry_manifest.get("stair_edge_source") or geometry_manifest.get("edge_source") or ( "regular_fallback" if used_regular_fallback else "detector_rows_with_manifest_confidence" ) ) if used_regular_fallback: trusted_rows: list[int] = [] policy = "reject_regular_fallback_rows_for_physical_step_count" elif confidence is None: trusted_rows = [] policy = "reject_rows_without_detector_confidence" elif confidence < minimum_confidence: trusted_rows = [] policy = "reject_low_confidence_detector_rows" else: trusted_rows = raw_rows policy = "accept_detector_rows_for_rgb_depth_filtering" return trusted_rows, { "raw_rows": raw_rows, "trusted_rows": trusted_rows, "used_regular_fallback": used_regular_fallback, "detector_confidence": confidence, "minimum_confidence": float(minimum_confidence), "edge_source": str(source), "policy": policy, } def stair_source_band_bounds( rows: list[int], top_y: int, bottom_y: int, ) -> list[int]: """Turn physical stair-edge centers into one source-texture band per step.""" if not rows: return [top_y, bottom_y] if len(rows) == 1: half_span = max(2, int(round(0.20 * max(bottom_y - top_y, 1)))) return [max(top_y, rows[0] - half_span), min(bottom_y, rows[0] + half_span)] midpoints = [ int(round(0.5 * (left + right))) for left, right in zip(rows[:-1], rows[1:]) ] first_gap = max(rows[1] - rows[0], 2) last_gap = max(rows[-1] - rows[-2], 2) bounds = [ max(top_y, int(round(rows[0] - 0.5 * first_gap))), *midpoints, min(bottom_y, int(round(rows[-1] + 0.5 * last_gap))), ] for index in range(1, len(bounds)): bounds[index] = max(bounds[index], bounds[index - 1] + 1) if bounds[-1] > bottom_y: return ( np.linspace(top_y, bottom_y, len(rows) + 1) .round() .astype(int) .tolist() ) return bounds def estimate_stair_dimensions( depth: np.ndarray, target_mask: np.ndarray, stair_rows: list[int], step_count: int, *, metric_style_depth: bool, ) -> dict[str, Any]: """Estimate staircase dimensions from depth points with conservative bounds.""" h, w = depth.shape fx, fy, cx, cy = camera_intrinsics(w, h, None, None, None, None) points = pixels_to_points(depth, fx, fy, cx, cy) scale = depth_scale_for_mask(depth, target_mask) row_points: list[np.ndarray] = [] for row in stair_rows: radius = max(2, int(round(h / 512.0))) row_mask = np.zeros_like(target_mask) row_mask[max(0, row - radius) : min(h, row + radius + 1)] = True valid = row_mask & target_mask & np.isfinite(depth) & (depth > 0) if np.any(valid): row_points.append(np.median(points[valid], axis=0)) raw_treads: list[float] = [] raw_rises: list[float] = [] for left, right in zip(row_points[:-1], row_points[1:]): raw_treads.append(abs(float(right[2] - left[2]))) raw_rises.append(abs(float(right[1] - left[1]))) count = max(int(step_count), 1) positive_treads = [value for value in raw_treads if value > 1e-6] positive_rises = [value for value in raw_rises if value > 1e-6] raw_tread = ( float(np.median(positive_treads)) if positive_treads else 0.0 ) raw_rise = ( float(np.median(positive_rises)) if positive_rises else 0.0 ) median_depth = max(float(scale["median_z"]), 1e-3) depth_discontinuity_threshold = ( max(0.03, 0.01 * median_depth) if metric_style_depth else 0.01 * median_depth ) depth_edge_evidence = bool( len(row_points) >= 2 and raw_tread >= depth_discontinuity_threshold ) if metric_style_depth: width = float(np.clip(float(scale["span_x"]), 0.75, 6.0)) units = "metric_style_depth_meters_not_calibrated_truth" if depth_edge_evidence: tread_depth = float(np.clip(raw_tread, 0.22, 0.55)) # Camera Y is not gravity-aligned without a calibrated camera pose. # It is therefore only bounded supporting evidence, not a direct # measurement of riser height. rise = float(np.clip(raw_rise, 0.08, 0.22)) source = ( "depth_edge_points_with_accessibility_plausibility_bounds" ) confidence = "medium" scale_status = ( "metric_style_model_with_unverified_monocular_scale" ) else: tread_depth = 0.30 rise = 0.15 source = ( "category_prior_due_to_insufficient_depth_discontinuity" ) confidence = "low" scale_status = ( "category_prior_fallback_no_metric_depth_edge_evidence" ) else: width = float( np.clip( float(scale["span_x"]), 0.18 * median_depth, 1.5 * median_depth, ) ) units = "relative_pseudo_depth" if depth_edge_evidence: tread_depth = float( np.clip( raw_tread, 0.04 * median_depth, 0.22 * median_depth, ) ) rise = float( np.clip( raw_rise, 0.018 * median_depth, 0.09 * median_depth, ) ) source = ( "relative_depth_edge_points_with_scale_relative_bounds" ) confidence = "low" scale_status = "relative_monocular_scale_only" else: tread_depth = 0.10 * median_depth rise = 0.045 * median_depth source = ( "scale_relative_category_prior_due_to_insufficient_depth_discontinuity" ) confidence = "low" scale_status = ( "relative_category_prior_no_depth_edge_evidence" ) total_length = tread_depth * count total_height = rise * count return { "width": width, "length": total_length, "height": total_height, "tread_depth": tread_depth, "riser_height": rise, "raw_tread_depth": raw_tread, "raw_riser_height": raw_rise, "row_point_count": len(row_points), "row_pair_count": max(len(row_points) - 1, 0), "depth_edge_evidence": depth_edge_evidence, "depth_discontinuity_threshold": depth_discontinuity_threshold, "dimension_confidence": confidence, "source_span_x": float(scale["span_x"]), "source_span_y": float(scale["span_y"]), "source_span_z": float(scale["span_z"]), "footprint_diag": float(scale["footprint_diag"]), "valid_points": int(scale["valid_points"]), "depth_units": units, "dimension_source": source, "physical_scale_status": scale_status, "intrinsics_source": "default_focal_equals_image_extent", "gravity_alignment": "unavailable", "camera_pose_assumption": ( "approximately_level_camera_only_for_bounded_projected_vertical_evidence" ), "calibrated_metric_truth": False, } def build_canonical_staircase_prism_mesh( depth: np.ndarray, colors: np.ndarray, target_mask: np.ndarray, *, stride: int, thickness: float | None, thickness_ratio: float, stair_edges_y: list[int] | None, step_count: int, stair_edge_slope: float = 0.0, metric_style_depth: bool = False, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, dict[str, Any]]: mask_ys, mask_xs = np.nonzero(target_mask) if mask_ys.size == 0: raise ValueError("No target mask pixels") h, w = target_mask.shape bbox_width = max(int(mask_xs.max() - mask_xs.min()), 1) bbox_height = max(int(mask_ys.max() - mask_ys.min()), 1) filtered_rows, stair_inference = infer_stair_edge_rows( depth, colors, target_mask, stair_edges_y, ) if step_count > 0: inferred_steps = step_count step_count_source = "explicit_review_override" elif filtered_rows: inferred_steps = len(filtered_rows) step_count_source = "filtered_rgb_depth_riser_evidence" else: inferred_steps = 1 step_count_source = "conservative_single_step_due_to_missing_edge_evidence" inferred_steps = int(np.clip(inferred_steps, 1, 18)) dimensions = estimate_stair_dimensions( depth, target_mask, filtered_rows, inferred_steps, metric_style_depth=metric_style_depth, ) target_colors = colors[target_mask] if target_colors.size == 0: base_color = np.array([176, 176, 168], dtype=np.float32) else: base_color = np.median(target_colors.astype(np.float32), axis=0) neutral = np.array([178, 178, 170], dtype=np.float32) base_color = 0.45 * base_color + 0.55 * neutral total_length = float(dimensions["length"]) total_width = float(dimensions["width"]) total_height = float(dimensions["height"]) side_drop = float(thickness) if thickness is not None else max( float(dimensions["riser_height"]) * 0.35, total_height * max(thickness_ratio * 0.30, 0.025), ) tread_depth = total_length / inferred_steps rise = total_height / inferred_steps x_left = -0.5 * total_width x_right = 0.5 * total_width vertices: list[np.ndarray] = [] vertex_colors: list[np.ndarray] = [] faces: list[tuple[int, int, int]] = [] front_faces = 0 side_faces = 0 def scene_to_camera(point: tuple[float, float, float]) -> np.ndarray: x, y, z = point return np.array([x, -z, y], dtype=np.float32) def add_vertex(point: tuple[float, float, float], color: np.ndarray) -> int: vertex_id = len(vertices) vertices.append(scene_to_camera(point)) vertex_colors.append(np.clip(color, 0, 255).astype(np.uint8)) return vertex_id def add_quad( a: tuple[float, float, float], b: tuple[float, float, float], c: tuple[float, float, float], d: tuple[float, float, float], color: np.ndarray, *, front: bool, ) -> None: nonlocal front_faces, side_faces ids = [add_vertex(point, color) for point in (a, b, c, d)] faces.append((ids[0], ids[1], ids[2])) faces.append((ids[0], ids[2], ids[3])) if front: front_faces += 2 else: side_faces += 2 texture_cross_samples = int( np.clip(math.ceil(bbox_width / max(float(stride), 1.0)), 24, 160) ) texture_depth_samples = int( np.clip( math.ceil( bbox_height / max(float(inferred_steps * max(stride, 1)), 1.0) ), 6, 32, ) ) def add_textured_quad( a: tuple[float, float, float], b: tuple[float, float, float], c: tuple[float, float, float], d: tuple[float, float, float], *, source_y0: int, source_y1: int, shade: float, ) -> None: nonlocal front_faces corner_a = np.asarray(a, dtype=np.float32) corner_b = np.asarray(b, dtype=np.float32) corner_c = np.asarray(c, dtype=np.float32) corner_d = np.asarray(d, dtype=np.float32) grid = np.empty( (texture_depth_samples + 1, texture_cross_samples + 1), dtype=np.int64, ) for depth_index in range(texture_depth_samples + 1): v = depth_index / texture_depth_samples left = (1.0 - v) * corner_a + v * corner_d right = (1.0 - v) * corner_b + v * corner_c source_y = int( np.clip( round((1.0 - v) * source_y0 + v * source_y1), 0, h - 1, ) ) for cross_index in range(texture_cross_samples + 1): u = cross_index / texture_cross_samples point = (1.0 - u) * left + u * right line_bounds = line_x_bounds_on_mask( float(source_y), float(stair_edge_slope), target_mask, ) source_left, source_right = ( line_bounds if line_bounds is not None else (int(mask_xs.min()), int(mask_xs.max())) ) source_x = int( np.clip( round( source_left + u * ( source_right - source_left ) ), 0, w - 1, ) ) color = colors[source_y, source_x].astype(np.float32) * shade grid[depth_index, cross_index] = add_vertex( tuple(float(value) for value in point), color, ) for depth_index in range(texture_depth_samples): for cross_index in range(texture_cross_samples): top_left = int(grid[depth_index, cross_index]) top_right = int(grid[depth_index, cross_index + 1]) bottom_left = int(grid[depth_index + 1, cross_index]) bottom_right = int( grid[depth_index + 1, cross_index + 1] ) faces.append((top_left, top_right, bottom_right)) faces.append((top_left, bottom_right, bottom_left)) front_faces += 2 row_edges = filtered_rows if step_count <= 0 and row_edges: row_bounds = stair_source_band_bounds( row_edges, int(mask_ys.min()), int(mask_ys.max()), ) else: row_bounds = np.linspace(int(mask_ys.min()), int(mask_ys.max()), inferred_steps + 1).round().astype(int).tolist() if len(row_bounds) < inferred_steps + 1: row_bounds = np.linspace(int(mask_ys.min()), int(mask_ys.max()), inferred_steps + 1).round().astype(int).tolist() def step_color(index: int, shade: float) -> np.ndarray: start = row_bounds[min(index, len(row_bounds) - 2)] end = row_bounds[min(index + 1, len(row_bounds) - 1)] y0, y1 = sorted((start, end)) band = np.zeros_like(target_mask, dtype=bool) center_x = (w - 1) * 0.5 ys, xs = np.indices(target_mask.shape, dtype=np.float32) aligned_y = ys - float(stair_edge_slope) * (xs - center_x) band = (aligned_y >= y0) & (aligned_y <= y1) & target_mask band_colors = colors[band] color = base_color if band_colors.size == 0 else 0.55 * np.median(band_colors.astype(np.float32), axis=0) + 0.45 * base_color return np.clip(color * shade, 0, 255) for step_index in range(inferred_steps): y0 = step_index * tread_depth y1 = (step_index + 1) * tread_depth z0 = step_index * rise z1 = (step_index + 1) * rise local_bottom_z = z0 - side_drop # Image rows run far/top -> near/bottom, while scene step index zero is # the near/lowest step. Reverse the band order to preserve perspective. source_band_index = inferred_steps - 1 - step_index side_color = step_color(source_band_index, 0.66) source_y_near = row_bounds[ min(source_band_index + 1, len(row_bounds) - 1) ] source_y_far = row_bounds[ min(source_band_index, len(row_bounds) - 2) ] add_textured_quad( (x_left, y0, z1), (x_right, y0, z1), (x_right, y1, z1), (x_left, y1, z1), source_y0=source_y_near, source_y1=source_y_far, shade=1.02, ) add_textured_quad( (x_left, y0, z0), (x_right, y0, z0), (x_right, y0, z1), (x_left, y0, z1), source_y0=source_y_near, source_y1=source_y_far, shade=0.82, ) # Each tread receives only a shallow local support skirt. A single # global bottom would turn a scene surface into a pyramid-like object # when viewed from the side. add_quad( (x_left, y0, local_bottom_z), (x_left, y0, z1), (x_left, y1, z1), (x_left, y1, local_bottom_z), side_color, front=False, ) add_quad( (x_right, y0, z1), (x_right, y0, local_bottom_z), (x_right, y1, local_bottom_z), (x_right, y1, z1), side_color, front=False, ) add_quad( (x_left, y0, local_bottom_z), (x_right, y0, local_bottom_z), (x_right, y0, z0), (x_left, y0, z0), side_color * 0.92, front=False, ) add_quad( (x_left, y1, local_bottom_z), (x_right, y1, local_bottom_z), (x_right, y0, local_bottom_z), (x_left, y0, local_bottom_z), side_color * 0.78, front=False, ) top_y = total_length top_z = total_height back_color = np.clip(base_color * 0.70, 0, 255) final_bottom_z = total_height - rise - side_drop add_quad( (x_right, top_y, final_bottom_z), (x_left, top_y, final_bottom_z), (x_left, top_y, top_z), (x_right, top_y, top_z), back_color, front=False, ) # Keep enough front and top landing to communicate that this is an # accessibility scene surface, not a freestanding manufactured object. if metric_style_depth: front_landing = max(0.75, 2.0 * tread_depth) top_landing = max(0.60, 1.5 * tread_depth) else: front_landing = max(2.0 * tread_depth, 0.50 * total_length) top_landing = max(1.5 * tread_depth, 0.35 * total_length) add_textured_quad( (x_left, -front_landing, 0.0), (x_right, -front_landing, 0.0), (x_right, 0.0, 0.0), (x_left, 0.0, 0.0), source_y0=int(mask_ys.max()), source_y1=row_bounds[-1], shade=1.0, ) add_textured_quad( (x_left, top_y, top_z), (x_right, top_y, top_z), (x_right, top_y + top_landing, top_z), (x_left, top_y + top_landing, top_z), source_y0=row_bounds[0], source_y1=int(mask_ys.min()), shade=0.98, ) metadata = { "front_faces": front_faces, "back_faces": 0, "side_faces": side_faces, "boundary_side_faces": side_faces, "wedge_support_faces": 0, "stair_sidewall_faces": side_faces, "stair_sidewall_bottom_faces": 2 * inferred_steps, "solid_thickness": side_drop, "max_depth_jump": None, "stride": stride, "support_mode": "open_world_local_support", "stair_surface_mode": "depth_scaled_staircase_prism", "stair_step_count": inferred_steps, "stair_step_count_source": step_count_source, "stair_inference": stair_inference, "stair_rows": row_bounds, "stair_physical_edge_rows": filtered_rows, "stair_cross_samples": texture_cross_samples, "stair_texture_depth_samples": texture_depth_samples, "texture_policy": "dense per-vertex frozen completed RGB sampled inside each perspective-aware source stair band", "texture_band_order": "near_low_step_uses_bottom_image_band", "texture_coordinate_policy": ( "image_x_maps_to_scene_width_and_image_y_maps_to_tread_depth_or_riser_height" ), "stair_edge_slope": float(stair_edge_slope), "depth_scaled_dimensions": { **dimensions, "side_drop": side_drop, }, "scene_semantics": "open_world_accessibility_surface_patch", "context_landings": { "front_length": front_landing, "top_length": top_landing, "purpose": "scene_context_not_navigability_proof", }, "automatic_passability_claim": False, "mesh_topology_role": "renderable_surface_patch_not_collision_mesh", "watertight_collision_mesh": False, } return np.asarray(vertices, dtype=np.float32), np.asarray(vertex_colors, dtype=np.uint8), np.asarray(faces, dtype=np.int64), metadata def build_solid_mesh( depth: np.ndarray, colors: np.ndarray, target_mask: np.ndarray, *, stride: int, max_depth_jump: float | None, thickness: float | None, thickness_ratio: float, support_mode: str, wedge_base_drop_ratio: float, category: str = "unknown", stair_edges_y: list[int] | None = None, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, dict[str, Any]]: h, w = depth.shape fx, fy, cx, cy = camera_intrinsics(w, h, None, None, None, None) points = pixels_to_points(depth, fx, fy, cx, cy) valid = target_mask & np.isfinite(depth) & (depth > 0) depth_values = depth[valid] if depth_values.size == 0: raise ValueError("No valid target depth values") depth_span = float(np.percentile(depth_values, 95) - np.percentile(depth_values, 5)) median_depth = float(np.median(depth_values)) mesh_jump = float(max_depth_jump) if max_depth_jump is not None else max(0.03, 0.08 * max(depth_span, 1e-3)) top_points = points[valid] scale = robust_extent_from_points(top_points) category_drop_ratio = { "tactile_paving": 0.035, "curb_cut": 0.060, "ramp": 0.080, "walkway": 0.055, }.get(category, 0.060) category_median_ratio = { "tactile_paving": 0.010, "curb_cut": 0.018, "ramp": 0.024, "walkway": 0.016, }.get(category, 0.018) footprint_drop = category_drop_ratio * max(float(scale["footprint_diag"]), 1e-3) depth_drop = max(thickness_ratio * 0.55, category_drop_ratio) * max(depth_span, 1e-3) median_drop = category_median_ratio * max(median_depth, 1e-3) solid_thickness = float(thickness) if thickness is not None else max(footprint_drop, depth_drop, median_drop) vertical_drop = solid_thickness if support_mode == "vertical_slab" else max(solid_thickness, wedge_base_drop_ratio * max(median_depth, 1e-3)) if top_points.size: top_y_high = float(np.percentile(top_points[:, 1], 98)) else: top_y_high = 0.0 vertical_base_y = top_y_high + vertical_drop ys = np.arange(0, h, stride) xs = np.arange(0, w, stride) front_id = -np.ones((len(ys), len(xs)), dtype=np.int64) back_id = -np.ones((len(ys), len(xs)), dtype=np.int64) vertices: list[np.ndarray] = [] vertex_colors: list[np.ndarray] = [] for iy, y in enumerate(ys): for ix, x in enumerate(xs): if not valid[y, x]: continue point = points[y, x].astype(np.float32) if not np.all(np.isfinite(point)) or point[2] <= 0: continue front_id[iy, ix] = len(vertices) vertices.append(point) vertex_colors.append(colors[y, x]) if support_mode == "vertical_slab": back_point = point.copy() back_point[1] = max(vertical_base_y, float(point[1] + 0.35 * vertical_drop)) else: back_depth = float(point[2] + solid_thickness) back_point = point * (back_depth / max(float(point[2]), 1e-6)) back_id[iy, ix] = len(vertices) vertices.append(back_point.astype(np.float32)) shade = 0.62 if support_mode == "vertical_slab" else 0.50 vertex_colors.append(np.clip(colors[y, x].astype(np.float32) * shade, 0, 255).astype(np.uint8)) front_faces: list[tuple[int, int, int]] = [] back_faces: list[tuple[int, int, int]] = [] edge_counts: defaultdict[tuple[int, int], int] = defaultdict(int) def add_front(face: tuple[int, int, int]) -> None: front_faces.append(face) for a, b in ((face[0], face[1]), (face[1], face[2]), (face[2], face[0])): edge_counts[tuple(sorted((a, b)))] += 1 for iy in range(len(ys) - 1): for ix in range(len(xs) - 1): ids = [ front_id[iy, ix], front_id[iy, ix + 1], front_id[iy + 1, ix], front_id[iy + 1, ix + 1], ] bids = [ back_id[iy, ix], back_id[iy, ix + 1], back_id[iy + 1, ix], back_id[iy + 1, ix + 1], ] if min(ids) < 0: continue z = np.array([vertices[i][2] for i in ids], dtype=np.float32) if float(z.max() - z.min()) > mesh_jump: continue add_front((ids[0], ids[2], ids[1])) add_front((ids[1], ids[2], ids[3])) back_faces.append((bids[1], bids[2], bids[0])) back_faces.append((bids[3], bids[2], bids[1])) boundary_side_faces: list[tuple[int, int, int]] = [] for (a, b), count in edge_counts.items(): if count != 1: continue a_back = a + 1 if a % 2 == 0 else a - 1 b_back = b + 1 if b % 2 == 0 else b - 1 if a_back < 0 or b_back < 0 or a_back >= len(vertices) or b_back >= len(vertices): continue boundary_side_faces.append((a, b, a_back)) boundary_side_faces.append((b, b_back, a_back)) wedge_support_faces: list[tuple[int, int, int]] = [] stair_sidewall_faces: list[tuple[int, int, int]] = [] stair_sidewall_bottom_faces: list[tuple[int, int, int]] = [] def append_triangular_side_panel(side_ids: list[int]) -> tuple[int, int] | None: ordered_ids: list[int] = [] for vertex_id in side_ids: if vertex_id < 0: continue if not ordered_ids or ordered_ids[-1] != int(vertex_id): ordered_ids.append(int(vertex_id)) if len(ordered_ids) < 2: return None ordered_ids = sorted( ordered_ids, key=lambda vertex_id: float(vertices[vertex_id][1]), ) side_points = np.asarray([vertices[vertex_id] for vertex_id in ordered_ids], dtype=np.float32) upper_id = ordered_ids[0] lower_id = ordered_ids[-1] upper_point = vertices[upper_id].astype(np.float32) lower_point = vertices[lower_id].astype(np.float32) upper_base_point = upper_point.copy() minimum_drop = max( solid_thickness * 1.5, wedge_base_drop_ratio * max(median_depth, 1e-3), 1e-3, ) upper_base_point[1] = max(float(lower_point[1]), float(upper_point[1] + minimum_drop)) base_color = np.clip( np.asarray([vertex_colors[vertex_id] for vertex_id in ordered_ids], dtype=np.float32).mean(axis=0) * 0.68, 0, 255, ).astype(np.uint8) bottom_ids: list[int] = [] denominator = max(len(ordered_ids) - 1, 1) for index, top_id in enumerate(ordered_ids): if top_id == lower_id: bottom_ids.append(lower_id) continue t = index / denominator bottom_point = ((1.0 - t) * upper_base_point + t * lower_point).astype(np.float32) bottom_color = np.clip( (1.0 - t) * base_color.astype(np.float32) + t * vertex_colors[lower_id].astype(np.float32) * 0.72, 0, 255, ).astype(np.uint8) bottom_id = len(vertices) vertices.append(bottom_point) vertex_colors.append(bottom_color) bottom_ids.append(bottom_id) for top_a, top_b, bottom_a, bottom_b in zip( ordered_ids[:-1], ordered_ids[1:], bottom_ids[:-1], bottom_ids[1:], ): if len({top_a, top_b, bottom_a}) == 3: wedge_support_faces.append((top_a, top_b, bottom_a)) if len({top_b, bottom_b, bottom_a}) == 3: wedge_support_faces.append((top_b, bottom_b, bottom_a)) return bottom_ids[0], lower_id def collect_side_rows(side: str) -> list[tuple[int, int]]: side_rows: list[tuple[int, int]] = [] for iy, y in enumerate(ys): valid_ix = np.flatnonzero(front_id[iy] >= 0) if valid_ix.size == 0: continue ix = int(valid_ix[0] if side == "left" else valid_ix[-1]) side_rows.append((int(y), int(front_id[iy, ix]))) return side_rows def thin_side_profile(side_rows: list[tuple[int, int]]) -> list[tuple[int, int]]: if len(side_rows) <= 2: return side_rows edge_rows = sorted( {int(edge) for edge in stair_edges_y or [] if 0 <= int(edge) < h} ) if not edge_rows: return side_rows selected_indices = {0, len(side_rows) - 1} row_values = np.asarray([row for row, _ in side_rows], dtype=np.int32) for edge in edge_rows: for offset in (-stride, 0, stride): index = int(np.argmin(np.abs(row_values - (edge + offset)))) selected_indices.add(index) return [side_rows[index] for index in sorted(selected_indices)] def append_stair_sidewall( left_rows: list[tuple[int, int]], right_rows: list[tuple[int, int]], ) -> tuple[int, int]: if len(left_rows) < 2 or len(right_rows) < 2: return 0, 0 by_y_left = {row: vertex_id for row, vertex_id in thin_side_profile(left_rows)} by_y_right = {row: vertex_id for row, vertex_id in thin_side_profile(right_rows)} common_rows = sorted(set(by_y_left) & set(by_y_right)) if len(common_rows) < 2: return 0, 0 left_ids = [by_y_left[row] for row in common_rows] right_ids = [by_y_right[row] for row in common_rows] top_ids = left_ids + right_ids top_points = np.asarray([vertices[vertex_id] for vertex_id in top_ids], dtype=np.float32) base_y = float(top_points[:, 1].max()) + max( solid_thickness * 1.5, wedge_base_drop_ratio * max(median_depth, 1e-3), 1e-3, ) def make_bottom(top_id: int) -> int: bottom = vertices[top_id].astype(np.float32).copy() bottom[1] = base_y color = np.clip(vertex_colors[top_id].astype(np.float32) * 0.70, 0, 255).astype(np.uint8) bottom_id = len(vertices) vertices.append(bottom) vertex_colors.append(color) return bottom_id left_bottom_ids = [make_bottom(vertex_id) for vertex_id in left_ids] right_bottom_ids = [make_bottom(vertex_id) for vertex_id in right_ids] side_count = 0 bottom_count = 0 for top_a, top_b, bottom_a, bottom_b in zip( left_ids[:-1], left_ids[1:], left_bottom_ids[:-1], left_bottom_ids[1:], ): if len({top_a, top_b, bottom_a}) == 3: stair_sidewall_faces.append((top_a, top_b, bottom_a)) side_count += 1 if len({top_b, bottom_b, bottom_a}) == 3: stair_sidewall_faces.append((top_b, bottom_b, bottom_a)) side_count += 1 for top_a, top_b, bottom_a, bottom_b in zip( right_ids[:-1], right_ids[1:], right_bottom_ids[:-1], right_bottom_ids[1:], ): if len({top_a, bottom_a, top_b}) == 3: stair_sidewall_faces.append((top_a, bottom_a, top_b)) side_count += 1 if len({top_b, bottom_a, bottom_b}) == 3: stair_sidewall_faces.append((top_b, bottom_a, bottom_b)) side_count += 1 for left_a, left_b, right_a, right_b in zip( left_bottom_ids[:-1], left_bottom_ids[1:], right_bottom_ids[:-1], right_bottom_ids[1:], ): if len({left_a, right_a, left_b}) == 3: stair_sidewall_bottom_faces.append((left_a, right_a, left_b)) bottom_count += 1 if len({left_b, right_a, right_b}) == 3: stair_sidewall_bottom_faces.append((left_b, right_a, right_b)) bottom_count += 1 return side_count, bottom_count if support_mode == "triangular_wedge": left_ids = [vertex_id for _, vertex_id in collect_side_rows("left")] right_ids = [vertex_id for _, vertex_id in collect_side_rows("right")] left_support = append_triangular_side_panel(left_ids) right_support = append_triangular_side_panel(right_ids) if left_support is not None and right_support is not None: left_base, left_lower = left_support right_base, right_lower = right_support if left_lower != right_lower and left_base != right_base: wedge_support_faces.append((left_lower, right_lower, left_base)) wedge_support_faces.append((right_lower, right_base, left_base)) elif support_mode == "stair_sidewall": append_stair_sidewall(collect_side_rows("left"), collect_side_rows("right")) side_faces = boundary_side_faces + wedge_support_faces + stair_sidewall_faces + stair_sidewall_bottom_faces face_rows = front_faces + back_faces + side_faces faces = ( np.asarray(face_rows, dtype=np.int64).reshape((-1, 3)) if face_rows else np.empty((0, 3), dtype=np.int64) ) metadata = { "front_faces": len(front_faces), "back_faces": len(back_faces), "side_faces": len(side_faces), "boundary_side_faces": len(boundary_side_faces), "wedge_support_faces": len(wedge_support_faces), "stair_sidewall_faces": len(stair_sidewall_faces), "stair_sidewall_bottom_faces": len(stair_sidewall_bottom_faces), "solid_thickness": solid_thickness, "max_depth_jump": mesh_jump, "stride": stride, "support_mode": support_mode, "vertical_drop": vertical_drop if support_mode == "vertical_slab" else None, "vertical_base_y": vertical_base_y if support_mode == "vertical_slab" else None, "depth_scaled_dimensions": { "source_span_x": float(scale["span_x"]), "source_span_y": float(scale["span_y"]), "source_span_z": float(scale["span_z"]), "footprint_diag": float(scale["footprint_diag"]), "median_depth": median_depth, "depth_span": depth_span, "solid_thickness": solid_thickness, "vertical_drop": vertical_drop if support_mode == "vertical_slab" else None, "category_drop_ratio": category_drop_ratio, "depth_units": "relative_pseudo_depth", }, } return np.asarray(vertices, dtype=np.float32), np.asarray(vertex_colors, dtype=np.uint8), faces, metadata def build_perspective_continuous_surface_mesh( depth: np.ndarray, colors: np.ndarray, target_mask: np.ndarray, *, category: str, stride: int, minimum_cell_coverage: float = 0.15, plane: tuple[np.ndarray, float] | None = None, plane_validation_mask: np.ndarray | None = None, texture_support_mask: np.ndarray | None = None, must_retain_mask: np.ndarray | None = None, bridge_evidence_mask: np.ndarray | None = None, maximum_bridge_distance_pixels: float | None = None, maximum_bridge_added_ratio: float = 0.005, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, dict[str, Any]]: """Build a dense, perspective-preserving road-surface patch. Unlike the legacy solid/slab builders, this representation has no inferred back, vertical wall, or wedge whose copied colors could visually stretch. Every front vertex retains an image-space coordinate and samples the frozen completed RGB at that coordinate (or the nearest supported boundary pixel). """ if stride < 1: raise ValueError("continuous surface stride must be positive") if not 0.0 < minimum_cell_coverage <= 1.0: raise ValueError("minimum_cell_coverage must be in (0, 1]") if colors.shape[:2] != depth.shape or target_mask.shape != depth.shape: raise ValueError("Depth, RGB, and continuous target mask must share a raster") for name, auxiliary in ( ("plane_validation_mask", plane_validation_mask), ("texture_support_mask", texture_support_mask), ("must_retain_mask", must_retain_mask), ("bridge_evidence_mask", bridge_evidence_mask), ): if auxiliary is not None and auxiliary.shape != depth.shape: raise ValueError(f"{name} must share the continuous surface raster") h, w = depth.shape bridge_maximum_distance_pixels = ( 3.0 * float(stride) * math.sqrt(2.0) if maximum_bridge_distance_pixels is None else float(maximum_bridge_distance_pixels) ) support = target_mask.astype(bool) texture_support = ( support if texture_support_mask is None else texture_support_mask.astype(bool) ) exact_geometry_support_texture = bool( np.array_equal(texture_support, support) ) if not np.any(texture_support): raise ValueError("No source pixels for continuous surface texture") mask_y, mask_x = np.nonzero(support) if mask_x.size == 0: raise ValueError("No target pixels for perspective continuous surface") x_min, x_max = int(mask_x.min()), int(mask_x.max()) y_min, y_max = int(mask_y.min()), int(mask_y.max()) def grid_axis(low: int, high: int) -> list[int]: values = list(range(low, high + 1, int(stride))) if not values: values = [low] if values[-1] != high: values.append(high) return values xs = grid_axis(x_min, x_max) ys = grid_axis(y_min, y_max) if len(xs) < 2 or len(ys) < 2: raise ValueError("Continuous surface bounding box is too small for meshing") raw_cell_mask = np.zeros((len(ys) - 1, len(xs) - 1), dtype=np.uint8) retained_evidence_cells = np.zeros_like(raw_cell_mask, dtype=bool) bridge_evidence_cells = np.zeros_like(raw_cell_mask, dtype=bool) expanded_bridge_evidence = None if bridge_evidence_mask is not None and np.any(bridge_evidence_mask): evidence_kernel_size = 4 * int(stride) + 1 evidence_kernel = cv2.getStructuringElement( cv2.MORPH_ELLIPSE, (evidence_kernel_size, evidence_kernel_size), ) expanded_bridge_evidence = cv2.dilate( bridge_evidence_mask.astype(np.uint8), evidence_kernel, ).astype(bool) for iy, (y0, y1) in enumerate(zip(ys[:-1], ys[1:])): for ix, (x0, x1) in enumerate(zip(xs[:-1], xs[1:])): patch = support[y0 : y1 + 1, x0 : x1 + 1] center_y = min((y0 + y1) // 2, h - 1) center_x = min((x0 + x1) // 2, w - 1) coverage = float(patch.mean()) if patch.size else 0.0 retain_evidence = bool( must_retain_mask is not None and np.any(must_retain_mask[y0 : y1 + 1, x0 : x1 + 1]) ) retained_evidence_cells[iy, ix] = retain_evidence if expanded_bridge_evidence is not None: bridge_evidence_cells[iy, ix] = bool( np.any(expanded_bridge_evidence[y0 : y1 + 1, x0 : x1 + 1]) ) if ( support[center_y, center_x] or coverage >= minimum_cell_coverage or retain_evidence ): raw_cell_mask[iy, ix] = 1 if not np.any(raw_cell_mask): raise ValueError("Perspective continuous surface produced no occupied cells") cell_kernel = np.ones((3, 3), dtype=np.uint8) closed_cell_candidate = cv2.morphologyEx( raw_cell_mask, cv2.MORPH_CLOSE, cell_kernel, ).astype(bool) cell_close_added = closed_cell_candidate & ~raw_cell_mask.astype(bool) cell_close_limit = max(1, int(math.floor(0.02 * int(raw_cell_mask.sum())))) cell_close_gate_passed = int(cell_close_added.sum()) <= cell_close_limit cell_mask = ( closed_cell_candidate if cell_close_gate_passed else raw_cell_mask.astype(bool) ) cell_mask_after_close = cell_mask.copy() # Remove only tiny raster islands that carry no hidden-surface evidence. # This replaces the legacy global-largest operation, which removed most of # the valid hidden walkway in one reviewed sample. component_count, component_labels, component_stats, _ = ( cv2.connectedComponentsWithStats( cell_mask.astype(np.uint8), connectivity=4, ) ) occupied_cells = int(cell_mask.sum()) minimum_component_cells = max(2, int(math.ceil(0.001 * occupied_cells))) main_component_label = ( 1 + int(np.argmax(component_stats[1:, cv2.CC_STAT_AREA])) if component_count > 1 else 0 ) kept_labels: set[int] = set() for label in range(1, component_count): component = component_labels == label area = int(component_stats[label, cv2.CC_STAT_AREA]) if ( label == main_component_label or area >= minimum_component_cells or np.any(component & retained_evidence_cells) ): kept_labels.add(label) filtered_cell_mask = np.isin(component_labels, list(kept_labels)) discarded_tiny_cells = int(cell_mask.sum() - filtered_cell_mask.sum()) cell_mask = filtered_cell_mask bridge_limit_cells = max( 1, int(math.floor(maximum_bridge_added_ratio * max(int(cell_mask.sum()), 1))), ) bridge_added_cells = 0 bridge_count = 0 bridge_rejected_count = 0 bridge_distances_pixels: list[float] = [] if main_component_label in kept_labels: anchor = component_labels == main_component_label retained_components = [ component_labels == label for label in sorted(kept_labels) if label != main_component_label and np.any((component_labels == label) & retained_evidence_cells) ] for component in retained_components: if np.any(component & anchor): continue distance, nearest_label_map = cv2.distanceTransformWithLabels( (~anchor).astype(np.uint8), cv2.DIST_L2, 5, labelType=cv2.DIST_LABEL_PIXEL, ) component_y, component_x = np.nonzero(component) nearest_index = int(np.argmin(distance[component_y, component_x])) source_y = int(component_y[nearest_index]) source_x = int(component_x[nearest_index]) distance_pixels = float(distance[source_y, source_x]) * float(stride) nearest_label = int(nearest_label_map[source_y, source_x]) anchor_y, anchor_x = np.nonzero(anchor) anchor_labels = nearest_label_map[anchor_y, anchor_x] matching = np.flatnonzero(anchor_labels == nearest_label) if matching.size == 0: bridge_rejected_count += 1 continue target_index = int(matching[0]) target_y = int(anchor_y[target_index]) target_x = int(anchor_x[target_index]) candidate = np.zeros_like(cell_mask, dtype=np.uint8) cv2.line( candidate, (source_x, source_y), (target_x, target_y), color=1, thickness=2, lineType=cv2.LINE_8, ) candidate = candidate.astype(bool) added_candidate = candidate & ~cell_mask evidence_overlap = ( float(np.count_nonzero(candidate & bridge_evidence_cells)) / max(int(candidate.sum()), 1) if expanded_bridge_evidence is not None else 0.0 ) if ( distance_pixels > bridge_maximum_distance_pixels or evidence_overlap < 0.25 or bridge_added_cells + int(added_candidate.sum()) > bridge_limit_cells ): bridge_rejected_count += 1 continue cell_mask |= candidate anchor |= component | candidate added = int(added_candidate.sum()) bridge_added_cells += added bridge_count += 1 bridge_distances_pixels.append(distance_pixels) # Deliver one coherent open-world surface. Detached components that # could not be connected through the full-obstacle evidence gate are # retained in the render-support audit, but are excluded from the mesh # instead of becoming floating road islands. detached_rejected_cells = int(np.count_nonzero(cell_mask & ~anchor)) detached_rejected_hidden_cells = int( np.count_nonzero(cell_mask & ~anchor & retained_evidence_cells) ) cell_mask &= anchor else: detached_rejected_cells = 0 detached_rejected_hidden_cells = 0 # Rasterizing a hole-free pixel footprint onto a coarser cell grid can # re-introduce a small enclosed hole even when the input support audit says # that every hole was filled. Close that representation gap here. This # does not join detached components or extrapolate the outer boundary; it # only fills cells that are topologically enclosed by the retained surface. cell_mask_before_final_hole_fill = cell_mask.copy() cell_mask = fill_enclosed_mask_holes(cell_mask) final_enclosed_hole_added_cells = int( np.count_nonzero(cell_mask & ~cell_mask_before_final_hole_fill) ) fx, fy, cx, cy = camera_intrinsics(w, h, None, None, None, None) camera_points = pixels_to_points(depth, fx, fy, cx, cy) validation_support = ( support if plane_validation_mask is None else plane_validation_mask.astype(bool) ) fit_mask = validation_support & np.isfinite(depth) & (depth > 0) if not np.any(fit_mask): fit_mask = support & np.isfinite(depth) & (depth > 0) surface_plane = plane if plane is not None else fit_plane(camera_points[fit_mask]) if surface_plane is None: raise ValueError("Could not fit the continuous surface plane") surface_depth = plane_depth_for_pixels((h, w), surface_plane, fx, fy, cx, cy) observed = depth[fit_mask] depth_low, depth_high = np.percentile(observed, [1, 99]) allowed_low = max(float(depth_low) * 0.5, 1e-4) allowed_high = max(float(depth_high) * 1.5, allowed_low + 1e-4) referenced_nodes = np.zeros((len(ys), len(xs)), dtype=bool) for iy, ix in np.argwhere(cell_mask): referenced_nodes[iy, ix] = True referenced_nodes[iy + 1, ix] = True referenced_nodes[iy, ix + 1] = True referenced_nodes[iy + 1, ix + 1] = True node_grid_y, node_grid_x = np.nonzero(referenced_nodes) node_image_y = np.asarray([ys[index] for index in node_grid_y], dtype=np.int32) node_image_x = np.asarray([xs[index] for index in node_grid_x], dtype=np.int32) node_plane_depth = surface_depth[node_image_y, node_image_x] finite_positive = np.isfinite(node_plane_depth) & (node_plane_depth > 0) if not np.all(finite_positive): raise ValueError("Continuous surface plane is invalid inside the mesh ROI") plausible_depth = ( (node_plane_depth >= allowed_low) & (node_plane_depth <= allowed_high) ) plane_plausible_fraction = float(np.mean(plausible_depth)) if plane_plausible_fraction < 0.95: raise ValueError( "Continuous surface plane failed the whole-ROI plausibility gate: " f"{plane_plausible_fraction:.6f}" ) nearest_distance, nearest_labels = cv2.distanceTransformWithLabels( (~texture_support).astype(np.uint8), cv2.DIST_L2, 5, labelType=cv2.DIST_LABEL_PIXEL, ) max_label = int(nearest_labels.max()) nearest_y = np.zeros(max_label + 1, dtype=np.int32) nearest_x = np.zeros(max_label + 1, dtype=np.int32) support_y, support_x = np.nonzero(texture_support) support_labels = nearest_labels[support_y, support_x] nearest_y[support_labels] = support_y nearest_x[support_labels] = support_x vertex_ids = -np.ones((len(ys), len(xs)), dtype=np.int64) vertices: list[np.ndarray] = [] vertex_colors: list[np.ndarray] = [] boundary_clamped_vertices = 0 texture_clamp_distances_pixels: list[float] = [] depth_clipped_vertices = 0 for iy, ix in np.argwhere(referenced_nodes): y, x = int(ys[iy]), int(xs[ix]) z = float(surface_depth[y, x]) point = np.asarray( [ (float(x) - cx) * z / fx, (float(y) - cy) * z / fy, z, ], dtype=np.float32, ) sample_y, sample_x = y, x if not texture_support[y, x]: label = int(nearest_labels[y, x]) if 0 < label <= max_label: sample_y = int(nearest_y[label]) sample_x = int(nearest_x[label]) boundary_clamped_vertices += 1 texture_clamp_distances_pixels.append( float(nearest_distance[y, x]) ) vertex_ids[iy, ix] = len(vertices) vertices.append(point) vertex_colors.append(colors[sample_y, sample_x].astype(np.uint8)) front_faces: list[tuple[int, int, int]] = [] for iy, ix in np.argwhere(cell_mask): ids = ( int(vertex_ids[iy, ix]), int(vertex_ids[iy, ix + 1]), int(vertex_ids[iy + 1, ix]), int(vertex_ids[iy + 1, ix + 1]), ) if min(ids) < 0: raise RuntimeError("Internal continuous-surface grid indexing error") front_faces.append((ids[0], ids[2], ids[1])) front_faces.append((ids[1], ids[2], ids[3])) vertex_array = np.asarray(vertices, dtype=np.float32) color_array = np.asarray(vertex_colors, dtype=np.uint8) face_array = np.asarray(front_faces, dtype=np.int64).reshape((-1, 3)) scale = robust_extent_from_points(vertex_array) metadata = { "front_faces": len(front_faces), "back_faces": 0, "side_faces": 0, "boundary_side_faces": 0, "wedge_support_faces": 0, "stair_sidewall_faces": 0, "stair_sidewall_bottom_faces": 0, "solid_thickness": 0.0, "max_depth_jump": None, "stride": int(stride), "support_mode": "open_world_surface_only", "vertical_drop": None, "vertical_base_y": None, "surface_model": "perspective_planar_accessibility_surface_patch", "category": category, "stair_surface_mode": "not_stairs", "stair_rows": [], "stair_cross_samples": None, "stair_edge_slope": None, "stair_step_count": None, "texture_policy": ( ( "dense frozen completed RGB sampled exactly at original image " "coordinates for reviewed and evidence-gated repair pixels; " "grid-boundary-only nodes use the nearest geometry texel; no " "copied side or back texture" ) if exact_geometry_support_texture else ( "dense frozen completed RGB sampled at original image coordinates; " "texture-repair nodes use the nearest explicit texture-support " "pixel; no copied side or back texture" ) ), "texture_coordinate_policy": "perspective_image_xy_reprojection_without_bbox_warp", "boundary_clamped_vertices": int(boundary_clamped_vertices), "texture_source_support_pixels": int(texture_support.sum()), "exact_geometry_support_texture": exact_geometry_support_texture, "texture_completion_vertices": int(boundary_clamped_vertices), "texture_completion_max_distance_pixels": ( max(texture_clamp_distances_pixels) if texture_clamp_distances_pixels else 0.0 ), "texture_completion_mean_distance_pixels": ( float(np.mean(texture_clamp_distances_pixels)) if texture_clamp_distances_pixels else 0.0 ), "texture_completion_policy": ( "nearest_reviewed_target_pixel_without_blur_or_bbox_warp" ), "depth_clipped_vertices": int(depth_clipped_vertices), "plane_plausible_fraction": plane_plausible_fraction, "plane_depth_policy": ( "whole_roi_validated_single_plane_without_per_vertex_clipping" ), "surface_cells_before_topology_close": int(raw_cell_mask.sum()), "surface_cells": int(cell_mask.sum()), "topology_close_added_cells": int( (cell_mask_after_close & ~raw_cell_mask.astype(bool)).sum() ), "topology_close_candidate_added_cells": int(cell_close_added.sum()), "topology_close_added_limit_cells": int(cell_close_limit), "topology_close_gate_passed": cell_close_gate_passed, "minimum_component_cells": int(minimum_component_cells), "discarded_tiny_cells_without_hidden_evidence": int( discarded_tiny_cells ), "hidden_evidence_cells": int(retained_evidence_cells.sum()), "hidden_evidence_covered_cells": int( np.count_nonzero(cell_mask & retained_evidence_cells) ), "hidden_cell_coverage_fraction": ( float( np.count_nonzero(cell_mask & retained_evidence_cells) / max(int(retained_evidence_cells.sum()), 1) ) if must_retain_mask is not None else None ), "bridge_policy": ( "hidden_component_shortest_cell_path_with_full_obstacle_evidence" ), "bridge_count": int(bridge_count), "bridge_rejected_count": int(bridge_rejected_count), "bridge_added_cells": int(bridge_added_cells), "bridge_added_limit_cells": int(bridge_limit_cells), "bridge_maximum_distance_pixels": bridge_maximum_distance_pixels, "bridge_distances_pixels": bridge_distances_pixels, "bridge_evidence_mask_used": expanded_bridge_evidence is not None, "detached_rejected_cells": int(detached_rejected_cells), "detached_rejected_hidden_cells": int( detached_rejected_hidden_cells ), "detached_component_policy": ( "exclude_when_not_connected_to_primary_surface_by_validated_obstacle_bridge" ), "final_enclosed_hole_added_cells": int( final_enclosed_hole_added_cells ), "final_cell_holes_filled": True, "cell_components_before": mask_component_count( raw_cell_mask.astype(bool), connectivity=4, ), "cell_components_after": mask_component_count( cell_mask, connectivity=4, ), "minimum_cell_coverage": float(minimum_cell_coverage), "source_pixel_bounds": { "x_min": x_min, "x_max": x_max, "y_min": y_min, "y_max": y_max, }, "plane_normal": [float(value) for value in surface_plane[0]], "plane_d": float(surface_plane[1]), "plane_source": ( "validated_visible_target_plane" if plane is not None else "robust_fit_on_repaired_support" ), "plane_validation_pixels": int(fit_mask.sum()), "depth_scaled_dimensions": { "source_span_x": float(scale["span_x"]), "source_span_y": float(scale["span_y"]), "source_span_z": float(scale["span_z"]), "footprint_diag": float(scale["footprint_diag"]), "median_depth": float(scale["median_z"]), "depth_units": "metric_style_depth_or_input_pseudo_depth_not_calibrated_truth", "calibrated_metric_truth": False, }, "scene_semantics": "open_world_accessibility_surface_patch", "automatic_passability_claim": False, "mesh_topology_role": "renderable_surface_patch_not_collision_mesh", "watertight_collision_mesh": False, } return vertex_array, color_array, face_array, metadata def build_hybrid_instance_footprint_mask( target_mask: np.ndarray, *, category: str, stride: int, visible_mask: np.ndarray | None = None, hidden_mask: np.ndarray | None = None, ) -> tuple[np.ndarray, dict[str, Any]]: """Return one smooth, hole-free instance footprint without using a hull. The reviewed amodal target remains the only semantic source. The component with the most visible-target evidence is retained (which rejects detached hidden islands such as a fence/car false positive), then a bounded close/open removes narrow obstacle cuts and render-only hairs. A simplified external contour removes raster stair-steps while preserving broad Y/T path branches. Convex hulls, row spans, bounding boxes, and generation masks are deliberately excluded. """ if target_mask.ndim != 2 or not np.any(target_mask): raise ValueError("Hybrid instance footprint needs a non-empty 2D mask") if stride < 1: raise ValueError("Hybrid instance footprint stride must be positive") for name, mask in (("visible_mask", visible_mask), ("hidden_mask", hidden_mask)): if mask is not None and mask.shape != target_mask.shape: raise ValueError(f"{name} must share the target raster") source = fill_enclosed_mask_holes(target_mask.astype(bool)) component_count, labels, stats, _ = cv2.connectedComponentsWithStats( source.astype(np.uint8), connectivity=8, ) if component_count <= 1: raise ValueError("Hybrid instance footprint has no connected target") visible = ( visible_mask.astype(bool) if visible_mask is not None else np.zeros_like(source) ) component_rows: list[tuple[int, int, int]] = [] for label in range(1, component_count): component = labels == label component_rows.append( ( int(np.count_nonzero(component & visible)), int(stats[label, cv2.CC_STAT_AREA]), label, ) ) # Visible evidence dominates area. This is fail-closed for a large remote # hidden-only island while remaining deterministic when no visible mask is # available. _, _, retained_label = max(component_rows) authoritative = fill_enclosed_mask_holes(labels == retained_label) source_pixels = int(authoritative.sum()) source_visible = int(np.count_nonzero(authoritative & visible)) hidden = ( hidden_mask.astype(bool) if hidden_mask is not None else np.zeros_like(authoritative) ) source_hidden = int(np.count_nonzero(authoritative & hidden)) category_multiplier = { "ramp": 8, "walkway": 10, "curb_cut": 12, "tactile_paving": 12, }.get(category, 8) requested_kernel = category_multiplier * int(stride) + 1 if requested_kernel % 2 == 0: requested_kernel += 1 kernel_candidates = [] for value in ( requested_kernel, max(4 * int(stride) + 1, requested_kernel - 2 * int(stride)), 4 * int(stride) + 1, ): if value % 2 == 0: value += 1 if value not in kernel_candidates: kernel_candidates.append(value) selected = authoritative.copy() selected_kernel = 1 selected_gate = "authoritative_component_fallback" selected_metrics: dict[str, float] = {} for kernel_size in kernel_candidates: kernel = cv2.getStructuringElement( cv2.MORPH_ELLIPSE, (kernel_size, kernel_size), ) candidate = cv2.morphologyEx( authoritative.astype(np.uint8), cv2.MORPH_CLOSE, kernel, ) candidate = cv2.morphologyEx(candidate, cv2.MORPH_OPEN, kernel) contours, _ = cv2.findContours( candidate, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE, ) if not contours: continue largest = max(contours, key=cv2.contourArea) candidate = np.zeros_like(candidate) cv2.drawContours(candidate, [largest], -1, color=1, thickness=-1) candidate_bool = fill_enclosed_mask_holes(candidate.astype(bool)) visible_retention = ( float(np.count_nonzero(candidate_bool & authoritative & visible)) / max(source_visible, 1) if source_visible > 0 else 1.0 ) hidden_retention = ( float(np.count_nonzero(candidate_bool & authoritative & hidden)) / max(source_hidden, 1) if source_hidden > 0 else 1.0 ) authoritative_retention = float( np.count_nonzero(candidate_bool & authoritative) ) / max(source_pixels, 1) growth_ratio = float( np.count_nonzero(candidate_bool & ~authoritative) ) / max(source_pixels, 1) if ( visible_retention >= 0.94 and hidden_retention >= 0.90 and authoritative_retention >= 0.84 and growth_ratio <= 0.20 ): selected = candidate_bool selected_kernel = kernel_size selected_gate = "bounded_close_open_retention_gate_passed" selected_metrics = { "visible_retention_fraction": visible_retention, "hidden_retention_fraction": hidden_retention, "authoritative_retention_fraction": authoritative_retention, "growth_fraction": growth_ratio, } break # Simplify only the retained external outline. Re-rasterization provides a # clean render boundary; the same evidence-retention gates prevent a broad # path branch from being simplified away. contours, _ = cv2.findContours( selected.astype(np.uint8), cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE, ) contour = max(contours, key=cv2.contourArea) raw_contour_vertices = int(len(contour)) perimeter = float(cv2.arcLength(contour, True)) epsilon = max(1.0, 0.0015 * perimeter) simplified_contour = cv2.approxPolyDP(contour, epsilon, True) simplified = np.zeros_like(selected, dtype=np.uint8) cv2.drawContours( simplified, [simplified_contour], -1, color=1, thickness=-1, ) simplified_bool = fill_enclosed_mask_holes(simplified.astype(bool)) simplify_visible_retention = ( float(np.count_nonzero(simplified_bool & authoritative & visible)) / max(source_visible, 1) if source_visible > 0 else 1.0 ) simplify_hidden_retention = ( float(np.count_nonzero(simplified_bool & authoritative & hidden)) / max(source_hidden, 1) if source_hidden > 0 else 1.0 ) if simplify_visible_retention >= 0.94 and simplify_hidden_retention >= 0.90: selected = simplified_bool contour_simplification_used = True else: contour_simplification_used = False return selected.astype(bool), { "role": "render_only_instance_footprint_not_annotation", "policy": ( "visible_anchored_component_plus_bounded_close_open_and_" "external_contour_simplification" ), "generation_mask_used": False, "convex_hull_used": False, "row_span_used": False, "bbox_footprint_used": False, "annotation_mask_modified": False, "source_components": int(component_count - 1), "retained_component_label": int(retained_label), "source_pixels": int(target_mask.sum()), "authoritative_component_pixels": source_pixels, "footprint_pixels": int(selected.sum()), "requested_smoothing_kernel": int(requested_kernel), "selected_smoothing_kernel": int(selected_kernel), "smoothing_gate": selected_gate, **selected_metrics, "raw_contour_vertices": raw_contour_vertices, "simplified_contour_vertices": int(len(simplified_contour)), "contour_simplification_used": contour_simplification_used, "simplified_visible_retention_fraction": simplify_visible_retention, "simplified_hidden_retention_fraction": simplify_hidden_retention, "final_components": mask_component_count(selected), "final_holes_filled": True, } def _contiguous_mask_runs(row: np.ndarray) -> list[tuple[int, int]]: """Return inclusive true runs for one binary mask row.""" columns = np.flatnonzero(row) if columns.size == 0: return [] split_indices = np.flatnonzero(np.diff(columns) > 1) + 1 return [ (int(run[0]), int(run[-1])) for run in np.split(columns, split_indices) if run.size > 0 ] def _robust_corridor_boundary( rows: np.ndarray, values: np.ndarray, *, side: str, ) -> tuple[np.ndarray, float]: """Fit a smooth boundary while treating obstacle bites as outliers.""" if side not in {"left", "right"}: raise ValueError("Corridor boundary side must be left or right") if len(rows) != len(values) or len(rows) < 2: raise ValueError("Corridor boundary needs at least two aligned samples") normalized_rows = (rows - float(rows.min())) / max( float(np.ptp(rows)), 1.0, ) degree = min(2, len(rows) - 1) retained = np.ones(len(rows), dtype=bool) coefficients = np.polyfit(normalized_rows, values, degree) for _ in range(4): retained_degree = min(degree, max(int(retained.sum()) - 1, 1)) coefficients = np.polyfit( normalized_rows[retained], values[retained], retained_degree, ) prediction = np.polyval(coefficients, normalized_rows) residual = values - prediction median = float(np.median(residual[retained])) mad = float( np.median(np.abs(residual[retained] - median)) ) + 1.0 if side == "left": # A foreground obstacle cutting in from the left moves the observed # boundary rightward. Reject that direction more aggressively. candidate = ( (residual < median + 2.2 * mad) & (residual > median - 4.0 * mad) ) else: # The mirrored case moves a right boundary leftward. candidate = ( (residual > median - 2.2 * mad) & (residual < median + 4.0 * mad) ) if int(candidate.sum()) < max(8, degree + 2): break retained = candidate return ( np.polyval(coefficients, normalized_rows).astype(np.float64), float(retained.mean()), ) def build_obstacle_centered_corridor_profile( footprint_mask: np.ndarray, *, stride: int, visible_mask: np.ndarray | None = None, hidden_mask: np.ndarray | None = None, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, dict[str, Any]]: """Extract one smooth local channel through the completed obstacle zone. A full perspective road mask often reaches the image horizon and produces extremely long triangular shards after monocular plane reprojection. For amodal review the useful object is instead the local, obstacle-free travel channel. This routine follows the mask run that crosses the hidden-region centre, keeps enough visible context on both sides, and robustly removes boundary notches caused by foreground silhouettes. It never edits the reviewed masks or the frozen completed RGB. """ if footprint_mask.ndim != 2 or not np.any(footprint_mask): raise ValueError("Obstacle-centred corridor needs a non-empty footprint") if stride < 1: raise ValueError("Obstacle-centred corridor stride must be positive") for name, mask in (("visible_mask", visible_mask), ("hidden_mask", hidden_mask)): if mask is not None and mask.shape != footprint_mask.shape: raise ValueError(f"{name} must share the corridor footprint raster") footprint = footprint_mask.astype(bool) visible = ( visible_mask.astype(bool) & footprint if visible_mask is not None else footprint.copy() ) if not np.any(visible): visible = footprint.copy() hidden = ( hidden_mask.astype(bool) & footprint if hidden_mask is not None else np.zeros_like(footprint) ) hidden_source = ( hidden if np.any(hidden) else ( hidden_mask.astype(bool) if hidden_mask is not None and np.any(hidden_mask) else np.zeros_like(footprint) ) ) visible_y, visible_x = np.nonzero(visible) visible_y_min = int(visible_y.min()) visible_y_max = int(visible_y.max()) visible_span = visible_y_max - visible_y_min + 1 hidden_y, hidden_x = np.nonzero(hidden_source) obstacle_centered = bool(hidden_x.size > 0) if obstacle_centered: anchor_x = float(np.median(hidden_x)) anchor_y = float(np.median(hidden_y)) hidden_y_min = int(hidden_y.min()) hidden_y_max = int(hidden_y.max()) hidden_span = hidden_y_max - hidden_y_min + 1 row_min = max( visible_y_min, int(round(hidden_y_min - 0.25 * hidden_span)), ) row_max = min( visible_y_max, int(round(hidden_y_max + 0.80 * hidden_span)), ) minimum_span = max( int(round(0.38 * visible_span)), 8 * int(stride), ) current_span = row_max - row_min + 1 if current_span < minimum_span: missing = minimum_span - current_span row_min = max(visible_y_min, row_min - missing // 3) current_span = row_max - row_min + 1 row_max = min( visible_y_max, row_max + (minimum_span - current_span), ) current_span = row_max - row_min + 1 if current_span < minimum_span: row_min = max( visible_y_min, row_min - (minimum_span - current_span), ) else: anchor_x = float(np.median(visible_x)) anchor_y = float(np.median(visible_y)) hidden_y_min = None hidden_y_max = None row_min = visible_y_min row_max = visible_y_max sampled_rows = np.arange( row_min, row_max + 1, int(stride), dtype=np.int32, ) if sampled_rows.size == 0: sampled_rows = np.asarray([row_min, row_max], dtype=np.int32) elif int(sampled_rows[-1]) != row_max: sampled_rows = np.concatenate( (sampled_rows, np.asarray([row_max], dtype=np.int32)) ) if sampled_rows.size < 2: raise ValueError("Obstacle-centred corridor is too short") row_runs = [ _contiguous_mask_runs(footprint[int(row)]) for row in sampled_rows ] available = [index for index, runs in enumerate(row_runs) if runs] if not available: raise ValueError("Obstacle-centred corridor has no occupied row") anchor_index = min( available, key=lambda index: abs(float(sampled_rows[index]) - anchor_y), ) selected_runs: list[tuple[int, int] | None] = [None] * len(sampled_rows) def choose_run( candidates: list[tuple[int, int]], reference_x: float, ) -> tuple[int, int]: def score(run: tuple[int, int]) -> tuple[float, int]: left, right = run distance = ( 0.0 if left <= reference_x <= right else min(abs(reference_x - left), abs(reference_x - right)) ) return distance, -(right - left + 1) return min(candidates, key=score) selected_runs[anchor_index] = choose_run( row_runs[anchor_index], anchor_x, ) for direction in (1, -1): reference_x = float(np.clip(anchor_x, *selected_runs[anchor_index])) index = anchor_index + direction while 0 <= index < len(sampled_rows): if row_runs[index]: chosen = choose_run(row_runs[index], reference_x) selected_runs[index] = chosen reference_x = float(np.clip(reference_x, *chosen)) index += direction valid_indices = np.asarray( [index for index, run in enumerate(selected_runs) if run is not None], dtype=np.int32, ) all_indices = np.arange(len(sampled_rows), dtype=np.float64) raw_left = np.interp( all_indices, valid_indices.astype(np.float64), np.asarray( [selected_runs[index][0] for index in valid_indices], dtype=np.float64, ), ) raw_right = np.interp( all_indices, valid_indices.astype(np.float64), np.asarray( [selected_runs[index][1] for index in valid_indices], dtype=np.float64, ), ) left, left_retention = _robust_corridor_boundary( sampled_rows.astype(np.float64), raw_left, side="left", ) right, right_retention = _robust_corridor_boundary( sampled_rows.astype(np.float64), raw_right, side="right", ) raw_width = np.maximum(raw_right - raw_left + 1.0, 1.0) minimum_width = max( 6.0 * float(stride), 0.45 * float(np.percentile(raw_width, 20.0)), ) center = 0.5 * (left + right) width = np.maximum(right - left + 1.0, minimum_width) left = center - 0.5 * (width - 1.0) right = center + 0.5 * (width - 1.0) image_width = footprint.shape[1] left = np.clip(left, 0.0, float(image_width - 2)) right = np.clip(right, left + 1.0, float(image_width - 1)) # Keep the reconstruction local to the travel channel crossing the hidden # zone. Very wide foreground branches and vanishingly narrow horizon tails # are useful scene context in 2D, but including both in one standalone mesh # creates extreme projective triangles. Cropping those tails changes only # the selected local ROI; it never rescales or rewrites the frozen texture. profile_width_before_crop = right - left + 1.0 anchor_profile_index = int( np.argmin(np.abs(sampled_rows.astype(np.float64) - anchor_y)) ) if obstacle_centered and hidden_y_min is not None and hidden_y_max is not None: reference_indices = np.flatnonzero( (sampled_rows >= hidden_y_min - int(stride)) & (sampled_rows <= hidden_y_max + int(stride)) ) else: reference_indices = np.asarray([anchor_profile_index], dtype=np.int64) if reference_indices.size == 0: reference_indices = np.asarray([anchor_profile_index], dtype=np.int64) reference_width = float(np.median(profile_width_before_crop[reference_indices])) maximum_local_width = float( max( 8.0 * float(stride), min(2.75 * reference_width, 0.45 * float(image_width)), ) ) minimum_tail_width = float( max(8.0 * float(stride), 0.18 * min(reference_width, maximum_local_width)) ) required_start = int(np.min(reference_indices)) required_end = int(np.max(reference_indices)) eligible_tail = profile_width_before_crop >= minimum_tail_width trim_start = required_start while trim_start > 0 and bool(eligible_tail[trim_start - 1]): trim_start -= 1 trim_end = required_end while trim_end + 1 < len(sampled_rows) and bool(eligible_tail[trim_end + 1]): trim_end += 1 minimum_sampled_rows = min(12, len(sampled_rows)) while trim_end - trim_start + 1 < minimum_sampled_rows: if trim_start > 0: trim_start -= 1 elif trim_end + 1 < len(sampled_rows): trim_end += 1 else: break original_sampled_row_count = len(sampled_rows) selection = slice(trim_start, trim_end + 1) sampled_rows = sampled_rows[selection] left = left[selection] right = right[selection] center = 0.5 * (left + right) selected_width = right - left + 1.0 cropped_width = np.minimum(selected_width, maximum_local_width) cropped_width_pixels = float(np.sum(selected_width - cropped_width)) left = center - 0.5 * (cropped_width - 1.0) right = center + 0.5 * (cropped_width - 1.0) left = np.clip(left, 0.0, float(image_width - 2)) right = np.clip(right, left + 1.0, float(image_width - 1)) row_min = int(sampled_rows[0]) row_max = int(sampled_rows[-1]) return ( sampled_rows.astype(np.float64), left.astype(np.float64), right.astype(np.float64), { "policy": ( "obstacle_centered_single_channel_with_robust_quadratic_" "boundary_completion" ), "obstacle_centered": obstacle_centered, "hidden_region_used_for_localization": obstacle_centered, "anchor_x": anchor_x, "anchor_y": anchor_y, "row_min": int(row_min), "row_max": int(row_max), "sampled_rows": int(len(sampled_rows)), "raw_valid_rows": int(len(valid_indices)), "left_fit_retention_fraction": left_retention, "right_fit_retention_fraction": right_retention, "raw_width_pixels_median": float(np.median(raw_width)), "profile_width_pixels_median": float(np.median(right - left + 1.0)), "profile_width_pixels_min": float(np.min(right - left + 1.0)), "profile_width_pixels_max": float(np.max(right - left + 1.0)), "local_roi_policy": ( "preserve_original_image_coordinates_while_cropping_wide_" "foreground_branches_and_narrow_horizon_tails" ), "local_roi_reference_width_pixels": reference_width, "local_roi_maximum_width_pixels": maximum_local_width, "local_roi_minimum_tail_width_pixels": minimum_tail_width, "local_roi_width_pixels_cropped_total": cropped_width_pixels, "local_roi_original_sampled_rows": original_sampled_row_count, "local_roi_tail_rows_trimmed": int( original_sampled_row_count - len(sampled_rows) ), "hidden_y_min": hidden_y_min, "hidden_y_max": hidden_y_max, "branch_policy": ( "follow_the_channel_crossing_the_completion_zone_instead_of_" "meshing_every_horizon_branch" ), "reviewed_masks_modified": False, }, ) def build_obstacle_centered_depth_scaled_corridor_mesh( depth: np.ndarray, colors: np.ndarray, footprint_mask: np.ndarray, *, category: str, stride: int, visible_mask: np.ndarray | None = None, hidden_mask: np.ndarray | None = None, plane_validation_mask: np.ndarray | None = None, metric_style_depth: bool = False, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, dict[str, Any]]: """Build a watertight, depth-scaled local context patch around an obstacle. A regular metric grid is created in the fitted local surface plane and each vertex is reprojected through the camera model into the frozen completed RGB. Width and length therefore select physical context rather than independently normalizing texture axes. The result keeps nearby path cues, covers the hidden obstacle zone, and avoids the horizon spikes and isolated ribbon appearance of whole-mask inverse projection. """ if colors.shape[:2] != depth.shape or footprint_mask.shape != depth.shape: raise ValueError("Depth, RGB, and corridor footprint must share a raster") for name, mask in ( ("visible_mask", visible_mask), ("hidden_mask", hidden_mask), ("plane_validation_mask", plane_validation_mask), ): if mask is not None and mask.shape != depth.shape: raise ValueError(f"{name} must share the corridor raster") profile_rows, profile_left, profile_right, profile_metadata = ( build_obstacle_centered_corridor_profile( footprint_mask, stride=stride, visible_mask=visible_mask, hidden_mask=hidden_mask, ) ) h, w = depth.shape fx, fy, cx, cy = camera_intrinsics(w, h, None, None, None, None) camera_points = pixels_to_points(depth, fx, fy, cx, cy) validation = ( plane_validation_mask.astype(bool) if plane_validation_mask is not None else footprint_mask.astype(bool) ) fit_mask = validation & np.isfinite(depth) & (depth > 0) if not np.any(fit_mask): fit_mask = footprint_mask.astype(bool) & np.isfinite(depth) & (depth > 0) plane = fit_plane(camera_points[fit_mask]) if plane is None: raise ValueError("Could not fit the obstacle-centred corridor plane") plane_normal = np.asarray(plane[0], dtype=np.float64) plane_normal_norm = float(np.linalg.norm(plane_normal)) if plane_normal_norm <= 1e-12: raise ValueError("Obstacle-centred corridor plane normal collapsed") plane_normal /= plane_normal_norm plane_d = float(plane[1]) / plane_normal_norm def exact_plane_points_for( image_x: np.ndarray, image_y: np.ndarray, ) -> tuple[np.ndarray, np.ndarray, np.ndarray]: image_x = np.asarray(image_x, dtype=np.float64) image_y = np.asarray(image_y, dtype=np.float64) rays = np.stack( ( (image_x - cx) / fx, (image_y - cy) / fy, np.ones_like(image_x), ), axis=-1, ) denominator = rays @ plane_normal with np.errstate(divide="ignore", invalid="ignore"): image_depth = -plane_d / denominator points = rays * image_depth[..., None] return points.astype(np.float64), image_depth, denominator profile_center = 0.5 * (profile_left + profile_right) left_points, left_depth, left_denominator = exact_plane_points_for( profile_left, profile_rows, ) right_points, right_depth, right_denominator = exact_plane_points_for( profile_right, profile_rows, ) center_points, center_depth, center_denominator = exact_plane_points_for( profile_center, profile_rows, ) control_depths = np.concatenate((left_depth, right_depth, center_depth)) if not np.all(np.isfinite(control_depths)) or np.any(control_depths <= 0): raise ValueError("Corridor ray/plane intersections are not positive and finite") segment_lengths = np.linalg.norm(np.diff(center_points, axis=0), axis=1) finite_segments = segment_lengths[ np.isfinite(segment_lengths) & (segment_lengths > 0) ] if finite_segments.size == 0: raise ValueError("Corridor has no valid depth-scaled length") segment_cap = float(np.percentile(finite_segments, 90.0)) robust_segments = np.minimum(segment_lengths, segment_cap) cumulative = np.concatenate( (np.asarray([0.0], dtype=np.float64), np.cumsum(robust_segments)) ) raw_centerline_length = float(cumulative[-1]) if raw_centerline_length <= 1e-6: raise ValueError("Corridor robust length collapsed") longitudinal_axis = center_points[-1] - center_points[0] longitudinal_axis -= float(np.dot(longitudinal_axis, plane_normal)) * plane_normal if float(np.linalg.norm(longitudinal_axis)) <= 1e-8: centered_points = center_points - np.mean(center_points, axis=0, keepdims=True) try: _, _, principal_axes = np.linalg.svd(centered_points, full_matrices=False) except np.linalg.LinAlgError as error: raise ValueError("Could not derive the corridor plane basis") from error longitudinal_axis = principal_axes[0] longitudinal_axis -= ( float(np.dot(longitudinal_axis, plane_normal)) * plane_normal ) longitudinal_axis /= max(float(np.linalg.norm(longitudinal_axis)), 1e-12) if float(np.dot(longitudinal_axis, center_points[-1] - center_points[0])) < 0: longitudinal_axis *= -1.0 lateral_hint = np.median(right_points - left_points, axis=0) lateral_axis = np.cross(plane_normal, longitudinal_axis) lateral_axis /= max(float(np.linalg.norm(lateral_axis)), 1e-12) if float(np.dot(lateral_axis, lateral_hint)) < 0: lateral_axis *= -1.0 anchor_x = float(profile_metadata["anchor_x"]) anchor_y = float(profile_metadata["anchor_y"]) anchor_points, anchor_depth_array, anchor_denominator_array = ( exact_plane_points_for( np.asarray([anchor_x], dtype=np.float64), np.asarray([anchor_y], dtype=np.float64), ) ) anchor_point = anchor_points[0] anchor_depth = float(anchor_depth_array[0]) if not np.isfinite(anchor_depth) or anchor_depth <= 0: raise ValueError("Obstacle anchor has no positive plane intersection") physical_widths = np.linalg.norm(right_points - left_points, axis=1) hidden_y_min = profile_metadata.get("hidden_y_min") hidden_y_max = profile_metadata.get("hidden_y_max") near_hidden_rows = np.ones(len(profile_rows), dtype=bool) if hidden_y_min is not None and hidden_y_max is not None: near_hidden_rows = ( (profile_rows >= float(hidden_y_min)) & (profile_rows <= float(hidden_y_max)) ) if not np.any(near_hidden_rows): near_hidden_rows = np.ones(len(profile_rows), dtype=bool) control_depth_median = float(np.median(control_depths)) stable_width_rows = ( (np.abs(left_denominator) >= 0.03) & (np.abs(right_denominator) >= 0.03) & (left_depth <= 4.0 * control_depth_median) & (right_depth <= 4.0 * control_depth_median) ) valid_widths = physical_widths[ near_hidden_rows & stable_width_rows & np.isfinite(physical_widths) & (physical_widths > 1e-6) ] if valid_widths.size == 0: valid_widths = physical_widths[ stable_width_rows & np.isfinite(physical_widths) & (physical_widths > 1e-6) ] if valid_widths.size == 0: raise ValueError("Corridor has no valid local physical width") channel_width_estimate = float(np.median(valid_widths)) hidden_bool = ( hidden_mask.astype(bool) if hidden_mask is not None else np.zeros_like(footprint_mask, dtype=bool) ) hidden_y_pixels, hidden_x_pixels = np.nonzero(hidden_bool) if hidden_x_pixels.size: hidden_points, hidden_depths, hidden_denominators = exact_plane_points_for( hidden_x_pixels.astype(np.float64), hidden_y_pixels.astype(np.float64), ) valid_hidden_points = ( np.isfinite(hidden_depths) & (hidden_depths > 0) & np.all(np.isfinite(hidden_points), axis=1) ) if int(np.count_nonzero(valid_hidden_points)) != len(hidden_x_pixels): raise ValueError( "Every reviewed hidden pixel must have a positive finite " "intersection with the fitted local surface plane" ) hidden_points = hidden_points[valid_hidden_points] hidden_depths = hidden_depths[valid_hidden_points] hidden_denominators = hidden_denominators[valid_hidden_points] hidden_point_image_x = hidden_x_pixels[valid_hidden_points] hidden_point_image_y = hidden_y_pixels[valid_hidden_points] else: hidden_depths = np.empty((0,), dtype=np.float64) hidden_denominators = np.empty((0,), dtype=np.float64) hidden_points = np.empty((0, 3), dtype=np.float64) hidden_point_image_x = np.empty((0,), dtype=np.int64) hidden_point_image_y = np.empty((0,), dtype=np.int64) if len(hidden_points): hidden_relative = hidden_points - anchor_point[None, :] hidden_local_x = hidden_relative @ lateral_axis hidden_local_z = hidden_relative @ longitudinal_axis stable_hidden_points = ( (np.abs(hidden_denominators) >= 0.03) & (hidden_depths <= 4.0 * control_depth_median) ) if int(np.count_nonzero(stable_hidden_points)) < max( 16, int(math.ceil(0.25 * len(hidden_points))), ): stable_hidden_points = np.ones(len(hidden_points), dtype=bool) stable_hidden_x = hidden_local_x[stable_hidden_points] stable_hidden_z = hidden_local_z[stable_hidden_points] # The full reviewed hidden support is a hard geometric requirement. # Stable subsets guide centering only; no percentile is allowed to # silently drop a small obstacle component or a thin occluded edge. hidden_x_low = float(np.min(hidden_local_x)) hidden_x_high = float(np.max(hidden_local_x)) hidden_z_low = float(np.min(hidden_local_z)) hidden_z_high = float(np.max(hidden_local_z)) context_center_x = float(np.median(stable_hidden_x)) context_center_z = float(np.median(stable_hidden_z)) else: stable_hidden_points = np.empty((0,), dtype=bool) hidden_x_low = hidden_x_high = 0.0 hidden_z_low = hidden_z_high = 0.0 context_center_x = 0.0 context_center_z = 0.0 hidden_width_span = float(max(hidden_x_high - hidden_x_low, 0.0)) hidden_length_span = float(max(hidden_z_high - hidden_z_low, 0.0)) # Keep local coordinates anchored at the robust obstacle ray. The desired # interval itself is asymmetric, so foreground and horizon context can be # retained without translating the camera-space plane or uniformly # shrinking both physical axes. context_origin = anchor_point.copy() center_relative = center_points - context_origin[None, :] center_longitudinal = center_relative @ longitudinal_axis center_lateral = center_relative @ lateral_axis stable_center_rows = ( (np.abs(center_denominator) >= 0.03) & (center_depth <= 4.0 * control_depth_median) ) if np.any(stable_center_rows): stable_center_longitudinal = center_longitudinal[stable_center_rows] stable_center_lateral = center_lateral[stable_center_rows] else: stable_center_longitudinal = center_longitudinal stable_center_lateral = center_lateral available_centerline_span = float( np.percentile(stable_center_longitudinal, 95.0) - np.percentile(stable_center_longitudinal, 5.0) ) raw_width = float( max(1.20 * channel_width_estimate, 1.15 * hidden_width_span) ) raw_length = float( max( 1.80 * channel_width_estimate, 1.35 * hidden_length_span, 0.55 * available_centerline_span, ) ) if raw_width <= 1e-6 or raw_length <= 1e-6: raise ValueError("Depth-scaled context patch dimensions collapsed") metric_dimension_bounds = { "curb_cut": { "width": (1.4, 4.5), "length": (2.5, 6.0), }, "tactile_paving": { "width": (0.8, 2.5), "length": (2.5, 8.0), }, "walkway": { "width": (1.4, 5.5), "length": (3.0, 8.0), }, "ramp": { "width": (1.2, 4.0), "length": (2.5, 7.0), }, } category_metric_bounds = metric_dimension_bounds.get(category) requested_width = raw_width requested_length = raw_length if metric_style_depth and category_metric_bounds is not None: requested_width = float( np.clip(requested_width, *category_metric_bounds["width"]) ) requested_length = float( np.clip(requested_length, *category_metric_bounds["length"]) ) def interval_containing_required_span( center: float, span: float, required_low: float, required_high: float, margin: float, ) -> tuple[float, float]: required_span = max(required_high - required_low, 0.0) + 2.0 * margin span = max(span, required_span, 1e-6) low = center - 0.5 * span high = center + 0.5 * span required_low -= margin required_high += margin if required_low < low: shift = required_low - low low += shift high += shift if required_high > high: shift = required_high - high low += shift high += shift return float(low), float(high) profile_center_x = float(np.median(stable_center_lateral)) if len(hidden_points): lateral_center = 0.65 * context_center_x + 0.35 * profile_center_x longitudinal_center = context_center_z else: lateral_center = profile_center_x longitudinal_center = float(np.median(stable_center_longitudinal)) lateral_margin = 0.08 * channel_width_estimate longitudinal_margin = 0.12 * channel_width_estimate desired_x_low, desired_x_high = interval_containing_required_span( lateral_center, requested_width, float(hidden_x_low), float(hidden_x_high), lateral_margin, ) desired_z_low, desired_z_high = interval_containing_required_span( longitudinal_center, requested_length, float(hidden_z_low), float(hidden_z_high), longitudinal_margin, ) def clip_convex_polygon_axis( polygon: np.ndarray, *, axis: int, bound: float, keep_greater: bool, ) -> np.ndarray: if len(polygon) == 0: return polygon clipped: list[np.ndarray] = [] def inside(point: np.ndarray) -> bool: value = float(point[axis]) return value >= bound - 1e-10 if keep_greater else value <= bound + 1e-10 previous = polygon[-1] previous_inside = inside(previous) for current in polygon: current_inside = inside(current) if current_inside != previous_inside: denominator = float(current[axis] - previous[axis]) if abs(denominator) > 1e-12: fraction = (bound - float(previous[axis])) / denominator clipped.append(previous + fraction * (current - previous)) if current_inside: clipped.append(current) previous = current previous_inside = current_inside return np.asarray(clipped, dtype=np.float64).reshape((-1, 2)) footprint_bool = footprint_mask.astype(bool) row_counts = np.count_nonzero(footprint_bool, axis=1) core_rows = np.flatnonzero(row_counts >= max(1, int(math.ceil(0.10 * w)))) footprint_y, footprint_x = np.nonzero(footprint_bool) if core_rows.size: roi_y_low = int(core_rows[0]) roi_y_high = int(core_rows[-1]) elif footprint_y.size: roi_y_low = int(np.min(footprint_y)) roi_y_high = int(np.max(footprint_y)) else: raise ValueError("Context ROI requires a non-empty reviewed footprint") if hidden_y_pixels.size: roi_y_low = min(roi_y_low, int(np.min(hidden_y_pixels))) roi_y_high = max(roi_y_high, int(np.max(hidden_y_pixels))) roi_y_margin = max(1, int(math.ceil(0.01 * h))) roi_y_low = max(0, roi_y_low - roi_y_margin) roi_y_high = min(h - 1, roi_y_high + roi_y_margin) roi_height = roi_y_high - roi_y_low + 1 column_counts = np.count_nonzero( footprint_bool[roi_y_low : roi_y_high + 1], axis=0, ) core_columns = np.flatnonzero( column_counts >= max(1, int(math.ceil(0.15 * roi_height))) ) if core_columns.size: roi_x_low = int(core_columns[0]) roi_x_high = int(core_columns[-1]) elif footprint_x.size: roi_x_low = int(np.min(footprint_x)) roi_x_high = int(np.max(footprint_x)) else: raise ValueError("Context ROI requires horizontal footprint support") if hidden_x_pixels.size: roi_x_low = min(roi_x_low, int(np.min(hidden_x_pixels))) roi_x_high = max(roi_x_high, int(np.max(hidden_x_pixels))) roi_x_margin = max(1, int(math.ceil(0.02 * w))) roi_x_low = max(0, roi_x_low - roi_x_margin) roi_x_high = min(w - 1, roi_x_high + roi_x_margin) image_margin = 0.5 image_context_roi = { "x_min": float(max(image_margin, roi_x_low)), "x_max": float(min(w - 1.0 - image_margin, roi_x_high)), "y_min": float(max(image_margin, roi_y_low)), "y_max": float(min(h - 1.0 - image_margin, roi_y_high)), } image_boundary_polygon = np.asarray( [ [image_margin, image_margin], [w - 1.0 - image_margin, image_margin], [w - 1.0 - image_margin, h - 1.0 - image_margin], [image_margin, h - 1.0 - image_margin], ], dtype=np.float64, ) plane_depth_sign = 1.0 if -plane_d >= 0.0 else -1.0 minimum_required_hidden_denominator = ( float(np.min(np.abs(hidden_denominators))) if len(hidden_denominators) else 0.03 ) plane_stability_floor = max( 1e-4, min(0.03, 0.98 * minimum_required_hidden_denominator), ) def stable_plane_halfspace_value(point: np.ndarray) -> float: image_x_value = float(point[0]) image_y_value = float(point[1]) denominator = ( plane_normal[0] * (image_x_value - cx) / fx + plane_normal[1] * (image_y_value - cy) / fy + plane_normal[2] ) return plane_depth_sign * float(denominator) - plane_stability_floor clipped_image_polygon: list[np.ndarray] = [] previous = image_boundary_polygon[-1] previous_value = stable_plane_halfspace_value(previous) for current in image_boundary_polygon: current_value = stable_plane_halfspace_value(current) previous_inside = previous_value >= -1e-10 current_inside = current_value >= -1e-10 if previous_inside != current_inside: value_delta = previous_value - current_value if abs(value_delta) <= 1e-12: raise ValueError("Plane-stability half-space edge collapsed") fraction = previous_value / value_delta clipped_image_polygon.append( previous + np.clip(fraction, 0.0, 1.0) * (current - previous) ) if current_inside: clipped_image_polygon.append(current) previous = current previous_value = current_value image_boundary_polygon = np.asarray( clipped_image_polygon, dtype=np.float64, ).reshape((-1, 2)) if len(image_boundary_polygon) < 3: raise ValueError("The reviewed context ROI has no stable positive plane region") frustum_points, frustum_depths, frustum_denominators = exact_plane_points_for( image_boundary_polygon[:, 0], image_boundary_polygon[:, 1], ) if ( not np.all(np.isfinite(frustum_points)) or not np.all(np.isfinite(frustum_depths)) or np.any(frustum_depths <= 0) ): raise ValueError("The local surface plane does not span the camera image") frustum_relative = frustum_points - context_origin[None, :] context_polygon = np.stack( ( frustum_relative @ lateral_axis, frustum_relative @ longitudinal_axis, ), axis=1, ) for axis, bound, keep_greater in ( (0, desired_x_low, True), (0, desired_x_high, False), (1, desired_z_low, True), (1, desired_z_high, False), ): context_polygon = clip_convex_polygon_axis( context_polygon, axis=axis, bound=bound, keep_greater=keep_greater, ) if len(context_polygon) < 3: raise ValueError("Depth-scaled context patch does not intersect the camera field of view") polygon_area = 0.5 * abs( float( np.sum( context_polygon[:, 0] * np.roll(context_polygon[:, 1], -1) - context_polygon[:, 1] * np.roll(context_polygon[:, 0], -1) ) ) ) if not np.isfinite(polygon_area) or polygon_area <= 1e-8: raise ValueError("Camera-clipped context polygon collapsed") context_polygon_camera = ( context_origin[None, :] + context_polygon[:, :1] * lateral_axis[None, :] + context_polygon[:, 1:] * longitudinal_axis[None, :] ) context_polygon_depth = context_polygon_camera[:, 2] context_image_polygon = np.stack( ( fx * context_polygon_camera[:, 0] / context_polygon_depth + cx, fy * context_polygon_camera[:, 1] / context_polygon_depth + cy, ), axis=1, ) if not np.all(np.isfinite(context_image_polygon)): raise ValueError("Camera-clipped context polygon did not reproject finitely") projected_width = float(np.ptp(context_image_polygon[:, 0])) projected_height = float(np.ptp(context_image_polygon[:, 1])) polygon_metric_width = float(np.ptp(context_polygon[:, 0])) polygon_metric_length = float(np.ptp(context_polygon[:, 1])) target_metric_cell = max( max(polygon_metric_width, polygon_metric_length) / 480.0, min(polygon_metric_width, polygon_metric_length) / 120.0, 1e-4, ) across_count = int( np.clip(math.ceil(polygon_metric_width / target_metric_cell) + 1, 24, 481) ) row_count = int( np.clip(math.ceil(polygon_metric_length / target_metric_cell) + 1, 24, 481) ) def polygon_x_bounds_at_z(z_value: float) -> tuple[float, float]: intersections: list[float] = [] for edge_index in range(len(context_polygon)): start = context_polygon[edge_index] end = context_polygon[(edge_index + 1) % len(context_polygon)] z_start = float(start[1]) z_end = float(end[1]) if abs(z_end - z_start) <= 1e-12: if abs(z_value - z_start) <= 1e-9: intersections.extend((float(start[0]), float(end[0]))) continue fraction = (z_value - z_start) / (z_end - z_start) if -1e-9 <= fraction <= 1.0 + 1e-9: intersections.append( float(start[0] + np.clip(fraction, 0.0, 1.0) * (end[0] - start[0])) ) if len(intersections) < 2: raise ValueError("Could not scan-convert the camera-clipped context polygon") return float(min(intersections)), float(max(intersections)) polygon_z_low = float(np.min(context_polygon[:, 1])) polygon_z_high = float(np.max(context_polygon[:, 1])) length = polygon_z_high - polygon_z_low if length <= 1e-6: raise ValueError("Camera-clipped context patch has no longitudinal extent") endpoint_low = polygon_x_bounds_at_z(polygon_z_low) endpoint_high = polygon_x_bounds_at_z(polygon_z_high) maximum_polygon_width = float(np.ptp(context_polygon[:, 0])) if min( endpoint_low[1] - endpoint_low[0], endpoint_high[1] - endpoint_high[0], ) < max(1e-5, 0.01 * maximum_polygon_width): endpoint_inset = min(0.0025 * length, 0.02 * channel_width_estimate) polygon_z_low += endpoint_inset polygon_z_high -= endpoint_inset length = polygon_z_high - polygon_z_low z_values = np.linspace(polygon_z_low, polygon_z_high, row_count, dtype=np.float64) row_x_bounds = np.asarray( [polygon_x_bounds_at_z(float(value)) for value in z_values], dtype=np.float64, ) if np.any(row_x_bounds[:, 1] - row_x_bounds[:, 0] <= 1e-6): raise ValueError("Camera-clipped context grid contains a collapsed row") across_fraction = np.linspace(0.0, 1.0, across_count, dtype=np.float64) longitudinal_fraction = np.linspace(0.0, 1.0, row_count, dtype=np.float64) local_x = ( row_x_bounds[:, :1] + across_fraction[None, :] * (row_x_bounds[:, 1:] - row_x_bounds[:, :1]) ) local_z = np.broadcast_to(z_values[:, None], (row_count, across_count)) width = float(np.ptp(local_x)) length = float(np.ptp(local_z)) final_x_low = float(np.min(local_x)) final_x_high = float(np.max(local_x)) final_z_low = float(np.min(local_z)) final_z_high = float(np.max(local_z)) mesh_context_polygon = np.concatenate( ( np.stack((row_x_bounds[:, 0], z_values), axis=1), np.stack((row_x_bounds[::-1, 1], z_values[::-1]), axis=1), ), axis=0, ) mesh_context_polygon_camera = ( context_origin[None, :] + mesh_context_polygon[:, :1] * lateral_axis[None, :] + mesh_context_polygon[:, 1:] * longitudinal_axis[None, :] ) mesh_context_polygon_depth = mesh_context_polygon_camera[:, 2] context_image_polygon = np.stack( ( fx * mesh_context_polygon_camera[:, 0] / mesh_context_polygon_depth + cx, fy * mesh_context_polygon_camera[:, 1] / mesh_context_polygon_depth + cy, ), axis=1, ) polygon_area = 0.5 * abs( float( np.sum( mesh_context_polygon[:, 0] * np.roll(mesh_context_polygon[:, 1], -1) - mesh_context_polygon[:, 1] * np.roll(mesh_context_polygon[:, 0], -1) ) ) ) grid_camera_points = ( context_origin[None, None, :] + local_x[..., None] * lateral_axis[None, None, :] + local_z[..., None] * longitudinal_axis[None, None, :] ) if np.any(grid_camera_points[..., 2] <= 1e-6): raise ValueError("Metric context grid crossed the camera plane") grid_image_x = fx * grid_camera_points[..., 0] / grid_camera_points[..., 2] + cx grid_image_y = fy * grid_camera_points[..., 1] / grid_camera_points[..., 2] + cy if ( np.any(grid_image_x < 0) or np.any(grid_image_x > w - 1) or np.any(grid_image_y < 0) or np.any(grid_image_y > h - 1) ): raise ValueError("Metric context grid projected outside the source image") sampled_grid_colors = cv2.remap( colors, grid_image_x.astype(np.float32), grid_image_y.astype(np.float32), interpolation=cv2.INTER_LINEAR, borderMode=cv2.BORDER_REPLICATE, ).astype(np.uint8) projected_patch_mask = np.zeros((h, w), dtype=np.uint8) cv2.fillConvexPoly( projected_patch_mask, np.rint(context_image_polygon).astype(np.int32), 1, ) projected_patch_bool = projected_patch_mask.astype(bool) hidden_pixel_coverage_fraction = float( np.count_nonzero(projected_patch_bool & hidden_bool) / max(int(np.count_nonzero(hidden_bool)), 1) ) context_target_fraction = float( np.count_nonzero(projected_patch_bool & footprint_mask.astype(bool)) / max(int(np.count_nonzero(projected_patch_bool)), 1) ) continuous_hidden_coverage_fraction = 1.0 continuous_hidden_membership = np.ones(len(hidden_points), dtype=bool) if len(hidden_points): polygon_orientation = np.sign( np.sum( mesh_context_polygon[:, 0] * np.roll(mesh_context_polygon[:, 1], -1) - mesh_context_polygon[:, 1] * np.roll(mesh_context_polygon[:, 0], -1) ) ) if polygon_orientation == 0: raise ValueError("Metric context polygon has no winding") hidden_local_points = np.stack((hidden_local_x, hidden_local_z), axis=1) edge_start = mesh_context_polygon edge_delta = np.roll(mesh_context_polygon, -1, axis=0) - edge_start coordinate_scale = max( float(np.max(np.abs(mesh_context_polygon))), 1.0, ) continuous_hidden_membership = np.ones(len(hidden_local_points), dtype=bool) for start, delta in zip(edge_start, edge_delta): relative = hidden_local_points - start[None, :] signed_cross = ( delta[0] * relative[:, 1] - delta[1] * relative[:, 0] ) continuous_hidden_membership &= ( polygon_orientation * signed_cross >= -1e-7 * coordinate_scale ) if not np.any(continuous_hidden_membership): break continuous_hidden_coverage_fraction = float( np.count_nonzero(continuous_hidden_membership) / len(continuous_hidden_membership) ) hidden_component_coverage: list[dict[str, Any]] = [] minimum_hidden_component_coverage = 1.0 minimum_continuous_hidden_component_coverage = 1.0 if np.any(hidden_bool): component_count, component_labels = cv2.connectedComponents( hidden_bool.astype(np.uint8), connectivity=8, ) minimum_component_pixels = max( 8, int(math.ceil(0.001 * np.count_nonzero(hidden_bool))), ) for component_id in range(1, component_count): component = component_labels == component_id component_pixels = int(np.count_nonzero(component)) if component_pixels < minimum_component_pixels: continue coverage = float( np.count_nonzero(projected_patch_bool & component) / component_pixels ) point_component = ( component_labels[hidden_point_image_y, hidden_point_image_x] == component_id ) continuous_coverage = float( np.count_nonzero( continuous_hidden_membership[point_component] ) / max(int(np.count_nonzero(point_component)), 1) ) hidden_component_coverage.append( { "component_id": component_id, "pixels": component_pixels, "coverage_fraction": coverage, "continuous_local_coordinate_coverage_fraction": ( continuous_coverage ), } ) if hidden_component_coverage: minimum_hidden_component_coverage = min( float(row["coverage_fraction"]) for row in hidden_component_coverage ) minimum_continuous_hidden_component_coverage = min( float(row["continuous_local_coordinate_coverage_fraction"]) for row in hidden_component_coverage ) observed_vertical = -center_points[:, 1] variance_longitudinal = float(np.var(center_longitudinal)) vertical_slope = ( float( np.mean( (center_longitudinal - float(np.mean(center_longitudinal))) * (observed_vertical - float(np.mean(observed_vertical))) ) / max(variance_longitudinal, 1e-12) ) if variance_longitudinal > 1e-12 else 0.0 ) raw_depth_rise = abs(vertical_slope) * length rise_source = "local_plane_camera_y_gradient_without_gravity_calibration" rise_confidence = "low" if category == "ramp": minimum_rise = 0.015 * length maximum_rise = (1.0 / 12.0) * length rise = float(np.clip(raw_depth_rise, minimum_rise, maximum_rise)) if raw_depth_rise < minimum_rise or raw_depth_rise > maximum_rise: rise_source += "_with_accessibility_bounds" rise_confidence = "low" category_signature = "depth_scaled_uniform_local_channel_slope" maximum_semantic_relief = 0.0 elif category == "curb_cut": minimum_rise = ( max(0.012 * length, 0.04) if metric_style_depth else 0.012 * length ) maximum_rise = ( min((1.0 / 12.0) * length, 0.20) if metric_style_depth else (1.0 / 12.0) * length ) rise = float(np.clip(raw_depth_rise, minimum_rise, maximum_rise)) if raw_depth_rise < minimum_rise or raw_depth_rise > maximum_rise: rise_source += "_with_accessibility_bounds" rise_confidence = "low" category_signature = "complete_central_low_channel_with_lateral_flares" maximum_semantic_relief = float(0.65 * rise) elif category == "tactile_paving": rise = 0.0 rise_source = "near_level_with_scale_bounded_tactile_relief" rise_confidence = "category_verified" category_signature = "complete_channel_with_visible_tactile_ribs" maximum_semantic_relief = float( np.clip(0.006 * length, 0.008, 0.025) ) else: rise = float(min(raw_depth_rise, 0.02 * length)) if raw_depth_rise > 0.02 * length: rise_source += "_near_level_bound" rise_confidence = "low" category_signature = "depth_scaled_near_level_complete_walkway_channel" maximum_semantic_relief = 0.0 front_vertices: list[np.ndarray] = [] front_colors: list[np.ndarray] = [] sampled_inside_footprint = 0 sampled_inside_hidden = 0 rib_count = int(np.clip(round(length / 0.45), 6, 24)) for row_index, v in enumerate(longitudinal_fraction): image_x = grid_image_x[row_index] image_y = grid_image_y[row_index] sample_x = np.clip(np.rint(image_x), 0, w - 1).astype(np.int32) sample_y = np.clip(np.rint(image_y), 0, h - 1).astype(np.int32) sampled_inside_footprint += int( np.count_nonzero(footprint_mask[sample_y, sample_x]) ) sampled_inside_hidden += int( np.count_nonzero(hidden_bool[sample_y, sample_x]) ) x_coordinate = local_x[row_index] z_coordinate = local_z[row_index] if category == "ramp": height = np.full_like(x_coordinate, (1.0 - v) * rise) elif category == "curb_cut": lateral = np.abs(2.0 * across_fraction - 1.0) height = ( np.full_like(x_coordinate, (1.0 - v) * rise) + 0.65 * rise * v * lateral**3 ) elif category == "tactile_paving": rib_phase = 2.0 * math.pi * float(rib_count) * v rib = max(0.0, math.cos(rib_phase)) ** 6 height = np.full_like( x_coordinate, maximum_semantic_relief * rib, ) else: height = np.full_like(x_coordinate, (1.0 - v) * rise) front_vertices.extend( np.stack((x_coordinate, -height, z_coordinate), axis=1).astype( np.float32 ) ) front_colors.extend(sampled_grid_colors[row_index]) front_faces: list[tuple[int, int, int]] = [] for row_index in range(row_count - 1): row_start = row_index * across_count next_start = (row_index + 1) * across_count for column_index in range(across_count - 1): a = row_start + column_index b = a + 1 c = next_start + column_index d = c + 1 front_faces.append((a, c, b)) front_faces.append((b, c, d)) transformed_top = np.asarray(front_vertices, dtype=np.float32) top_colors = np.asarray(front_colors, dtype=np.uint8) footprint_diag = float(math.hypot(width, length)) thickness = float(np.clip(0.018 * footprint_diag, 0.025, 0.18)) back_vertices = transformed_top.copy() back_vertices[:, 1] += thickness front_vertex_count = len(transformed_top) back_faces = [ ( face[2] + front_vertex_count, face[1] + front_vertex_count, face[0] + front_vertex_count, ) for face in front_faces ] edge_counts: defaultdict[tuple[int, int], int] = defaultdict(int) for face in front_faces: for a, b in ( (face[0], face[1]), (face[1], face[2]), (face[2], face[0]), ): edge_counts[tuple(sorted((a, b)))] += 1 boundary_edges = [edge for edge, count in edge_counts.items() if count == 1] side_faces: list[tuple[int, int, int]] = [] for a, b in boundary_edges: side_faces.append((a, b, a + front_vertex_count)) side_faces.append((b, b + front_vertex_count, a + front_vertex_count)) all_faces = front_faces + back_faces + side_faces full_edge_counts: defaultdict[tuple[int, int], int] = defaultdict(int) for face in all_faces: for a, b in ( (face[0], face[1]), (face[1], face[2]), (face[2], face[0]), ): full_edge_counts[tuple(sorted((a, b)))] += 1 watertight_edge_check = bool( full_edge_counts and all(value == 2 for value in full_edge_counts.values()) ) if not watertight_edge_check: raise ValueError("Obstacle-centred corridor failed the watertight edge check") neutral = np.clip( np.median(top_colors.astype(np.float32), axis=0) * 0.52, 0, 255, ).astype(np.uint8) front_face_array = np.asarray(front_faces, dtype=np.int64) front_triangles = transformed_top[front_face_array].astype(np.float64) triangle_edges = np.stack( ( front_triangles[:, 1] - front_triangles[:, 0], front_triangles[:, 2] - front_triangles[:, 1], front_triangles[:, 0] - front_triangles[:, 2], ), axis=1, ) triangle_edge_lengths = np.linalg.norm(triangle_edges, axis=2) triangle_twice_area = np.linalg.norm( np.cross( front_triangles[:, 1] - front_triangles[:, 0], front_triangles[:, 2] - front_triangles[:, 0], ), axis=1, ) if np.any(~np.isfinite(triangle_twice_area)) or np.any( triangle_twice_area <= 1e-12 ): raise ValueError("Perspective-preserving corridor has degenerate front faces") triangle_area = 0.5 * triangle_twice_area triangle_longest_edge = np.max(triangle_edge_lengths, axis=1) triangle_aspect = triangle_longest_edge**2 / triangle_twice_area edge_length_median = float(np.median(triangle_edge_lengths)) triangle_area_median = float(np.median(triangle_area)) front_xz = transformed_top[:, (0, 2)] front_xz_triangles = front_xz[front_face_array] signed_twice_area_xz = ( (front_xz_triangles[:, 1, 0] - front_xz_triangles[:, 0, 0]) * (front_xz_triangles[:, 2, 1] - front_xz_triangles[:, 0, 1]) - (front_xz_triangles[:, 1, 1] - front_xz_triangles[:, 0, 1]) * (front_xz_triangles[:, 2, 0] - front_xz_triangles[:, 0, 0]) ) winding_sign = -1.0 if float(np.median(signed_twice_area_xz)) < 0 else 1.0 flipped_front_faces = int( np.count_nonzero(winding_sign * signed_twice_area_xz <= 1e-12) ) source_pixel_points = np.stack((grid_image_x, grid_image_y), axis=-1) source_pixel_edges = np.concatenate( ( np.linalg.norm(np.diff(source_pixel_points, axis=1), axis=2).reshape(-1), np.linalg.norm(np.diff(source_pixel_points, axis=0), axis=2).reshape(-1), ) ) source_pixel_edge_median = float(np.median(source_pixel_edges)) control_denominators = np.abs( np.concatenate((left_denominator, right_denominator, center_denominator)) ) grid_rays = np.stack( ( (grid_image_x - cx) / fx, (grid_image_y - cy) / fy, np.ones_like(grid_image_x), ), axis=-1, ) grid_denominators = np.abs(grid_rays @ plane_normal) grid_depths = grid_camera_points[..., 2] fit_residuals = np.abs(camera_points[fit_mask] @ plane_normal + plane_d) desired_rectangle_area = max( (desired_x_high - desired_x_low) * (desired_z_high - desired_z_low), 1e-12, ) field_of_view_retained_area_fraction = float( np.clip(polygon_area / desired_rectangle_area, 0.0, 1.0) ) field_of_view_scale = float(math.sqrt(field_of_view_retained_area_fraction)) geometry_quality = { "front_triangle_area_max_to_median": float( np.max(triangle_area) / max(triangle_area_median, 1e-12) ), "front_edge_length_max_to_median": float( np.max(triangle_edge_lengths) / max(edge_length_median, 1e-12) ), "front_triangle_aspect_p99": float(np.percentile(triangle_aspect, 99.0)), "front_triangle_aspect_max": float(np.max(triangle_aspect)), "front_triangle_min_area": float(np.min(triangle_area)), "front_triangle_median_area": triangle_area_median, "front_edge_median_length": edge_length_median, "actual_width_span": float(np.ptp(transformed_top[:, 0])), "actual_length_span": float(np.ptp(transformed_top[:, 2])), "reported_width_absolute_error": float( abs(float(np.ptp(transformed_top[:, 0])) - width) ), "reported_length_absolute_error": float( abs(float(np.ptp(transformed_top[:, 2])) - length) ), "flipped_or_degenerate_front_faces_xz": flipped_front_faces, "basis_lateral_longitudinal_dot_absolute": float( abs(float(np.dot(lateral_axis, longitudinal_axis))) ), "basis_lateral_normal_dot_absolute": float( abs(float(np.dot(lateral_axis, plane_normal))) ), "basis_longitudinal_normal_dot_absolute": float( abs(float(np.dot(longitudinal_axis, plane_normal))) ), "source_pixel_edge_max_to_median": float( np.max(source_pixel_edges) / max(source_pixel_edge_median, 1e-12) ), "source_pixel_edge_min_to_median": float( np.min(source_pixel_edges) / max(source_pixel_edge_median, 1e-12) ), "source_pixel_edge_median": source_pixel_edge_median, "source_pixel_edge_p99": float( np.percentile(source_pixel_edges, 99.0) ), "source_pixel_edge_max": float(np.max(source_pixel_edges)), "control_ray_plane_denominator_min_abs": float( np.min(control_denominators) ), "control_plane_depth_max_to_median": float( np.max(control_depths) / max(float(np.median(control_depths)), 1e-12) ), "context_grid_ray_plane_denominator_min_abs": float( np.min(grid_denominators) ), "context_grid_plane_depth_max_to_median": float( np.max(grid_depths) / max(float(np.median(grid_depths)), 1e-12) ), "plane_fit_residual_median": float(np.median(fit_residuals)), "plane_fit_residual_p95": float(np.percentile(fit_residuals, 95.0)), "field_of_view_retained_area_fraction": ( field_of_view_retained_area_fraction ), "hidden_pixel_coverage_fraction": hidden_pixel_coverage_fraction, "continuous_local_coordinate_hidden_coverage_fraction": ( continuous_hidden_coverage_fraction ), "minimum_hidden_component_coverage_fraction": ( minimum_hidden_component_coverage ), "minimum_continuous_hidden_component_coverage_fraction": ( minimum_continuous_hidden_component_coverage ), "context_target_fraction": context_target_fraction, "profile_anchor_y_inside_sampled_rows": bool( float(profile_metadata["row_min"]) <= float(profile_metadata["anchor_y"]) <= float(profile_metadata["row_max"]) ), } back_colors = np.repeat(neutral[None, :], front_vertex_count, axis=0) vertices = np.concatenate((transformed_top, back_vertices), axis=0) vertex_colors = np.concatenate((top_colors, back_colors), axis=0) sample_count = row_count * across_count width_bound_applied = bool( metric_style_depth and category_metric_bounds is not None and not ( category_metric_bounds["width"][0] <= raw_width <= category_metric_bounds["width"][1] ) ) length_bound_applied = bool( metric_style_depth and category_metric_bounds is not None and not ( category_metric_bounds["length"][0] <= raw_length <= category_metric_bounds["length"][1] ) ) return vertices, vertex_colors, np.asarray(all_faces, dtype=np.int64), { "front_faces": len(front_faces), "back_faces": len(back_faces), "side_faces": len(side_faces), "boundary_side_faces": len(side_faces), "wedge_support_faces": 0, "stair_sidewall_faces": 0, "stair_sidewall_bottom_faces": 0, "solid_thickness": thickness, "max_depth_jump": None, "stride": int(stride), "support_mode": "obstacle_centered_metric_context_patch_slab", "surface_model": "obstacle_centered_depth_scaled_context_surface", "complete_footprint": False, "complete_obstacle_zone": bool( not np.any(hidden_bool) or ( hidden_pixel_coverage_fraction >= 0.98 and minimum_hidden_component_coverage >= 0.90 and continuous_hidden_coverage_fraction >= 0.999 and minimum_continuous_hidden_component_coverage >= 0.999 ) ), "open_world_channel_model": False, "local_context_patch_model": True, "open_world_scene_compatible": True, "open_world_topology_claim": False, "context_pixels_are_target_claim": False, "footprint_policy": ( "asymmetric_depth_scaled_local_plane_context_polygon_clipped_to_" "the_camera_frustum_and_required_hidden_extent" ), "texture_coordinate_policy": ( "regular_metric_plane_grid_reprojected_to_frozen_completed_rgb_" "original_image_xy_with_bilinear_sampling_without_extrapolation" ), "texture_mapping": "metric_plane_grid_reprojected_to_original_image", "texture_geometry_warp": ( "metric_plane_pinhole_reprojection_plus_bounded_semantic_height" ), "independent_axis_normalization_used": False, "uniform_geometry_scale": 1.0, "uniform_geometry_scale_applied": False, "texture_resampling": "bilinear_original_image_pixel", "texture_extrapolation_used": False, "texture_extension_max_distance_pixels": 0.0, "texture_profile_inside_reviewed_footprint_fraction": float( sampled_inside_footprint / max(sample_count, 1) ), "texture_profile_hidden_completion_fraction": float( sampled_inside_hidden / max(sample_count, 1) ), "projected_context_hidden_pixel_coverage_fraction": ( hidden_pixel_coverage_fraction ), "projected_context_target_fraction": context_target_fraction, "projected_context_hidden_component_coverage": ( hidden_component_coverage ), "projected_context_minimum_hidden_component_coverage_fraction": ( minimum_hidden_component_coverage ), "continuous_local_coordinate_hidden_coverage_fraction": ( continuous_hidden_coverage_fraction ), "minimum_continuous_hidden_component_coverage_fraction": ( minimum_continuous_hidden_component_coverage ), "all_hidden_components_covered": bool( continuous_hidden_coverage_fraction >= 0.999 and minimum_continuous_hidden_component_coverage >= 0.999 ), "projected_context_image_corners": [ [float(value) for value in corner] for corner in context_image_polygon ], "projected_context_image_polygon": [ [float(value) for value in point] for point in context_image_polygon ], "context_roi_image_xyxy": image_context_roi, "context_roi_policy": ( "reviewed_target_core_plus_all_hidden_components_for_diagnostics_" "while_camera_frustum_clipping_uses_the_full_source_image" ), "category_geometry_signature": category_signature, "maximum_semantic_relief": maximum_semantic_relief, "front_boundary_loops": 1, "final_cell_holes_filled": True, "regular_grid_has_no_internal_topological_holes": True, "final_enclosed_hole_added_cells": 0, "watertight_edge_check": watertight_edge_check, "calibrated_metric_truth": False, "metric_style_depth_input": bool(metric_style_depth), "automatic_passability_claim": False, "dimension_source": ( "asymmetric_hidden_component_local_ray_plane_extent_with_category_" "roi_bounds_and_camera_frustum_polygon_clip" ), "rise_source": rise_source, "rise_confidence": rise_confidence, "raw_depth_rise": float(raw_depth_rise), "raw_depth_scaled_dimensions": { "width": raw_width, "length": raw_length, "robust_centerline_length": raw_centerline_length, "channel_width_estimate": channel_width_estimate, "hidden_width_span_full_reviewed_support": hidden_width_span, "hidden_length_span_full_reviewed_support": hidden_length_span, "stable_hidden_point_fraction": float( np.count_nonzero(stable_hidden_points) / max(len(stable_hidden_points), 1) ), }, "dimension_bounds": { "independent_texture_axis_normalization_used": False, "physical_roi_axis_bounds_used": bool( width_bound_applied or length_bound_applied ), "metric_style_width_min": ( category_metric_bounds["width"][0] if metric_style_depth and category_metric_bounds is not None else None ), "metric_style_width_max": ( category_metric_bounds["width"][1] if metric_style_depth and category_metric_bounds is not None else None ), "metric_style_length_min": ( category_metric_bounds["length"][0] if metric_style_depth and category_metric_bounds is not None else None ), "metric_style_length_max": ( category_metric_bounds["length"][1] if metric_style_depth and category_metric_bounds is not None else None ), "width_bound_applied": width_bound_applied, "length_bound_applied": length_bound_applied, "hard_required_local_extents": { "x": [float(hidden_x_low), float(hidden_x_high)], "z": [float(hidden_z_low), float(hidden_z_high)], }, "desired_local_extents": { "x": [desired_x_low, desired_x_high], "z": [desired_z_low, desired_z_high], }, "final_local_extents": { "x": [final_x_low, final_x_high], "z": [final_z_low, final_z_high], }, "required_core_clipped": bool( continuous_hidden_coverage_fraction < 0.999 ), "required_width_exceeds_recommended_max": bool( metric_style_depth and category_metric_bounds is not None and hidden_width_span > category_metric_bounds["width"][1] ), "required_length_exceeds_recommended_max": bool( metric_style_depth and category_metric_bounds is not None and hidden_length_span > category_metric_bounds["length"][1] ), "camera_field_of_view_scale": field_of_view_scale, "camera_field_of_view_retained_area_fraction": ( field_of_view_retained_area_fraction ), "plane_stability_denominator_floor": plane_stability_floor, "minimum_required_hidden_denominator_abs": ( minimum_required_hidden_denominator ), "camera_field_of_view_crop_applied": not math.isclose( field_of_view_retained_area_fraction, 1.0, rel_tol=1e-8, ), "width_below_recommended_minimum": bool( metric_style_depth and category_metric_bounds is not None and width < category_metric_bounds["width"][0] ), "length_below_recommended_minimum": bool( metric_style_depth and category_metric_bounds is not None and length < category_metric_bounds["length"][0] ), }, "surface_dimensions": { "width": width, "length": length, "rise": float(rise), "thickness": thickness, "footprint_diag": footprint_diag, "units": ( "metric_style_depth_units_not_calibrated_truth" if metric_style_depth else "relative_pseudo_depth_units" ), }, "depth_scaled_dimensions": { "width": width, "length": length, "height": float(rise), "thickness": thickness, "footprint_diag": footprint_diag, "raw_width": raw_width, "raw_length": raw_length, "dimension_source": ( "asymmetric_hidden_component_local_ray_plane_extent_with_" "category_roi_bounds_and_camera_frustum_polygon_clip" ), "calibrated_metric_truth": False, "depth_units": ( "metric_style_depth_units_not_calibrated_truth" if metric_style_depth else "relative_pseudo_depth_units" ), }, "corridor_profile": profile_metadata, "corridor_grid": { "rows": row_count, "columns": across_count, "front_vertices": front_vertex_count, "rib_count": rib_count if category == "tactile_paving" else 0, "metric_cell_width": float( np.median(row_x_bounds[:, 1] - row_x_bounds[:, 0]) / max(across_count - 1, 1) ), "metric_cell_length": float(length / max(row_count - 1, 1)), "target_metric_cell_size": target_metric_cell, "projected_width_pixels": projected_width, "projected_height_pixels": projected_height, "row_width_min": float( np.min(row_x_bounds[:, 1] - row_x_bounds[:, 0]) ), "row_width_max": float( np.max(row_x_bounds[:, 1] - row_x_bounds[:, 0]) ), "regular_longitudinal_spacing": True, "variable_lateral_width_due_to_camera_frustum_clip": bool( field_of_view_retained_area_fraction < 1.0 - 1e-8 ), }, "geometry_quality": geometry_quality, "local_plane_basis": { "lateral_axis": [float(value) for value in lateral_axis], "longitudinal_axis": [float(value) for value in longitudinal_axis], "normal_axis": [float(value) for value in plane_normal], "context_origin_camera": [float(value) for value in context_origin], "anchor_point_camera": [float(value) for value in anchor_point], "anchor_depth": anchor_depth, "hidden_median_center_offset": [ context_center_x, context_center_z, ], "desired_local_bounds": { "x": [desired_x_low, desired_x_high], "z": [desired_z_low, desired_z_high], }, }, "plane_normal": [float(value) for value in plane_normal], "plane_d": plane_d, "plane_fit_pixels": int(fit_mask.sum()), "plane_fit_residual_median": float(np.median(fit_residuals)), "plane_fit_residual_p95": float(np.percentile(fit_residuals, 95.0)), } def build_hybrid_depth_scaled_surface_mesh( depth: np.ndarray, colors: np.ndarray, footprint_mask: np.ndarray, *, category: str, stride: int, minimum_cell_coverage: float = 0.15, plane_validation_mask: np.ndarray | None = None, metric_style_depth: bool = False, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, dict[str, Any]]: """Build a watertight, depth-scaled instance surface with direct RGB. The front topology and every color sample retain the reviewed image-space footprint. Ray/plane reconstruction supplies physical-like lateral and longitudinal dimensions; only accessibility-plausibility bounds are used when the monocular vertical gradient is weak or implausible. A thin neutral underside makes the open-world patch stable in oblique views without stretching photographic texture down legacy walls. """ top_vertices, top_colors, top_faces, perspective = ( build_perspective_continuous_surface_mesh( depth, colors, footprint_mask, category=category, stride=stride, minimum_cell_coverage=minimum_cell_coverage, plane_validation_mask=plane_validation_mask, texture_support_mask=footprint_mask, ) ) if len(top_vertices) < 4 or len(top_faces) < 2: raise ValueError("Hybrid instance surface is too small") h, w = depth.shape fx, fy, cx, cy = camera_intrinsics(w, h, None, None, None, None) image_x = np.clip( np.rint(top_vertices[:, 0] * fx / top_vertices[:, 2] + cx), 0, w - 1, ).astype(np.int32) image_y = np.clip( np.rint(top_vertices[:, 1] * fy / top_vertices[:, 2] + cy), 0, h - 1, ).astype(np.int32) center = top_vertices.mean(axis=0) centered = top_vertices - center pixels = np.stack( ( image_x.astype(np.float64) - float(np.mean(image_x)), image_y.astype(np.float64) - float(np.mean(image_y)), ), axis=1, ) coefficients, _, rank, _ = np.linalg.lstsq( pixels, centered.astype(np.float64), rcond=None, ) axis_x = coefficients[0].astype(np.float64) axis_y = coefficients[1].astype(np.float64) if rank < 2 or np.linalg.norm(axis_x) < 1e-8: _, _, vh = np.linalg.svd(centered, full_matrices=False) axis_x = vh[0].astype(np.float64) axis_y = vh[1].astype(np.float64) axis_x /= max(float(np.linalg.norm(axis_x)), 1e-8) axis_y -= axis_x * float(np.dot(axis_y, axis_x)) if np.linalg.norm(axis_y) < 1e-8: _, _, vh = np.linalg.svd(centered, full_matrices=False) axis_y = vh[1].astype(np.float64) axis_y -= axis_x * float(np.dot(axis_y, axis_x)) axis_y /= max(float(np.linalg.norm(axis_y)), 1e-8) local_x = centered @ axis_x local_y = centered @ axis_y if np.corrcoef(local_x, image_x.astype(np.float64))[0, 1] < 0: axis_x *= -1.0 local_x *= -1.0 if np.corrcoef(local_y, image_y.astype(np.float64))[0, 1] < 0: axis_y *= -1.0 local_y *= -1.0 x_low, x_high = np.percentile(local_x, [2.0, 98.0]) y_low, y_high = np.percentile(local_y, [2.0, 98.0]) width = max(float(x_high - x_low), 1e-4) length = max(float(y_high - y_low), 1e-4) local_x_center = 0.5 * float(x_low + x_high) local_y_center = 0.5 * float(y_low + y_high) canonical_x = local_x - local_x_center canonical_y = local_y - local_y_center v = np.clip((local_y - y_low) / length, 0.0, 1.0) u = np.clip((local_x - x_low) / width, 0.0, 1.0) observed_vertical = -top_vertices[:, 1].astype(np.float64) variance_y = float(np.var(local_y)) vertical_slope = ( float( np.mean( (local_y - float(np.mean(local_y))) * (observed_vertical - float(np.mean(observed_vertical))) ) / max(variance_y, 1e-12) ) if variance_y > 1e-12 else 0.0 ) raw_depth_rise = abs(vertical_slope) * length rise_source = "depth_plane_vertical_gradient" rise_confidence = "medium" if category == "ramp": minimum_rise = 0.015 * length maximum_rise = 0.12 * length rise = float(np.clip(raw_depth_rise, minimum_rise, maximum_rise)) if raw_depth_rise < minimum_rise or raw_depth_rise > maximum_rise: rise_source += "_with_accessibility_bounds" rise_confidence = "low" height = (1.0 - v) * rise category_signature = "depth_scaled_uniform_longitudinal_slope" maximum_semantic_relief = 0.0 elif category == "curb_cut": minimum_rise = 0.012 * length maximum_rise = 0.10 * length rise = float(np.clip(raw_depth_rise, minimum_rise, maximum_rise)) if raw_depth_rise < minimum_rise or raw_depth_rise > maximum_rise: rise_source += "_with_accessibility_bounds" rise_confidence = "low" lateral = np.abs(2.0 * u - 1.0) flare = 0.65 * rise * v * lateral**3 height = (1.0 - v) * rise + flare category_signature = "depth_scaled_central_low_channel_with_lateral_flares" maximum_semantic_relief = float(np.max(flare)) elif category == "tactile_paving": rise = 0.0 rise_source = "near_level_with_scale_bounded_tactile_relief" rise_confidence = "category_verified" rib_height = float(np.clip(0.006 * length, 0.006, 0.018)) rib_phase = 2.0 * math.pi * 10.0 * v height = rib_height * np.maximum(0.0, np.cos(rib_phase)) ** 6 category_signature = "instance_footprint_with_low_transverse_tactile_ribs" maximum_semantic_relief = rib_height else: rise = float(min(raw_depth_rise, 0.02 * length)) height = (1.0 - v) * rise category_signature = "depth_scaled_near_level_instance_walkway" maximum_semantic_relief = 0.0 if raw_depth_rise > 0.02 * length: rise_source += "_near_level_bound" rise_confidence = "low" transformed_top = np.stack( ( canonical_x, -height, canonical_y, ), axis=1, ).astype(np.float32) footprint_diag = float(math.hypot(width, length)) thickness = float(np.clip(0.018 * footprint_diag, 0.025, 0.22)) back_vertices = transformed_top.copy() back_vertices[:, 1] += thickness front_vertex_count = len(transformed_top) front_faces = [tuple(int(value) for value in face) for face in top_faces] edge_counts: defaultdict[tuple[int, int], int] = defaultdict(int) for face in front_faces: for a, b in ( (face[0], face[1]), (face[1], face[2]), (face[2], face[0]), ): edge_counts[tuple(sorted((a, b)))] += 1 boundary_edges = [edge for edge, count in edge_counts.items() if count == 1] boundary_adjacency: defaultdict[int, set[int]] = defaultdict(set) for a, b in boundary_edges: boundary_adjacency[a].add(b) boundary_adjacency[b].add(a) if not boundary_adjacency or any( len(neighbors) != 2 for neighbors in boundary_adjacency.values() ): raise ValueError("Hybrid front boundary is not a simple closed loop") remaining_boundary = set(boundary_adjacency) boundary_loop_count = 0 while remaining_boundary: boundary_loop_count += 1 stack = [next(iter(remaining_boundary))] while stack: current = stack.pop() if current not in remaining_boundary: continue remaining_boundary.remove(current) stack.extend(boundary_adjacency[current]) if boundary_loop_count != 1: raise ValueError( "Hybrid instance surface must have exactly one outer boundary; " f"found {boundary_loop_count}" ) back_faces = [ ( face[2] + front_vertex_count, face[1] + front_vertex_count, face[0] + front_vertex_count, ) for face in front_faces ] side_faces: list[tuple[int, int, int]] = [] for a, b in boundary_edges: side_faces.append((a, b, a + front_vertex_count)) side_faces.append((b, b + front_vertex_count, a + front_vertex_count)) all_faces = front_faces + back_faces + side_faces full_edge_counts: defaultdict[tuple[int, int], int] = defaultdict(int) for face in all_faces: for a, b in ( (face[0], face[1]), (face[1], face[2]), (face[2], face[0]), ): full_edge_counts[tuple(sorted((a, b)))] += 1 watertight_edge_check = bool( full_edge_counts and all(value == 2 for value in full_edge_counts.values()) ) if not watertight_edge_check: raise ValueError("Hybrid instance extrusion failed the watertight edge check") neutral = np.clip( np.median(top_colors.astype(np.float32), axis=0) * 0.52, 0, 255, ).astype(np.uint8) back_colors = np.repeat(neutral[None, :], front_vertex_count, axis=0) vertices = np.concatenate((transformed_top, back_vertices), axis=0) vertex_colors = np.concatenate((top_colors, back_colors), axis=0) return vertices, vertex_colors, np.asarray(all_faces, dtype=np.int64), { "front_faces": len(front_faces), "back_faces": len(back_faces), "side_faces": len(side_faces), "boundary_side_faces": len(side_faces), "wedge_support_faces": 0, "stair_sidewall_faces": 0, "stair_sidewall_bottom_faces": 0, "solid_thickness": thickness, "max_depth_jump": None, "stride": int(stride), "support_mode": "hybrid_depth_scaled_instance_surface_slab", "surface_model": "hybrid_depth_scaled_instance_accessibility_surface", "complete_footprint": True, "footprint_policy": ( "smoothed_visible_anchored_reviewed_instance_topology_without_hull" ), "texture_coordinate_policy": ( "direct_frozen_completed_rgb_at_original_image_xy_without_donor_" "extrapolation" ), "texture_extrapolation_used": False, "texture_extension_max_distance_pixels": 0.0, "category_geometry_signature": category_signature, "maximum_semantic_relief": maximum_semantic_relief, "front_boundary_loops": int(boundary_loop_count), "front_boundary_edges": int(len(boundary_edges)), "final_cell_holes_filled": bool( perspective.get("final_cell_holes_filled") ), "final_enclosed_hole_added_cells": int( perspective.get("final_enclosed_hole_added_cells") or 0 ), "watertight_edge_check": watertight_edge_check, "calibrated_metric_truth": False, "metric_style_depth_input": bool(metric_style_depth), "automatic_passability_claim": False, "dimension_source": ( "ray_plane_depth_reprojection_robust_extents_with_" "accessibility_plausibility_bounds" ), "rise_source": rise_source, "rise_confidence": rise_confidence, "raw_depth_rise": float(raw_depth_rise), "surface_dimensions": { "width": width, "length": length, "rise": float(rise), "thickness": thickness, "footprint_diag": footprint_diag, "units": ( "metric_style_depth_units_not_calibrated_truth" if metric_style_depth else "relative_pseudo_depth_units" ), }, "depth_scaled_dimensions": { "width": width, "length": length, "height": float(rise), "thickness": thickness, "footprint_diag": footprint_diag, "dimension_source": ( "ray_plane_depth_reprojection_robust_extents_with_" "accessibility_plausibility_bounds" ), "calibrated_metric_truth": False, "depth_units": ( "metric_style_depth_units_not_calibrated_truth" if metric_style_depth else "relative_pseudo_depth_units" ), }, "local_basis": { "image_x_axis": [float(value) for value in axis_x], "image_y_axis": [float(value) for value in axis_y], }, "perspective_front_audit": perspective, } def build_canonical_textured_surface_mesh( colors: np.ndarray, target_mask: np.ndarray, *, category: str, stride: int, thickness_ratio: float, complete_footprint: bool = False, contextual_crop: bool = False, context_margin_ratio: float = 0.06, ) -> tuple[np.ndarray, np.ndarray, np.ndarray, dict[str, Any]]: """Build a clean category surface while retaining image-scale RGB texture. ``complete_footprint`` is a qualitative presentation prior for categories whose reviewed mask is a fragmented perspective corridor. It uses the largest rectangular patch fully contained by the hole-filled reviewed surface, so every vertex samples the accepted completed RGB at its original image coordinate. No texture extrapolation or bbox warp is used. ``contextual_crop`` instead keeps a bounded rectangular crop around the reviewed target. This is useful for non-stair surfaces whose semantics are lost when an isolated patch of pavement is rendered: the accepted completed RGB supplies the obstacle-free surroundings, while every mesh vertex still samples its original image coordinate. It is a qualitative scene-context presentation, not a claim that all pixels in the crop belong to the target. """ if contextual_crop and not complete_footprint: raise ValueError("contextual_crop requires complete_footprint") if context_margin_ratio < 0.0 or context_margin_ratio > 0.5: raise ValueError("context_margin_ratio must be in [0, 0.5]") texture_support = target_mask.astype(bool) if contextual_crop: texture_support = fill_enclosed_mask_holes(texture_support) support_y, support_x = np.nonzero(texture_support) if support_x.size == 0: raise ValueError("No target pixels for contextual textured surface") image_height, image_width = target_mask.shape raw_x_min, raw_x_max = int(support_x.min()), int(support_x.max()) raw_y_min, raw_y_max = int(support_y.min()), int(support_y.max()) margin_x = int(round(image_width * context_margin_ratio)) margin_y = int(round(image_height * context_margin_ratio)) x_min = max(0, raw_x_min - margin_x) x_max = min(image_width - 1, raw_x_max + margin_x) y_min = max(0, raw_y_min - margin_y) y_max = min(image_height - 1, raw_y_max + margin_y) mask_y, mask_x = np.nonzero( np.ones((y_max - y_min + 1, x_max - x_min + 1), dtype=bool) ) elif complete_footprint: texture_support = fill_enclosed_mask_holes(texture_support) x_min, x_max, y_min, y_max = largest_axis_aligned_rectangle( texture_support ) mask_y, mask_x = np.nonzero(texture_support[y_min : y_max + 1, x_min : x_max + 1]) else: mask_y, mask_x = np.nonzero(texture_support) if mask_x.size == 0: raise ValueError("No target pixels for canonical textured surface") if not complete_footprint: x_min, x_max = int(mask_x.min()), int(mask_x.max()) y_min, y_max = int(mask_y.min()), int(mask_y.max()) bbox_width = max(x_max - x_min + 1, 2) bbox_height = max(y_max - y_min + 1, 2) aspect = float(bbox_width / bbox_height) surface_length = 2.4 surface_width = float(surface_length * aspect) category_rise_ratio = { "ramp": 0.24, "curb_cut": 0.10, "walkway": 0.015, "tactile_paving": 0.010, }.get(category, 0.03) total_rise = surface_length * category_rise_ratio thickness = max(surface_length * thickness_ratio * 0.35, 0.035) xs = list(range(x_min, x_max + 1, max(stride, 1))) ys = list(range(y_min, y_max + 1, max(stride, 1))) if xs[-1] != x_max: xs.append(x_max) if ys[-1] != y_max: ys.append(y_max) vertex_ids: dict[tuple[int, int], int] = {} vertices: list[np.ndarray] = [] vertex_colors: list[np.ndarray] = [] front_faces: list[tuple[int, int, int]] = [] edge_counts: defaultdict[tuple[int, int], int] = defaultdict(int) extended_texture_vertices = 0 def scene_to_camera(point: tuple[float, float, float]) -> np.ndarray: x, y, z = point return np.array([x, -z, y], dtype=np.float32) def add_vertex(grid_y: int, grid_x: int) -> int: nonlocal extended_texture_vertices key = (grid_y, grid_x) if key in vertex_ids: return vertex_ids[key] u = (grid_x - x_min) / max(float(bbox_width - 1), 1.0) v = (grid_y - y_min) / max(float(bbox_height - 1), 1.0) x = (u - 0.5) * surface_width y = v * surface_length z = (1.0 - v) * total_rise if complete_footprint and category == "curb_cut": # A curb cut is not merely a weak uniform ramp: retain a low # central channel while the two lateral flare edges rise toward # the sidewalk. This is a qualitative category signature only. lateral = abs(2.0 * u - 1.0) z += total_rise * 0.65 * v * lateral**3 elif complete_footprint and category == "tactile_paving": # Low transverse ribs make tactile paving distinguishable from a # generic walkway without modifying its photographic texture. rib_phase = 2.0 * math.pi * 10.0 * v ridge = max(0.0, math.cos(rib_phase)) ** 6 z += 0.018 * ridge sample_y, sample_x = grid_y, grid_x if complete_footprint and not target_mask[grid_y, grid_x]: # This coordinate lies in an enclosed reviewed hole. The accepted # completed RGB already contains its surface appearance, so sample # it directly instead of copying a distant donor texel. extended_texture_vertices += 1 vertex_id = len(vertices) vertex_ids[key] = vertex_id vertices.append(scene_to_camera((x, y, z))) vertex_colors.append(colors[sample_y, sample_x].astype(np.uint8)) return vertex_id for row_index in range(len(ys) - 1): for column_index in range(len(xs) - 1): y0, y1 = ys[row_index], ys[row_index + 1] x0, x1 = xs[column_index], xs[column_index + 1] center_y = min((y0 + y1) // 2, target_mask.shape[0] - 1) center_x = min((x0 + x1) // 2, target_mask.shape[1] - 1) if ( not complete_footprint and not target_mask[center_y, center_x] ): continue ids = ( add_vertex(y0, x0), add_vertex(y0, x1), add_vertex(y1, x0), add_vertex(y1, x1), ) cell_faces = ( (ids[0], ids[2], ids[1]), (ids[1], ids[2], ids[3]), ) front_faces.extend(cell_faces) for face in cell_faces: for a, b in ( (face[0], face[1]), (face[1], face[2]), (face[2], face[0]), ): edge_counts[tuple(sorted((a, b)))] += 1 if not front_faces: raise ValueError("Canonical textured surface produced no faces") front_vertex_count = len(vertices) for vertex_id in range(front_vertex_count): point = vertices[vertex_id].copy() point[1] += thickness vertices.append(point) vertex_colors.append( np.clip( vertex_colors[vertex_id].astype(np.float32) * 0.58, 0, 255, ).astype(np.uint8) ) back_faces = [ ( face[2] + front_vertex_count, face[1] + front_vertex_count, face[0] + front_vertex_count, ) for face in front_faces ] side_faces: list[tuple[int, int, int]] = [] for (a, b), count in edge_counts.items(): if count != 1: continue side_faces.append((a, b, a + front_vertex_count)) side_faces.append( (b, b + front_vertex_count, a + front_vertex_count) ) all_faces = front_faces + back_faces + side_faces metadata = { "front_faces": len(front_faces), "back_faces": len(back_faces), "side_faces": len(side_faces), "boundary_side_faces": len(side_faces), "wedge_support_faces": 0, "stair_sidewall_faces": 0, "stair_sidewall_bottom_faces": 0, "solid_thickness": thickness, "max_depth_jump": None, "stride": stride, "support_mode": "canonical_textured_surface_slab", "stair_surface_mode": "not_stairs", "stair_rows": [], "stair_cross_samples": None, "surface_model": ( "canonical_textured_context_surface" if contextual_crop else "canonical_textured_category_surface" ), "footprint_policy": ( "bounded_context_rectangle_around_hole_filled_reviewed_surface" if contextual_crop else ( "largest_rectangle_inside_hole_filled_reviewed_surface" if complete_footprint else "reviewed_target_mask_cells" ) ), "complete_footprint": bool(complete_footprint), "contextual_crop": bool(contextual_crop), "context_margin_ratio": float(context_margin_ratio), "texture_coordinate_policy": ( "direct_completed_rgb_image_xy_context_crop_without_extrapolation_or_bbox_warp" if contextual_crop else "direct_completed_rgb_image_xy_without_extrapolation_or_bbox_warp" ), "extended_texture_vertices": int(extended_texture_vertices), "texture_extension_max_distance_pixels": 0.0, "texture_extension_mean_distance_pixels": 0.0, "texture_extrapolation_used": False, "context_pixels_are_target_claim": False, "contextual_crop_pixel_count": int(bbox_width * bbox_height), "contextual_crop_target_fraction": float( target_mask[y_min : y_max + 1, x_min : x_max + 1].sum() / max(bbox_width * bbox_height, 1) ), "footprint_source_bounds": { "x_min": int(x_min), "x_max": int(x_max), "y_min": int(y_min), "y_max": int(y_max), }, "calibrated_metric_truth": False, "automatic_passability_claim": False, "category_rise_ratio": category_rise_ratio, "category_geometry_signature": { "ramp": "uniform_longitudinal_slope", "curb_cut": "central_low_channel_with_lateral_flares", "walkway": "near_level_continuous_surface", "tactile_paving": "near_level_surface_with_low_transverse_ribs", }.get(category, "generic_continuous_surface"), "maximum_semantic_relief": ( 0.65 * total_rise if complete_footprint and category == "curb_cut" else ( 0.018 if complete_footprint and category == "tactile_paving" else 0.0 ) ), "surface_dimensions": { "width": surface_width, "length": surface_length, "rise": total_rise, "thickness": thickness, }, } return ( np.asarray(vertices, dtype=np.float32), np.asarray(vertex_colors, dtype=np.uint8), np.asarray(all_faces, dtype=np.int64), metadata, ) def write_ply(path: Path, vertices: np.ndarray, colors: np.ndarray, faces: np.ndarray) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="ascii") as handle: handle.write("ply\nformat ascii 1.0\n") handle.write(f"element vertex {len(vertices)}\n") handle.write("property float x\nproperty float y\nproperty float z\n") handle.write("property uchar red\nproperty uchar green\nproperty uchar blue\n") handle.write(f"element face {len(faces)}\n") handle.write("property list uchar int vertex_indices\n") handle.write("end_header\n") for point, color in zip(vertices, colors): handle.write( f"{point[0]:.6f} {point[1]:.6f} {point[2]:.6f} " f"{int(color[0])} {int(color[1])} {int(color[2])}\n" ) for face in faces: handle.write(f"3 {int(face[0])} {int(face[1])} {int(face[2])}\n") def write_vertex_color_glb( path: Path, vertices: np.ndarray, colors: np.ndarray, faces: np.ndarray, ) -> dict[str, Any]: import trimesh rgba = np.concatenate( ( colors.astype(np.uint8), np.full((len(colors), 1), 255, dtype=np.uint8), ), axis=1, ) mesh = trimesh.Trimesh( vertices=vertices, faces=faces, vertex_colors=rgba, process=False, ) path.parent.mkdir(parents=True, exist_ok=True) mesh.export(path) return { "watertight": bool(mesh.is_watertight), "euler_number": int(mesh.euler_number), "vertex_color_texture": True, } def render_mesh_frame( vertices: np.ndarray, colors: np.ndarray, faces: np.ndarray, *, size: tuple[int, int], yaw: float, pitch: float, title: str, ) -> Image.Image: width, height = size centered = vertices.astype(np.float32).copy() centered[:, 1] *= -1.0 centered -= np.median(centered, axis=0, keepdims=True) rotated = centered @ rotation_matrix(yaw, pitch).T finite = np.all(np.isfinite(rotated), axis=1) if not np.any(finite): return Image.new("RGB", size, (24, 24, 24)) if faces.size == 0: image = Image.new("RGB", size, (24, 24, 24)) draw = ImageDraw.Draw(image) draw.rectangle((0, 0, width, 18), fill=(245, 245, 245)) draw.text((6, 4), f"{title} | no faces", fill=(0, 0, 0)) return image x_values = rotated[:, 0] y_values = rotated[:, 1] x_low, x_high = np.percentile(x_values[finite], [1, 99]) y_low, y_high = np.percentile(y_values[finite], [1, 99]) x_scale = (width - 36) / max(float(x_high - x_low), 1e-6) y_scale = (height - 46) / max(float(y_high - y_low), 1e-6) scale = min(x_scale, y_scale) x_center = (x_low + x_high) * 0.5 y_center = (y_low + y_high) * 0.5 projected = np.empty((len(vertices), 2), dtype=np.float32) projected[:, 0] = (rotated[:, 0] - x_center) * scale + width * 0.5 projected[:, 1] = height * 0.5 - (rotated[:, 1] - y_center) * scale + 8 image = Image.new("RGB", size, (24, 24, 24)) draw = ImageDraw.Draw(image) face_depth = rotated[faces].mean(axis=1)[:, 2] order = np.argsort(face_depth) z_low, z_high = np.percentile(face_depth, [5, 95]) if len(face_depth) else (0.0, 1.0) for face_index in order: face = faces[face_index] polygon = [(float(projected[i, 0]), float(projected[i, 1])) for i in face] if all((x < -5 or x > width + 5 or y < 13 or y > height + 5) for x, y in polygon): continue tri = rotated[face] normal = np.cross(tri[1] - tri[0], tri[2] - tri[0]) normal_norm = float(np.linalg.norm(normal)) facing = abs(float(normal[2]) / normal_norm) if normal_norm > 1e-6 else 0.0 depth_weight = np.clip((face_depth[face_index] - z_low) / max(float(z_high - z_low), 1e-6), 0.0, 1.0) shade = 0.52 + 0.28 * float(depth_weight) + 0.20 * facing color = np.clip(colors[face].astype(np.float32).mean(axis=0) * shade, 0, 255).astype(np.uint8) draw.polygon(polygon, fill=tuple(int(v) for v in color)) draw.rectangle((0, 0, width, 18), fill=(245, 245, 245)) draw.text((6, 4), title, fill=(0, 0, 0)) return image def write_turntable_gif( path: Path, vertices: np.ndarray, colors: np.ndarray, faces: np.ndarray, *, sample_id: str, category: str, frames: int, size: tuple[int, int], pitch: float, duration_ms: int, orbit_mode: str = "front_arc", orbit_span: float = 140.0, ) -> None: if orbit_mode not in {"front_arc", "full_360"}: raise ValueError(f"Unsupported preview orbit mode: {orbit_mode}") if orbit_mode == "front_arc" and not 20.0 <= orbit_span <= 240.0: raise ValueError("front_arc preview span must be in [20, 240] degrees") images = [] for index in range(frames): yaw = ( 0.5 * orbit_span * math.sin(2.0 * math.pi * index / frames) if orbit_mode == "front_arc" else 360.0 * index / frames ) images.append( render_mesh_frame( vertices, colors, faces, size=size, yaw=yaw, pitch=pitch, title=f"{sample_id} | {category} | open-world surface", ) ) path.parent.mkdir(parents=True, exist_ok=True) images[0].save( path, save_all=True, append_images=images[1:], optimize=True, duration=duration_ms, loop=0, ) def build_html(output_dir: Path, rows: list[dict[str, Any]], summary_rows: list[dict[str, Any]]) -> None: page_title = output_dir.name.replace("_", " ").replace("-", " ").title() cards = [] for row in summary_rows: sample_id = row["sample_id"] gif_markup = ( f'' if row.get("gif") else '
Turntable render pending
' ) cards.append( f"""

{html.escape(sample_id)} {html.escape(row['category'])}

{gif_markup}

solid_mesh.ply solid_mesh.glb source completed_mesh.ply {row['vertices']} vertices / {row['faces']} faces {html.escape(row.get('geometry_model') or 'untyped')} {html.escape(row.get('surface_model') or 'depth_surface')}

""" ) page = f""" {html.escape(page_title)}

{html.escape(page_title)}

Geometry-first open-world accessibility surface patches with trusted stair-count evidence, depth-scaled dimensions, and dense colors sampled from the selected completion. These are review artifacts, not calibrated metric truth, collision meshes, or passability certification.

{''.join(cards)}
""" (output_dir / "index.html").write_text(page, encoding="utf-8") def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--results-dir", required=True) parser.add_argument("--manifest", required=True) parser.add_argument("--quality-summary", default=None) parser.add_argument("--taxonomy", default="configs/accessibility_taxonomy.json") parser.add_argument( "--category-overrides", default="configs/accessibility_review_overrides.json", help="Human-reviewed sample category overrides applied before geometry routing.", ) parser.add_argument( "--allow-unreviewed-level-change-categories", action="store_true", help="Legacy escape hatch. By default stairs/ramp/curb categories require explicit review.", ) parser.add_argument("--output-dir", required=True) parser.add_argument( "--texture-package", default=None, help=( "Optional frozen review package whose " "results//03_2d_completion/completed_rgb_selected.png " "is used only as the mesh texture. The 2D files are read-only." ), ) parser.add_argument( "--sample-ids", default="", help="Optional comma-separated sample ids. When set, preserves this order and ignores limit/per-category selection.", ) parser.add_argument("--limit", type=int, default=12) parser.add_argument("--per-category-limit", type=int, default=6) parser.add_argument("--include-review", action="store_true") parser.add_argument("--stride", type=int, default=6) parser.add_argument("--max-depth-jump", type=float, default=None) parser.add_argument("--thickness", type=float, default=None) parser.add_argument("--thickness-ratio", type=float, default=0.08) parser.add_argument( "--surface-coordinate-mode", choices=[ "relative_depth", "perspective_surface", "hybrid_instance_complete", "canonical_textured", "canonical_textured_complete", "canonical_textured_context", ], default="perspective_surface", help=( "For non-stair continuous categories, use the legacy solid depth " "mesh, a perspective-preserving surface-only patch, a watertight " "depth-scaled instance surface, the legacy canonical presentation " "fallback, a complete category surface, or a bounded contextual " "crop. Stair routing is independent." ), ) parser.add_argument( "--continuous-stride", type=int, default=3, help="Image-grid stride for the perspective continuous front surface.", ) parser.add_argument( "--continuous-minimum-cell-coverage", type=float, default=0.15, help="Minimum target coverage for a perspective surface grid cell.", ) parser.add_argument( "--continuous-support-close-kernel", type=int, default=0, help="Render-only support close kernel; 0 uses 4*--continuous-stride+1.", ) parser.add_argument( "--continuous-obstacle-texture-cleanup", action=argparse.BooleanOptionalAction, default=False, help=( "Optional render-only experiment that replaces obstacle appearance " "with same-surface exemplar texels in memory. Disabled by default: " "the accepted completed RGB is already obstacle-completed, and a " "second texture repair can introduce repeated or blocky pavement. " "The frozen 2D file always remains read-only." ), ) parser.add_argument( "--support-mode", choices=["boundary", "auto", "triangular_wedge", "stair_sidewall", "vertical_slab"], default="boundary", help="Add category-aware structural support faces. auto applies stair_sidewall to stairs and vertical_slab to continuous categories, unless wedge categories are requested.", ) parser.add_argument("--wedge-support-categories", default="ramp,curb_cut") parser.add_argument("--stair-sidewall-categories", default="stairs") parser.add_argument("--wedge-base-drop-ratio", type=float, default=0.10) parser.add_argument( "--stair-surface-mode", choices=["depth_grid", "continuous_step_strips", "canonical_staircase_prism"], default="depth_grid", help="For stair categories, choose the presentation mesh surface model.", ) parser.add_argument( "--stair-cross-samples", type=int, default=0, help="Number of samples across each continuous stair strip. 0 derives it from stride and mask width.", ) parser.add_argument( "--stair-step-count", type=int, default=0, help="Step count for canonical_staircase_prism. 0 derives it from detected edges with a visual minimum.", ) parser.add_argument("--hidden-tint", type=float, default=0.0) parser.add_argument("--mask-close-kernel", type=int, default=5) parser.add_argument("--keep-largest-component", action=argparse.BooleanOptionalAction, default=True) parser.add_argument( "--continuous-surface-categories", default="walkway,ramp,curb_cut,raised_curb,tactile_paving", ) parser.add_argument("--raised-curb-categories", default="raised_curb") parser.add_argument("--raised-curb-height", type=float, default=None) parser.add_argument("--raised-curb-height-ratio", type=float, default=0.18) parser.add_argument("--raised-curb-contour-epsilon-ratio", type=float, default=0.015) parser.add_argument("--frames", type=int, default=48) parser.add_argument( "--render-gif", action=argparse.BooleanOptionalAction, default=False, help=( "Explicitly render the dependency-light CPU preview GIF. The " "production default is GPU-required Open3D/EGL rendering through " "tools/render_accessibility_turntable.py under Slurm." ), ) parser.add_argument("--duration-ms", type=int, default=70) parser.add_argument("--width", type=int, default=420) parser.add_argument("--height", type=int, default=320) parser.add_argument("--pitch", type=float, default=-25.0) args = parser.parse_args() if ( args.surface_coordinate_mode == "hybrid_instance_complete" and args.render_gif ): parser.error( "hybrid_instance_complete is a production GPU-only path; " "render it with tools/render_accessibility_turntable.py " "inside a Slurm GPU allocation" ) results_dir = Path(args.results_dir) texture_package = ( Path(args.texture_package).expanduser().resolve() if args.texture_package else None ) if texture_package is not None and not texture_package.is_dir(): raise FileNotFoundError( f"Frozen 2D texture package does not exist: {texture_package}" ) output_dir = Path(args.output_dir) mesh_dir = output_dir / "meshes" gif_dir = output_dir / "gifs" surface_texture_dir = output_dir / "surface_textures" output_dir.mkdir(parents=True, exist_ok=True) mesh_dir.mkdir(parents=True, exist_ok=True) if args.render_gif: gif_dir.mkdir(parents=True, exist_ok=True) surface_texture_dir.mkdir(parents=True, exist_ok=True) taxonomy = load_taxonomy(Path(args.taxonomy) if args.taxonomy else None) category_overrides = load_reviewed_category_overrides( Path(args.category_overrides) if args.category_overrides else None ) rows = select_rows( read_jsonl(Path(args.manifest)), quality_review_ids(Path(args.quality_summary) if args.quality_summary else None), args.limit, args.per_category_limit, args.include_review, [item.strip() for item in args.sample_ids.split(",") if item.strip()], ) summary_rows = [] global CONTINUOUS_SURFACE_CATEGORIES CONTINUOUS_SURFACE_CATEGORIES = { item.strip() for item in args.continuous_surface_categories.split(",") if item.strip() } wedge_support_categories = { item.strip() for item in args.wedge_support_categories.split(",") if item.strip() } stair_sidewall_categories = { item.strip() for item in args.stair_sidewall_categories.split(",") if item.strip() } global RAISED_CURB_CATEGORIES RAISED_CURB_CATEGORIES = { item.strip() for item in args.raised_curb_categories.split(",") if item.strip() } for row in rows: sample_id = row["sample_id"] source_category = str( row.get("source_category") or row.get("category") or "unknown" ) category_override = category_overrides.get(sample_id, {}) category = resolve_geometry_category_for_modeling( sample_id, source_category, row, category_overrides, allow_unreviewed_level_change=( args.allow_unreviewed_level_change_categories or category_override.get("geometry_profile_reviewed") is True ), ) geometry_profile = category_override.get("geometry_profile") category_shape_enabled = bool( category_override.get("category_shape_enabled", True) ) modeled_geometry_category = category if not category_shape_enabled and geometry_profile in { "generic_walkway_no_synthetic_tactile_relief", "platform_walkway_without_synthetic_curb_cut", "crossing_walkway_without_synthetic_curb_cut", }: modeled_geometry_category = "walkway" category_info = taxonomy.get(category, {}) sample_dir = results_dir / sample_id geometry_dir = sample_dir / "geometry" depth = np.load(geometry_dir / "completed_depth.npy").astype(np.float32) h, w = depth.shape raw_target_mask = read_mask(geometry_dir / "amodal_target_mask.png", (h, w)) visible_path = geometry_dir / "target_visible_mask.png" visible_mask = read_mask(visible_path, (h, w)) if visible_path.is_file() else None hidden_path = geometry_dir / "hidden_completion_mask.png" hidden_mask = read_mask(hidden_path, (h, w)) if hidden_path.is_file() else None target_mask = clean_target_mask( raw_target_mask, args.mask_close_kernel, args.keep_largest_component, ) mesh_target_mask = target_mask surface_support_metadata: dict[str, Any] = { "role": "not_applied", "support_pixels": int(target_mask.sum()), "support_added_pixels": 0, "generation_mask_used": False, "annotation_mask_modified": False, } full_obstacle_mask: np.ndarray | None = None if ( args.surface_coordinate_mode == "perspective_surface" and category in CONTINUOUS_SURFACE_CATEGORIES and category not in RAISED_CURB_CATEGORIES ): full_obstacle_path = sample_dir / "obstacle_mask.png" full_obstacle_mask = ( read_mask_nearest(full_obstacle_path, (h, w)) if full_obstacle_path.is_file() else None ) mesh_target_mask, surface_support_metadata = ( build_continuous_surface_support_mask( raw_target_mask, stride=args.continuous_stride, close_kernel=args.continuous_support_close_kernel, visible_mask=visible_mask, hidden_mask=hidden_mask, obstacle_mask=full_obstacle_mask, ) ) depth, surface_metadata = continuous_depth_for_category( depth, mesh_target_mask, visible_mask, category, ) surface_metadata = { **surface_metadata, "geometry_support": surface_support_metadata, } frozen_texture_path = ( texture_package / "results" / sample_id / "03_2d_completion" / "completed_rgb_selected.png" if texture_package is not None else None ) colors, texture_source = prepare_texture( sample_dir, geometry_dir, (h, w), hidden_tint=args.hidden_tint, texture_override=frozen_texture_path, ) surface_texture_path: Path | None = None surface_texture_metadata: dict[str, Any] = { "role": "not_applied", "source_file_modified": False, "frozen_2d_consumed_read_only": True, } if ( args.continuous_obstacle_texture_cleanup and args.surface_coordinate_mode == "perspective_surface" and category in CONTINUOUS_SURFACE_CATEGORIES and category not in RAISED_CURB_CATEGORIES ): colors, surface_texture_metadata = ( complete_continuous_surface_texture( colors, mesh_target_mask, full_obstacle_mask, stride=args.continuous_stride, ) ) surface_texture_path = ( surface_texture_dir / f"{sample_id}_surface_texture.png" ) Image.fromarray(colors, mode="RGB").save(surface_texture_path) texture_source += " + render_only_same_surface_obstacle_exemplar" surface_metadata = { **surface_metadata, "render_only_texture_completion": surface_texture_metadata, } if args.support_mode == "auto": if category in stair_sidewall_categories: support_mode = "stair_sidewall" elif category in wedge_support_categories: support_mode = "triangular_wedge" elif category in CONTINUOUS_SURFACE_CATEGORIES: support_mode = "vertical_slab" else: support_mode = "boundary" else: support_mode = args.support_mode geometry_manifest_path = geometry_dir / "geometry_manifest.json" geometry_manifest = read_json(geometry_manifest_path) if geometry_manifest_path.is_file() else {} depth_manifest_path = sample_dir / "depth_manifest.json" depth_manifest = ( read_json(depth_manifest_path) if depth_manifest_path.is_file() else {} ) metric_style_depth = bool(depth_manifest.get("metric_depth", False)) stair_edges_y, stair_edge_provenance = ( trusted_stair_edges_from_manifest(geometry_manifest) ) depth_output = depth_manifest.get("output_depth") geometry_depth_source = geometry_manifest.get("depth_source") depth_source_matches_manifest = bool( depth_output and geometry_depth_source and Path(depth_output).expanduser().resolve() == Path(geometry_depth_source).expanduser().resolve() ) checkpoint_value = depth_manifest.get("checkpoint") checkpoint_path = ( Path(checkpoint_value).expanduser() if checkpoint_value else None ) depth_model_identity = { "engine": depth_manifest.get("engine"), "encoder": depth_manifest.get("encoder"), "checkpoint": ( str(checkpoint_path) if checkpoint_path is not None else None ), "checkpoint_sha256": ( sha256_file(checkpoint_path) if checkpoint_path is not None and checkpoint_path.is_file() else None ), "metric_style_depth": metric_style_depth, "max_depth": depth_manifest.get("max_depth"), "geometry_depth_source": geometry_depth_source, "manifest_output_depth": depth_output, "geometry_depth_source_matches_manifest": ( depth_source_matches_manifest ), "selection_policy": ( "best_available_local_single_image_metric_depth_model" if ( depth_manifest.get("engine") == "depth_anything_v2" and metric_style_depth and checkpoint_path is not None and "metric_hypersim_vitl" in checkpoint_path.name ) else "input_pipeline_depth_model" ), } if category in RAISED_CURB_CATEGORIES: sample_curb_height = category_override.get("raised_curb_height", args.raised_curb_height) sample_curb_height_ratio = float( category_override.get("raised_curb_height_ratio", args.raised_curb_height_ratio) ) sample_contour_epsilon_ratio = float( category_override.get( "raised_curb_contour_epsilon_ratio", args.raised_curb_contour_epsilon_ratio, ) ) vertices, vertex_colors, faces, metadata = build_raised_curb_prism_mesh( depth, colors, target_mask, height=None if sample_curb_height is None else float(sample_curb_height), height_ratio=sample_curb_height_ratio, contour_epsilon_ratio=sample_contour_epsilon_ratio, stride=args.stride, ) surface_metadata = { **surface_metadata, "surface_model": metadata["surface_model"], "curb_height": metadata["curb_height"], "contour_vertices": metadata["contour_vertices"], "raw_contour_vertices": metadata["raw_contour_vertices"], } elif args.stair_surface_mode == "canonical_staircase_prism" and category in stair_sidewall_categories: vertices, vertex_colors, faces, metadata = build_canonical_staircase_prism_mesh( depth, colors, target_mask, stride=args.stride, thickness=args.thickness, thickness_ratio=args.thickness_ratio, stair_edges_y=stair_edges_y, step_count=int( category_override.get( "stair_step_count", args.stair_step_count, ) ), stair_edge_slope=float( geometry_manifest.get("stair_edge_slope", 0.0) ), metric_style_depth=metric_style_depth, ) elif ( args.stair_surface_mode == "continuous_step_strips" and category in stair_sidewall_categories and len(stair_edges_y) > 0 ): vertices, vertex_colors, faces, metadata = build_continuous_stair_mesh( depth, colors, target_mask, stride=args.stride, thickness=args.thickness, thickness_ratio=args.thickness_ratio, support_mode=support_mode, wedge_base_drop_ratio=args.wedge_base_drop_ratio, stair_edges_y=stair_edges_y, stair_edge_slope=float(geometry_manifest.get("stair_edge_slope", 0.0)), cross_samples=args.stair_cross_samples, ) elif ( args.surface_coordinate_mode == "perspective_surface" and category in CONTINUOUS_SURFACE_CATEGORIES ): vertices, vertex_colors, faces, metadata = ( build_perspective_continuous_surface_mesh( depth, colors, mesh_target_mask, category=category, stride=args.continuous_stride, minimum_cell_coverage=args.continuous_minimum_cell_coverage, plane=( ( np.asarray( surface_metadata["plane_normal"], dtype=np.float32, ), float(surface_metadata["plane_d"]), ) if ( surface_metadata.get("plane_normal") is not None and surface_metadata.get("plane_d") is not None ) else None ), plane_validation_mask=( visible_mask if visible_mask is not None and np.any(visible_mask) else raw_target_mask ), # The frozen 2D completion is the accepted appearance for # both reviewed and evidence-gated repair pixels. Only # grid-boundary nodes outside the geometry support borrow # their nearest supported texel. texture_support_mask=mesh_target_mask, must_retain_mask=hidden_mask, bridge_evidence_mask=full_obstacle_mask, ) ) surface_metadata = { **surface_metadata, "surface_model": metadata["surface_model"], "texture_policy": metadata["texture_policy"], "texture_coordinate_policy": metadata[ "texture_coordinate_policy" ], "surface_cells": metadata["surface_cells"], "cell_components_after": metadata["cell_components_after"], "hidden_cell_coverage_fraction": metadata[ "hidden_cell_coverage_fraction" ], "bridge_count": metadata["bridge_count"], "bridge_added_cells": metadata["bridge_added_cells"], "discarded_tiny_cells_without_hidden_evidence": metadata[ "discarded_tiny_cells_without_hidden_evidence" ], "detached_rejected_cells": metadata[ "detached_rejected_cells" ], "detached_rejected_hidden_cells": metadata[ "detached_rejected_hidden_cells" ], "boundary_clamped_vertices": metadata[ "boundary_clamped_vertices" ], "texture_completion_vertices": metadata[ "texture_completion_vertices" ], "texture_completion_max_distance_pixels": metadata[ "texture_completion_max_distance_pixels" ], "texture_completion_mean_distance_pixels": metadata[ "texture_completion_mean_distance_pixels" ], "texture_completion_policy": metadata[ "texture_completion_policy" ], "depth_clipped_vertices": metadata[ "depth_clipped_vertices" ], "plane_plausible_fraction": metadata[ "plane_plausible_fraction" ], "plane_depth_policy": metadata["plane_depth_policy"], } elif ( args.surface_coordinate_mode == "hybrid_instance_complete" and category in CONTINUOUS_SURFACE_CATEGORIES and category not in RAISED_CURB_CATEGORIES ): mesh_target_mask, surface_support_metadata = ( build_hybrid_instance_footprint_mask( target_mask, category=modeled_geometry_category, stride=args.continuous_stride, visible_mask=visible_mask, hidden_mask=hidden_mask, ) ) vertices, vertex_colors, faces, metadata = ( build_obstacle_centered_depth_scaled_corridor_mesh( depth, colors, mesh_target_mask, category=modeled_geometry_category, stride=args.continuous_stride, visible_mask=visible_mask, hidden_mask=hidden_mask, plane_validation_mask=( visible_mask & mesh_target_mask if visible_mask is not None and np.any(visible_mask) else mesh_target_mask ), metric_style_depth=metric_style_depth, ) ) surface_metadata = { **surface_metadata, "geometry_support": surface_support_metadata, "source_semantic_category": category, "modeled_geometry_category": modeled_geometry_category, "geometry_profile": geometry_profile, "geometry_profile_source": category_override.get( "geometry_profile_source" ), "category_shape_enabled": category_shape_enabled, "surface_model": metadata["surface_model"], "support_mode": metadata["support_mode"], "footprint_policy": metadata["footprint_policy"], "complete_footprint": metadata["complete_footprint"], "category_geometry_signature": metadata[ "category_geometry_signature" ], "maximum_semantic_relief": metadata[ "maximum_semantic_relief" ], "texture_coordinate_policy": metadata[ "texture_coordinate_policy" ], "texture_mapping": metadata["texture_mapping"], "texture_geometry_warp": metadata[ "texture_geometry_warp" ], "independent_axis_normalization_used": metadata[ "independent_axis_normalization_used" ], "uniform_geometry_scale": metadata[ "uniform_geometry_scale" ], "uniform_geometry_scale_applied": metadata[ "uniform_geometry_scale_applied" ], "texture_extrapolation_used": metadata[ "texture_extrapolation_used" ], "texture_resampling": metadata["texture_resampling"], "texture_extension_max_distance_pixels": metadata[ "texture_extension_max_distance_pixels" ], "surface_dimensions": metadata["surface_dimensions"], "dimension_source": metadata["dimension_source"], "rise_source": metadata["rise_source"], "rise_confidence": metadata["rise_confidence"], "raw_depth_rise": metadata["raw_depth_rise"], "calibrated_metric_truth": metadata[ "calibrated_metric_truth" ], "metric_style_depth_input": metadata[ "metric_style_depth_input" ], "automatic_passability_claim": metadata[ "automatic_passability_claim" ], "front_boundary_loops": metadata["front_boundary_loops"], "final_cell_holes_filled": metadata[ "final_cell_holes_filled" ], "final_enclosed_hole_added_cells": metadata[ "final_enclosed_hole_added_cells" ], "watertight_edge_check": metadata[ "watertight_edge_check" ], "complete_obstacle_zone": metadata[ "complete_obstacle_zone" ], "open_world_channel_model": metadata[ "open_world_channel_model" ], "local_context_patch_model": metadata[ "local_context_patch_model" ], "open_world_scene_compatible": metadata[ "open_world_scene_compatible" ], "open_world_topology_claim": metadata[ "open_world_topology_claim" ], "context_pixels_are_target_claim": metadata[ "context_pixels_are_target_claim" ], "projected_context_hidden_pixel_coverage_fraction": metadata[ "projected_context_hidden_pixel_coverage_fraction" ], "projected_context_target_fraction": metadata[ "projected_context_target_fraction" ], "projected_context_hidden_component_coverage": metadata[ "projected_context_hidden_component_coverage" ], "projected_context_minimum_hidden_component_coverage_fraction": metadata[ "projected_context_minimum_hidden_component_coverage_fraction" ], "continuous_local_coordinate_hidden_coverage_fraction": metadata[ "continuous_local_coordinate_hidden_coverage_fraction" ], "minimum_continuous_hidden_component_coverage_fraction": metadata[ "minimum_continuous_hidden_component_coverage_fraction" ], "all_hidden_components_covered": metadata[ "all_hidden_components_covered" ], "projected_context_image_corners": metadata[ "projected_context_image_corners" ], "projected_context_image_polygon": metadata[ "projected_context_image_polygon" ], "context_roi_image_xyxy": metadata["context_roi_image_xyxy"], "context_roi_policy": metadata["context_roi_policy"], "regular_grid_has_no_internal_topological_holes": metadata[ "regular_grid_has_no_internal_topological_holes" ], "corridor_profile": metadata["corridor_profile"], "corridor_grid": metadata["corridor_grid"], "geometry_quality": metadata["geometry_quality"], "local_plane_basis": metadata["local_plane_basis"], "raw_depth_scaled_dimensions": metadata[ "raw_depth_scaled_dimensions" ], "dimension_bounds": metadata["dimension_bounds"], "plane_fit_residual_median": metadata[ "plane_fit_residual_median" ], "plane_fit_residual_p95": metadata["plane_fit_residual_p95"], "texture_profile_inside_reviewed_footprint_fraction": metadata[ "texture_profile_inside_reviewed_footprint_fraction" ], "texture_profile_hidden_completion_fraction": metadata[ "texture_profile_hidden_completion_fraction" ], } elif ( args.surface_coordinate_mode in { "canonical_textured", "canonical_textured_complete", "canonical_textured_context", } and category in CONTINUOUS_SURFACE_CATEGORIES ): vertices, vertex_colors, faces, metadata = ( build_canonical_textured_surface_mesh( colors, target_mask, category=category, stride=args.stride, thickness_ratio=args.thickness_ratio, complete_footprint=( args.surface_coordinate_mode in { "canonical_textured_complete", "canonical_textured_context", } ), contextual_crop=( args.surface_coordinate_mode == "canonical_textured_context" ), ) ) surface_metadata = { **surface_metadata, "surface_model": metadata["surface_model"], "category_rise_ratio": metadata["category_rise_ratio"], "category_geometry_signature": metadata[ "category_geometry_signature" ], "maximum_semantic_relief": metadata[ "maximum_semantic_relief" ], "surface_dimensions": metadata["surface_dimensions"], "footprint_policy": metadata["footprint_policy"], "complete_footprint": metadata["complete_footprint"], "contextual_crop": metadata["contextual_crop"], "context_margin_ratio": metadata["context_margin_ratio"], "context_pixels_are_target_claim": metadata[ "context_pixels_are_target_claim" ], "contextual_crop_pixel_count": metadata[ "contextual_crop_pixel_count" ], "contextual_crop_target_fraction": metadata[ "contextual_crop_target_fraction" ], "texture_coordinate_policy": metadata[ "texture_coordinate_policy" ], "extended_texture_vertices": metadata[ "extended_texture_vertices" ], "texture_extension_max_distance_pixels": metadata[ "texture_extension_max_distance_pixels" ], "texture_extension_mean_distance_pixels": metadata[ "texture_extension_mean_distance_pixels" ], "texture_extrapolation_used": metadata[ "texture_extrapolation_used" ], "footprint_source_bounds": metadata[ "footprint_source_bounds" ], "calibrated_metric_truth": metadata[ "calibrated_metric_truth" ], "automatic_passability_claim": metadata[ "automatic_passability_claim" ], } else: vertices, vertex_colors, faces, metadata = build_solid_mesh( depth, colors, target_mask, stride=args.stride, max_depth_jump=args.max_depth_jump, thickness=args.thickness, thickness_ratio=args.thickness_ratio, support_mode=support_mode, wedge_base_drop_ratio=args.wedge_base_drop_ratio, category=category, stair_edges_y=stair_edges_y, ) metadata["stair_edge_provenance"] = stair_edge_provenance mesh_path = mesh_dir / f"{sample_id}_solid_mesh.ply" glb_path = mesh_dir / f"{sample_id}_solid_mesh.glb" gif_path = gif_dir / f"{sample_id}_solid_mesh.gif" write_ply(mesh_path, vertices, vertex_colors, faces) glb_info = write_vertex_color_glb( glb_path, vertices, vertex_colors, faces, ) if args.render_gif: write_turntable_gif( gif_path, vertices, vertex_colors, faces, sample_id=sample_id, category=category, frames=args.frames, size=(args.width, args.height), pitch=args.pitch, duration_ms=args.duration_ms, ) summary_rows.append( { "sample_id": sample_id, "category": category, "source_category": source_category, "category_override_applied": category != source_category, "geometry_profile": category_override.get("geometry_profile"), "vertices": int(len(vertices)), "faces": int(len(faces)), "front_faces": metadata["front_faces"], "back_faces": metadata["back_faces"], "side_faces": metadata["side_faces"], "boundary_side_faces": metadata["boundary_side_faces"], "wedge_support_faces": metadata["wedge_support_faces"], "stair_sidewall_faces": metadata["stair_sidewall_faces"], "stair_sidewall_bottom_faces": metadata["stair_sidewall_bottom_faces"], "solid_thickness": metadata["solid_thickness"], "max_depth_jump": metadata["max_depth_jump"], "stride": metadata["stride"], "support_mode": metadata["support_mode"], "vertical_drop": metadata.get("vertical_drop"), "vertical_base_y": metadata.get("vertical_base_y"), "depth_scaled_dimensions": metadata.get("depth_scaled_dimensions", {}), "stair_surface_mode": metadata.get("stair_surface_mode", "depth_grid"), "stair_rows": metadata.get("stair_rows", []), "stair_cross_samples": metadata.get("stair_cross_samples"), "stair_edge_slope": metadata.get("stair_edge_slope"), "stair_step_count": metadata.get("stair_step_count"), "stair_step_count_source": metadata.get("stair_step_count_source"), "stair_inference": metadata.get("stair_inference"), "stair_edge_provenance": stair_edge_provenance, "stair_physical_edge_rows": metadata.get( "stair_physical_edge_rows", [], ), "curb_height": metadata.get("curb_height"), "curb_height_ratio": metadata.get("curb_height_ratio"), "curb_height_source": metadata.get("curb_height_source"), "contour_vertices": metadata.get("contour_vertices"), "raw_contour_vertices": metadata.get("raw_contour_vertices"), "surface_model": surface_metadata.get("surface_model"), "surface_metadata": surface_metadata, "stair_edges_y": stair_edges_y, "raw_stair_edges_y": stair_edge_provenance["raw_rows"], "geometry_model": category_info.get("geometry_model", "untyped_depth_surface"), "primary_users": category_info.get("primary_users", []), "texture_source": texture_source, "texture_sha256": ( sha256_file(frozen_texture_path) if frozen_texture_path is not None else None ), "texture_package": ( str(texture_package) if texture_package is not None else None ), "surface_texture": ( str(Path("surface_textures") / surface_texture_path.name) if surface_texture_path is not None else None ), "surface_texture_sha256": ( sha256_file(surface_texture_path) if surface_texture_path is not None else None ), "surface_texture_metadata": surface_texture_metadata, "raw_target_pixels": int(raw_target_mask.sum()), "clean_target_pixels": int(target_mask.sum()), "render_support_pixels": int(mesh_target_mask.sum()), "render_support_added_pixels": int( mesh_target_mask.sum() - raw_target_mask.sum() ), "render_support_metadata": surface_support_metadata, "hidden_cell_coverage_fraction": metadata.get( "hidden_cell_coverage_fraction" ), "surface_cell_components": metadata.get( "cell_components_after" ), "surface_bridge_count": metadata.get("bridge_count"), "surface_bridge_added_cells": metadata.get( "bridge_added_cells" ), "surface_detached_rejected_cells": metadata.get( "detached_rejected_cells" ), "surface_detached_rejected_hidden_cells": metadata.get( "detached_rejected_hidden_cells" ), "surface_boundary_clamped_vertices": metadata.get( "boundary_clamped_vertices" ), "surface_depth_clipped_vertices": metadata.get( "depth_clipped_vertices" ), "surface_plane_plausible_fraction": metadata.get( "plane_plausible_fraction" ), "mesh": str(Path("meshes") / mesh_path.name), "glb": str(Path("meshes") / glb_path.name), "glb_info": glb_info, "gif": ( str(Path("gifs") / gif_path.name) if args.render_gif else None ), "source_mesh": os.path.relpath(geometry_dir / "completed_mesh.ply", output_dir), "depth_is_metric": False, "depth_model_is_metric_style": metric_style_depth, "depth_model_identity": depth_model_identity, "amodal_3d_completion_is_ground_truth": False, } ) summary = { "sample_count": len(summary_rows), "rows": summary_rows, "depth_is_metric": False, "amodal_3d_completion_is_ground_truth": False, "continuous_surface_categories": sorted(CONTINUOUS_SURFACE_CATEGORIES), "wedge_support_categories": sorted(wedge_support_categories), "stair_sidewall_categories": sorted(stair_sidewall_categories), "raised_curb_categories": sorted(RAISED_CURB_CATEGORIES), "stair_surface_mode": args.stair_surface_mode, "surface_coordinate_mode": args.surface_coordinate_mode, "taxonomy": str(Path(args.taxonomy)) if args.taxonomy else None, "texture_policy": ( "frozen quality-gated completed RGB used read-only; perspective-aware " "dense vertex sampling; no semantic tint" if texture_package is not None and args.hidden_tint <= 0 else ( "clean completed RGB with hidden-region inpainting; no mask-color tint" if args.hidden_tint <= 0 else f"clean completed RGB with hidden-region inpainting and hidden tint {args.hidden_tint:.3f}" ) ), "texture_package": ( str(texture_package) if texture_package is not None else None ), "solid_mesh_version": ( "v14_obstacle_centered_metric_context_surface_gpu_required" if args.surface_coordinate_mode == "hybrid_instance_complete" else "v10_continuous_surface_gpu_required" ), "preview_orbit": { "mode": "front_arc", "span_degrees": 140.0, "fixed_projection_scale": False, "role": ( "cpu_preview_forbidden_for_hybrid_gpu_only_production_path" if args.surface_coordinate_mode == "hybrid_instance_complete" else "optional_dependency_light_cpu_preview_disabled_by_default" ), }, "production_renderer_policy": ( "physical_nvidia_gpu_required_fail_closed_no_implicit_cpu_fallback" ), "allow_unreviewed_level_change_categories": args.allow_unreviewed_level_change_categories, "note": ( ( "Hybrid non-stair surfaces retain a smoothed, visible-anchored " "reviewed instance, then follow the local travel channel crossing " "the hidden obstacle zone instead of turning every horizon branch " "into a long standalone shard. A regular depth-scaled metric grid is " "built in the fitted local surface plane and reprojected through the " "camera model into the frozen accepted 2D completion. This keeps a " "bounded amount of path context without independently normalizing " "texture axes. The local context slab is watertight and uses no " "second obstacle cleanup, donor extrapolation, fog, or repeated " "texture. It does not claim that all context pixels are target " "surface or that the patch recovers full open-world topology. " "Production rendering requires a verified physical " "NVIDIA GPU and fails closed instead of falling back to CPU. Values " "remain monocular metric-style estimates rather " "than calibrated physical truth or passability certification." ) if args.surface_coordinate_mode == "hybrid_instance_complete" else ( "Geometry-first accessibility review surfaces. Stair count is " "filtered from trusted detector provenance plus RGB/depth riser " "evidence. Frozen 2D output is used read-only for appearance; " "geometry and dimensions remain model estimates rather than " "calibrated physical truth." ) ), } (output_dir / "summary.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) build_html(output_dir, rows, summary_rows) print(json.dumps({"output": str(output_dir), "sample_count": len(summary_rows)}, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())