"""Hardened, run-scoped dual-camera dataset capture.

This module is deliberately independent of the HTTP and LVGL layers.  It owns
the invariants that must not be weakened by a UI request:

* every run lasts exactly fifteen seconds;
* one strictly-new, freshness- and skew-checked pair is captured at each of
  the offsets 0..14 seconds;
* both tool and workspace JPEGs are durable before a timepoint is published in
  the manifest; and
* success requires evidence from an injected, commissioned motion runner.

The motion runner is an integration seam, not a camera-only fallback.  A live
deployment must inject an implementation backed by the reviewed xArm command
path.  No default or simulated motion implementation is provided here.
"""

from __future__ import annotations

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

CAPTURE_SCHEMA = "labcobot.capture-dataset.v1"
STATUS_SCHEMA_VERSION = 1
RUN_DURATION_SECONDS = 15.0
CAPTURE_INTERVAL_SECONDS = 1.0
CAPTURE_TIMEPOINTS = 15
CAPTURE_IMAGE_COUNT = 30
REQUIRED_ROLES = ("tool", "workspace")
MAX_JPEG_BYTES = 8 * 1024 * 1024
MAX_MANIFEST_BYTES = 1024 * 1024
DEFAULT_MAX_FRAME_AGE_SECONDS = 0.5
DEFAULT_MAX_PAIR_SKEW_SECONDS = 0.15
DEFAULT_FRAME_TIMEOUT_SECONDS = 0.75
DEFAULT_MOTION_COMPLETION_GRACE_SECONDS = 1.0
# Live Lite6 preparation performs several acknowledged controller calls before
# the first 100 Hz command. A 50 ms lead was only realistic for test doubles;
# keep a bounded margin below the motion runner's hard two-second maximum.
DEFAULT_START_LEAD_SECONDS = 1.5
DEFAULT_MAX_SCHEDULE_LATENESS_SECONDS = 0.25
DEFAULT_FREE_SPACE_RESERVE_BYTES = 128 * 1024 * 1024

_RUN_ID = re.compile(r"^[0-9a-f]{32}$")
_SOURCE_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,255}$")
_SHA256 = re.compile(r"^[0-9a-f]{64}$")
_TRANSIENT_MANIFEST_STATES = frozenset(("starting", "running", "stopping"))
_FINAL_MANIFEST_STATES = frozenset(
    ("completed", "cancelled", "failed", "interrupted", "fault_locked")
)
_PUBLIC_STATES = frozenset(
    (
        "idle",
        "starting",
        "running",
        "stopping",
        "completed",
        "cancelled",
        "failed",
        "interrupted",
        "fault_locked",
    )
)


class CaptureRunError(RuntimeError):
    """A bounded, machine-classifiable capture-run failure."""

    def __init__(self, code: str, message: str = "") -> None:
        self.code = _safe_error_code(code)
        self.safe_message = _safe_text(message, 160)
        super().__init__(self.safe_message or self.code)


class CaptureConfigurationError(CaptureRunError):
    """The capture core was not given a commissioned dependency."""


class CaptureBusyError(CaptureRunError):
    """A second run was requested while another run is active."""


class CaptureCancelled(CaptureRunError):
    """The current run was cancelled or the service was stopped."""


class DatasetSecurityError(CaptureRunError):
    """A dataset path or file failed a no-follow/exclusive-write check."""


@dataclass(frozen=True)
class MotionExecutionEvidence:
    """Evidence returned by a commissioned motion session at completion."""

    executed: bool
    profile_sha256: str
    started_at_monotonic_s: float
    finished_at_monotonic_s: float
    command_count: int

    @property
    def duration_seconds(self) -> float:
        return self.finished_at_monotonic_s - self.started_at_monotonic_s


class CommissionedMotionSession(Protocol):
    """One already-started asynchronous execution of the reviewed motion."""

    def is_active(self) -> bool:
        """Return true while the motion execution remains healthy and active."""

    def wait(self, timeout_seconds: float) -> MotionExecutionEvidence:
        """Return bounded completion evidence or raise on timeout/failure."""


class CommissionedMotionRunner(Protocol):
    """Required live-motion integration seam.

    ``commissioned`` must be the literal boolean ``True``.  The profile hash
    and duration are checked both before starting and in the returned evidence.
    ``halt`` must be idempotent because it is the final fail-closed operation
    for success, cancellation, and every error path.
    """

    commissioned: bool
    profile_sha256: str
    duration_seconds: float

    def start(
        self,
        *,
        start_at_monotonic_s: float,
        duration_seconds: float,
        stop_requested: Callable[[], bool],
    ) -> CommissionedMotionSession:
        """Start only the reviewed motion profile and return its session."""

    def halt(self) -> None:
        """Close command gates and request a controller halt."""


class PairedFrameProvider(Protocol):
    """Fetch a pair newer than the supplied per-role sequence numbers."""

    def __call__(
        self,
        *,
        after_sequences: Mapping[str, int],
        timeout_seconds: float,
        max_age_seconds: float,
        max_skew_seconds: float,
    ) -> Mapping[Any, Any]:
        """Return exactly one tool and one workspace frame."""


@dataclass(frozen=True)
class _EncodedFrame:
    role: str
    source_id: str
    sequence: int
    captured_at_s: float
    received_monotonic_s: float
    jpeg: bytes
    sha256: str


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 _safe_error_code(value: Any) -> str:
    code = str(value or "capture_failed").strip().lower()
    code = re.sub(r"[^a-z0-9_]+", "_", code)[:64].strip("_")
    return code or "capture_failed"


def _finite(value: Any, name: str) -> float:
    try:
        result = float(value)
    except (TypeError, ValueError, OverflowError) as exc:
        raise CaptureRunError(
            "invalid_frame", "{} is not numeric".format(name)
        ) from exc
    if not math.isfinite(result):
        raise CaptureRunError("invalid_frame", "{} is not finite".format(name))
    return result


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


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


def _file_read_flags() -> int:
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
    flags |= getattr(os, "O_NOFOLLOW", 0)
    return flags


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


def _fsync_directory(descriptor: int) -> None:
    os.fsync(descriptor)


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(errno.EIO, "short dataset write")
        written += count


