"""Read-only dual-camera hand-training capture with typed tool labels.

The recorder never enables motors, changes xArm mode, or emits a motion
command.  It only starts while the already connected Lite6 reports controller
teach mode, records synchronized camera pairs plus bounded telemetry, and
optionally invokes the existing typed sampling-tool capability.  A completed
session is uploaded through one injected API-gateway client.
"""

from __future__ import annotations

import hashlib
import json
import math
import os
import re
import stat
import threading
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, Mapping, Optional, Protocol, Tuple


HAND_TRAINING_SCHEMA = "labcobot.hand-training-recording.v1"
HAND_TRAINING_STATUS_SCHEMA_VERSION = 1
HAND_TRAINING_COORDINATE_FRAME = "xarm_base"
HAND_TRAINING_REQUIRED_MODE = 2
HAND_TRAINING_ROLES = ("tool", "workspace")
HAND_TRAINING_VIDEO_CONTENT_TYPE = "video/x-motion-jpeg"
DEFAULT_SAMPLE_RATE_HZ = 5.0
DEFAULT_MAX_DURATION_SECONDS = 600.0
DEFAULT_MAX_PAIR_SKEW_SECONDS = 0.15
DEFAULT_MAX_FRAME_AGE_SECONDS = 0.5
DEFAULT_PAIR_TIMEOUT_SECONDS = 0.75
DEFAULT_MAX_VIDEO_BYTES = 256 * 1024 * 1024
MAX_MANIFEST_BYTES = 16 * 1024 * 1024

_SESSION_ID = re.compile(r"^[0-9a-f]{32}$")
_SOURCE_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,255}$")
_FINAL_STATES = frozenset({"completed", "failed", "cancelled"})
_ACTIVE_STATES = frozenset({"starting", "recording", "stopping", "uploading"})
_PUBLIC_STATES = frozenset({"idle"}) | _ACTIVE_STATES | _FINAL_STATES


class HandTrainingError(RuntimeError):
    """One bounded, machine-readable recording failure."""

    def __init__(self, code: str, message: str = "") -> None:
        normalized = re.sub(r"[^a-z0-9_]+", "_", str(code).lower()).strip("_")
        self.code = (normalized or "hand_training_failed")[:64]
        self.safe_message = _safe_text(message, 160)
        super().__init__(self.safe_message or self.code)


class HandTrainingBusy(HandTrainingError):
    """Raised when another recording owns the controller."""


class HandTrainingUploader(Protocol):
    """Upload one immutable local recording to the configured API gateway."""

    def upload(
        self,
        *,
        manifest_path: Path,
        video_paths: Mapping[str, Path],
        idempotency_key: str,
    ) -> Mapping[str, Any]:
        """Return a bounded server receipt or raise on delivery failure."""


def _safe_text(value: Any, limit: int) -> str:
    text = str(value or "")
    return "".join(
        character if 32 <= ord(character) <= 126 else "?"
        for character in text[: max(0, int(limit))]
    )


def _finite(value: Any, name: str) -> float:
    if isinstance(value, bool):
        raise HandTrainingError("telemetry_invalid", "{} is invalid".format(name))
    try:
        result = float(value)
    except (TypeError, ValueError, OverflowError) as exc:
        raise HandTrainingError(
            "telemetry_invalid", "{} is invalid".format(name)
        ) from exc
    if not math.isfinite(result):
        raise HandTrainingError("telemetry_invalid", "{} is invalid".format(name))
    return result


def _positive(value: Any, name: str, maximum: float) -> float:
    result = _finite(value, name)
    if not 0.0 < result <= maximum:
        raise HandTrainingError("configuration_invalid", "{} is invalid".format(name))
    return result


def _utc_timestamp(value: float) -> str:
    return (
        datetime.fromtimestamp(_finite(value, "wall clock"), tz=timezone.utc)
        .isoformat(timespec="milliseconds")
        .replace("+00:00", "Z")
    )


