"""Offline-only commissioning for the fixed Lite6 dual-camera capture run.

This module deliberately has no xArm, DepthAI, HTTP, subprocess, or socket
imports.  It cannot discover, connect to, configure, arm, or move hardware.
It consumes already-reviewed files, verifies all of their exact SHA-256
digests, and emits a production-loader-compatible safety-evidence artifact.

The physical observations remain human safety evidence.  This tool validates
their completeness, freshness, identity separation, and internal consistency;
it cannot replace an onsite risk assessment or an emergency-stop test.
"""

from __future__ import annotations

import argparse
import hashlib
import hmac
import ipaddress
import json
import math
import os
import re
import stat
import sys
import tempfile
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, Mapping, Optional, Sequence, Tuple, Union

try:
    from .capture_motion import (
        CONTROLLER_DEVICE_TYPE,
        JOINT_COUNT,
        REQUIRED_FRAME_COUNT,
        ROBOT_MODEL as MOTION_ROBOT_MODEL,
        CaptureMotionError,
        CaptureMotionPrimitive,
        load_commissioned_motion_primitive,
    )
    from .capture_safety import (
        MAX_CLOCK_SKEW,
        MAX_EVIDENCE_LIFETIME,
        MAX_WITNESS_AGE_AT_ISSUE,
        ROBOT_MODEL as SAFETY_ROBOT_MODEL,
        SAFETY_EVIDENCE_SCHEMA,
        SafetyEvidenceError,
        load_capture_safety_evidence,
    )
except ImportError:  # Direct ``python main/capture_commissioning.py`` use.
    from capture_motion import (  # type: ignore[no-redef]
        CONTROLLER_DEVICE_TYPE,
        JOINT_COUNT,
        REQUIRED_FRAME_COUNT,
        ROBOT_MODEL as MOTION_ROBOT_MODEL,
        CaptureMotionError,
        CaptureMotionPrimitive,
        load_commissioned_motion_primitive,
    )
    from capture_safety import (  # type: ignore[no-redef]
        MAX_CLOCK_SKEW,
        MAX_EVIDENCE_LIFETIME,
        MAX_WITNESS_AGE_AT_ISSUE,
        ROBOT_MODEL as SAFETY_ROBOT_MODEL,
        SAFETY_EVIDENCE_SCHEMA,
        SafetyEvidenceError,
        load_capture_safety_evidence,
    )


OBSERVATIONS_SCHEMA = "labcobot.capture-commissioning-observations.v1"
FK_TRACE_SCHEMA = "labcobot.capture-fk-trace.v1"
RECEIPT_SCHEMA = "labcobot.capture-commissioning-receipt.v1"
VERIFICATION_SCHEMA = "labcobot.capture-commissioning-verification.v1"

MAX_JSON_BYTES = 8 * 1024 * 1024
MAX_IMAGE_BYTES = 50 * 1024 * 1024
MAX_LOCKED_OBSERVATION_AGE = timedelta(minutes=5)
MAX_CAMERA_IMAGE_AGE_AT_WITNESS = timedelta(minutes=15)
MIN_CAMERA_WIDTH = 320
MIN_CAMERA_HEIGHT = 240
MAX_CAMERA_DIMENSION = 8192

_SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
_SAFE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,127}$")
_MXID_RE = re.compile(r"^[A-Za-z0-9_-]{8,128}$")
_TIMESTAMP_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
_JPEG_SOF_MARKERS = frozenset(
    (0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF)
)


class OfflineCommissioningError(ValueError):
    """One offline input is incomplete, inconsistent, or unsafe."""

    def __init__(self, code: str, message: str) -> None:
        self.code = str(code)
        super().__init__(str(message))


@dataclass(frozen=True)
class CameraObservation:
    mxid: str
    role: str
    image_sha256: str
    captured_at: datetime
    width_pixels: int
    height_pixels: int


@dataclass(frozen=True)
class ValidatedCommissioningInputs:
    raw: Mapping[str, Any]
    observations_sha256: str
    motion: CaptureMotionPrimitive
    fk_trace_sha256: str
    tool_camera: CameraObservation
    workspace_camera: CameraObservation
    issued_at: datetime


def _error(code: str, message: str) -> OfflineCommissioningError:
    return OfflineCommissioningError(code, message)


def _exact_mapping(value: Any, fields: Sequence[str], name: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping) or not all(isinstance(key, str) for key in value):
        raise _error(
            "commissioning_schema_invalid", "{} must be a JSON object".format(name)
        )
    if set(value) != set(fields):
        raise _error(
            "commissioning_schema_invalid",
            "{} fields do not match the offline commissioning schema".format(name),
        )
    return value


def _reject_duplicate_keys(pairs: Sequence[Tuple[str, Any]]) -> Mapping[str, Any]:
    result: Dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise _error("commissioning_json_invalid", "JSON contains duplicate fields")
        result[key] = value
    return result


def _reject_nonstandard_constant(_value: str) -> None:
    raise _error("commissioning_json_invalid", "JSON contains a non-standard number")


def _parse_json_float(value: str) -> float:
    try:
        result = float(value)
    except (TypeError, ValueError, OverflowError) as exc:
        raise _error("commissioning_json_invalid", "JSON number is invalid") from exc
    if not math.isfinite(result):
        raise _error("commissioning_json_invalid", "JSON number must be finite")
    return result


def _strict_json(payload: bytes, name: str) -> Any:
    try:
        return json.loads(
            payload.decode("utf-8"),
            object_pairs_hook=_reject_duplicate_keys,
            parse_constant=_reject_nonstandard_constant,
            parse_float=_parse_json_float,
        )
    except OfflineCommissioningError:
        raise
    except (UnicodeDecodeError, ValueError) as exc:
        raise _error(
            "commissioning_json_invalid", "{} is not strict UTF-8 JSON".format(name)
        ) from exc