def _map_os_error(exc: OSError) -> CaptureRunError:
    if exc.errno in {errno.ENOSPC, getattr(errno, "EDQUOT", errno.ENOSPC)}:
        return CaptureRunError("dataset_full", "Dataset storage is full")
    if exc.errno in {
        errno.ELOOP,
        errno.EEXIST,
        errno.ENOTDIR,
        errno.EISDIR,
    }:
        return DatasetSecurityError(
            "dataset_unsafe_path", "Dataset path failed an exclusive no-follow check"
        )
    return CaptureRunError("dataset_io", "Dataset storage operation failed")


class CaptureDatasetStore:
    """No-follow, crash-evident storage for capture manifests and JPEGs."""

    def __init__(
        self,
        root: Path,
        *,
        wall_clock: Callable[[], float] = time.time,
        free_bytes_provider: Optional[Callable[[Path], int]] = None,
        free_space_reserve_bytes: int = DEFAULT_FREE_SPACE_RESERVE_BYTES,
        max_jpeg_bytes: int = MAX_JPEG_BYTES,
    ) -> None:
        self.root = Path(root)
        self.wall_clock = wall_clock
        self.free_bytes_provider = free_bytes_provider or (
            lambda path: int(shutil.disk_usage(path).free)
        )
        self.free_space_reserve_bytes = int(free_space_reserve_bytes)
        self.max_jpeg_bytes = int(max_jpeg_bytes)
        if not self.root.is_absolute():
            raise DatasetSecurityError(
                "dataset_unsafe_path", "Dataset root must be absolute"
            )
        if (
            self.free_space_reserve_bytes < 0
            or not 1024 <= self.max_jpeg_bytes <= MAX_JPEG_BYTES
        ):
            raise CaptureConfigurationError(
                "dataset_configuration", "Dataset size limits are invalid"
            )
        self._prepare_root()

    @property
    def maximum_run_bytes(self) -> int:
        return CAPTURE_IMAGE_COUNT * self.max_jpeg_bytes + MAX_MANIFEST_BYTES

    def _prepare_root(self) -> None:
        try:
            self.root.mkdir(mode=0o750, parents=True, exist_ok=True)
            info = self.root.lstat()
            if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
                raise DatasetSecurityError(
                    "dataset_unsafe_path", "Dataset root is not a real directory"
                )
            descriptor = os.open(str(self.root), _directory_flags())
            try:
                os.fchmod(descriptor, 0o750)
                _fsync_directory(descriptor)
            finally:
                os.close(descriptor)
        except CaptureRunError:
            raise
        except OSError as exc:
            raise _map_os_error(exc) from exc

    def ensure_capacity(self) -> None:
        try:
            free_bytes = int(self.free_bytes_provider(self.root))
        except Exception as exc:
            raise CaptureRunError(
                "dataset_capacity_unknown", "Could not verify free dataset space"
            ) from exc
        required = self.maximum_run_bytes + self.free_space_reserve_bytes
        if free_bytes < required:
            raise CaptureRunError(
                "dataset_full", "Insufficient free space for a complete capture run"
            )

    def begin_run(self, *, profile_sha256: str) -> Tuple[str, Dict[str, Any]]:
        self.ensure_capacity()
        profile_sha256 = str(profile_sha256).lower()
        if not _SHA256.fullmatch(profile_sha256):
            raise CaptureConfigurationError(
                "motion_not_commissioned", "Motion profile hash is invalid"
            )
        root_fd = self._open_root()
        try:
            for _attempt in range(8):
                run_id = uuid.uuid4().hex
                try:
                    os.mkdir(run_id, mode=0o750, dir_fd=root_fd)
                    break
                except FileExistsError:
                    continue
            else:  # pragma: no cover - cryptographically implausible
                raise CaptureRunError("dataset_io", "Could not allocate a run id")
            run_fd = os.open(run_id, _directory_flags(), dir_fd=root_fd)
            try:
                os.mkdir("frames", mode=0o750, dir_fd=run_fd)
                _fsync_directory(run_fd)
            finally:
                os.close(run_fd)
            _fsync_directory(root_fd)
        except CaptureRunError:
            raise
        except OSError as exc:
            raise _map_os_error(exc) from exc
        finally:
            os.close(root_fd)

        manifest: Dict[str, Any] = {
            "schema": CAPTURE_SCHEMA,
            "run_id": run_id,
            "status": "starting",
            "created_at": _utc_timestamp(self.wall_clock()),
            "started_at": "",
            "finished_at": "",
            "requested": {
                "duration_seconds": RUN_DURATION_SECONDS,
                "interval_seconds": CAPTURE_INTERVAL_SECONDS,
                "timepoints": CAPTURE_TIMEPOINTS,
                "images": CAPTURE_IMAGE_COUNT,
                "roles": list(REQUIRED_ROLES),
            },
            "motion": {
                "required": True,
                "profile_sha256": profile_sha256,
                "executed": False,
                "command_count": 0,
                "duration_seconds": 0.0,
            },
            "samples": [],
            "error": {"code": ""},
        }
        self.write_manifest(run_id, manifest)
        return run_id, manifest

    def write_sample(
        self,
        run_id: str,
        index: int,
        frames: Mapping[str, _EncodedFrame],
    ) -> Mapping[str, Any]:
        self._validate_run_id(run_id)
        if isinstance(index, bool) or not 0 <= int(index) < CAPTURE_TIMEPOINTS:
            raise CaptureRunError("invalid_sample", "Timepoint index is invalid")
        if set(frames) != set(REQUIRED_ROLES):
            raise CaptureRunError(
                "camera_pair_incomplete", "Both camera roles are required"
            )
        for role in REQUIRED_ROLES:
            self._validate_encoded_frame(role, frames[role])

        run_fd = self._open_run(run_id)
        completed_names = []
        try:
            frames_fd = os.open("frames", _directory_flags(), dir_fd=run_fd)
            try:
                records = []
                for role in REQUIRED_ROLES:
                    frame = frames[role]
                    filename = "{:03d}-{}.jpg".format(int(index), role)
                    self._write_exclusive_file(frames_fd, filename, frame.jpeg)
                    completed_names.append(filename)
                    records.append(
                        {
                            "role": role,
                            "source_id": frame.source_id,
                            "sequence": frame.sequence,
                            "captured_at_s": frame.captured_at_s,
                            "received_monotonic_s": frame.received_monotonic_s,
                            "path": "frames/{}".format(filename),
                            "content_type": "image/jpeg",
                            "byte_length": len(frame.jpeg),
                            "sha256": frame.sha256,
                        }
                    )
                _fsync_directory(frames_fd)
            except Exception:
                for filename in completed_names:
                    try:
                        os.unlink(filename, dir_fd=frames_fd)
                    except OSError:
                        pass
                try:
                    _fsync_directory(frames_fd)
                except OSError:
                    pass
                raise
            finally:
                os.close(frames_fd)
        except CaptureRunError:
            raise
        except OSError as exc:
            raise _map_os_error(exc) from exc
        finally:
            os.close(run_fd)

        skew = abs(frames["tool"].captured_at_s - frames["workspace"].captured_at_s)
        return {
            "index": int(index),
            "scheduled_offset_seconds": int(index) * CAPTURE_INTERVAL_SECONDS,
            "pair_skew_seconds": skew,
            "frames": records,
        }

    def _validate_encoded_frame(self, role: str, frame: _EncodedFrame) -> None:
        if not isinstance(frame, _EncodedFrame) or frame.role != role:
            raise CaptureRunError(
                "invalid_sample", "Encoded frame does not match its camera role"
            )
        if not _SOURCE_ID.fullmatch(frame.source_id):
            raise CaptureRunError(
                "invalid_sample", "Encoded frame source identity is invalid"
            )
        if (
            isinstance(frame.sequence, bool)
            or not isinstance(frame.sequence, int)
            or frame.sequence < 1
        ):
            raise CaptureRunError("invalid_sample", "Encoded frame sequence is invalid")
        _finite(frame.captured_at_s, "encoded capture timestamp")
        _finite(frame.received_monotonic_s, "encoded receive timestamp")
        if (
            not isinstance(frame.jpeg, bytes)
            or not 4 <= len(frame.jpeg) <= self.max_jpeg_bytes
            or not frame.jpeg.startswith(b"\xff\xd8")
            or not frame.jpeg.endswith(b"\xff\xd9")
        ):
            raise CaptureRunError("invalid_sample", "Encoded frame JPEG is invalid")
        if (
            not _SHA256.fullmatch(frame.sha256)
            or hashlib.sha256(frame.jpeg).hexdigest() != frame.sha256
        ):
            raise CaptureRunError(
                "invalid_sample", "Encoded frame checksum does not match"
            )

    def write_manifest(self, run_id: str, manifest: Mapping[str, Any]) -> None:
        self._validate_run_id(run_id)
        self._validate_manifest(run_id, manifest)
        try:
            payload = json.dumps(
                dict(manifest),
                ensure_ascii=False,
                allow_nan=False,
                separators=(",", ":"),
                sort_keys=True,
            ).encode("utf-8")
        except (TypeError, ValueError) as exc:
            raise CaptureRunError(
                "manifest_invalid", "Capture manifest is not valid JSON"
            ) from exc
        if len(payload) > MAX_MANIFEST_BYTES:
            raise CaptureRunError(
                "manifest_too_large", "Capture manifest exceeds its size limit"
            )

        run_fd = self._open_run(run_id)
        temporary = ".manifest.{}.tmp".format(uuid.uuid4().hex)
        descriptor: Optional[int] = None
        try:
            try:
                existing = os.stat(
                    "manifest.json", dir_fd=run_fd, follow_symlinks=False
                )
            except FileNotFoundError:
                existing = None
            if existing is not None and not stat.S_ISREG(existing.st_mode):
                raise DatasetSecurityError(
                    "dataset_unsafe_path", "Manifest target is not a regular file"
                )
            descriptor = os.open(
                temporary,
                _file_create_flags(),
                0o640,
                dir_fd=run_fd,
            )
            _write_all(descriptor, payload)
            os.fsync(descriptor)
            os.close(descriptor)
            descriptor = None
            os.replace(
                temporary,
                "manifest.json",
                src_dir_fd=run_fd,
                dst_dir_fd=run_fd,
            )
            _fsync_directory(run_fd)
        except CaptureRunError:
            raise
        except OSError as exc:
            raise _map_os_error(exc) from exc
        finally:
            if descriptor is not None:
                os.close(descriptor)
            try:
                os.unlink(temporary, dir_fd=run_fd)
            except OSError:
                pass
            os.close(run_fd)

    def read_manifest(self, run_id: str) -> Mapping[str, Any]:
        self._validate_run_id(run_id)
        run_fd = self._open_run(run_id)
        descriptor: Optional[int] = None
        try:
            descriptor = os.open("manifest.json", _file_read_flags(), dir_fd=run_fd)
            info = os.fstat(descriptor)
            if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_MANIFEST_BYTES:
                raise DatasetSecurityError(
                    "dataset_unsafe_path", "Manifest is not a bounded regular file"
                )
            payload = bytearray()
            while len(payload) <= MAX_MANIFEST_BYTES:
                chunk = os.read(
                    descriptor, min(65536, MAX_MANIFEST_BYTES + 1 - len(payload))
                )
                if not chunk:
                    break
                payload.extend(chunk)
            if len(payload) > MAX_MANIFEST_BYTES:
                raise CaptureRunError(
                    "manifest_too_large", "Capture manifest exceeds its size limit"
                )
            value = json.loads(bytes(payload).decode("utf-8"))
            if not isinstance(value, Mapping):
                raise ValueError("manifest is not an object")
            result = dict(value)
            self._validate_manifest(run_id, result)
            return result
        except CaptureRunError:
            raise
        except (OSError, UnicodeError, ValueError, json.JSONDecodeError) as exc:
            raise CaptureRunError(
                "manifest_invalid", "Capture manifest could not be read"
            ) from exc
        finally:
            if descriptor is not None:
                os.close(descriptor)
            os.close(run_fd)

    def recover_incomplete_runs(self) -> Tuple[str, ...]:
        """Mark stale transient manifests interrupted; never infer success."""

        recovered = []
        try:
            entries = tuple(os.scandir(self.root))
        except OSError as exc:
            raise _map_os_error(exc) from exc
        for entry in entries:
            if not _RUN_ID.fullmatch(entry.name):
                continue
            if entry.is_symlink() or not entry.is_dir(follow_symlinks=False):
                raise DatasetSecurityError(
                    "dataset_unsafe_path", "Capture run entry is not a real directory"
                )
            manifest = dict(self.read_manifest(entry.name))
            if manifest.get("status") not in _TRANSIENT_MANIFEST_STATES:
                continue
            manifest["status"] = "interrupted"
            manifest["finished_at"] = _utc_timestamp(self.wall_clock())
            manifest["error"] = {"code": "process_restart"}
            self.write_manifest(entry.name, manifest)
            recovered.append(entry.name)
        return tuple(recovered)

    def _open_root(self) -> int:
        try:
            return os.open(str(self.root), _directory_flags())
        except OSError as exc:
            raise _map_os_error(exc) from exc

    def _open_run(self, run_id: str) -> int:
        self._validate_run_id(run_id)
        root_fd = self._open_root()
        try:
            return os.open(run_id, _directory_flags(), dir_fd=root_fd)
        except OSError as exc:
            raise _map_os_error(exc) from exc
        finally:
            os.close(root_fd)

    @staticmethod
    def _validate_run_id(run_id: str) -> None:
        if not _RUN_ID.fullmatch(str(run_id)):
            raise DatasetSecurityError(
                "dataset_unsafe_path", "Capture run id is invalid"
            )

    @staticmethod
    def _validate_manifest(run_id: str, manifest: Mapping[str, Any]) -> None:
        if manifest.get("schema") != CAPTURE_SCHEMA:
            raise CaptureRunError(
                "manifest_invalid", "Capture manifest schema is invalid"
            )
        if manifest.get("run_id") != run_id:
            raise DatasetSecurityError(
                "dataset_unsafe_path", "Capture manifest belongs to another run"
            )
        status_value = str(manifest.get("status") or "")
        if status_value not in _TRANSIENT_MANIFEST_STATES | _FINAL_MANIFEST_STATES:
            raise CaptureRunError(
                "manifest_invalid", "Capture manifest status is invalid"
            )
        samples = manifest.get("samples")
        if not isinstance(samples, list) or len(samples) > CAPTURE_TIMEPOINTS:
            raise CaptureRunError("manifest_invalid", "Capture sample list is invalid")

    @staticmethod
    def _write_exclusive_file(directory_fd: int, filename: str, payload: bytes) -> None:
        temporary = ".{}.{}.tmp".format(filename, uuid.uuid4().hex)
        descriptor: Optional[int] = None
        try:
            descriptor = os.open(
                temporary,
                _file_create_flags(),
                0o640,
                dir_fd=directory_fd,
            )
            _write_all(descriptor, payload)
            os.fsync(descriptor)
            os.close(descriptor)
            descriptor = None
            # Linking a complete, fsynced temporary inode provides portable
            # no-replace publication.  It fails atomically if the final name
            # already exists, unlike os.replace which would overwrite evidence.
            os.link(
                temporary,
                filename,
                src_dir_fd=directory_fd,
                dst_dir_fd=directory_fd,
                follow_symlinks=False,
            )
            _fsync_directory(directory_fd)
            os.unlink(temporary, dir_fd=directory_fd)
            _fsync_directory(directory_fd)
        except OSError as exc:
            raise _map_os_error(exc) from exc
        finally:
            if descriptor is not None:
                os.close(descriptor)
            try:
                os.unlink(temporary, dir_fd=directory_fd)
            except OSError:
                pass