def _json_safe(value: Any, *, depth: int = 0) -> Any:
    """Return bounded JSON data without accepting NaN, bytes, or objects."""

    if depth > 8:
        raise HandTrainingError("telemetry_invalid", "Telemetry nesting is too deep")
    if value is None or isinstance(value, (str, bool, int)):
        return value
    if isinstance(value, float):
        if not math.isfinite(value):
            raise HandTrainingError("telemetry_invalid", "Telemetry is not finite")
        return value
    if isinstance(value, Mapping):
        if len(value) > 512:
            raise HandTrainingError("telemetry_invalid", "Telemetry has too many keys")
        return {
            _safe_text(key, 120): _json_safe(item, depth=depth + 1)
            for key, item in value.items()
        }
    if isinstance(value, (tuple, list)):
        if len(value) > 4096:
            raise HandTrainingError("telemetry_invalid", "Telemetry list is too long")
        return [_json_safe(item, depth=depth + 1) for item in value]
    raise HandTrainingError("telemetry_invalid", "Telemetry contains an object")


def _directory_flags() -> int:
    return (
        os.O_RDONLY
        | getattr(os, "O_DIRECTORY", 0)
        | getattr(os, "O_CLOEXEC", 0)
        | getattr(os, "O_NOFOLLOW", 0)
    )


def _create_flags() -> int:
    return (
        os.O_WRONLY
        | os.O_CREAT
        | os.O_EXCL
        | getattr(os, "O_CLOEXEC", 0)
        | getattr(os, "O_NOFOLLOW", 0)
    )


def _write_all(descriptor: int, payload: bytes) -> None:
    view = memoryview(payload)
    written = 0
    while written < len(view):
        count = os.write(descriptor, view[written:])
        if count <= 0:
            raise OSError("short hand-training write")
        written += count