def _read_regular(path: Path, *, maximum_bytes: int, name: str) -> bytes:
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    descriptor: Optional[int] = None
    try:
        before = os.lstat(str(path))
        if stat.S_ISLNK(before.st_mode):
            raise _error(
                "commissioning_unsafe_path",
                "{} may not be a symbolic link".format(name),
            )
        descriptor = os.open(str(path), flags)
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode):
            raise _error(
                "commissioning_unsafe_path", "{} must be a regular file".format(name)
            )
        if info.st_size < 2 or info.st_size > maximum_bytes:
            raise _error(
                "commissioning_size_invalid", "{} has an invalid size".format(name)
            )
        chunks = []
        remaining = maximum_bytes + 1
        while remaining > 0:
            chunk = os.read(descriptor, min(65536, remaining))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        payload = b"".join(chunks)
        if len(payload) != info.st_size or len(payload) > maximum_bytes:
            raise _error(
                "commissioning_size_invalid", "{} changed while being read".format(name)
            )
        return payload
    except OfflineCommissioningError:
        raise
    except OSError as exc:
        raise _error(
            "commissioning_read_failed", "{} could not be read".format(name)
        ) from exc
    finally:
        if descriptor is not None:
            os.close(descriptor)


def _canonical_json(value: Mapping[str, Any]) -> bytes:
    try:
        return (
            json.dumps(value, allow_nan=False, sort_keys=True, separators=(",", ":"))
            + "\n"
        ).encode("utf-8")
    except (TypeError, ValueError, OverflowError) as exc:
        raise _error(
            "commissioning_output_invalid", "Output could not be serialized"
        ) from exc


def _sha256(value: Any, name: str) -> str:
    if not isinstance(value, str) or not _SHA256_RE.fullmatch(value):
        raise _error(
            "commissioning_hash_invalid",
            "{} must be 64 lowercase hexadecimal characters".format(name),
        )
    return value


def _verify_digest(payload: bytes, expected: Any, name: str) -> str:
    expected_digest = _sha256(expected, "{} SHA-256".format(name))
    actual = hashlib.sha256(payload).hexdigest()
    if not hmac.compare_digest(actual, expected_digest):
        raise _error(
            "commissioning_hash_mismatch",
            "{} does not match its approved SHA-256".format(name),
        )
    return actual


def _safe_id(value: Any, name: str) -> str:
    if not isinstance(value, str) or not _SAFE_ID_RE.fullmatch(value):
        raise _error("commissioning_identity_invalid", "{} is invalid".format(name))
    return value


def _mxid(value: Any, name: str) -> str:
    if not isinstance(value, str) or not _MXID_RE.fullmatch(value):
        raise _error(
            "commissioning_identity_invalid", "{} MXID is invalid".format(name)
        )
    return value


def _timestamp(value: Any, name: str) -> datetime:
    if not isinstance(value, str) or not _TIMESTAMP_RE.fullmatch(value):
        raise _error(
            "commissioning_time_invalid",
            "{} must use canonical UTC ISO-8601 seconds".format(name),
        )
    try:
        parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
    except ValueError as exc:
        raise _error(
            "commissioning_time_invalid", "{} is invalid".format(name)
        ) from exc
    return parsed.replace(tzinfo=timezone.utc)


def _utc_now(value: Optional[datetime]) -> datetime:
    result = datetime.now(timezone.utc) if value is None else value
    if not isinstance(result, datetime) or result.tzinfo is None:
        raise _error(
            "commissioning_time_invalid", "Current time must be timezone-aware"
        )
    try:
        return result.astimezone(timezone.utc).replace(microsecond=0)
    except (TypeError, ValueError, OverflowError) as exc:
        raise _error("commissioning_time_invalid", "Current time is invalid") from exc


def _format_timestamp(value: datetime) -> str:
    return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _finite_number(value: Any, name: str) -> float:
    if isinstance(value, bool):
        raise _error("commissioning_value_invalid", "{} must be numeric".format(name))
    try:
        result = float(value)
    except (TypeError, ValueError, OverflowError) as exc:
        raise _error(
            "commissioning_value_invalid", "{} must be numeric".format(name)
        ) from exc
    if not math.isfinite(result):
        raise _error("commissioning_value_invalid", "{} must be finite".format(name))
    return result


def _finite_vector(value: Any, length: int, name: str) -> Tuple[float, ...]:
    if not isinstance(value, (list, tuple)) or len(value) != length:
        raise _error(
            "commissioning_value_invalid",
            "{} must contain exactly {} values".format(name, length),
        )
    return tuple(
        _finite_number(item, "{}[{}]".format(name, index))
        for index, item in enumerate(value)
    )


def _bool_vector(value: Any, expected: bool, name: str) -> Tuple[bool, ...]:
    if (
        not isinstance(value, (list, tuple))
        or len(value) != JOINT_COUNT
        or not all(type(item) is bool and item is expected for item in value)
    ):
        raise _error(
            "commissioning_controller_invalid",
            "{} must contain exactly six {} values".format(name, expected),
        )
    return tuple(value)


def _require_true(value: Mapping[str, Any], fields: Sequence[str], name: str) -> None:
    for field in fields:
        if value[field] is not True:
            raise _error(
                "commissioning_witness_invalid",
                "{} assertion {} was not witnessed".format(name, field),
            )


def _fresh_at_issue(
    observed_at: datetime,
    issued_at: datetime,
    *,
    maximum_age: timedelta,
    name: str,
) -> None:
    if (
        observed_at > issued_at + MAX_CLOCK_SKEW
        or issued_at - observed_at > maximum_age
    ):
        raise _error(
            "commissioning_time_invalid",
            "{} is stale or in the future".format(name),
        )