def _manifest_has_complete_local_evidence(
    run_id: str,
    manifest: Mapping[str, Any],
) -> bool:
    """Validate the closed structural evidence required for local completion.

    This check deliberately exposes only a boolean. Paths and frame metadata
    remain inside the durable manifest and the exporter performs the separate
    byte/hash verification before any remote publication.
    """

    error = manifest.get("error")
    requested = manifest.get("requested")
    motion = manifest.get("motion")
    samples = manifest.get("samples")
    if (
        manifest.get("schema") != CAPTURE_SCHEMA
        or manifest.get("run_id") != run_id
        or manifest.get("status") != "completed"
        or not isinstance(manifest.get("finished_at"), str)
        or not 1 <= len(str(manifest.get("finished_at"))) <= 64
        or not isinstance(error, Mapping)
        or error.get("code") not in ("", None)
        or not isinstance(requested, Mapping)
        or requested.get("duration_seconds") != RUN_DURATION_SECONDS
        or requested.get("interval_seconds") != CAPTURE_INTERVAL_SECONDS
        or requested.get("timepoints") != CAPTURE_TIMEPOINTS
        or requested.get("images") != CAPTURE_IMAGE_COUNT
        or requested.get("roles") != list(REQUIRED_ROLES)
        or not isinstance(motion, Mapping)
        or motion.get("required") is not True
        or motion.get("executed") is not True
        or not isinstance(motion.get("profile_sha256"), str)
        or _SHA256.fullmatch(str(motion.get("profile_sha256"))) is None
        or isinstance(motion.get("command_count"), bool)
        or not isinstance(motion.get("command_count"), int)
        or int(motion.get("command_count")) < 1
        or isinstance(motion.get("duration_seconds"), bool)
        or not isinstance(motion.get("duration_seconds"), (int, float))
        or not math.isfinite(float(motion.get("duration_seconds")))
        or abs(float(motion.get("duration_seconds")) - RUN_DURATION_SECONDS) > 0.1
        or not isinstance(samples, list)
        or len(samples) != CAPTURE_TIMEPOINTS
    ):
        return False

    for expected_index, sample in enumerate(samples):
        if not isinstance(sample, Mapping):
            return False
        if (
            sample.get("index") != expected_index
            or sample.get("scheduled_offset_seconds") != expected_index
        ):
            return False
        frames = sample.get("frames")
        if not isinstance(frames, list) or len(frames) != len(REQUIRED_ROLES):
            return False
        roles = set()
        for frame in frames:
            if not isinstance(frame, Mapping):
                return False
            role = frame.get("role")
            if role not in REQUIRED_ROLES or role in roles:
                return False
            roles.add(role)
            sequence = frame.get("sequence")
            byte_length = frame.get("byte_length")
            source_id = frame.get("source_id")
            checksum = frame.get("sha256")
            if (
                not isinstance(source_id, str)
                or _SOURCE_ID.fullmatch(source_id) is None
                or isinstance(sequence, bool)
                or not isinstance(sequence, int)
                or sequence < 1
                or frame.get("path")
                != "frames/{:03d}-{}.jpg".format(expected_index, role)
                or frame.get("content_type") != "image/jpeg"
                or isinstance(byte_length, bool)
                or not isinstance(byte_length, int)
                or not 4 <= byte_length <= MAX_JPEG_BYTES
                or not isinstance(checksum, str)
                or _SHA256.fullmatch(checksum) is None
            ):
                return False
        if roles != set(REQUIRED_ROLES):
            return False
    return True


