"""Commissioning and readiness commands without commanded robot motion.

The optional demonstration recorder reads a connected robot while an onsite
operator owns teach mode; no function here enables motors or sends motion.
Other commands build reviewed artifacts and inspect deployment configuration.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple
from urllib.parse import urlsplit

if __package__:
    from .functions.trajectory_training import (
        Demonstration,
        DemonstrationRecorder,
        DemonstrationStore,
        TrajectoryError,
        ValveMotionPrimitive,
        ValveMotionPrimitiveStore,
        ValveTarget,
    )
    from .integrations.http import TransportError, UrllibTransport
    from .integrations.readiness import DeviceReadinessError
    from .integrations.settings import IntegrationSettings
    from .integrations.workcell import open_xarm_metric_source
    from .safety_boundary import load_lite6_safety_profile
    from .vision.eye_in_hand import (
        CalibrationError,
        CameraIntrinsics,
        EyeInHandCalibrationStore,
        EyeInHandCalibrator,
        HandEyeSample,
        RigidTransform,
    )
    from .vision.reviewed_valve_model import load_reviewed_valve_model
else:  # Direct ``python main/commissioning.py`` execution on the Pi.
    from functions.trajectory_training import (  # type: ignore[no-redef]
        Demonstration,
        DemonstrationRecorder,
        DemonstrationStore,
        TrajectoryError,
        ValveMotionPrimitive,
        ValveMotionPrimitiveStore,
        ValveTarget,
    )
    from integrations.http import (  # type: ignore[no-redef]
        TransportError,
        UrllibTransport,
    )
    from integrations.readiness import (  # type: ignore[no-redef]
        DeviceReadinessError,
    )
    from integrations.settings import IntegrationSettings  # type: ignore[no-redef]
    from integrations.workcell import (  # type: ignore[no-redef]
        open_xarm_metric_source,
    )
    from safety_boundary import (  # type: ignore[no-redef]
        load_lite6_safety_profile,
    )
    from vision.eye_in_hand import (  # type: ignore[no-redef]
        CalibrationError,
        CameraIntrinsics,
        EyeInHandCalibrationStore,
        EyeInHandCalibrator,
        HandEyeSample,
        RigidTransform,
    )
    from vision.reviewed_valve_model import (  # type: ignore[no-redef]
        load_reviewed_valve_model,
    )


class CommissioningError(ValueError):
    """Invalid or incomplete operator commissioning input."""


_MAC_ADDRESS_RE = re.compile(
    r"^(?P<octet>[0-9a-fA-F]{2})(?P<separator>[:-])"
    r"(?:[0-9a-fA-F]{2}(?P=separator)){4}[0-9a-fA-F]{2}$"
)
_INTERFACE_NAME_RE = re.compile(r"^[A-Za-z0-9_.:-]{1,32}$")
_DEVICE_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$")


def read_network_interface_mac(
    interface: str = "eth0",
    *,
    sys_class_net: Path = Path("/sys/class/net"),
) -> str:
    """Read and validate one unicast interface MAC from Linux sysfs."""

    interface_name = str(interface).strip()
    if not _INTERFACE_NAME_RE.fullmatch(interface_name):
        raise CommissioningError("Network interface name is invalid")
    address_path = Path(sys_class_net) / interface_name / "address"
    try:
        raw_address = address_path.read_bytes()
    except OSError as exc:
        raise CommissioningError(
            "Cannot read MAC address for interface {}".format(interface_name)
        ) from exc
    if not raw_address or len(raw_address) > 128:
        raise CommissioningError(
            "MAC address for interface {} has an invalid size".format(interface_name)
        )
    try:
        address = raw_address.decode("ascii").strip()
    except UnicodeError as exc:
        raise CommissioningError(
            "MAC address for interface {} is not ASCII".format(interface_name)
        ) from exc
    return _normalize_registration_mac(address)


def build_device_registration_request(
    *,
    manufacturer_id: str,
    location: str,
    interface: str = "eth0",
    mac_address: Optional[str] = None,
    uid_device: Optional[str] = None,
    name: str = "AINA LabCobot",
    model: str = "LabCobot AINA",
    hardware_revision: Optional[str] = None,
    serial_number: Optional[str] = None,
    sys_class_net: Path = Path("/sys/class/net"),
) -> Mapping[str, Any]:
    """Build the exact authenticated Supraworx MAC-registration request."""

    interface_name = str(interface).strip()
    if not _INTERFACE_NAME_RE.fullmatch(interface_name):
        raise CommissioningError("Network interface name is invalid")
    normalized_mac = (
        _normalize_registration_mac(mac_address)
        if mac_address is not None
        else read_network_interface_mac(
            interface_name,
            sys_class_net=sys_class_net,
        )
    )
    compact_mac = normalized_mac.replace(":", "")
    normalized_uid = _device_identifier(
        uid_device or "labcobot-{}".format(compact_mac),
        "uid_device",
    )
    body: Dict[str, Any] = {
        "eth0": normalized_mac,
        "name": _registration_text(name, "name", maximum_length=100),
        "manufacturer_id": _device_identifier(
            manufacturer_id,
            "manufacturer_id",
        ),
        "uid_device": normalized_uid,
        "device_type": "robotic-stand",
        "location": _registration_text(
            location,
            "location",
            maximum_length=100,
        ),
        "model": _registration_text(
            model,
            "model",
            maximum_length=100,
        ),
    }
    if hardware_revision is not None:
        body["hardware_revision"] = _registration_text(
            hardware_revision,
            "hardware_revision",
            maximum_length=50,
        )
    if serial_number is not None:
        body["serial_number"] = _registration_text(
            serial_number,
            "serial_number",
            maximum_length=100,
        )
    return {
        "schema": "labcobot.device-registration-request.v1",
        "source_interface": interface_name,
        "request": {
            "method": "POST",
            "path": ("/device-objects-manager/api/devices/register-by-mac/"),
            "content_type": "application/json",
            "body": body,
        },
    }


def _normalize_registration_mac(value: Any) -> str:
    raw_value = str(value or "").strip()
    match = _MAC_ADDRESS_RE.fullmatch(raw_value)
    if match is None:
        raise CommissioningError("MAC address must contain six hexadecimal octets")
    normalized = raw_value.replace("-", ":").lower()
    octets = [int(octet, 16) for octet in normalized.split(":")]
    if not any(octets) or all(octet == 255 for octet in octets):
        raise CommissioningError("MAC address cannot be all-zero or broadcast")
    if octets[0] & 1:
        raise CommissioningError("MAC address must identify a unicast device")
    return normalized


def _device_identifier(value: Any, name: str) -> str:
    result = _registration_text(value, name, maximum_length=100)
    if not _DEVICE_IDENTIFIER_RE.fullmatch(result):
        raise CommissioningError("{} must be a URL-safe device identifier".format(name))
    return result


def _registration_text(
    value: Any,
    name: str,
    *,
    maximum_length: int,
) -> str:
    result = str(value or "").strip()
    if not result:
        raise CommissioningError("{} is required".format(name))
    if len(result) > maximum_length:
        raise CommissioningError(
            "{} must be at most {} characters".format(name, maximum_length)
        )
    if any(ord(character) < 32 or ord(character) == 127 for character in result):
        raise CommissioningError("{} cannot contain control characters".format(name))
    return result


def record_xarm_demonstration(
    arm: Any,
    output_path: Path,
    *,
    duration_seconds: float,
    sample_rate_hz: float = 25.0,
    name: str = "valve-demonstration",
) -> Mapping[str, Any]:
    """Record read-only joint/TCP snapshots from an already connected xArm."""

    duration = float(duration_seconds)
    sample_rate = float(sample_rate_hz)
    if not 0.1 <= duration <= 3600.0:
        raise CommissioningError(
            "Recording duration must be between 0.1 and 3600 seconds"
        )
    if not 1.0 <= sample_rate <= 100.0:
        raise CommissioningError("Recording sample rate must be between 1 and 100 Hz")
    if not bool(getattr(arm, "connected", False)):
        raise CommissioningError("xArm is not connected")
    reported_axis = getattr(arm, "axis", None)
    if reported_axis is not None and int(reported_axis) != 6:
        raise CommissioningError("Read-only recorder requires a Lite6")

    def read_state() -> Mapping[str, Any]:
        error_warning = _sdk_payload(
            arm.get_err_warn_code(),
            "get_err_warn_code",
        )
        if (
            not isinstance(error_warning, (list, tuple))
            or len(error_warning) < 2
            or int(error_warning[0]) != 0
            or int(error_warning[1]) != 0
        ):
            raise CommissioningError("xArm controller reports an error or warning")
        joints = _numeric_vector(
            _sdk_payload(
                arm.get_servo_angle(is_radian=True),
                "get_servo_angle",
            ),
            length=6,
            name="joint angles",
        )
        tcp_pose = _numeric_vector(
            _sdk_payload(
                arm.get_position(is_radian=True),
                "get_position",
            ),
            length=6,
            name="TCP pose",
        )
        return {"joints": joints, "tcp_pose": tcp_pose}

    try:
        demonstration = DemonstrationRecorder(
            read_state,
            dry_run=False,
        ).record_live(
            duration=duration,
            sample_rate_hz=sample_rate,
            allow_hardware=True,
            name=str(name),
            metadata={
                "source": "xarm-read-only-teach-capture",
                "units": {
                    "joints": "radian",
                    "tcp_position": "millimetre",
                    "tcp_orientation": "radian",
                    "time": "second",
                },
            },
        )
        destination = DemonstrationStore().save(
            output_path,
            demonstration,
        )
    except (TypeError, ValueError, TrajectoryError) as exc:
        if isinstance(exc, CommissioningError):
            raise
        raise CommissioningError(
            "Read-only xArm demonstration capture failed: {}".format(exc)
        ) from exc
    return {
        "artifact": str(destination),
        "sha256": _sha256_file(destination),
        "frames": len(demonstration.frames),
        "duration_seconds": demonstration.duration,
        "joint_count": demonstration.joint_count,
    }


def solve_eye_in_hand_bundle(
    bundle_path: Path,
    output_path: Path,
    *,
    hand_eye_solver: Optional[Callable[[Sequence[HandEyeSample]], Any]] = None,
) -> Mapping[str, Any]:
    """Solve and save one quality-gated eye-in-hand calibration bundle."""

    payload = _read_json_object(bundle_path, maximum_bytes=8 * 1024 * 1024)
    if payload.get("schema") != "labcobot.hand-eye-samples.v1":
        raise CommissioningError("Unsupported hand-eye sample bundle schema")
    raw_samples = payload.get("samples")
    if not isinstance(raw_samples, list):
        raise CommissioningError("Hand-eye bundle samples must be a list")
    if len(raw_samples) > 1_000:
        raise CommissioningError("Hand-eye bundle contains too many samples")
    try:
        calibrator = EyeInHandCalibrator(
            CameraIntrinsics.from_dict(_mapping(payload, "intrinsics")),
            camera_source_id=str(payload["camera_source_id"]),
            hand_eye_solver=hand_eye_solver,
        )
        for raw_sample in raw_samples:
            sample = _require_mapping(raw_sample, "hand-eye sample")
            calibrator.add_sample(
                HandEyeSample(
                    base_T_tool=RigidTransform.from_array(sample["base_T_tool"]),
                    camera_T_target=RigidTransform.from_array(
                        sample["camera_T_target"]
                    ),
                    image_captured_at_s=float(sample["image_captured_at_s"]),
                    tool_pose_at_s=float(sample["tool_pose_at_s"]),
                )
            )
        calibration = calibrator.solve()
        destination = EyeInHandCalibrationStore().save(
            output_path,
            calibration,
        )
    except (KeyError, TypeError, ValueError, CalibrationError) as exc:
        raise CommissioningError(
            "Hand-eye sample bundle was rejected: {}".format(exc)
        ) from exc
    return {
        "artifact": str(destination),
        "sha256": _sha256_file(destination),
        "camera_source_id": calibration.camera_source_id,
        "quality": calibration.quality.to_dict(),
    }


def build_valve_primitive_artifact(
    demonstration_path: Path,
    target_path: Path,
    output_path: Path,
) -> Mapping[str, Any]:
    """Convert a reviewed six-axis demonstration into a relative primitive."""

    demonstration_payload = _read_json_object(
        demonstration_path,
        maximum_bytes=16 * 1024 * 1024,
    )
    target_payload = _read_json_object(
        target_path,
        maximum_bytes=1024 * 1024,
    )
    if target_payload.get("schema") != "labcobot.valve-target.v1":
        raise CommissioningError("Unsupported valve target schema")
    try:
        demonstration = Demonstration.from_dict(demonstration_payload)
        if demonstration.joint_count != 6:
            raise CommissioningError(
                "Live Lite6 demonstrations must contain six joints"
            )
        if any(
            frame.tcp_pose is None or len(frame.tcp_pose) < 6
            for frame in demonstration.frames
        ):
            raise CommissioningError(
                "Every demonstration frame needs a six-value TCP pose"
            )
        if any(frame.gripper is not None for frame in demonstration.frames):
            raise CommissioningError(
                "Gripper demonstrations are not supported by the live adapter"
            )
        target = ValveTarget(
            target_id=str(target_payload["target_id"]),
            center_xyz=target_payload["center_xyz"],
            normal_xyz=target_payload["normal_xyz"],
            confidence=float(target_payload.get("confidence", 1.0)),
            timestamp=float(target_payload.get("timestamp", 0.0)),
            coordinate_frame=str(target_payload.get("coordinate_frame", "robot")),
            orientation_radians=target_payload.get("orientation_radians"),
        )
        if target.coordinate_frame != "robot":
            raise CommissioningError(
                "Primitive source target must use the robot coordinate frame"
            )
        primitive = ValveMotionPrimitive.from_demonstration(
            demonstration,
            target,
        )
        destination = ValveMotionPrimitiveStore().save(
            output_path,
            primitive,
        )
    except (KeyError, TypeError, ValueError, TrajectoryError) as exc:
        if isinstance(exc, CommissioningError):
            raise
        raise CommissioningError(
            "Valve demonstration or target was rejected: {}".format(exc)
        ) from exc
    return {
        "artifact": str(destination),
        "sha256": _sha256_file(destination),
        "frames": len(primitive.frames),
        "duration_seconds": (
            primitive.frames[-1].timestamp - primitive.frames[0].timestamp
        ),
        "source_target_id": primitive.source_target_id,
    }


def deployment_preflight(
    environment: Mapping[str, str],
    *,
    install_root: Path,
    require_live_motion: bool = True,
) -> Mapping[str, Any]:
    """Return secret-free evidence for each configured deployment requirement."""

    try:
        settings = IntegrationSettings.from_environment(
            environment,
            install_root=install_root,
        )
    except ValueError as exc:
        return {
            "ready": False,
            "mode": "live" if require_live_motion else "dry-run",
            "checks": [
                {
                    "id": "configuration",
                    "ok": False,
                    "detail": str(exc),
                }
            ],
        }

    checks: List[Dict[str, Any]] = []
    deployment_mode = settings.deployment_mode

    def add(identifier: str, ok: bool, detail: str) -> None:
        checks.append({"id": identifier, "ok": bool(ok), "detail": detail})

    cloud_required = deployment_mode == "production"
    if cloud_required or settings.registry_configured:
        registered = bool(
            settings.device_uid and settings.device_secret and settings.manufacturer_id
        )
        add(
            "registered_device",
            registered,
            (
                "registered stand identity is provisioned"
                if registered
                else "device UID, secret, or manufacturer ID is missing"
            ),
        )
        add(
            "device_registry_probe",
            settings.registry_configured,
            (
                "authenticated non-mutating registry verification configured"
                if settings.registry_configured
                else "Supraworx device readiness URL is missing"
            ),
        )
    for identifier, required, configured, ready_detail, missing_detail in (
        (
            "datahub",
            deployment_mode in {"production", "workcell"},
            settings.datahub_configured,
            "device-authenticated time-series ingestion configured",
            "DataHub ingestion URL is missing",
        ),
        (
            "signed_ota",
            cloud_required,
            settings.ota_configured,
            "signed OTA manifest configured",
            "signed OTA manifest is not configured",
        ),
        (
            "compute_offload",
            cloud_required,
            settings.offload_configured,
            "correlated ShortcutJob submit/status endpoints configured",
            "compute offload endpoints are missing",
        ),
    ):
        if required or configured:
            add(
                identifier,
                configured,
                ready_detail if configured else missing_detail,
            )
    workspace_camera_configured = bool(
        settings.workspace_camera_device_id
        if settings.workspace_camera_backend == "depthai"
        else settings.workspace_camera_source
    )
    tool_camera_configured = bool(
        settings.tool_camera_device_id
        if settings.tool_camera_backend == "depthai"
        else settings.tool_camera_source
    )
    add(
        "dual_cameras",
        bool(
            settings.cameras_enabled
            and tool_camera_configured
            and workspace_camera_configured
        ),
        (
            "tool and static workspace cameras configured"
            if (
                settings.cameras_enabled
                and tool_camera_configured
                and workspace_camera_configured
            )
            else "dual camera configuration is incomplete"
        ),
    )
    add(
        "video_streams",
        settings.mjpeg_enabled,
        (
            "tool and workspace MJPEG streams enabled"
            if settings.mjpeg_enabled
            else "MJPEG streams are disabled"
        ),
    )
    add(
        "xarm_telemetry",
        settings.xarm_metrics_enabled,
        (
            "xArm telemetry and synchronized pose sampling configured"
            if settings.xarm_metrics_enabled
            else "xArm telemetry is disabled"
        ),
    )
    valve_mode = deployment_mode in {"operator", "production"}
    if valve_mode or settings.eye_in_hand_enabled:
        add(
            "eye_in_hand",
            settings.eye_in_hand_enabled,
            (
                "moving-camera workspace mapping configured"
                if settings.eye_in_hand_enabled
                else "eye-in-hand mapping is disabled"
            ),
        )
        calibration_ok, calibration_detail = _check_calibration(settings)
        add("calibration_artifact", calibration_ok, calibration_detail)
    if valve_mode or settings.dynamic_valve_enabled:
        add(
            "dynamic_valves",
            settings.dynamic_valve_enabled,
            (
                "dynamic valve detection/imitation configured"
                if settings.dynamic_valve_enabled
                else "dynamic valve controller is disabled"
            ),
        )
        primitive_ok, primitive_detail = _check_primitive(settings)
        add("valve_primitive", primitive_ok, primitive_detail)
    if valve_mode:
        safety_ok, safety_detail = _check_safety_profile(settings)
        add("safety_profile", safety_ok, safety_detail)
    if settings.local_valve_model_configured:
        model_ok, model_detail = _check_local_valve_model(settings)
        add("local_valve_model", model_ok, model_detail)
    elif valve_mode:
        add(
            "valve_confirmation_owner",
            settings.offload_configured,
            (
                "reviewed remote valve-confirmation owner configured"
                if settings.offload_configured
                else "reviewed valve-confirmation owner is missing"
            ),
        )

    if require_live_motion and deployment_mode == "gateway":
        add(
            "live_motion",
            False,
            "gateway deployment is intentionally read-only",
        )
    elif require_live_motion:
        motor_enable_configured = bool(
            settings.xarm_motion_enabled or settings.capture_run_enabled
        )
        add(
            "live_motion",
            motor_enable_configured and settings.xarm_model == "lite6",
            (
                "explicit Lite6 motor-enable path configured"
                if motor_enable_configured and settings.xarm_model == "lite6"
                else "live Lite6 motion is not enabled"
            ),
        )
        if motor_enable_configured and deployment_mode in {
            "operator",
            "production",
            "workcell",
        }:
            add(
                "motion_release_incomplete",
                False,
                "signed runtime motion-release admission is not integrated",
            )

    requirements_path = install_root / "requirements" / "legacy.txt"
    pinned_sdk = False
    try:
        pinned_sdk = "xarm-python-sdk==1.18.4" in requirements_path.read_text(
            encoding="utf-8"
        )
    except OSError:
        pass
    add(
        "pinned_xarm_sdk",
        pinned_sdk,
        (
            "xArm SDK is pinned"
            if pinned_sdk
            else "pinned xArm SDK requirement is missing"
        ),
    )

    return {
        "ready": all(check["ok"] for check in checks),
        "mode": "live" if require_live_motion else "dry-run",
        "deployment_mode": deployment_mode,
        "checks": checks,
    }


def verify_cloud_registration(
    environment: Mapping[str, str],
    *,
    install_root: Path,
    readiness_client: Any = None,
) -> Mapping[str, Any]:
    """Verify the stand and required cloud features without creating records."""

    try:
        settings = IntegrationSettings.from_environment(
            environment,
            install_root=install_root,
        )
        client = readiness_client or settings.build_readiness_client()
    except ValueError as exc:
        raise CommissioningError(
            "Cloud verification configuration is invalid: {}".format(exc)
        ) from exc
    if client is None:
        raise CommissioningError("LABCOBOT_DEVICE_READINESS_URL is required")
    try:
        result = dict(client.verify())
    except DeviceReadinessError as exc:
        raise CommissioningError(str(exc)) from exc
    features = result.get("features") or {}
    if not isinstance(features, Mapping):
        raise CommissioningError("Supraworx readiness features are invalid")
    datahub_value = features.get("datahub_timeseries")
    ota_value = features.get("ota")
    compute_value = features.get("compute_offload")
    motion_release_value = features.get("motion_release")
    datahub = datahub_value if isinstance(datahub_value, Mapping) else {}
    ota = ota_value if isinstance(ota_value, Mapping) else {}
    compute = compute_value if isinstance(compute_value, Mapping) else {}
    motion_release = (
        motion_release_value if isinstance(motion_release_value, Mapping) else {}
    )
    raw_job_types = compute.get("job_types")
    job_types = {
        str(value)
        for value in (raw_job_types if isinstance(raw_job_types, (list, tuple)) else ())
    }
    checks = [
        {
            "id": "registered_device",
            "ok": bool(
                (result.get("device") or {}).get("registered")
                and result.get("verified")
            ),
            "detail": "Supraworx authenticated the configured stand",
        },
        {
            "id": "datahub_timeseries",
            "ok": datahub.get("ready") is True,
            "detail": "DataHub time-series endpoint is provisioned",
        },
        {
            "id": "signed_ota",
            "ok": ota.get("ready") is True,
            "detail": "signed OTA release metadata is provisioned",
        },
        {
            "id": "trajectory_offload",
            "ok": (compute.get("ready") is True and "trajectory_training" in job_types),
            "detail": "trajectory training ShortcutJob is allowlisted",
        },
        {
            "id": "valve_detection_offload",
            "ok": (compute.get("ready") is True and "valve_detection" in job_types),
            "detail": "valve detection ShortcutJob is allowlisted",
        },
    ]
    if settings.xarm_motion_enabled or settings.capture_run_enabled:
        checks.append(
            {
                "id": "device_manager_motion_release",
                "ok": bool(
                    motion_release.get("schema") == "labcobot.device-motion-release.v1"
                    and motion_release.get("ready") is True
                    and motion_release.get("revoked") is not True
                ),
                "detail": (
                    "professional acceptance and customer handover are recorded "
                    "in Device Manager"
                ),
            }
        )
    return {
        "ready": all(check["ok"] for check in checks),
        "device": dict(result.get("device") or {}),
        "server_time": str(result.get("server_time") or ""),
        "checks": checks,
    }


def collect_deployment_evidence(
    environment: Mapping[str, str],
    *,
    install_root: Path,
    unit_path: Path = Path("/etc/systemd/system/labcobot.service"),
    enabled_link: Path = Path(
        "/etc/systemd/system/graphical.target.wants/labcobot.service"
    ),
    health_url: str = "http://127.0.0.1:8765/healthz",
    readiness_client: Any = None,
    health_transport: Any = None,
) -> Mapping[str, Any]:
    """Collect secret-free proof from a configured, running Raspberry Pi."""

    preflight = deployment_preflight(
        environment,
        install_root=install_root,
        require_live_motion=True,
    )
    try:
        settings = IntegrationSettings.from_environment(
            environment,
            install_root=install_root,
        )
    except ValueError:
        settings = None
    cloud_required = bool(
        settings is not None
        and (
            settings.deployment_mode == "production"
            or settings.registry_configured
            or settings.datahub_configured
            or settings.ota_configured
            or settings.offload_configured
        )
    )
    if cloud_required:
        try:
            cloud = verify_cloud_registration(
                environment,
                install_root=install_root,
                readiness_client=readiness_client,
            )
        except CommissioningError as exc:
            cloud = {
                "ready": False,
                "error": str(exc),
                "checks": [],
            }
    else:
        cloud = {
            "ready": True,
            "status": "not_configured",
            "checks": [],
        }

    unit_ok = False
    try:
        if (
            unit_path.is_file()
            and not unit_path.is_symlink()
            and unit_path.stat().st_size <= 1024 * 1024
        ):
            unit_text = unit_path.read_text(encoding="utf-8")
            unit_ok = str(install_root / "main" / "runtime.py") in unit_text
    except (OSError, UnicodeError):
        unit_ok = False

    enabled_ok = False
    try:
        enabled_ok = enabled_link.is_symlink() and enabled_link.resolve(
            strict=True
        ) == unit_path.resolve(strict=True)
    except OSError:
        enabled_ok = False

    parsed_health = urlsplit(str(health_url))
    if (
        parsed_health.scheme not in {"http", "https"}
        or parsed_health.hostname not in {"127.0.0.1", "::1", "localhost"}
        or parsed_health.username is not None
        or parsed_health.password is not None
    ):
        raise CommissioningError(
            "Health URL must be an uncredentialed loopback HTTP(S) URL"
        )
    transport = health_transport or UrllibTransport()
    health: Mapping[str, Any]
    health_status = 0
    try:
        response = transport.request(
            "GET",
            str(health_url),
            headers={
                "Accept": "application/json",
                "Cache-Control": "no-cache",
                "User-Agent": "LabCobot-Commissioning/1",
            },
            timeout=10.0,
            max_response_bytes=512 * 1024,
        )
        health_status = int(response.status_code)
        value = response.json()
        if not isinstance(value, Mapping):
            raise TransportError("Health response must be an object")
        health = dict(value)
    except (TransportError, OSError) as exc:
        health = {
            "ready": False,
            "status": "unavailable",
            "reason": type(exc).__name__,
        }
    health_ok = health_status == 200 and health.get("ready") is True

    checks = [
        {
            "id": "configuration_preflight",
            "ok": preflight.get("ready") is True,
            "detail": "live configuration and approved artifacts",
        },
        {
            "id": "systemd_unit",
            "ok": unit_ok,
            "detail": "unit targets this installed runtime",
        },
        {
            "id": "systemd_autostart",
            "ok": enabled_ok,
            "detail": "graphical target enablement link",
        },
        {
            "id": "runtime_health",
            "ok": health_ok,
            "detail": "HTTP {} from local workcell health".format(
                health_status or "unavailable"
            ),
        },
    ]
    if cloud_required:
        checks.insert(
            1,
            {
                "id": "cloud_readiness",
                "ok": cloud.get("ready") is True,
                "detail": "registered stand and configured cloud features",
            },
        )
    return {
        "schema_version": "labcobot.deployment-evidence.v1",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "install_root": str(install_root),
        "ready": all(check["ok"] for check in checks),
        "checks": checks,
        "preflight": preflight,
        "cloud": cloud,
        "runtime_health": dict(health),
    }


def load_environment_file(path: Path) -> Mapping[str, str]:
    """Parse the assignment-only subset used by the systemd environment file."""

    if path.is_symlink() or not path.is_file():
        raise CommissioningError("Environment path must be a regular, non-symlink file")
    if path.stat().st_size > 1024 * 1024:
        raise CommissioningError("Environment file exceeds 1 MiB")
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except (OSError, UnicodeError) as exc:
        raise CommissioningError("Environment file could not be read as UTF-8") from exc
    result: Dict[str, str] = {}
    for line_number, raw_line in enumerate(lines, start=1):
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if "=" not in line:
            raise CommissioningError(
                "Environment line {} is not an assignment".format(line_number)
            )
        name, value = line.split("=", 1)
        name = name.strip()
        value = value.strip()
        if (
            not name
            or not (name[0].isalpha() or name[0] == "_")
            or not all(character.isalnum() or character == "_" for character in name)
        ):
            raise CommissioningError(
                "Environment line {} has an invalid name".format(line_number)
            )
        if len(value) >= 2 and value[0] == value[-1] and value[0] in "'\"":
            value = value[1:-1]
        if "\x00" in value:
            raise CommissioningError(
                "Environment line {} contains a NUL byte".format(line_number)
            )
        result[name] = value
    return result


def _check_calibration(settings: IntegrationSettings) -> Tuple[bool, str]:
    if not settings.eye_in_hand_calibration_path:
        return False, "calibration artifact path is missing"
    if not settings.eye_in_hand_calibration_sha256:
        return False, "calibration artifact SHA-256 is missing"
    try:
        calibration = EyeInHandCalibrationStore().load(
            settings.eye_in_hand_calibration_path,
            expected_sha256=settings.eye_in_hand_calibration_sha256,
        )
    except Exception as exc:
        return False, "calibration rejected: {}".format(type(exc).__name__)
    if calibration.camera_source_id != settings.tool_camera_source_id:
        return False, "calibration camera identity does not match configuration"
    if (
        calibration.intrinsics.width != settings.camera_width
        or calibration.intrinsics.height != settings.camera_height
    ):
        return False, "calibration resolution does not match configuration"
    return True, "approved calibration artifact matches camera and resolution"


def _check_primitive(settings: IntegrationSettings) -> Tuple[bool, str]:
    if not settings.dynamic_valve_primitive_path:
        return False, "valve primitive artifact path is missing"
    if not settings.dynamic_valve_primitive_sha256:
        return False, "valve primitive artifact SHA-256 is missing"
    try:
        primitive = ValveMotionPrimitiveStore().load(
            settings.dynamic_valve_primitive_path,
            expected_sha256=settings.dynamic_valve_primitive_sha256,
        )
    except Exception as exc:
        return False, "valve primitive rejected: {}".format(type(exc).__name__)
    if any(frame.gripper is not None for frame in primitive.frames):
        return False, "valve primitive needs an unavailable gripper adapter"
    return True, "approved valve primitive artifact is valid"


def _check_safety_profile(settings: IntegrationSettings) -> Tuple[bool, str]:
    if not settings.safety_profile_path or not settings.safety_profile_sha256:
        return False, "Lite6 safety profile path or SHA-256 is missing"
    try:
        profile = load_lite6_safety_profile(
            settings.safety_profile_path,
            expected_sha256=settings.safety_profile_sha256,
        )
    except Exception as exc:
        return False, "safety profile rejected: {}".format(type(exc).__name__)
    if (
        profile.eye_in_hand_calibration_sha256
        != settings.eye_in_hand_calibration_sha256
        or profile.motion_primitive_sha256 != settings.dynamic_valve_primitive_sha256
    ):
        return False, "safety profile artifact bindings do not match configuration"
    return True, "root-owned Lite6 safety profile and bindings are valid"


def _check_local_valve_model(
    settings: IntegrationSettings,
) -> Tuple[bool, str]:
    try:
        load_reviewed_valve_model(
            model_path=settings.dynamic_valve_local_model_path,
            expected_model_sha256=settings.dynamic_valve_local_model_sha256,
            manifest_path=settings.dynamic_valve_local_model_manifest_path,
            expected_manifest_sha256=(
                settings.dynamic_valve_local_model_manifest_sha256
            ),
            tool_camera_source_id=settings.tool_camera_source_id,
            camera_width=settings.camera_width,
            camera_height=settings.camera_height,
            eye_in_hand_calibration_sha256=(settings.eye_in_hand_calibration_sha256),
            confidence_threshold=settings.dynamic_valve_min_confidence,
        )
    except Exception as exc:
        return False, "reviewed local valve model rejected: {}".format(
            type(exc).__name__
        )
    return True, "reviewed local ONNX model and manifest are valid"


def _read_json_object(path: Path, *, maximum_bytes: int) -> Mapping[str, Any]:
    if path.is_symlink() or not path.is_file():
        raise CommissioningError(
            "Input path must be a regular, non-symlink file: {}".format(path)
        )
    raw_payload = path.read_bytes()
    if not raw_payload or len(raw_payload) > maximum_bytes:
        raise CommissioningError("Input JSON file has an invalid size")
    try:
        payload = json.loads(raw_payload.decode("utf-8"))
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise CommissioningError("Input file is not valid JSON") from exc
    return _require_mapping(payload, "input JSON")


def _require_mapping(value: Any, name: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise CommissioningError("{} must be a JSON object".format(name))
    return value


def _mapping(value: Mapping[str, Any], name: str) -> Mapping[str, Any]:
    try:
        return _require_mapping(value[name], name)
    except KeyError as exc:
        raise CommissioningError("{} is missing".format(name)) from exc


def _sha256_file(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _sdk_payload(result: Any, operation: str) -> Any:
    if (
        not isinstance(result, (tuple, list))
        or len(result) != 2
        or isinstance(result[0], bool)
        or not isinstance(result[0], int)
    ):
        raise CommissioningError(
            "xArm {} returned an invalid response".format(operation)
        )
    if result[0] != 0:
        raise CommissioningError(
            "xArm {} returned code {}".format(operation, result[0])
        )
    return result[1]


def _numeric_vector(value: Any, *, length: int, name: str) -> List[float]:
    try:
        result = [float(item) for item in value]
    except (TypeError, ValueError) as exc:
        raise CommissioningError("{} must be numeric".format(name)) from exc
    if len(result) != length or not all(
        item == item and abs(item) != float("inf") for item in result
    ):
        raise CommissioningError(
            "{} must contain {} finite values".format(name, length)
        )
    return result


def _write_result(payload: Mapping[str, Any]) -> None:
    print(json.dumps(payload, allow_nan=False, indent=2, sort_keys=True))


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="LabCobot non-motion commissioning tools"
    )
    commands = parser.add_subparsers(dest="command", required=True)

    registration = commands.add_parser(
        "device-registration-request",
        help="read the Pi identity and emit a Supraworx registration request",
    )
    registration.add_argument("--interface", default="eth0")
    registration.add_argument("--manufacturer-id", required=True)
    registration.add_argument("--location", required=True)
    registration.add_argument("--uid-device")
    registration.add_argument("--name", default="AINA LabCobot")
    registration.add_argument("--model", default="LabCobot AINA")
    registration.add_argument("--hardware-revision")
    registration.add_argument("--serial-number")
    registration.add_argument(
        "--mac-address",
        help="explicit override for bench testing; omit on the production Pi",
    )

    recording = commands.add_parser(
        "record-demonstration",
        help="capture read-only Lite6 joints/TCP while using controller teach mode",
    )
    recording.add_argument("--xarm-ip", required=True)
    recording.add_argument("--duration", type=float, required=True)
    recording.add_argument("--sample-rate", type=float, default=25.0)
    recording.add_argument("--name", default="valve-demonstration")
    recording.add_argument("--output", type=Path, required=True)
    recording.add_argument(
        "--confirm-read-only-capture",
        action="store_true",
        help="confirm that teach mode and physical safety are operator-managed",
    )

    calibration = commands.add_parser(
        "solve-calibration",
        help="solve a versioned eye-in-hand sample bundle",
    )
    calibration.add_argument("--bundle", type=Path, required=True)
    calibration.add_argument("--output", type=Path, required=True)

    primitive = commands.add_parser(
        "build-primitive",
        help="build a valve-relative primitive from a demonstration",
    )
    primitive.add_argument("--demonstration", type=Path, required=True)
    primitive.add_argument("--target", type=Path, required=True)
    primitive.add_argument("--output", type=Path, required=True)

    preflight = commands.add_parser(
        "preflight",
        help="check a Pi environment file without contacting hardware",
    )
    preflight.add_argument("--env-file", type=Path, required=True)
    preflight.add_argument(
        "--install-root",
        type=Path,
        default=Path(__file__).resolve().parent.parent,
    )
    preflight.add_argument(
        "--mode",
        choices=("dry-run", "live"),
        default="live",
    )
    cloud = commands.add_parser(
        "verify-cloud",
        help="authenticate the registered stand and verify cloud provisioning",
    )
    cloud.add_argument("--env-file", type=Path, required=True)
    cloud.add_argument(
        "--install-root",
        type=Path,
        default=Path(__file__).resolve().parent.parent,
    )
    evidence = commands.add_parser(
        "deployment-evidence",
        help="prove configuration, cloud identity, autostart, and live health",
    )
    evidence.add_argument("--env-file", type=Path, required=True)
    evidence.add_argument(
        "--install-root",
        type=Path,
        default=Path(__file__).resolve().parent.parent,
    )
    evidence.add_argument(
        "--unit-path",
        type=Path,
        default=Path("/etc/systemd/system/labcobot.service"),
    )
    evidence.add_argument(
        "--enabled-link",
        type=Path,
        default=Path("/etc/systemd/system/graphical.target.wants/labcobot.service"),
    )
    evidence.add_argument(
        "--health-url",
        default="http://127.0.0.1:8765/healthz",
    )
    return parser


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = _parser().parse_args(argv)
    try:
        if args.command == "device-registration-request":
            result = build_device_registration_request(
                manufacturer_id=args.manufacturer_id,
                location=args.location,
                interface=args.interface,
                mac_address=args.mac_address,
                uid_device=args.uid_device,
                name=args.name,
                model=args.model,
                hardware_revision=args.hardware_revision,
                serial_number=args.serial_number,
            )
        elif args.command == "record-demonstration":
            if not args.confirm_read_only_capture:
                raise CommissioningError("--confirm-read-only-capture is required")
            try:
                source = open_xarm_metric_source(args.xarm_ip)
            except RuntimeError as exc:
                raise CommissioningError(
                    "xArm SDK is required for demonstration capture"
                ) from exc
            try:
                result = record_xarm_demonstration(
                    source.arm,
                    args.output,
                    duration_seconds=args.duration,
                    sample_rate_hz=args.sample_rate,
                    name=args.name,
                )
            finally:
                source.close()
        elif args.command == "solve-calibration":
            result = solve_eye_in_hand_bundle(args.bundle, args.output)
        elif args.command == "build-primitive":
            result = build_valve_primitive_artifact(
                args.demonstration,
                args.target,
                args.output,
            )
        elif args.command == "preflight":
            environment = load_environment_file(args.env_file)
            result = deployment_preflight(
                environment,
                install_root=args.install_root,
                require_live_motion=args.mode == "live",
            )
            _write_result(result)
            return 0 if result["ready"] else 2
        elif args.command == "verify-cloud":
            environment = load_environment_file(args.env_file)
            result = verify_cloud_registration(
                environment,
                install_root=args.install_root,
            )
            _write_result(result)
            return 0 if result["ready"] else 2
        else:
            environment = load_environment_file(args.env_file)
            result = collect_deployment_evidence(
                environment,
                install_root=args.install_root,
                unit_path=args.unit_path,
                enabled_link=args.enabled_link,
                health_url=args.health_url,
            )
            _write_result(result)
            return 0 if result["ready"] else 2
    except (CommissioningError, CalibrationError, TrajectoryError) as exc:
        print(
            json.dumps(
                {"ok": False, "error": str(exc)},
                allow_nan=False,
                sort_keys=True,
            ),
            file=sys.stderr,
        )
        return 2
    _write_result({"ok": True, **dict(result)})
    return 0


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