def _jpeg_dimensions(payload: bytes, name: str) -> Tuple[int, int]:
    if len(payload) < 16 or payload[:2] != b"\xff\xd8" or payload[-2:] != b"\xff\xd9":
        raise _error(
            "commissioning_image_invalid", "{} is not a complete JPEG".format(name)
        )
    offset = 2
    saw_start_of_scan = False
    while offset < len(payload) - 1:
        if payload[offset] != 0xFF:
            if saw_start_of_scan:
                offset += 1
                continue
            raise _error(
                "commissioning_image_invalid",
                "{} has malformed JPEG markers".format(name),
            )
        while offset < len(payload) and payload[offset] == 0xFF:
            offset += 1
        if offset >= len(payload):
            break
        marker = payload[offset]
        offset += 1
        if marker == 0x00:
            if saw_start_of_scan:
                continue
            raise _error(
                "commissioning_image_invalid",
                "{} has invalid JPEG stuffing".format(name),
            )
        if marker == 0xD9:
            break
        if marker == 0xDA:
            saw_start_of_scan = True
        if marker in (0x01, 0xD8) or 0xD0 <= marker <= 0xD7:
            continue
        if offset + 2 > len(payload):
            raise _error(
                "commissioning_image_invalid",
                "{} has a truncated JPEG marker".format(name),
            )
        segment_length = int.from_bytes(payload[offset : offset + 2], "big")
        if segment_length < 2 or offset + segment_length > len(payload):
            raise _error(
                "commissioning_image_invalid",
                "{} has a truncated JPEG segment".format(name),
            )
        if marker in _JPEG_SOF_MARKERS:
            if segment_length < 7:
                raise _error(
                    "commissioning_image_invalid",
                    "{} has an invalid JPEG frame".format(name),
                )
            height = int.from_bytes(payload[offset + 3 : offset + 5], "big")
            width = int.from_bytes(payload[offset + 5 : offset + 7], "big")
            if (
                width < MIN_CAMERA_WIDTH
                or height < MIN_CAMERA_HEIGHT
                or width > MAX_CAMERA_DIMENSION
                or height > MAX_CAMERA_DIMENSION
            ):
                raise _error(
                    "commissioning_image_invalid",
                    "{} JPEG dimensions are outside the commissioning range".format(
                        name
                    ),
                )
            return width, height
        offset += segment_length
    raise _error(
        "commissioning_image_invalid", "{} lacks a JPEG frame header".format(name)
    )


class _TraceForwardKinematics:
    """Sequential FK callable accepted by the production motion loader."""

    def __init__(self, frames: Sequence[Tuple[Tuple[float, ...], Tuple[float, ...]]]):
        self.frames = tuple(frames)
        self.index = 0

    def __call__(self, joints: Sequence[float]) -> Tuple[float, ...]:
        if self.index >= len(self.frames):
            raise _error("commissioning_fk_invalid", "Motion requested extra FK frames")
        expected_joints, pose = self.frames[self.index]
        actual_joints = _finite_vector(joints, JOINT_COUNT, "motion FK request")
        if actual_joints != expected_joints:
            raise _error(
                "commissioning_fk_mismatch",
                "FK trace frame {} does not match the motion primitive".format(
                    self.index
                ),
            )
        self.index += 1
        return pose


def _load_motion_and_trace(
    motion_path: Path,
    motion_sha256: str,
    fk_trace_path: Path,
    fk_trace_sha256: str,
) -> Tuple[CaptureMotionPrimitive, str, Mapping[str, Any]]:
    motion_payload = _read_regular(
        Path(motion_path), maximum_bytes=MAX_JSON_BYTES, name="motion primitive"
    )
    motion_digest = _verify_digest(motion_payload, motion_sha256, "motion primitive")
    trace_payload = _read_regular(
        Path(fk_trace_path), maximum_bytes=MAX_JSON_BYTES, name="FK trace"
    )
    trace_digest = _verify_digest(trace_payload, fk_trace_sha256, "FK trace")
    trace = _exact_mapping(
        _strict_json(trace_payload, "FK trace"),
        (
            "schema",
            "motion_profile_sha256",
            "xarm",
            "recorded_by",
            "recorded_at",
            "frames",
        ),
        "FK trace",
    )
    if trace["schema"] != FK_TRACE_SCHEMA:
        raise _error("commissioning_schema_invalid", "FK trace schema is unsupported")
    if (
        _sha256(trace["motion_profile_sha256"], "FK trace motion profile")
        != motion_digest
    ):
        raise _error(
            "commissioning_fk_mismatch", "FK trace is bound to another motion profile"
        )
    trace_xarm = _exact_mapping(
        trace["xarm"],
        ("ip", "model", "axis_count", "controller_device_type"),
        "FK trace xArm",
    )
    try:
        trace_ip = ipaddress.ip_address(trace_xarm["ip"])
    except (TypeError, ValueError) as exc:
        raise _error(
            "commissioning_identity_invalid", "FK trace xArm IP is invalid"
        ) from exc
    if (
        trace_ip.version != 4
        or str(trace_ip) != trace_xarm["ip"]
        or trace_xarm["model"] != SAFETY_ROBOT_MODEL
        or type(trace_xarm["axis_count"]) is not int
        or trace_xarm["axis_count"] != JOINT_COUNT
        or type(trace_xarm["controller_device_type"]) is not int
        or trace_xarm["controller_device_type"] != CONTROLLER_DEVICE_TYPE
    ):
        raise _error(
            "commissioning_identity_invalid",
            "FK trace must identify the commissioned six-axis Lite6 type 9",
        )
    _safe_id(trace["recorded_by"], "FK trace recorder")
    _timestamp(trace["recorded_at"], "FK trace recording time")
    raw_frames = trace["frames"]
    if not isinstance(raw_frames, list) or len(raw_frames) != REQUIRED_FRAME_COUNT:
        raise _error(
            "commissioning_fk_invalid",
            "FK trace must contain exactly {} frames".format(REQUIRED_FRAME_COUNT),
        )
    frames = []
    for index, raw_frame in enumerate(raw_frames):
        frame = _exact_mapping(
            raw_frame,
            ("time_seconds", "joints", "tcp_pose"),
            "FK frame {}".format(index),
        )
        timestamp = _finite_number(frame["time_seconds"], "FK frame time")
        if abs(timestamp - index * 0.01) > 1e-12:
            raise _error(
                "commissioning_fk_invalid",
                "FK trace does not use the exact 100 Hz schedule",
            )
        frames.append(
            (
                _finite_vector(frame["joints"], JOINT_COUNT, "FK frame joints"),
                _finite_vector(frame["tcp_pose"], 6, "FK frame TCP pose"),
            )
        )
    forward_kinematics = _TraceForwardKinematics(frames)
    try:
        primitive = load_commissioned_motion_primitive(
            motion_path,
            expected_sha256=motion_digest,
            forward_kinematics=forward_kinematics,
        )
    except OfflineCommissioningError:
        raise
    except CaptureMotionError as exc:
        cause = exc.__cause__
        if isinstance(cause, OfflineCommissioningError):
            raise cause
        raise
    if forward_kinematics.index != REQUIRED_FRAME_COUNT:
        raise _error(
            "commissioning_fk_invalid",
            "Production validation did not consume all FK frames",
        )
    return primitive, trace_digest, trace