class CaptureRunController:
    """Execute one commissioned-motion, dual-OAK dataset run at a time."""

    def __init__(
        self,
        *,
        store: CaptureDatasetStore,
        pair_provider: PairedFrameProvider,
        jpeg_encoder: Callable[[Any], bytes],
        motion_runner: CommissionedMotionRunner,
        monotonic: Callable[[], float] = time.monotonic,
        wall_clock: Callable[[], float] = time.time,
        sleeper: Callable[[float], None] = time.sleep,
        max_frame_age_seconds: float = DEFAULT_MAX_FRAME_AGE_SECONDS,
        max_pair_skew_seconds: float = DEFAULT_MAX_PAIR_SKEW_SECONDS,
        frame_timeout_seconds: float = DEFAULT_FRAME_TIMEOUT_SECONDS,
        motion_completion_grace_seconds: float = (
            DEFAULT_MOTION_COMPLETION_GRACE_SECONDS
        ),
        start_lead_seconds: float = DEFAULT_START_LEAD_SECONDS,
        max_schedule_lateness_seconds: float = (DEFAULT_MAX_SCHEDULE_LATENESS_SECONDS),
    ) -> None:
        self.store = store
        self.pair_provider = pair_provider
        self.jpeg_encoder = jpeg_encoder
        self.motion_runner = motion_runner
        self.monotonic = monotonic
        self.wall_clock = wall_clock
        self.sleeper = sleeper
        self.max_frame_age_seconds = _positive_bounded(
            max_frame_age_seconds, "max_frame_age_seconds", 10.0
        )
        self.max_pair_skew_seconds = _positive_bounded(
            max_pair_skew_seconds, "max_pair_skew_seconds", 2.0
        )
        self.frame_timeout_seconds = _positive_bounded(
            frame_timeout_seconds, "frame_timeout_seconds", 10.0
        )
        self.motion_completion_grace_seconds = _positive_bounded(
            motion_completion_grace_seconds,
            "motion_completion_grace_seconds",
            10.0,
        )
        self.start_lead_seconds = _nonnegative_bounded(
            start_lead_seconds, "start_lead_seconds", 2.0
        )
        self.max_schedule_lateness_seconds = _positive_bounded(
            max_schedule_lateness_seconds,
            "max_schedule_lateness_seconds",
            1.0,
        )
        self._profile_sha256 = self._validate_motion_runner(motion_runner)

        self._state_lock = threading.RLock()
        self._halt_lock = threading.Lock()
        self._thread: Optional[threading.Thread] = None
        self._cancel_event = threading.Event()
        self._halt_attempted = False
        self._halt_error = False
        self._run_id = ""
        self._state = "idle"
        self._captured_timepoints = 0
        self._error_code = ""
        self._last_recovered_run = ""
        self._motion_started = False
        self._stop_requested = False

        recovered = self.store.recover_incomplete_runs()
        if recovered:
            self._state = "interrupted"
            self._error_code = "process_restart"
            self._last_recovered_run = recovered[-1]
            self._run_id = recovered[-1]

    @staticmethod
    def _validate_motion_runner(runner: CommissionedMotionRunner) -> str:
        if getattr(runner, "commissioned", None) is not True:
            raise CaptureConfigurationError(
                "motion_not_commissioned", "Capture motion runner is not commissioned"
            )
        profile_sha256 = str(getattr(runner, "profile_sha256", "")).lower()
        if not _SHA256.fullmatch(profile_sha256):
            raise CaptureConfigurationError(
                "motion_not_commissioned", "Capture motion profile hash is invalid"
            )
        duration = _finite(getattr(runner, "duration_seconds", None), "motion duration")
        if abs(duration - RUN_DURATION_SECONDS) > 1e-9:
            raise CaptureConfigurationError(
                "motion_profile_duration", "Capture motion profile must last 15 seconds"
            )
        for method_name in ("start", "halt"):
            if not callable(getattr(runner, method_name, None)):
                raise CaptureConfigurationError(
                    "motion_not_commissioned", "Capture motion runner is incomplete"
                )
        return profile_sha256

    def start(
        self,
        *,
        external_stop_requested: Optional[Callable[[], bool]] = None,
    ) -> str:
        """Create and asynchronously start one fixed capture run.

        ``external_stop_requested`` binds the new worker to the workcell's
        global STOP generation.  It is deliberately carried into the worker
        and the commissioned motion runner: a STOP that lands after the
        caller's final admission check, but before this thread starts, cannot
        be erased when the controller creates its per-run cancellation event.
        Any callback failure is treated as a STOP request.
        """

        if external_stop_requested is not None and not callable(
            external_stop_requested
        ):
            raise CaptureConfigurationError(
                "capture_configuration",
                "External capture STOP predicate must be callable",
            )

        with self._state_lock:
            if self._thread is not None and self._thread.is_alive():
                raise CaptureBusyError(
                    "capture_busy", "A capture run is already active"
                )
            if self._state == "fault_locked":
                raise CaptureRunError(
                    "motion_fault_locked",
                    "Capture motion remains fault-locked after a failed halt",
                )
            self.store.ensure_capacity()
            run_id, manifest = self.store.begin_run(profile_sha256=self._profile_sha256)
            self._cancel_event = threading.Event()
            self._halt_attempted = False
            self._halt_error = False
            self._run_id = run_id
            self._state = "starting"
            self._captured_timepoints = 0
            self._error_code = ""
            self._motion_started = False
            self._stop_requested = False
            thread = threading.Thread(
                target=self._run,
                args=(run_id, manifest, external_stop_requested),
                name="capture-run-{}".format(run_id[:8]),
                daemon=True,
            )
            self._thread = thread
            try:
                thread.start()
            except Exception as exc:
                self._thread = None
                self._state = "failed"
                self._error_code = "worker_start_failed"
                manifest["status"] = "failed"
                manifest["finished_at"] = _utc_timestamp(self.wall_clock())
                manifest["error"] = {"code": "worker_start_failed"}
                self.store.write_manifest(run_id, manifest)
                raise CaptureRunError(
                    "worker_start_failed", "Capture worker could not start"
                ) from exc
            return run_id

    def cancel(self, run_id: str = "") -> bool:
        """Request cancellation and immediately invoke the fail-closed halt seam."""

        with self._state_lock:
            thread = self._thread
            if thread is None or not thread.is_alive():
                return False
            if run_id and run_id != self._run_id:
                return False
            self._state = "stopping"
            self._stop_requested = True
            self._cancel_event.set()
        if not self._halt_motion_once(retry_failed=True):
            raise CaptureRunError(
                "motion_halt_failed",
                "Capture cancellation could not confirm the motion halt",
            )
        return True

    def shutdown(self, timeout_seconds: float = 3.0) -> bool:
        """Cancel the active run, halt motion, and wait a bounded interval."""

        self.cancel()
        self._halt_motion_once(retry_failed=True)
        return self.wait(timeout_seconds)

    def wait(self, timeout_seconds: Optional[float] = None) -> bool:
        with self._state_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]:
        """Return bounded, path-free state suitable for the local LVGL API."""

        with self._state_lock:
            state = self._state if self._state in _PUBLIC_STATES else "failed"
            active = self._thread is not None and self._thread.is_alive()
            run_id = self._run_id
            captured = min(max(int(self._captured_timepoints), 0), CAPTURE_TIMEPOINTS)
            error_code = _safe_error_code(self._error_code) if self._error_code else ""
            motion_started = bool(self._motion_started)
            stop_requested = bool(self._stop_requested)
            message = self._status_message(state, captured)
        manifest_state = "none"
        manifest_complete = False
        if _RUN_ID.fullmatch(run_id):
            try:
                manifest = self.store.read_manifest(run_id)
            except Exception:
                manifest_state = "invalid"
            else:
                raw_manifest_state = str(manifest.get("status") or "")
                manifest_state = (
                    raw_manifest_state
                    if raw_manifest_state
                    in _TRANSIENT_MANIFEST_STATES | _FINAL_MANIFEST_STATES
                    else "invalid"
                )
                manifest_complete = bool(
                    state == "completed"
                    and captured == CAPTURE_TIMEPOINTS
                    and _manifest_has_complete_local_evidence(run_id, manifest)
                )
        return {
            "schema_version": STATUS_SCHEMA_VERSION,
            "configured": True,
            "state": state,
            "active": active,
            "action_allowed": not active and state != "fault_locked",
            "can_cancel": active,
            "run_id": run_id if _RUN_ID.fullmatch(run_id) else "",
            "planned_timepoints": CAPTURE_TIMEPOINTS,
            "completed_timepoints": captured,
            "saved_images": captured * len(REQUIRED_ROLES),
            "tool_saved": captured,
            "workspace_saved": captured,
            "captured_timepoints": captured,
            "total_timepoints": CAPTURE_TIMEPOINTS,
            "captured_images": captured * len(REQUIRED_ROLES),
            "total_images": CAPTURE_IMAGE_COUNT,
            "duration_seconds": RUN_DURATION_SECONDS,
            "interval_seconds": CAPTURE_INTERVAL_SECONDS,
            "message": message,
            "motion_started": motion_started,
            "stop_requested": stop_requested,
            "error_code": error_code,
            "local_manifest_state": manifest_state,
            "local_manifest_complete": manifest_complete,
        }

    @staticmethod
    def _status_message(state: str, completed: int) -> str:
        messages = {
            "idle": "Ready for a 15-second capture run",
            "starting": "Preparing commissioned motion and both cameras",
            "stopping": "Stopping motion and capture",
            "completed": "Saved all 30 camera images",
            "cancelled": "Capture cancelled; motion halt requested",
            "failed": "Capture failed; motion halt requested",
            "interrupted": "An interrupted capture run requires review",
            "fault_locked": (
                "A motion safety fault is latched; inspect the workcell and use "
                "the physical emergency stop if motion persists"
            ),
        }
        if state == "running":
            return "Saved paired timepoint {} of {}".format(
                min(max(int(completed), 0), CAPTURE_TIMEPOINTS),
                CAPTURE_TIMEPOINTS,
            )
        return _safe_text(messages.get(state, "Capture status unavailable"), 160)

    def _run(
        self,
        run_id: str,
        manifest: Dict[str, Any],
        external_stop_requested: Optional[Callable[[], bool]],
    ) -> None:
        final_state = "failed"
        error_code = "capture_failed"
        motion_evidence: Optional[MotionExecutionEvidence] = None
        session: Optional[CommissionedMotionSession] = None
        start_at = float(self.monotonic()) + self.start_lead_seconds
        last_sequences: Dict[str, int] = {}
        source_ids: Dict[str, str] = {}

        def stop_requested() -> bool:
            if self._cancel_event.is_set():
                return True
            if external_stop_requested is None:
                return False
            try:
                stopped = external_stop_requested() is True
            except Exception:
                stopped = True
            if stopped:
                self._cancel_event.set()
            return stopped

        try:
            self._raise_if_cancelled(stop_requested)
            session = self.motion_runner.start(
                start_at_monotonic_s=start_at,
                duration_seconds=RUN_DURATION_SECONDS,
                stop_requested=stop_requested,
            )
            if (
                session is None
                or not callable(getattr(session, "is_active", None))
                or not callable(getattr(session, "wait", None))
            ):
                raise CaptureRunError(
                    "motion_start_failed", "Motion runner returned no valid session"
                )
            if session.is_active() is not True:
                raise CaptureRunError(
                    "motion_start_failed", "Commissioned motion did not become active"
                )
            with self._state_lock:
                self._motion_started = True
            manifest["status"] = "running"
            manifest["started_at"] = _utc_timestamp(self.wall_clock())
            self.store.write_manifest(run_id, manifest)
            self._set_status("running")

            for index in range(CAPTURE_TIMEPOINTS):
                scheduled_at = start_at + index * CAPTURE_INTERVAL_SECONDS
                self._sleep_until(scheduled_at, stop_requested=stop_requested)
                self._raise_if_cancelled(stop_requested)
                if float(self.monotonic()) - scheduled_at > (
                    self.max_schedule_lateness_seconds
                ):
                    raise CaptureRunError(
                        "capture_deadline_missed",
                        "A one-second camera timepoint deadline was missed",
                    )
                if session.is_active() is not True:
                    raise CaptureRunError(
                        "motion_ended_early",
                        "Motion ended before all images were captured",
                    )
                raw_pair = self.pair_provider(
                    after_sequences=dict(last_sequences),
                    timeout_seconds=self.frame_timeout_seconds,
                    max_age_seconds=self.max_frame_age_seconds,
                    max_skew_seconds=self.max_pair_skew_seconds,
                )
                self._raise_if_cancelled(stop_requested)
                if float(self.monotonic()) - scheduled_at > (
                    self.max_schedule_lateness_seconds
                ):
                    raise CaptureRunError(
                        "capture_deadline_missed",
                        "A camera pair arrived after its one-second deadline",
                    )
                if session.is_active() is not True:
                    raise CaptureRunError(
                        "motion_ended_early",
                        "Motion ended before the camera pair was accepted",
                    )
                encoded = self._validate_and_encode_pair(
                    raw_pair,
                    after_sequences=last_sequences,
                    stable_source_ids=source_ids,
                )
                self._raise_if_cancelled(stop_requested)
                if float(self.monotonic()) - scheduled_at > (
                    self.max_schedule_lateness_seconds
                ):
                    raise CaptureRunError(
                        "capture_deadline_missed",
                        "JPEG encoding exceeded the timepoint deadline",
                    )
                if session.is_active() is not True:
                    raise CaptureRunError(
                        "motion_ended_early",
                        "Motion ended before the camera pair was persisted",
                    )
                sample = self.store.write_sample(run_id, index, encoded)
                manifest["samples"].append(sample)
                self.store.write_manifest(run_id, manifest)
                for role in REQUIRED_ROLES:
                    last_sequences[role] = encoded[role].sequence
                    source_ids.setdefault(role, encoded[role].source_id)
                self._set_progress(index + 1)

            self._sleep_until(
                start_at + RUN_DURATION_SECONDS,
                stop_requested=stop_requested,
            )
            self._raise_if_cancelled(stop_requested)
            motion_evidence = session.wait(self.motion_completion_grace_seconds)
            self._validate_motion_evidence(motion_evidence, start_at)
            final_state = "completed"
            error_code = ""
        except CaptureCancelled:
            self._cancel_event.set()
            with self._state_lock:
                self._stop_requested = True
            final_state = "cancelled"
            error_code = "cancelled"
        except CaptureRunError as exc:
            self._cancel_event.set()
            with self._state_lock:
                self._stop_requested = True
            final_state = "failed"
            error_code = exc.code
        except Exception as exc:
            self._cancel_event.set()
            with self._state_lock:
                self._stop_requested = True
            error_code = _exception_code(exc)
            # A live commissioning/safety guard mismatch is different from an
            # ordinary camera or storage failure.  The guard has already
            # requested a controller halt, but the process must remain locked
            # until an operator inspects the workcell.  Importing the optional
            # safety module here would create an unnecessary dependency cycle,
            # so recognize its closed, machine-readable error namespace.
            final_state = (
                "fault_locked" if error_code.startswith("runtime_safety_") else "failed"
            )
        finally:
            self._cancel_event.set()
            self._set_status("stopping", error_code=error_code)
            self._halt_motion_once(retry_failed=True)
            if self._halt_error:
                final_state = "fault_locked"
                error_code = "motion_halt_failed"

            if motion_evidence is not None:
                manifest["motion"] = {
                    "required": True,
                    "profile_sha256": self._profile_sha256,
                    "executed": bool(motion_evidence.executed),
                    "command_count": int(motion_evidence.command_count),
                    "duration_seconds": motion_evidence.duration_seconds,
                }
            manifest["status"] = final_state
            manifest["finished_at"] = _utc_timestamp(self.wall_clock())
            manifest["error"] = {"code": error_code}
            try:
                self.store.write_manifest(run_id, manifest)
            except Exception:
                final_state = "fault_locked" if self._halt_error else "failed"
                error_code = "manifest_finalize_failed"
            self._set_status(final_state, error_code=error_code)

    def _sleep_until(
        self,
        deadline: float,
        *,
        stop_requested: Optional[Callable[[], bool]] = None,
    ) -> None:
        while True:
            self._raise_if_cancelled(stop_requested)
            remaining = float(deadline) - float(self.monotonic())
            if remaining <= 0:
                return
            self.sleeper(min(remaining, 0.1))

    def _raise_if_cancelled(
        self,
        stop_requested: Optional[Callable[[], bool]] = None,
    ) -> None:
        if self._cancel_event.is_set() or (
            stop_requested is not None and stop_requested()
        ):
            raise CaptureCancelled("cancelled", "Capture run was cancelled")

    def _validate_and_encode_pair(
        self,
        pair: Mapping[Any, Any],
        *,
        after_sequences: Mapping[str, int],
        stable_source_ids: Mapping[str, str],
    ) -> Mapping[str, _EncodedFrame]:
        if not isinstance(pair, Mapping):
            raise CaptureRunError(
                "camera_pair_incomplete", "Camera provider did not return a pair"
            )
        normalized: Dict[str, Any] = {}
        for key, frame in pair.items():
            role = getattr(key, "value", key)
            role = str(role)
            if role in normalized:
                raise CaptureRunError(
                    "camera_pair_incomplete", "Camera pair contains a duplicate role"
                )
            normalized[role] = frame
        if set(normalized) != set(REQUIRED_ROLES):
            raise CaptureRunError(
                "camera_pair_incomplete", "Camera pair must contain tool and workspace"
            )

        now = float(self.monotonic())
        result: Dict[str, _EncodedFrame] = {}
        for role in REQUIRED_ROLES:
            frame = normalized[role]
            frame_role = str(
                getattr(getattr(frame, "role", ""), "value", getattr(frame, "role", ""))
            )
            if frame_role != role:
                raise CaptureRunError(
                    "camera_role_mismatch",
                    "Camera frame role does not match its pair key",
                )
            source_id = str(getattr(frame, "source_id", ""))
            if not _SOURCE_ID.fullmatch(source_id):
                raise CaptureRunError(
                    "camera_identity_invalid", "Camera source identity is invalid"
                )
            expected_source = stable_source_ids.get(role)
            if expected_source is not None and source_id != expected_source:
                raise CaptureRunError(
                    "camera_identity_changed", "Camera source changed during the run"
                )
            sequence = getattr(frame, "sequence", None)
            if (
                isinstance(sequence, bool)
                or not isinstance(sequence, int)
                or sequence < 1
            ):
                raise CaptureRunError(
                    "camera_sequence_invalid", "Camera sequence is invalid"
                )
            if sequence <= int(after_sequences.get(role, 0)):
                raise CaptureRunError(
                    "camera_frame_reused",
                    "Camera frame is not newer than the prior sample",
                )
            captured_at = _finite(
                getattr(frame, "captured_at_s", None), "capture timestamp"
            )
            received_at = _finite(
                getattr(frame, "received_monotonic_s", None), "receive timestamp"
            )
            age = now - received_at
            if age < -0.05 or age > self.max_frame_age_seconds:
                raise CaptureRunError(
                    "camera_frame_stale", "Camera frame is outside the freshness window"
                )
            try:
                jpeg = self.jpeg_encoder(getattr(frame, "image"))
            except Exception as exc:
                raise CaptureRunError(
                    "jpeg_encode_failed", "Camera frame could not be encoded"
                ) from exc
            if not isinstance(jpeg, bytes):
                raise CaptureRunError(
                    "jpeg_encode_failed", "JPEG encoder did not return bytes"
                )
            if not 4 <= len(jpeg) <= self.store.max_jpeg_bytes:
                raise CaptureRunError(
                    "jpeg_size_invalid", "Encoded camera image has an invalid size"
                )
            if not (jpeg.startswith(b"\xff\xd8") and jpeg.endswith(b"\xff\xd9")):
                raise CaptureRunError(
                    "jpeg_invalid", "Encoded camera image is not a JPEG"
                )
            result[role] = _EncodedFrame(
                role=role,
                source_id=source_id,
                sequence=sequence,
                captured_at_s=captured_at,
                received_monotonic_s=received_at,
                jpeg=jpeg,
                sha256=hashlib.sha256(jpeg).hexdigest(),
            )

        skew = abs(result["tool"].captured_at_s - result["workspace"].captured_at_s)
        if skew > self.max_pair_skew_seconds:
            raise CaptureRunError(
                "camera_pair_skew", "Camera pair exceeds the allowed timestamp skew"
            )
        return result

    def _validate_motion_evidence(
        self,
        evidence: MotionExecutionEvidence,
        requested_start: float,
    ) -> None:
        if not isinstance(evidence, MotionExecutionEvidence):
            raise CaptureRunError(
                "motion_evidence_invalid", "Motion runner returned invalid evidence"
            )
        if evidence.executed is not True:
            raise CaptureRunError(
                "motion_not_executed", "Commissioned motion was not executed"
            )
        if evidence.profile_sha256 != self._profile_sha256:
            raise CaptureRunError(
                "motion_profile_mismatch", "Motion evidence profile does not match"
            )
        started = _finite(evidence.started_at_monotonic_s, "motion start")
        finished = _finite(evidence.finished_at_monotonic_s, "motion finish")
        if abs(started - requested_start) > 0.1:
            raise CaptureRunError(
                "motion_timing_invalid", "Motion did not start at the capture epoch"
            )
        duration = finished - started
        if duration < RUN_DURATION_SECONDS - 0.05 or duration > (
            RUN_DURATION_SECONDS + self.motion_completion_grace_seconds
        ):
            raise CaptureRunError(
                "motion_timing_invalid", "Motion evidence does not cover 15 seconds"
            )
        if (
            isinstance(evidence.command_count, bool)
            or not isinstance(evidence.command_count, int)
            or evidence.command_count < 1
        ):
            raise CaptureRunError(
                "motion_evidence_invalid", "Motion evidence contains no commands"
            )

    def _halt_motion_once(self, *, retry_failed: bool = False) -> bool:
        with self._halt_lock:
            if self._halt_attempted and not (retry_failed and self._halt_error):
                return not self._halt_error
            self._halt_attempted = True
            try:
                self.motion_runner.halt()
            except Exception:
                self._halt_error = True
                return False
            self._halt_error = False
            return True

    def _set_progress(self, value: int) -> None:
        with self._state_lock:
            self._captured_timepoints = min(max(int(value), 0), CAPTURE_TIMEPOINTS)

    def _set_status(self, state: str, *, error_code: str = "") -> None:
        with self._state_lock:
            self._state = state if state in _PUBLIC_STATES else "failed"
            self._error_code = _safe_error_code(error_code) if error_code else ""


