tools/build_accessibility_review_bundle.py

#!/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())