def _parse_camera(
    raw: Any,
    *,
    role: str,
    observation_path: Path,
) -> CameraObservation:
    value = _exact_mapping(
        raw,
        (
            "mxid",
            "source_role",
            "image_path",
            "image_sha256",
            "captured_at",
            "width_pixels",
            "height_pixels",
        ),
        "{} camera observation".format(role),
    )
    if value["source_role"] != role:
        raise _error(
            "commissioning_identity_invalid", "{} camera role is invalid".format(role)
        )
    mxid = _mxid(value["mxid"], role)
    raw_path = value["image_path"]
    if not isinstance(raw_path, str) or not raw_path or "\x00" in raw_path:
        raise _error(
            "commissioning_unsafe_path", "{} image path is invalid".format(role)
        )
    image_path = Path(raw_path)
    if not image_path.is_absolute():
        image_path = observation_path.parent / image_path
    payload = _read_regular(
        image_path, maximum_bytes=MAX_IMAGE_BYTES, name="{} camera image".format(role)
    )
    image_digest = _verify_digest(
        payload, value["image_sha256"], "{} camera image".format(role)
    )
    width, height = _jpeg_dimensions(payload, "{} camera image".format(role))
    if (
        type(value["width_pixels"]) is not int
        or type(value["height_pixels"]) is not int
    ):
        raise _error(
            "commissioning_image_invalid", "Camera dimensions must be integers"
        )
    if (width, height) != (value["width_pixels"], value["height_pixels"]):
        raise _error(
            "commissioning_image_invalid",
            "{} camera dimensions do not match the JPEG".format(role),
        )
    return CameraObservation(
        mxid=mxid,
        role=role,
        image_sha256=image_digest,
        captured_at=_timestamp(
            value["captured_at"], "{} image capture time".format(role)
        ),
        width_pixels=width,
        height_pixels=height,
    )


