#!/usr/bin/env python3 """Build one orderly 2D/3D accessibility-completion quick-review directory.""" from __future__ import annotations import argparse import hashlib import json import math import os import shutil from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable, Mapping, Sequence from PIL import Image, ImageDraw, ImageOps PROFILES = ("walking", "blind_low_vision", "wheelchair_wheeled", "cyclist") PRIMARY_3D_ROLES = ( "category_geometry", "open_world_accessibility_surface", "learned_amodal3d_gaussian", ) CORE_OUTPUT_NAMES = { "original": "01_original.jpg", "completion_2d": "02_2d_completion.png", "geometry_turntable": "03_3d_turntable.gif", "geometry_multiview": "04_3d_multiview.jpg", "geometry_mesh": "05_mesh.ply", "overview": "overview.jpg", "accessibility_review": "accessibility_review.json", } VISUAL_OUTPUT_NAMES = { "visual_turntable": "05_visual_3d_turntable.gif", "visual_multiview": "06_visual_3d_multiview.jpg", } METRIC_ALIASES = { "width": ( "clear_width_m", "minimum_clear_width_m", "min_clear_width_m", "path_width_m", "walkable_width_m", "width_m", ), "slope": ( "slope_degrees", "slope_percent", "longitudinal_slope_percent", "cross_slope_percent", "slope_ratio", "grade_percent", ), "clearance": ( "clearance_m", "minimum_clearance_m", "min_clearance_m", "vertical_clearance_m", "obstacle_clearance_m", ), } def build_argument_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source", "--original", dest="source", required=True) parser.add_argument( "--completion-2d", "--selected-2d", "--2d-selected", dest="completion_2d", required=True, ) parser.add_argument( "--turntable-gif", "--geometry-turntable", dest="turntable_gif", required=True, ) parser.add_argument( "--multiview", "--geometry-multiview", dest="multiview", required=True, ) parser.add_argument("--mesh", "--geometry-mesh", dest="mesh", required=True) parser.add_argument( "--primary-3d-role", choices=PRIMARY_3D_ROLES, default="category_geometry", help=( "Declares whether the core rotating review files are diagnostic " "category geometry, an open-world accessibility surface, or the " "Accessibility3D CUDA Gaussian/dense-mesh result." ), ) parser.add_argument("--completion-manifest", required=True) parser.add_argument( "--geometry-manifest", default=None, help="Optional geometry/depth manifest containing metric-evidence declarations.", ) parser.add_argument( "--verification-manifest", default=None, help=( "Optional generated-3D verification report. Its exact bytes are bound " "into the quick-review manifest so the selected primary role remains auditable." ), ) parser.add_argument( "--visual-turntable", default=None, help="Optional learned visual 3D candidate; never used as passability evidence.", ) parser.add_argument( "--visual-multiview", default=None, help="Optional learned visual 3D candidate; never used as passability evidence.", ) parser.add_argument("--category", required=True) parser.add_argument("--sample-id", required=True) parser.add_argument( "--output-dir", required=True, help="Sample directory; files are written below its 00_quick_review child.", ) return parser def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def relative_path(path: Path, anchor: Path) -> str: return Path(os.path.relpath(path.resolve(), anchor.resolve())).as_posix() def file_record(path: Path, anchor: Path) -> dict[str, Any]: return { "path": relative_path(path, anchor), "sha256": sha256_file(path), "bytes": path.stat().st_size, } def read_json_object(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise ValueError(f"Expected a JSON object: {path}") return value def write_json(path: Path, payload: Mapping[str, Any]) -> None: path.write_text( json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) def _walk_json(value: Any, prefix: tuple[str, ...] = ()) -> Iterable[tuple[tuple[str, ...], Any]]: if isinstance(value, dict): for key, child in value.items(): key_path = (*prefix, str(key)) yield key_path, child yield from _walk_json(child, key_path) elif isinstance(value, list): for index, child in enumerate(value): yield from _walk_json(child, (*prefix, str(index))) def _number(value: Any) -> float | None: if isinstance(value, bool): return None if isinstance(value, (int, float)) and math.isfinite(float(value)): return float(value) if isinstance(value, dict): return _number(value.get("value")) return None def _measurement_unit(field: str) -> str: if field.endswith("_m"): return "m" if field.endswith("_degrees"): return "degrees" if field.endswith("_percent") or field == "grade_percent": return "percent" if field.endswith("_ratio"): return "ratio" return "declared_metric" def _find_measurement( manifests: Sequence[tuple[str, Mapping[str, Any]]], aliases: Sequence[str], ) -> dict[str, Any] | None: alias_set = set(aliases) for source_name, payload in manifests: for key_path, value in _walk_json(payload): field = key_path[-1] if field not in alias_set: continue numeric_value = _number(value) if numeric_value is None: continue unit = _measurement_unit(field) if isinstance(value, dict) and isinstance(value.get("unit"), str): unit = str(value["unit"]) return { "value": numeric_value, "unit": unit, "source_manifest": source_name, "source_field": ".".join(key_path), } return None def _metric_truth_declarations( manifests: Sequence[tuple[str, Mapping[str, Any]]], ) -> list[dict[str, Any]]: truth_fields = { "depth_is_metric_truth", "depth_is_metric", "geometry_is_metric", "metric_geometry", "metric_calibrated", } declarations: list[dict[str, Any]] = [] for source_name, payload in manifests: for key_path, value in _walk_json(payload): if key_path[-1] in truth_fields and isinstance(value, bool): declarations.append( { "source_manifest": source_name, "source_field": ".".join(key_path), "value": value, } ) return declarations def extract_metric_evidence( completion_manifest: Mapping[str, Any], geometry_manifest: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Extract only explicitly named physical measurements. A width-like pixel count is deliberately not treated as metric evidence. Measurements are trusted for passability review only when a supplied manifest explicitly declares calibrated/metric geometry. """ manifests: list[tuple[str, Mapping[str, Any]]] = [ ("completion_manifest", completion_manifest) ] if geometry_manifest is not None: manifests.append(("geometry_manifest", geometry_manifest)) declarations = _metric_truth_declarations(manifests) trusted_metric_geometry = any(item["value"] is True for item in declarations) measurements = { name: _find_measurement(manifests, aliases) for name, aliases in METRIC_ALIASES.items() } missing = [name for name, measurement in measurements.items() if measurement is None] complete = trusted_metric_geometry and not missing return { "trusted_metric_geometry": trusted_metric_geometry, "metric_truth_declarations": declarations, "measurements": measurements, "missing_measurements": missing, "complete_width_slope_clearance": complete, "automatic_passability_allowed": False, "reason": ( "Metric width, slope, and clearance are present, but thresholds and field " "validation still require human review." if complete else "Trusted metric width, slope, and clearance are incomplete." ), } def is_stairs_category(category: str) -> bool: normalized = category.strip().lower().replace("-", "_").replace(" ", "_") return normalized in {"stairs", "stair", "staircase", "steps"} or normalized.startswith( "stairs_" ) def build_population_assessments( category: str, metric_evidence: Mapping[str, Any], ) -> dict[str, dict[str, Any]]: """Return conservative per-population decisions without a safety claim.""" assessments: dict[str, dict[str, Any]] = {} complete_metrics = bool(metric_evidence.get("complete_width_slope_clearance", False)) for profile in PROFILES: if is_stairs_category(category) and profile == "wheelchair_wheeled": assessments[profile] = { "status": "blocked", "decision": "block", "can_pass": False, "review_action": "stop_and_replan", "reason": "Stair geometry blocks a wheeled route; use an alternate reviewed route.", "basis": "category_rule_stairs_wheelchair", "human_review_required": True, } continue assessments[profile] = { "status": "unknown", "decision": "unknown", "can_pass": None, "review_action": "manual_review", "reason": ( "Metric evidence exists, but no jurisdiction-specific thresholds or " "field validation were supplied." if complete_metrics else "Trusted metric width, slope, and clearance are incomplete." ), "basis": ( "metric_evidence_requires_threshold_review" if complete_metrics else "insufficient_metric_width_slope_clearance" ), "human_review_required": True, } return assessments def _provenance_summary(payload: Mapping[str, Any]) -> dict[str, Any]: """Keep truthful source identifiers without copying absolute paths.""" summary: dict[str, Any] = {} scalar_fields = ( "pipeline", "backend", "output_kind", "completion_method", "warning", "device", "category", "sample_id", ) for field in scalar_fields: value = payload.get(field) if isinstance(value, (str, int, float, bool)) or value is None: summary[field] = value model = payload.get("model") if isinstance(model, str) and model.strip(): # A backend/model identifier is useful provenance. If it is a local # filesystem path, retain the final identifier rather than publishing an # absolute host path. summary["model_identifier"] = Path(model).name if ("/" in model or "\\" in model) else model return summary def _save_image(source: Path, destination: Path, image_format: str) -> None: image = ImageOps.exif_transpose(Image.open(source)).convert("RGB") if image_format == "JPEG": image.save(destination, format=image_format, quality=95, subsampling=0) else: image.save(destination, format=image_format) def copy_review_assets( *, source: Path, completion_2d: Path, turntable_gif: Path, multiview: Path, mesh: Path, review_dir: Path, visual_turntable: Path | None = None, visual_multiview: Path | None = None, ) -> dict[str, Path]: review_dir.mkdir(parents=True, exist_ok=True) outputs = { role: review_dir / filename for role, filename in CORE_OUTPUT_NAMES.items() } _save_image(source, outputs["original"], "JPEG") _save_image(completion_2d, outputs["completion_2d"], "PNG") shutil.copy2(turntable_gif, outputs["geometry_turntable"]) _save_image(multiview, outputs["geometry_multiview"], "JPEG") shutil.copy2(mesh, outputs["geometry_mesh"]) if visual_turntable is not None: destination = review_dir / VISUAL_OUTPUT_NAMES["visual_turntable"] shutil.copy2(visual_turntable, destination) outputs["visual_turntable"] = destination if visual_multiview is not None: destination = review_dir / VISUAL_OUTPUT_NAMES["visual_multiview"] _save_image(visual_multiview, destination, "JPEG") outputs["visual_multiview"] = destination return outputs def _panel(path: Path, label: str, size: tuple[int, int]) -> Image.Image: image = ImageOps.exif_transpose(Image.open(path)).convert("RGB") header_height = 42 body = ImageOps.contain(image, (size[0], size[1] - header_height)) panel = Image.new("RGB", size, "white") draw = ImageDraw.Draw(panel) draw.rectangle((0, 0, size[0], header_height), fill=(241, 241, 238)) draw.text((13, 14), label, fill=(18, 18, 18)) panel.paste( body, ( (size[0] - body.width) // 2, header_height + (size[1] - header_height - body.height) // 2, ), ) return panel def build_overview( *, original: Path, completion_2d: Path, geometry_multiview: Path, destination: Path, category: str, primary_3d_role: str = "category_geometry", ) -> None: if primary_3d_role == "learned_amodal3d_gaussian": review_label = "03-05 Accessibility3D CUDA 3D review" elif primary_3d_role == "open_world_accessibility_surface": review_label = "03-05 Open-world accessibility surface review" else: review_label = "03-05 Diagnostic geometry review" panel_size = (480, 500) panels = [ _panel(original, "01 Original image", panel_size), _panel(completion_2d, "02 2D completion candidate", panel_size), _panel( geometry_multiview, review_label, panel_size, ), ] footer_height = 64 canvas = Image.new( "RGB", (panel_size[0] * len(panels), panel_size[1] + footer_height), (231, 231, 228), ) for index, panel in enumerate(panels): canvas.paste(panel, (index * panel_size[0], 0)) draw = ImageDraw.Draw(canvas) draw.text( (14, panel_size[1] + 12), ( f"Category: {category}. Review artifact only: do not infer safe passage " "without metric width, slope, clearance, and human validation." ), fill=(90, 35, 28), ) draw.text( (14, panel_size[1] + 36), ( "The 2D result is generative; the Accessibility3D rotation is a nonmetric " "visual reconstruction, not a navigation certification." if primary_3d_role == "learned_amodal3d_gaussian" else "The 2D result is generative; the rotating geometry is not a navigation certification." ), fill=(70, 70, 70), ) canvas.save(destination, quality=94, subsampling=0) def build_accessibility_review( *, sample_id: str, category: str, completion_manifest: Mapping[str, Any], metric_evidence: Mapping[str, Any], visual_candidate_present: bool, primary_3d_role: str = "category_geometry", ) -> dict[str, Any]: assessments = build_population_assessments(category, metric_evidence) if primary_3d_role == "learned_amodal3d_gaussian": render_backend = "Accessibility3D CUDA Gaussian rasterization" mesh_representation = "dense FlexiCubes triangle faces" elif primary_3d_role == "open_world_accessibility_surface": render_backend = "category-constrained open-world surface renderer" mesh_representation = "open-world accessibility triangle surface patch" else: render_backend = "category-constrained geometry renderer" mesh_representation = "category-constrained triangle mesh" primary_review_name = ( "Accessibility3D learned nonmetric reconstruction" if primary_3d_role == "learned_amodal3d_gaussian" else "category-constrained diagnostic 3D geometry" ) return { "schema_version": "accessibilityamodal_review_v1", "created_at_utc": datetime.now(timezone.utc).isoformat(), "sample_id": sample_id, "category": category, "overall_status": "manual_review_required", "safe_passage_claim": False, "population_assessments": assessments, "metric_evidence": dict(metric_evidence), "primary_3d_evidence": { "turntable": CORE_OUTPUT_NAMES["geometry_turntable"], "multiview": CORE_OUTPUT_NAMES["geometry_multiview"], "mesh": CORE_OUTPUT_NAMES["geometry_mesh"], "role": primary_3d_role, "render_backend": render_backend, "mesh_representation": mesh_representation, "is_metric_geometry": False, "is_navigation_certification": False, }, "completion_2d": { "path": CORE_OUTPUT_NAMES["completion_2d"], "role": "generative_visual_hypothesis", "is_ground_truth": False, }, "visual_3d_candidate": { "present": visual_candidate_present, "role": "learned_visual_candidate_only", "metric_evidence": False, "passability_evidence": False, "warning": ( "The learned visual candidate must not be used to infer dimensions or passage." if visual_candidate_present else None ), }, "completion_provenance": _provenance_summary(completion_manifest), "required_human_checks": [ "Verify that the hidden ground/support surface is completed continuously, without fog, haze, or ghost obstacles.", f"Verify that the 2D completion agrees with the selected {primary_review_name}.", "Measure and validate route width, slope, and clearance before any passage decision.", ], "limitations": [ "No population is declared safely passable by this automatic bundle.", "A blocked stairs/wheelchair rule is a route-level constraint, not a complete site assessment.", ], } def build_bundle( *, source: Path, completion_2d: Path, turntable_gif: Path, multiview: Path, mesh: Path, completion_manifest_path: Path, category: str, sample_id: str, output_dir: Path, geometry_manifest_path: Path | None = None, verification_manifest_path: Path | None = None, visual_turntable: Path | None = None, visual_multiview: Path | None = None, primary_3d_role: str = "category_geometry", ) -> Path: if primary_3d_role not in PRIMARY_3D_ROLES: raise ValueError( f"Unsupported primary_3d_role={primary_3d_role!r}; " f"expected one of {PRIMARY_3D_ROLES}" ) input_paths = { "source": source, "completion_2d": completion_2d, "geometry_turntable": turntable_gif, "geometry_multiview": multiview, "geometry_mesh": mesh, "completion_manifest": completion_manifest_path, } if geometry_manifest_path is not None: input_paths["geometry_manifest"] = geometry_manifest_path if verification_manifest_path is not None: input_paths["verification_manifest"] = verification_manifest_path if visual_turntable is not None: input_paths["visual_turntable"] = visual_turntable if visual_multiview is not None: input_paths["visual_multiview"] = visual_multiview for role, path in input_paths.items(): if not path.is_file(): raise FileNotFoundError(f"Missing {role}: {path}") if (visual_turntable is None) != (visual_multiview is None): raise ValueError( "--visual-turntable and --visual-multiview must be supplied together" ) review_dir = ( output_dir if output_dir.name == "00_quick_review" else output_dir / "00_quick_review" ) completion_manifest = read_json_object(completion_manifest_path) geometry_manifest = ( read_json_object(geometry_manifest_path) if geometry_manifest_path is not None else None ) verification_manifest = ( read_json_object(verification_manifest_path) if verification_manifest_path is not None else None ) metric_evidence = extract_metric_evidence(completion_manifest, geometry_manifest) outputs = copy_review_assets( source=source, completion_2d=completion_2d, turntable_gif=turntable_gif, multiview=multiview, mesh=mesh, review_dir=review_dir, visual_turntable=visual_turntable, visual_multiview=visual_multiview, ) build_overview( original=outputs["original"], completion_2d=outputs["completion_2d"], geometry_multiview=outputs["geometry_multiview"], destination=outputs["overview"], category=category, primary_3d_role=primary_3d_role, ) review = build_accessibility_review( sample_id=sample_id, category=category, completion_manifest=completion_manifest, metric_evidence=metric_evidence, visual_candidate_present=( visual_turntable is not None or primary_3d_role == "learned_amodal3d_gaussian" ), primary_3d_role=primary_3d_role, ) write_json(outputs["accessibility_review"], review) output_records = { role: file_record(path, review_dir) for role, path in outputs.items() if path.is_file() } input_records = { role: file_record(path, review_dir) for role, path in input_paths.items() } manifest = { "schema_version": "accessibilityamodal_quick_review_manifest_v1", "created_at_utc": datetime.now(timezone.utc).isoformat(), "sample_id": sample_id, "category": category, "purpose": "fast_human_review_of_original_2d_and_3d_completion", "primary_3d_role": primary_3d_role, "inputs": input_records, "files": output_records, "review_summary": { "overall_status": review["overall_status"], "safe_passage_claim": False, "wheelchair_wheeled": review["population_assessments"][ "wheelchair_wheeled" ]["status"], }, "visual_candidate_policy": { "present": ( visual_turntable is not None or primary_3d_role == "learned_amodal3d_gaussian" ), "role": "learned_visual_candidate", "is_metric_evidence": False, "is_passability_evidence": False, }, "provenance": { "completion_manifest": _provenance_summary(completion_manifest), "geometry_manifest_supplied": geometry_manifest is not None, "verification_manifest_supplied": verification_manifest is not None, "verification_decision": ( verification_manifest.get("decision") if verification_manifest is not None else None ), }, "path_policy": "All filesystem paths in this manifest are relative to manifest.json.", "integrity_note": "manifest.json omits its own hash to avoid recursive self-hashing.", } write_json(review_dir / "manifest.json", manifest) return review_dir def main(argv: Sequence[str] | None = None) -> int: parser = build_argument_parser() args = parser.parse_args(argv) category = str(args.category).strip().lower() sample_id = str(args.sample_id).strip() if not category: parser.error("--category must not be empty") if not sample_id: parser.error("--sample-id must not be empty") review_dir = build_bundle( source=Path(args.source).expanduser().resolve(), completion_2d=Path(args.completion_2d).expanduser().resolve(), turntable_gif=Path(args.turntable_gif).expanduser().resolve(), multiview=Path(args.multiview).expanduser().resolve(), mesh=Path(args.mesh).expanduser().resolve(), completion_manifest_path=Path(args.completion_manifest).expanduser().resolve(), geometry_manifest_path=( Path(args.geometry_manifest).expanduser().resolve() if args.geometry_manifest else None ), verification_manifest_path=( Path(args.verification_manifest).expanduser().resolve() if args.verification_manifest else None ), visual_turntable=( Path(args.visual_turntable).expanduser().resolve() if args.visual_turntable else None ), visual_multiview=( Path(args.visual_multiview).expanduser().resolve() if args.visual_multiview else None ), primary_3d_role=args.primary_3d_role, category=category, sample_id=sample_id, output_dir=Path(args.output_dir).expanduser().resolve(), ) print( json.dumps( { "status": "built", "sample_id": sample_id, "review_dir": str(review_dir), "safe_passage_claim": False, }, ensure_ascii=False, sort_keys=True, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())