def _positive_bounded(value: Any, name: str, maximum: float) -> float:
    result = _finite(value, name)
    if result <= 0.0 or result > maximum:
        raise CaptureConfigurationError(
            "capture_configuration", "{} is outside its safe range".format(name)
        )
    return result


def _nonnegative_bounded(value: Any, name: str, maximum: float) -> float:
    result = _finite(value, name)
    if result < 0.0 or result > maximum:
        raise CaptureConfigurationError(
            "capture_configuration", "{} is outside its safe range".format(name)
        )
    return result


def _exception_code(exc: BaseException) -> str:
    if isinstance(exc, OSError) and exc.errno in {
        errno.ENOSPC,
        getattr(errno, "EDQUOT", errno.ENOSPC),
    }:
        return "dataset_full"
    code = getattr(exc, "code", "")
    if isinstance(code, str) and code:
        return _safe_error_code(code)
    name = type(exc).__name__
    return _safe_error_code("capture_{}".format(name))


__all__ = [
    "CAPTURE_IMAGE_COUNT",
    "CAPTURE_INTERVAL_SECONDS",
    "CAPTURE_SCHEMA",
    "CAPTURE_TIMEPOINTS",
    "CaptureBusyError",
    "CaptureCancelled",
    "CaptureConfigurationError",
    "CaptureDatasetStore",
    "CaptureRunController",
    "CaptureRunError",
    "CommissionedMotionRunner",
    "CommissionedMotionSession",
    "DatasetSecurityError",
    "MAX_JPEG_BYTES",
    "MotionExecutionEvidence",
    "REQUIRED_ROLES",
    "RUN_DURATION_SECONDS",
]