def _validate_observations(
    observation_path: Path,
    observations_sha256: str,
    primitive: CaptureMotionPrimitive,
    issued_at: datetime,
) -> Tuple[Mapping[str, Any], str, CameraObservation, CameraObservation]:
    payload = _read_regular(
        observation_path,
        maximum_bytes=MAX_JSON_BYTES,
        name="commissioning observations",
    )
    digest = _verify_digest(payload, observations_sha256, "commissioning observations")
    value = _exact_mapping(
        _strict_json(payload, "commissioning observations"),
        (
            "schema",
            "device_uid",
            "xarm",
            "cameras",
            "emergency_stop_witness",
            "camera_visibility_witness",
            "locked_controller_observation",
        ),
        "commissioning observations",
    )
    if value["schema"] != OBSERVATIONS_SCHEMA:
        raise _error(
            "commissioning_schema_invalid", "Observation schema is unsupported"
        )
    _safe_id(value["device_uid"], "device UID")
    xarm = _exact_mapping(
        value["xarm"], ("ip", "model", "axis_count", "controller_device_type"), "xArm"
    )
    try:
        parsed_ip = ipaddress.ip_address(xarm["ip"])
    except (TypeError, ValueError) as exc:
        raise _error("commissioning_identity_invalid", "xArm IP is invalid") from exc
    if parsed_ip.version != 4 or str(parsed_ip) != xarm["ip"]:
        raise _error("commissioning_identity_invalid", "xArm IP must be canonical IPv4")
    if (
        xarm["model"] != SAFETY_ROBOT_MODEL
        or type(xarm["axis_count"]) is not int
        or xarm["axis_count"] != JOINT_COUNT
        or type(xarm["controller_device_type"]) is not int
        or xarm["controller_device_type"] != CONTROLLER_DEVICE_TYPE
        or primitive.robot_model != MOTION_ROBOT_MODEL
    ):
        raise _error(
            "commissioning_identity_invalid",
            "Commissioning requires one six-axis Lite6 controller type 9",
        )
    cameras = _exact_mapping(value["cameras"], ("tool", "workspace"), "cameras")
    tool = _parse_camera(
        cameras["tool"], role="tool", observation_path=observation_path
    )
    workspace = _parse_camera(
        cameras["workspace"], role="workspace", observation_path=observation_path
    )
    if tool.mxid == workspace.mxid or tool.image_sha256 == workspace.image_sha256:
        raise _error(
            "commissioning_camera_invalid",
            "Tool and workspace evidence must be distinct camera views",
        )

    emergency = _exact_mapping(
        value["emergency_stop_witness"],
        (
            "performed_by",
            "witnessed_by",
            "witnessed_at",
            "physical_estop_pressed",
            "motion_inhibited",
            "manual_reset_required",
            "no_automatic_restart",
        ),
        "emergency-stop witness",
    )
    _require_true(
        emergency,
        (
            "physical_estop_pressed",
            "motion_inhibited",
            "manual_reset_required",
            "no_automatic_restart",
        ),
        "emergency-stop witness",
    )
    performer = _safe_id(emergency["performed_by"], "emergency-stop performer")
    emergency_witness = _safe_id(emergency["witnessed_by"], "emergency-stop witness")
    if performer == emergency_witness:
        raise _error(
            "commissioning_witness_invalid",
            "Emergency-stop performer and witness must be different people",
        )
    emergency_at = _timestamp(emergency["witnessed_at"], "emergency-stop witness time")
    _fresh_at_issue(
        emergency_at,
        issued_at,
        maximum_age=MAX_WITNESS_AGE_AT_ISSUE,
        name="emergency-stop witness",
    )

    visibility = _exact_mapping(
        value["camera_visibility_witness"],
        (
            "witnessed_by",
            "witnessed_at",
            "tool_image_sha256",
            "workspace_image_sha256",
            "full_workcell_visible",
            "entire_robot_envelope_visible",
            "both_oak_views_verified",
            "lighting_sufficient",
            "robot_base_visible",
            "tool_tip_visible",
            "workspace_boundary_visible",
            "no_blind_zone_in_motion_envelope",
        ),
        "camera-visibility witness",
    )
    _require_true(
        visibility,
        (
            "full_workcell_visible",
            "entire_robot_envelope_visible",
            "both_oak_views_verified",
            "lighting_sufficient",
            "robot_base_visible",
            "tool_tip_visible",
            "workspace_boundary_visible",
            "no_blind_zone_in_motion_envelope",
        ),
        "camera-visibility witness",
    )
    visibility_witness = _safe_id(
        visibility["witnessed_by"], "camera-visibility witness"
    )
    visibility_at = _timestamp(
        visibility["witnessed_at"], "camera-visibility witness time"
    )
    _fresh_at_issue(
        visibility_at,
        issued_at,
        maximum_age=MAX_WITNESS_AGE_AT_ISSUE,
        name="camera-visibility witness",
    )
    if (
        _sha256(visibility["tool_image_sha256"], "witnessed tool image")
        != tool.image_sha256
        or _sha256(visibility["workspace_image_sha256"], "witnessed workspace image")
        != workspace.image_sha256
    ):
        raise _error(
            "commissioning_camera_invalid",
            "Visibility witness references different images",
        )
    for camera in (tool, workspace):
        if (
            camera.captured_at > visibility_at + MAX_CLOCK_SKEW
            or visibility_at - camera.captured_at > MAX_CAMERA_IMAGE_AGE_AT_WITNESS
        ):
            raise _error(
                "commissioning_time_invalid",
                "{} camera image is stale relative to its witness".format(camera.role),
            )

    locked = _exact_mapping(
        value["locked_controller_observation"],
        (
            "observed_by",
            "observed_at",
            "connected",
            "state",
            "mode",
            "error_code",
            "warn_code",
            "device_type",
            "axis_count",
            "moving",
            "motor_enable_states",
            "motor_brake_states",
            "joint_angles_radians",
            "tcp_pose",
            "reduced_mode",
        ),
        "locked controller observation",
    )
    controller_observer = _safe_id(locked["observed_by"], "controller observer")
    locked_at = _timestamp(locked["observed_at"], "controller observation time")
    _fresh_at_issue(
        locked_at,
        issued_at,
        maximum_age=MAX_LOCKED_OBSERVATION_AGE,
        name="locked controller observation",
    )
    if (
        locked["connected"] is not True
        or type(locked["state"]) is not int
        or locked["state"] != 4
        or type(locked["mode"]) is not int
        or locked["mode"] != 0
        or type(locked["error_code"]) is not int
        or locked["error_code"] != 0
        or type(locked["warn_code"]) is not int
        or locked["warn_code"] != 0
        or type(locked["device_type"]) is not int
        or locked["device_type"] != CONTROLLER_DEVICE_TYPE
        or type(locked["axis_count"]) is not int
        or locked["axis_count"] != JOINT_COUNT
        or locked["moving"] is not False
    ):
        raise _error(
            "commissioning_controller_invalid",
            "Controller must be healthy, stationary, disabled, "
            "and locked in state 4/mode 0",
        )
    _bool_vector(locked["motor_enable_states"], False, "motor-enable states")
    _bool_vector(locked["motor_brake_states"], False, "motor-brake states")
    current_joints = _finite_vector(
        locked["joint_angles_radians"], JOINT_COUNT, "current joint angles"
    )
    current_tcp = _finite_vector(locked["tcp_pose"], 6, "current TCP pose")
    safety = primitive.safety
    for index, (joint, first, low, high, delta) in enumerate(
        zip(
            current_joints,
            primitive.frames[0].joints,
            safety.joint_lower_radians,
            safety.joint_upper_radians,
            safety.max_start_delta_radians,
        )
    ):
        if not low <= joint <= high:
            raise _error(
                "commissioning_start_invalid",
                "Current joint {} is outside the reviewed primitive".format(index + 1),
            )
        if abs(joint - first) > delta:
            raise _error(
                "commissioning_start_invalid",
                "Current joint {} is too far from the reviewed start".format(index + 1),
            )
    if not all(
        low <= coordinate <= high
        for coordinate, low, high in zip(
            current_tcp[:3],
            safety.workspace_lower_xyz_mm,
            safety.workspace_upper_xyz_mm,
        )
    ):
        raise _error(
            "commissioning_start_invalid",
            "Current TCP is outside the reviewed workspace",
        )

    reduced = _exact_mapping(
        locked["reduced_mode"],
        (
            "commissioned_by",
            "commissioned_at",
            "enabled",
            "tcp_boundary_mm",
            "max_tcp_speed_mm_per_second",
            "max_joint_speed_radians_per_second",
            "joint_ranges_radians",
            "safety_boundary_enabled",
            "collision_rebound_enabled",
        ),
        "reduced mode",
    )
    commissioner = _safe_id(reduced["commissioned_by"], "reduced-mode commissioner")
    if commissioner != controller_observer:
        raise _error(
            "commissioning_witness_invalid",
            "Reduced-mode commissioner must make the locked controller observation",
        )
    if emergency_witness == performer or visibility_witness == commissioner:
        raise _error(
            "commissioning_witness_invalid",
            "Physical witnesses must be independent of the operator",
        )
    reduced_at = _timestamp(
        reduced["commissioned_at"], "reduced-mode commissioning time"
    )
    _fresh_at_issue(
        reduced_at,
        issued_at,
        maximum_age=MAX_WITNESS_AGE_AT_ISSUE,
        name="reduced-mode commissioning",
    )
    if (
        reduced["enabled"] is not True
        or reduced["safety_boundary_enabled"] is not True
        or reduced["collision_rebound_enabled"] is not False
    ):
        raise _error(
            "commissioning_reduced_mode_invalid",
            "Reduced mode and its safety boundary must be on; "
            "collision rebound must be off",
        )
    boundary = _finite_vector(reduced["tcp_boundary_mm"], 6, "reduced TCP boundary")
    joint_ranges = _finite_vector(
        reduced["joint_ranges_radians"], 14, "reduced joint ranges"
    )
    _finite_number(reduced["max_tcp_speed_mm_per_second"], "reduced TCP speed")
    _finite_number(reduced["max_joint_speed_radians_per_second"], "reduced joint speed")
    for index, frame in enumerate(primitive.frames):
        for axis, joint in enumerate(frame.joints):
            if not joint_ranges[axis * 2] <= joint <= joint_ranges[axis * 2 + 1]:
                raise _error(
                    "commissioning_reduced_mode_invalid",
                    "Motion frame {} leaves reduced joint range {}".format(
                        index, axis + 1
                    ),
                )
        pose = primitive.tcp_poses[index]
        for axis, coordinate in enumerate(pose[:3]):
            maximum = boundary[axis * 2]
            minimum = boundary[axis * 2 + 1]
            if not minimum <= coordinate <= maximum:
                raise _error(
                    "commissioning_reduced_mode_invalid",
                    "Motion frame {} leaves the reduced TCP boundary".format(index),
                )
    for axis, joint in enumerate(current_joints):
        if not joint_ranges[axis * 2] <= joint <= joint_ranges[axis * 2 + 1]:
            raise _error(
                "commissioning_start_invalid",
                "Current joint {} is outside the reduced range".format(axis + 1),
            )
    for axis, coordinate in enumerate(current_tcp[:3]):
        if not boundary[axis * 2 + 1] <= coordinate <= boundary[axis * 2]:
            raise _error(
                "commissioning_start_invalid", "Current TCP is outside reduced mode"
            )
    return value, digest, tool, workspace