class HandTrainingRecordingController:
    """Record two MJPEG tracks and all read-only robot observations.

    ``pair_provider`` follows ``DualCameraRig.capture_pair_for_run``.  The
    injected ``tool_control`` is the capability object issued by the existing
    commissioned sampling-tool adapter; raw GPIO is intentionally impossible.
    """

    def __init__(
        self,
        *,
        root: Path,
        pair_provider: Callable[..., Mapping[Any, Any]],
        jpeg_encoder: Callable[[Any], bytes],
        telemetry_provider: Callable[[], Mapping[str, Any]],
        uploader: HandTrainingUploader,
        device_uid: str,
        tool_control: Any = None,
        sample_rate_hz: float = DEFAULT_SAMPLE_RATE_HZ,
        max_duration_seconds: float = DEFAULT_MAX_DURATION_SECONDS,
        max_pair_skew_seconds: float = DEFAULT_MAX_PAIR_SKEW_SECONDS,
        max_frame_age_seconds: float = DEFAULT_MAX_FRAME_AGE_SECONDS,
        pair_timeout_seconds: float = DEFAULT_PAIR_TIMEOUT_SECONDS,
        max_video_bytes: int = DEFAULT_MAX_VIDEO_BYTES,
        monotonic: Callable[[], float] = time.monotonic,
        wall_clock: Callable[[], float] = time.time,
    ) -> None:
        self.root = Path(root)
        if not self.root.is_absolute() or ".." in self.root.parts:
            raise HandTrainingError(
                "configuration_invalid", "Recording root must be absolute"
            )
        if not callable(pair_provider) or not callable(jpeg_encoder):
            raise HandTrainingError(
                "configuration_invalid", "Both camera seams are required"
            )
        if not callable(telemetry_provider):
            raise HandTrainingError(
                "configuration_invalid", "Telemetry provider is required"
            )
        if not callable(getattr(uploader, "upload", None)):
            raise HandTrainingError(
                "configuration_invalid", "API-gateway uploader is required"
            )
        if not isinstance(device_uid, str) or not device_uid.strip():
            raise HandTrainingError("configuration_invalid", "Device UID is required")
        self.sample_rate_hz = _positive(sample_rate_hz, "sample rate", 30.0)
        self.max_duration_seconds = _positive(
            max_duration_seconds, "maximum duration", 3600.0
        )
        self.max_pair_skew_seconds = _positive(
            max_pair_skew_seconds, "pair skew", 2.0
        )
        self.max_frame_age_seconds = _positive(
            max_frame_age_seconds, "frame age", 5.0
        )
        self.pair_timeout_seconds = _positive(
            pair_timeout_seconds, "pair timeout", 10.0
        )
        if (
            isinstance(max_video_bytes, bool)
            or not isinstance(max_video_bytes, int)
            or not 1024 * 1024 <= max_video_bytes <= 2 * 1024 * 1024 * 1024
        ):
            raise HandTrainingError(
                "configuration_invalid", "Video byte limit is invalid"
            )
        self.max_video_bytes = max_video_bytes
        self.pair_provider = pair_provider
        self.jpeg_encoder = jpeg_encoder
        self.telemetry_provider = telemetry_provider
        self.uploader = uploader
        self.device_uid = device_uid.strip()
        self.tool_control = tool_control
        self.monotonic = monotonic
        self.wall_clock = wall_clock
        self._lock = threading.RLock()
        self._tool_lock = threading.Lock()
        self._stop_event = threading.Event()
        self._thread: Optional[threading.Thread] = None
        self._state = "idle"
        self._session_id = ""
        self._sample_count = 0
        self._video_bytes = {role: 0 for role in HAND_TRAINING_ROLES}
        self._tool_on = False
        self._tool_events = 0
        self._started_monotonic = 0.0
        self._error_code = ""
        self._receipt: Dict[str, Any] = {}
        self._prepare_root()

    def bind_tool_control(self, tool_control: Any) -> None:
        """Bind the one canonical commissioned tool capability before use."""

        if tool_control is None or not all(
            callable(getattr(tool_control, name, None))
            for name in ("grip", "release", "status")
        ):
            raise HandTrainingError(
                "configuration_invalid", "Typed tool control is incomplete"
            )
        with self._lock:
            if self._thread is not None or self._state != "idle":
                raise HandTrainingError(
                    "hand_training_busy", "Tool control cannot change during a session"
                )
            if self.tool_control not in (None, tool_control):
                raise HandTrainingError(
                    "configuration_invalid", "Typed tool control is already bound"
                )
            self.tool_control = tool_control

    def _prepare_root(self) -> None:
        self.root.mkdir(mode=0o750, parents=True, exist_ok=True)
        metadata = self.root.lstat()
        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
            raise HandTrainingError("recording_path_unsafe", "Recording root is unsafe")
        descriptor = os.open(str(self.root), _directory_flags())
        try:
            os.fchmod(descriptor, 0o750)
            os.fsync(descriptor)
        finally:
            os.close(descriptor)

    def start(self) -> str:
        """Start one recording after proving teach mode and both cameras."""

        with self._lock:
            if self._thread is not None and self._thread.is_alive():
                raise HandTrainingBusy("hand_training_busy", "Recording is active")
            telemetry = self._validated_telemetry(self.telemetry_provider())
            if telemetry["moving"] is not False:
                raise HandTrainingError(
                    "teach_mode_not_ready", "Robot telemetry is not stationary"
                )
            tool = self._tool_status()
            if tool.get("grasp_confirmed") is True:
                raise HandTrainingError(
                    "tool_not_safe", "Tool must be off before recording starts"
                )
            initial_pair = self._normalized_pair(
                self.pair_provider(
                    after_sequences={},
                    timeout_seconds=self.pair_timeout_seconds,
                    max_age_seconds=self.max_frame_age_seconds,
                    max_skew_seconds=self.max_pair_skew_seconds,
                )
            )
            session_id = uuid.uuid4().hex
            session_dir = self.root / session_id
            session_dir.mkdir(mode=0o750)
            self._stop_event = threading.Event()
            self._state = "starting"
            self._session_id = session_id
            self._sample_count = 0
            self._video_bytes = {role: 0 for role in HAND_TRAINING_ROLES}
            self._tool_on = False
            self._tool_events = 0
            self._started_monotonic = float(self.monotonic())
            self._error_code = ""
            self._receipt = {}
            thread = threading.Thread(
                target=self._record,
                args=(session_id, initial_pair),
                name="hand-training-{}".format(session_id[:8]),
                daemon=True,
            )
            self._thread = thread
            try:
                thread.start()
            except Exception as exc:
                self._state = "failed"
                self._error_code = "worker_start_failed"
                raise HandTrainingError(
                    "worker_start_failed", "Recording worker could not start"
                ) from exc
            return session_id

    def set_tool(self, active: bool) -> Mapping[str, Any]:
        """Toggle only the commissioned typed tool while recording."""

        if type(active) is not bool:
            raise HandTrainingError("tool_request_invalid", "Tool state is invalid")
        with self._lock:
            if self._state != "recording" or self._stop_event.is_set():
                raise HandTrainingError(
                    "hand_training_not_recording", "Recording is not active"
                )
        control = self.tool_control
        callback = getattr(control, "grip" if active else "release", None)
        if not callable(callback):
            raise HandTrainingError("tool_unavailable", "Typed tool control is unavailable")
        if active:
            self._validated_telemetry(self.telemetry_provider())
        with self._tool_lock:
            try:
                result = callback()
            except Exception as exc:
                with self._lock:
                    self._error_code = "tool_command_failed"
                raise HandTrainingError(
                    "tool_command_failed", "Commissioned tool command failed"
                ) from exc
            if not isinstance(result, Mapping):
                raise HandTrainingError(
                    "tool_status_invalid", "Typed tool returned invalid status"
                )
            confirmed = result.get("grasp_confirmed") is True
            if confirmed is not active:
                raise HandTrainingError(
                    "tool_feedback_mismatch", "Tool feedback did not confirm the state"
                )
            with self._lock:
                self._tool_on = active
                self._tool_events += 1
            return self.status()

    def end(self) -> bool:
        """End the current recording and keep final upload asynchronous."""

        with self._lock:
            thread = self._thread
            if thread is None or not thread.is_alive():
                return False
            self._state = "stopping"
            self._stop_event.set()
        return True

    def cancel(self) -> bool:
        """Cancel locally; no incomplete recording is uploaded."""

        with self._lock:
            thread = self._thread
            if thread is None or not thread.is_alive():
                return False
            self._error_code = "cancelled"
            self._state = "stopping"
            self._stop_event.set()
        return True

    def wait(self, timeout_seconds: Optional[float] = None) -> bool:
        with self._lock:
            thread = self._thread
        if thread is None:
            return True
        thread.join(timeout_seconds)
        return not thread.is_alive()

    def status(self) -> Mapping[str, Any]:
        with self._lock:
            state = self._state if self._state in _PUBLIC_STATES else "failed"
            active = self._thread is not None and self._thread.is_alive()
            elapsed = (
                max(0.0, float(self.monotonic()) - self._started_monotonic)
                if self._started_monotonic
                else 0.0
            )
            receipt = self._receipt
            return {
                "schema_version": HAND_TRAINING_STATUS_SCHEMA_VERSION,
                "configured": True,
                "state": state,
                "active": active,
                "start_allowed": not active,
                "end_allowed": active and state in {"starting", "recording"},
                "tool_allowed": active and state == "recording" and self.tool_control is not None,
                "tool_on": self._tool_on,
                "tool_events": self._tool_events,
                "session_id": self._session_id if _SESSION_ID.fullmatch(self._session_id) else "",
                "sample_count": self._sample_count,
                "tool_frames": self._sample_count,
                "workspace_frames": self._sample_count,
                "elapsed_seconds": min(elapsed, self.max_duration_seconds),
                "maximum_duration_seconds": self.max_duration_seconds,
                "sample_rate_hz": self.sample_rate_hz,
                "video_bytes": dict(self._video_bytes),
                "error_code": self._error_code,
                "uploaded": state == "completed" and bool(receipt),
                "recording_id": _safe_text(receipt.get("recording_id"), 64),
                "workbench_rows": int(receipt.get("workbench_rows") or 0),
                "shortcut_jobs": int(receipt.get("shortcut_jobs") or 0),
            }

    def _record(self, session_id: str, initial_pair: Mapping[str, Any]) -> None:
        directory = self.root / session_id
        video_paths = {
            role: directory / "{}.mjpeg".format(role) for role in HAND_TRAINING_ROLES
        }
        descriptors: Dict[str, int] = {}
        digests = {role: hashlib.sha256() for role in HAND_TRAINING_ROLES}
        samples = []
        tool_events = []
        last_sequences: Dict[str, int] = {}
        started_wall = float(self.wall_clock())
        started_mono = float(self.monotonic())
        final_state = "failed"
        error_code = "hand_training_failed"
        try:
            for role, path in video_paths.items():
                descriptors[role] = os.open(str(path), _create_flags(), 0o640)
            with self._lock:
                self._state = "recording"
            next_sample = started_mono
            interval = 1.0 / self.sample_rate_hz
            last_tool_on = False
            pending_pair: Optional[Mapping[str, Any]] = initial_pair
            while not self._stop_event.is_set():
                now = float(self.monotonic())
                if now - started_mono >= self.max_duration_seconds:
                    self._stop_event.set()
                    break
                remaining = next_sample - now
                if remaining > 0.0:
                    self._stop_event.wait(min(remaining, 0.05))
                    continue
                if pending_pair is not None:
                    normalized = pending_pair
                    pending_pair = None
                else:
                    pair = self.pair_provider(
                        after_sequences=dict(last_sequences),
                        timeout_seconds=self.pair_timeout_seconds,
                        max_age_seconds=self.max_frame_age_seconds,
                        max_skew_seconds=self.max_pair_skew_seconds,
                    )
                    normalized = self._normalized_pair(pair)
                telemetry = self._validated_telemetry(self.telemetry_provider())
                tool_status = self._tool_status()
                tool_on = tool_status.get("grasp_confirmed") is True
                if tool_on != last_tool_on:
                    tool_events.append(
                        {
                            "sample_index": len(samples),
                            "observed_at": _utc_timestamp(float(self.wall_clock())),
                            "tool_on": tool_on,
                        }
                    )
                    last_tool_on = tool_on
                frame_records = {}
                for role in HAND_TRAINING_ROLES:
                    frame = normalized[role]
                    jpeg = self.jpeg_encoder(getattr(frame, "image", None))
                    if (
                        not isinstance(jpeg, bytes)
                        or len(jpeg) < 4
                        or not jpeg.startswith(b"\xff\xd8")
                        or not jpeg.endswith(b"\xff\xd9")
                    ):
                        raise HandTrainingError(
                            "jpeg_invalid", "Camera encoder returned invalid JPEG"
                        )
                    next_size = self._video_bytes[role] + len(jpeg)
                    if next_size > self.max_video_bytes:
                        raise HandTrainingError(
                            "video_limit_reached", "Recording reached its video limit"
                        )
                    offset = self._video_bytes[role]
                    _write_all(descriptors[role], jpeg)
                    digests[role].update(jpeg)
                    self._video_bytes[role] = next_size
                    sequence = int(getattr(frame, "sequence", 0))
                    last_sequences[role] = sequence
                    frame_records[role] = {
                        "source_id": str(getattr(frame, "source_id", "")),
                        "sequence": sequence,
                        "captured_at_s": _finite(
                            getattr(frame, "captured_at_s", None), "capture time"
                        ),
                        "byte_offset": offset,
                        "byte_length": len(jpeg),
                        "sha256": hashlib.sha256(jpeg).hexdigest(),
                    }
                sample = {
                    "index": len(samples),
                    "elapsed_seconds": max(0.0, float(self.monotonic()) - started_mono),
                    "observed_at": _utc_timestamp(float(self.wall_clock())),
                    "frames": frame_records,
                    "tool": _json_safe(tool_status),
                    "coordinates": {
                        "frame": HAND_TRAINING_COORDINATE_FRAME,
                        "joint_angles_rad": telemetry["metrics"].get("joint_angles_rad"),
                        "tcp_position_mm": telemetry["metrics"].get("tcp_position_mm"),
                        "tcp_orientation_rad": telemetry["metrics"].get("tcp_orientation_rad"),
                    },
                    "metrics": telemetry,
                }
                encoded_sample = json.dumps(
                    sample, allow_nan=False, separators=(",", ":"), sort_keys=True
                ).encode("utf-8")
                if len(encoded_sample) > 256 * 1024:
                    raise HandTrainingError(
                        "telemetry_too_large", "One telemetry sample is too large"
                    )
                samples.append(sample)
                with self._lock:
                    self._sample_count = len(samples)
                    self._tool_on = tool_on
                next_sample += interval
                if next_sample < float(self.monotonic()) - interval:
                    next_sample = float(self.monotonic()) + interval

            if not samples:
                raise HandTrainingError(
                    "recording_empty", "Recording ended before the first camera pair"
                )
            self._force_tool_off()
            final_state = "cancelled" if self._error_code == "cancelled" else "completed"
            error_code = "cancelled" if final_state == "cancelled" else ""
        except HandTrainingError as exc:
            error_code = exc.code
        except Exception as exc:
            error_code = "recording_{}".format(type(exc).__name__.lower())[:64]
        finally:
            try:
                self._force_tool_off()
            except Exception:
                final_state = "failed"
                error_code = "tool_safe_state_failed"
            for descriptor in descriptors.values():
                try:
                    os.fsync(descriptor)
                finally:
                    os.close(descriptor)

        finished_wall = float(self.wall_clock())
        manifest = {
            "schema": HAND_TRAINING_SCHEMA,
            "session_id": session_id,
            "device_uid": self.device_uid,
            "status": final_state,
            "created_at": _utc_timestamp(started_wall),
            "finished_at": _utc_timestamp(finished_wall),
            "duration_seconds": max(0.0, float(self.monotonic()) - started_mono),
            "sample_rate_hz": self.sample_rate_hz,
            "coordinate_frame": HAND_TRAINING_COORDINATE_FRAME,
            "required_controller_mode": HAND_TRAINING_REQUIRED_MODE,
            "videos": {
                role: {
                    "path": "{}.mjpeg".format(role),
                    "content_type": HAND_TRAINING_VIDEO_CONTENT_TYPE,
                    "byte_length": self._video_bytes[role],
                    "sha256": digests[role].hexdigest(),
                    "frame_count": len(samples),
                }
                for role in HAND_TRAINING_ROLES
            },
            "tool_events": tool_events,
            "samples": samples,
            "error": {"code": error_code},
        }
        manifest_path = directory / "manifest.json"
        try:
            payload = json.dumps(
                manifest, allow_nan=False, separators=(",", ":"), sort_keys=True
            ).encode("utf-8")
            if len(payload) > MAX_MANIFEST_BYTES:
                raise HandTrainingError(
                    "manifest_too_large", "Recording manifest is too large"
                )
            descriptor = os.open(str(manifest_path), _create_flags(), 0o640)
            try:
                _write_all(descriptor, payload)
                os.fsync(descriptor)
            finally:
                os.close(descriptor)
            if final_state == "completed":
                with self._lock:
                    self._state = "uploading"
                receipt = self.uploader.upload(
                    manifest_path=manifest_path,
                    video_paths=video_paths,
                    idempotency_key=session_id,
                )
                if not isinstance(receipt, Mapping):
                    raise HandTrainingError(
                        "upload_receipt_invalid", "API gateway returned invalid receipt"
                    )
                with self._lock:
                    self._receipt = dict(receipt)
            with self._lock:
                self._state = final_state
                self._error_code = error_code
        except HandTrainingError as exc:
            with self._lock:
                self._state = "failed"
                self._error_code = exc.code
        except Exception:
            with self._lock:
                self._state = "failed"
                self._error_code = "upload_failed"

    def _force_tool_off(self) -> None:
        control = self.tool_control
        callback = getattr(control, "release", None)
        if control is None or not callable(callback):
            with self._lock:
                self._tool_on = False
            return
        status = self._tool_status()
        if status.get("grasp_confirmed") is True:
            result = callback()
            if not isinstance(result, Mapping) or result.get("grasp_confirmed") is not False:
                raise HandTrainingError(
                    "tool_safe_state_failed", "Tool off was not confirmed"
                )
        with self._lock:
            self._tool_on = False

    def _tool_status(self) -> Mapping[str, Any]:
        if self.tool_control is None:
            return {
                "configured": False,
                "ready": False,
                "grasp_confirmed": False,
                "state": "unavailable",
            }
        callback = getattr(self.tool_control, "status", None)
        if not callable(callback):
            raise HandTrainingError(
                "tool_unavailable", "Typed tool status is unavailable"
            )
        result = callback()
        if not isinstance(result, Mapping):
            raise HandTrainingError("tool_status_invalid", "Tool status is invalid")
        return dict(result)

    @staticmethod
    def _validated_telemetry(value: Any) -> Mapping[str, Any]:
        if not isinstance(value, Mapping):
            raise HandTrainingError("telemetry_unavailable", "Telemetry is unavailable")
        if value.get("connected") is not True or value.get("healthy") is not True:
            raise HandTrainingError("xarm_not_ready", "Lite6 is not healthy")
        if value.get("fresh") is not True:
            raise HandTrainingError("telemetry_stale", "Telemetry is stale")
        if value.get("mode") != HAND_TRAINING_REQUIRED_MODE or value.get("state") != 0:
            raise HandTrainingError(
                "teach_mode_not_ready", "Lite6 teach mode is not active"
            )
        if value.get("error_code") != 0 or value.get("warning_code") != 0:
            raise HandTrainingError("xarm_fault", "Lite6 reports a fault")
        metrics = value.get("metrics")
        if not isinstance(metrics, Mapping):
            raise HandTrainingError("telemetry_invalid", "Metric payload is missing")
        for name, length in (
            ("joint_angles_rad", 6),
            ("tcp_position_mm", 3),
            ("tcp_orientation_rad", 3),
        ):
            vector = metrics.get(name)
            if not isinstance(vector, (tuple, list)) or len(vector) != length:
                raise HandTrainingError(
                    "telemetry_invalid", "{} is missing".format(name)
                )
            for item in vector:
                _finite(item, name)
        sanitized = _json_safe(value)
        encoded = json.dumps(
            sanitized, allow_nan=False, separators=(",", ":"), sort_keys=True
        ).encode("utf-8")
        if len(encoded) > 192 * 1024:
            raise HandTrainingError("telemetry_too_large", "Telemetry is too large")
        return sanitized

    def _normalized_pair(self, pair: Any) -> Mapping[str, Any]:
        if not isinstance(pair, Mapping):
            raise HandTrainingError("camera_pair_incomplete", "Camera pair is missing")
        normalized = {}
        for key, frame in pair.items():
            role = str(getattr(key, "value", key))
            if role in normalized:
                raise HandTrainingError(
                    "camera_pair_incomplete", "Camera pair has a duplicate role"
                )
            normalized[role] = frame
        if set(normalized) != set(HAND_TRAINING_ROLES):
            raise HandTrainingError(
                "camera_pair_incomplete", "Both camera streams are required"
            )
        captures = {}
        for role in HAND_TRAINING_ROLES:
            frame = normalized[role]
            frame_role = str(
                getattr(getattr(frame, "role", ""), "value", getattr(frame, "role", ""))
            )
            source_id = str(getattr(frame, "source_id", ""))
            sequence = getattr(frame, "sequence", None)
            if frame_role != role or _SOURCE_ID.fullmatch(source_id) is None:
                raise HandTrainingError(
                    "camera_identity_invalid", "Camera identity is invalid"
                )
            if isinstance(sequence, bool) or not isinstance(sequence, int) or sequence < 1:
                raise HandTrainingError(
                    "camera_sequence_invalid", "Camera sequence is invalid"
                )
            captures[role] = _finite(
                getattr(frame, "captured_at_s", None), "camera capture time"
            )
        if abs(captures["tool"] - captures["workspace"]) > self.max_pair_skew_seconds:
            raise HandTrainingError(
                "camera_pair_skew", "Camera pair exceeds commissioned skew"
            )
        return normalized


__all__ = [
    "HAND_TRAINING_COORDINATE_FRAME",
    "HAND_TRAINING_REQUIRED_MODE",
    "HAND_TRAINING_ROLES",
    "HAND_TRAINING_SCHEMA",
    "HAND_TRAINING_VIDEO_CONTENT_TYPE",
    "HandTrainingBusy",
    "HandTrainingError",
    "HandTrainingRecordingController",
]