def validate_commissioning_inputs(
    *,
    observations_path: Union[os.PathLike, str],
    observations_sha256: str,
    motion_path: Union[os.PathLike, str],
    motion_sha256: str,
    fk_trace_path: Union[os.PathLike, str],
    fk_trace_sha256: str,
    now: Optional[datetime] = None,
) -> ValidatedCommissioningInputs:
    """Validate all offline inputs without creating files or touching hardware."""

    issued_at = _utc_now(now)
    primitive, trace_digest, _trace = _load_motion_and_trace(
        Path(motion_path), motion_sha256, Path(fk_trace_path), fk_trace_sha256
    )
    observations, observation_digest, tool, workspace = _validate_observations(
        Path(observations_path), observations_sha256, primitive, issued_at
    )
    if dict(_trace["xarm"]) != dict(observations["xarm"]):
        raise _error(
            "commissioning_identity_invalid",
            "FK trace and observations identify different xArm controllers",
        )
    return ValidatedCommissioningInputs(
        raw=observations,
        observations_sha256=observation_digest,
        motion=primitive,
        fk_trace_sha256=trace_digest,
        tool_camera=tool,
        workspace_camera=workspace,
        issued_at=issued_at,
    )


def _source_binding_sha256(inputs: ValidatedCommissioningInputs) -> str:
    payload = _canonical_json(
        {
            "observations_sha256": inputs.observations_sha256,
            "motion_profile_sha256": inputs.motion.profile_sha256,
            "fk_trace_sha256": inputs.fk_trace_sha256,
            "tool_image_sha256": inputs.tool_camera.image_sha256,
            "workspace_image_sha256": inputs.workspace_camera.image_sha256,
        }
    )
    return hashlib.sha256(payload).hexdigest()


def _evidence_value(
    inputs: ValidatedCommissioningInputs, valid_for: timedelta
) -> Mapping[str, Any]:
    if valid_for <= timedelta(0) or valid_for > MAX_EVIDENCE_LIFETIME:
        raise _error(
            "commissioning_lifetime_invalid",
            "Evidence lifetime must be positive and at most seven days",
        )
    raw = inputs.raw
    xarm = raw["xarm"]
    emergency = raw["emergency_stop_witness"]
    visibility = raw["camera_visibility_witness"]
    locked = raw["locked_controller_observation"]
    reduced = locked["reduced_mode"]
    safety = inputs.motion.safety
    return {
        "schema": SAFETY_EVIDENCE_SCHEMA,
        "evidence_id": "capture-commission-{}".format(_source_binding_sha256(inputs)),
        "device_uid": raw["device_uid"],
        "issued_at": _format_timestamp(inputs.issued_at),
        "expires_at": _format_timestamp(inputs.issued_at + valid_for),
        "xarm": {
            "ip": xarm["ip"],
            "model": xarm["model"],
            "axis_count": xarm["axis_count"],
            "controller_device_type": xarm["controller_device_type"],
        },
        "cameras": {
            "tool": {"mxid": inputs.tool_camera.mxid, "source_role": "tool"},
            "workspace": {
                "mxid": inputs.workspace_camera.mxid,
                "source_role": "workspace",
            },
        },
        "motion_profile_sha256": inputs.motion.profile_sha256,
        "workspace_lower_xyz_mm": list(safety.workspace_lower_xyz_mm),
        "workspace_upper_xyz_mm": list(safety.workspace_upper_xyz_mm),
        "joint_limit_policy": inputs.motion.joint_limit_policy,
        "emergency_stop_witness": {
            "witnessed_by": emergency["witnessed_by"],
            "witnessed_at": emergency["witnessed_at"],
            "physical_estop_pressed": True,
            "motion_inhibited": True,
        },
        "camera_visibility_witness": {
            "witnessed_by": visibility["witnessed_by"],
            "witnessed_at": visibility["witnessed_at"],
            "full_workcell_visible": True,
            "entire_robot_envelope_visible": True,
            "both_oak_views_verified": True,
        },
        "reduced_mode": dict(reduced),
        "controller_phases": {
            "preflight": {
                "state": 4,
                "mode": 0,
                "motor_enable_states": [False] * JOINT_COUNT,
                "motor_brake_states": [False] * JOINT_COUNT,
            },
            "active": {
                "state": 0,
                "mode": 1,
                "motor_enable_states": [True] * JOINT_COUNT,
                "motor_brake_states": [True] * JOINT_COUNT,
            },
        },
    }


def _production_load(
    evidence_path: Path,
    evidence_sha256: str,
    inputs: ValidatedCommissioningInputs,
) -> Any:
    xarm = inputs.raw["xarm"]
    safety = inputs.motion.safety
    return load_capture_safety_evidence(
        evidence_path,
        expected_sha256=evidence_sha256,
        device_uid=inputs.raw["device_uid"],
        xarm_ip=xarm["ip"],
        xarm_model=xarm["model"],
        tool_camera_mxid=inputs.tool_camera.mxid,
        workspace_camera_mxid=inputs.workspace_camera.mxid,
        motion_profile_sha256=inputs.motion.profile_sha256,
        workspace_lower_xyz_mm=safety.workspace_lower_xyz_mm,
        workspace_upper_xyz_mm=safety.workspace_upper_xyz_mm,
        joint_limit_policy=inputs.motion.joint_limit_policy,
        now=inputs.issued_at,
    )


def _prevalidate_evidence(
    payload: bytes,
    digest: str,
    inputs: ValidatedCommissioningInputs,
    parent: Path,
) -> None:
    temporary_path: Optional[Path] = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="wb",
            prefix=".labcobot-capture-evidence-",
            dir=str(parent),
            delete=False,
        ) as temporary:
            temporary_path = Path(temporary.name)
            os.chmod(temporary.name, 0o600)
            temporary.write(payload)
            temporary.flush()
            os.fsync(temporary.fileno())
        _production_load(temporary_path, digest, inputs)
    finally:
        if temporary_path is not None:
            try:
                temporary_path.unlink()
            except FileNotFoundError:
                pass


def _ensure_new_target(path: Path, name: str) -> None:
    if not path.parent.is_dir():
        raise _error(
            "commissioning_output_invalid", "{} parent does not exist".format(name)
        )
    try:
        os.lstat(str(path))
    except FileNotFoundError:
        return
    except OSError as exc:
        raise _error(
            "commissioning_output_invalid", "{} cannot be checked".format(name)
        ) from exc
    raise _error("commissioning_output_exists", "{} already exists".format(name))


def _write_new_regular(path: Path, payload: bytes, name: str) -> None:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0)
    flags |= getattr(os, "O_NOFOLLOW", 0)
    descriptor: Optional[int] = None
    created = False
    try:
        descriptor = os.open(str(path), flags, 0o600)
        created = True
        view = memoryview(payload)
        while view:
            written = os.write(descriptor, view)
            if written <= 0:
                raise OSError("short write")
            view = view[written:]
        os.fsync(descriptor)
    except OSError as exc:
        if created:
            try:
                path.unlink()
            except OSError:
                pass
        raise _error(
            "commissioning_write_failed", "{} could not be written".format(name)
        ) from exc
    finally:
        if descriptor is not None:
            os.close(descriptor)


def _receipt_value(
    inputs: ValidatedCommissioningInputs,
    evidence_sha256: str,
    expires_at: datetime,
) -> Mapping[str, Any]:
    emergency = inputs.raw["emergency_stop_witness"]
    visibility = inputs.raw["camera_visibility_witness"]
    reduced = inputs.raw["locked_controller_observation"]["reduced_mode"]
    return {
        "schema": RECEIPT_SCHEMA,
        "offline_only": True,
        "device_uid": inputs.raw["device_uid"],
        "issued_at": _format_timestamp(inputs.issued_at),
        "expires_at": _format_timestamp(expires_at),
        "observations_sha256": inputs.observations_sha256,
        "source_binding_sha256": _source_binding_sha256(inputs),
        "motion_profile_sha256": inputs.motion.profile_sha256,
        "fk_trace_sha256": inputs.fk_trace_sha256,
        "safety_evidence_sha256": evidence_sha256,
        "cameras": {
            "tool": {
                "mxid": inputs.tool_camera.mxid,
                "image_sha256": inputs.tool_camera.image_sha256,
                "captured_at": _format_timestamp(inputs.tool_camera.captured_at),
                "width_pixels": inputs.tool_camera.width_pixels,
                "height_pixels": inputs.tool_camera.height_pixels,
            },
            "workspace": {
                "mxid": inputs.workspace_camera.mxid,
                "image_sha256": inputs.workspace_camera.image_sha256,
                "captured_at": _format_timestamp(inputs.workspace_camera.captured_at),
                "width_pixels": inputs.workspace_camera.width_pixels,
                "height_pixels": inputs.workspace_camera.height_pixels,
            },
        },
        "participants": {
            "emergency_stop_performed_by": emergency["performed_by"],
            "emergency_stop_witnessed_by": emergency["witnessed_by"],
            "camera_visibility_witnessed_by": visibility["witnessed_by"],
            "reduced_mode_commissioned_by": reduced["commissioned_by"],
        },
        "production_loader_verified": True,
    }


def issue_capture_safety_evidence(
    *,
    observations_path: Union[os.PathLike, str],
    observations_sha256: str,
    motion_path: Union[os.PathLike, str],
    motion_sha256: str,
    fk_trace_path: Union[os.PathLike, str],
    fk_trace_sha256: str,
    output_path: Union[os.PathLike, str],
    receipt_path: Union[os.PathLike, str],
    valid_hours: float = 24.0,
    now: Optional[datetime] = None,
) -> Mapping[str, Any]:
    """Issue canonical evidence after the production loaders accept all inputs."""

    if isinstance(valid_hours, bool):
        raise _error("commissioning_lifetime_invalid", "Evidence lifetime is invalid")
    try:
        valid_for = timedelta(hours=float(valid_hours))
    except (TypeError, ValueError, OverflowError) as exc:
        raise _error(
            "commissioning_lifetime_invalid", "Evidence lifetime is invalid"
        ) from exc
    output = Path(output_path)
    receipt = Path(receipt_path)
    if output == receipt:
        raise _error(
            "commissioning_output_invalid", "Evidence and receipt paths must differ"
        )
    _ensure_new_target(output, "evidence output")
    _ensure_new_target(receipt, "receipt output")
    inputs = validate_commissioning_inputs(
        observations_path=observations_path,
        observations_sha256=observations_sha256,
        motion_path=motion_path,
        motion_sha256=motion_sha256,
        fk_trace_path=fk_trace_path,
        fk_trace_sha256=fk_trace_sha256,
        now=now,
    )
    evidence_value = _evidence_value(inputs, valid_for)
    evidence_payload = _canonical_json(evidence_value)
    evidence_digest = hashlib.sha256(evidence_payload).hexdigest()
    _prevalidate_evidence(evidence_payload, evidence_digest, inputs, output.parent)
    expires_at = inputs.issued_at + valid_for
    receipt_value = _receipt_value(inputs, evidence_digest, expires_at)
    receipt_payload = _canonical_json(receipt_value)
    receipt_digest = hashlib.sha256(receipt_payload).hexdigest()
    _write_new_regular(output, evidence_payload, "evidence output")
    try:
        _write_new_regular(receipt, receipt_payload, "receipt output")
    except Exception:
        try:
            output.unlink()
        except OSError:
            pass
        raise
    _production_load(output, evidence_digest, inputs)
    return {
        "ok": True,
        "schema": VERIFICATION_SCHEMA,
        "offline_only": True,
        "production_loader_verified": True,
        "safety_evidence_sha256": evidence_digest,
        "commissioning_receipt_sha256": receipt_digest,
        "motion_profile_sha256": inputs.motion.profile_sha256,
        "observations_sha256": inputs.observations_sha256,
        "source_binding_sha256": _source_binding_sha256(inputs),
        "fk_trace_sha256": inputs.fk_trace_sha256,
        "expires_at": _format_timestamp(expires_at),
    }


def verify_capture_safety_evidence(
    *,
    observations_path: Union[os.PathLike, str],
    observations_sha256: str,
    motion_path: Union[os.PathLike, str],
    motion_sha256: str,
    fk_trace_path: Union[os.PathLike, str],
    fk_trace_sha256: str,
    evidence_path: Union[os.PathLike, str],
    evidence_sha256: str,
    now: Optional[datetime] = None,
) -> Mapping[str, Any]:
    """Revalidate evidence plus its source observations without side effects."""

    inputs = validate_commissioning_inputs(
        observations_path=observations_path,
        observations_sha256=observations_sha256,
        motion_path=motion_path,
        motion_sha256=motion_sha256,
        fk_trace_path=fk_trace_path,
        fk_trace_sha256=fk_trace_sha256,
        now=now,
    )
    evidence_payload = _read_regular(
        Path(evidence_path), maximum_bytes=MAX_JSON_BYTES, name="safety evidence"
    )
    digest = _verify_digest(evidence_payload, evidence_sha256, "safety evidence")
    loaded = _production_load(Path(evidence_path), digest, inputs)
    expected_id = "capture-commission-{}".format(_source_binding_sha256(inputs))
    if loaded.evidence_id != expected_id:
        raise _error(
            "commissioning_binding_mismatch",
            "Safety evidence is not bound to the reviewed observation bundle",
        )
    return {
        "ok": True,
        "schema": VERIFICATION_SCHEMA,
        "offline_only": True,
        "production_loader_verified": True,
        "safety_evidence_sha256": digest,
        "motion_profile_sha256": inputs.motion.profile_sha256,
        "observations_sha256": inputs.observations_sha256,
        "source_binding_sha256": _source_binding_sha256(inputs),
        "fk_trace_sha256": inputs.fk_trace_sha256,
        "expires_at": _format_timestamp(loaded.expires_at),
    }


def _add_source_arguments(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--observations", type=Path, required=True)
    parser.add_argument("--observations-sha256", required=True)
    parser.add_argument("--motion", type=Path, required=True)
    parser.add_argument("--motion-sha256", required=True)
    parser.add_argument("--fk-trace", type=Path, required=True)
    parser.add_argument("--fk-trace-sha256", required=True)


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Offline-only LabCobot capture commissioning evidence tool"
    )
    commands = parser.add_subparsers(dest="command", required=True)
    issue = commands.add_parser(
        "issue", help="validate observations and issue new evidence"
    )
    _add_source_arguments(issue)
    issue.add_argument("--output", type=Path, required=True)
    issue.add_argument("--receipt", type=Path, required=True)
    issue.add_argument("--valid-hours", type=float, default=24.0)
    verify = commands.add_parser("verify", help="revalidate existing evidence offline")
    _add_source_arguments(verify)
    verify.add_argument("--evidence", type=Path, required=True)
    verify.add_argument("--evidence-sha256", required=True)
    return parser


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = _parser().parse_args(argv)
    try:
        if args.command == "issue":
            result = issue_capture_safety_evidence(
                observations_path=args.observations,
                observations_sha256=args.observations_sha256,
                motion_path=args.motion,
                motion_sha256=args.motion_sha256,
                fk_trace_path=args.fk_trace,
                fk_trace_sha256=args.fk_trace_sha256,
                output_path=args.output,
                receipt_path=args.receipt,
                valid_hours=args.valid_hours,
            )
        else:
            result = verify_capture_safety_evidence(
                observations_path=args.observations,
                observations_sha256=args.observations_sha256,
                motion_path=args.motion,
                motion_sha256=args.motion_sha256,
                fk_trace_path=args.fk_trace,
                fk_trace_sha256=args.fk_trace_sha256,
                evidence_path=args.evidence,
                evidence_sha256=args.evidence_sha256,
            )
    except (OfflineCommissioningError, CaptureMotionError, SafetyEvidenceError) as exc:
        code = getattr(exc, "code", "commissioning_failed")
        print(
            json.dumps(
                {"ok": False, "code": str(code), "message": str(exc)},
                sort_keys=True,
                separators=(",", ":"),
            ),
            file=sys.stderr,
        )
        return 2
    print(json.dumps(result, sort_keys=True, separators=(",", ":")))
    return 0


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


__all__ = [
    "FK_TRACE_SCHEMA",
    "OBSERVATIONS_SCHEMA",
    "OfflineCommissioningError",
    "RECEIPT_SCHEMA",
    "ValidatedCommissioningInputs",
    "issue_capture_safety_evidence",
    "main",
    "validate_commissioning_inputs",
    "verify_capture_safety_evidence",
]
