"""Camera capture and streaming primitives for the robot workcell.

The concurrency in this module is independent of OpenCV and DepthAI.  Hardware
drivers are imported lazily so a developer can exercise the stream logic on a
machine with neither camera SDK installed.
"""

from __future__ import annotations

import logging
import math
import re
import threading
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Dict, Iterator, Mapping, Optional, Protocol, Sequence

logger = logging.getLogger(__name__)

CAMERA_STATUS_SCHEMA_VERSION = 1
_PUBLIC_CAMERA_IDENTIFIER = re.compile(r"^[A-Za-z0-9_.:+-]{1,255}$")
_PUBLIC_CAMERA_BACKENDS = frozenset({"depthai", "opencv", "unknown"})
_MAX_PUBLIC_SAMPLE_AGE_SECONDS = 86_400.0
_MAX_PUBLIC_COUNTER = 10_000_000_000


class CameraRole(str, Enum):
    """Physical purpose of a camera in the workcell."""

    TOOL = "tool"
    WORKSPACE = "workspace"


class StreamState(str, Enum):
    """Lifecycle/health state reported by :class:`CameraStream`."""

    STOPPED = "stopped"
    STARTING = "starting"
    RUNNING = "running"
    DEGRADED = "degraded"


class CameraError(RuntimeError):
    """Base exception for camera operations."""


class CameraUnavailable(CameraError):
    """Raised when a requested camera stream is unavailable."""


class FrameTimeout(CameraError):
    """Raised when no suitable frame arrives before a deadline."""


class OptionalCameraDependencyError(CameraError):
    """Raised when the selected hardware driver has not been installed."""


@dataclass(frozen=True)
class CapturedImage:
    """Image returned by a camera driver before stream metadata is attached."""

    image: Any
    captured_at_s: Optional[float] = None
    metadata: Mapping[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class CameraFrame:
    """One immutable frame envelope.

    ``image`` is intentionally untyped: production drivers normally provide a
    NumPy BGR array, while tests and remote adapters may provide another image
    representation.
    """

    source_id: str
    role: CameraRole
    sequence: int
    captured_at_s: float
    received_monotonic_s: float
    image: Any
    metadata: Mapping[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class AlignedDepthPacket:
    """One RGB-sequence-bound depth array handed to a bounded local buffer.

    The array is intentionally excluded from ``CameraFrame.metadata`` and from
    every public health payload.  A receiving buffer must take ownership of a
    copy because DepthAI may recycle the SDK backing storage.
    """

    source_id: str
    role: CameraRole
    sequence: int
    captured_at_s: float
    received_monotonic_s: float
    depth_mm: Any = field(repr=False, compare=False)

    def __post_init__(self) -> None:
        if not isinstance(
            self.source_id, str
        ) or not _PUBLIC_CAMERA_IDENTIFIER.fullmatch(self.source_id):
            raise ValueError("aligned depth source_id is invalid")
        object.__setattr__(self, "role", CameraRole(self.role))
        if isinstance(self.sequence, bool) or not isinstance(self.sequence, int):
            raise ValueError("aligned depth sequence must be an integer")
        if self.sequence < 0:
            raise ValueError("aligned depth sequence must be non-negative")
        for name in ("captured_at_s", "received_monotonic_s"):
            value = float(getattr(self, name))
            if not math.isfinite(value):
                raise ValueError("{} must be finite".format(name))
            object.__setattr__(self, name, value)


@dataclass(frozen=True)
class StreamStats:
    state: StreamState
    frames_captured: int
    reconnects: int
    consecutive_errors: int
    last_error: Optional[str]
    last_frame_at_s: Optional[float]


def _public_camera_identifier(value: Any) -> str:
    normalized = str(value or "")
    return normalized if _PUBLIC_CAMERA_IDENTIFIER.fullmatch(normalized) else ""


def _bounded_camera_counter(value: Any) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        return 0
    return min(max(value, 0), _MAX_PUBLIC_COUNTER)


class CameraSource(Protocol):
    """Minimal interface implemented by local and remote camera drivers."""

    source_id: str
    role: CameraRole

    def open(self) -> None:
        """Open the device or network stream."""

    def read(self) -> Optional[CapturedImage]:
        """Return a frame, or ``None`` if a frame is not ready yet."""

    def close(self) -> None:
        """Close the device and unblock an outstanding read where possible."""


class CameraStream:
    """Reconnectable latest-frame stream around one :class:`CameraSource`.

    Consumers do not queue unbounded images: they ask for the newest frame
    after a known sequence number.  This is important on a Raspberry Pi where a
    slow UI or inference worker must not build an ever-growing frame backlog.
    """

    def __init__(
        self,
        source: CameraSource,
        *,
        reconnect_delay_s: float = 0.25,
        idle_delay_s: float = 0.01,
        max_consecutive_empty_reads: int = 100,
        wall_clock: Callable[[], float] = time.time,
        monotonic_clock: Callable[[], float] = time.monotonic,
    ) -> None:
        if reconnect_delay_s < 0 or idle_delay_s < 0:
            raise ValueError("retry delays must be non-negative")
        if max_consecutive_empty_reads < 1:
            raise ValueError("max_consecutive_empty_reads must be positive")
        self.source = source
        self.reconnect_delay_s = reconnect_delay_s
        self.idle_delay_s = idle_delay_s
        self.max_consecutive_empty_reads = max_consecutive_empty_reads
        self._wall_clock = wall_clock
        self._monotonic_clock = monotonic_clock

        self._condition = threading.Condition()
        self._stop_event = threading.Event()
        self._thread: Optional[threading.Thread] = None
        self._latest: Optional[CameraFrame] = None
        self._sequence = 0
        self._state = StreamState.STOPPED
        self._frames_captured = 0
        self._reconnects = 0
        self._consecutive_errors = 0
        self._last_error: Optional[str] = None
        self._last_frame_at_s: Optional[float] = None
        self._mjpeg_cache_lock = threading.Lock()
        self._mjpeg_cache_sequence: Optional[int] = None
        self._mjpeg_cache_encoder: Optional[Callable[[Any], bytes]] = None
        self._mjpeg_cache_bytes: Optional[bytes] = None

    @property
    def role(self) -> CameraRole:
        return self.source.role

    @property
    def source_id(self) -> str:
        return self.source.source_id

    def start(self) -> None:
        """Start capture. Calling this more than once is harmless."""

        with self._condition:
            if self._thread is not None and self._thread.is_alive():
                return
            self._stop_event.clear()
            self._state = StreamState.STARTING
            self._thread = threading.Thread(
                target=self._run,
                name="camera-{}-{}".format(self.role.value, self.source_id),
                daemon=True,
            )
            self._thread.start()
            self._condition.notify_all()

    def stop(self, join_timeout_s: float = 2.0) -> None:
        """Stop capture and release the source.

        ``close`` is invoked before joining so drivers that support it can
        unblock a pending read.
        """

        self._stop_event.set()
        try:
            self.source.close()
        except Exception:
            logger.debug("Error closing camera %s", self.source_id, exc_info=True)
        thread = self._thread
        if thread is not None and thread is not threading.current_thread():
            thread.join(max(0.0, join_timeout_s))
        with self._condition:
            self._state = StreamState.STOPPED
            self._condition.notify_all()

    def latest(self, *, max_age_s: Optional[float] = None) -> Optional[CameraFrame]:
        """Return the latest frame without waiting."""

        with self._condition:
            frame = self._latest
        if frame is None:
            return None
        if max_age_s is not None:
            if max_age_s < 0:
                raise ValueError("max_age_s must be non-negative")
            if self._monotonic_clock() - frame.received_monotonic_s > max_age_s:
                return None
        return frame

    def wait_for_frame(
        self,
        *,
        after_sequence: Optional[int] = None,
        timeout_s: float = 1.0,
        max_age_s: Optional[float] = None,
    ) -> CameraFrame:
        """Wait for a fresh-enough frame newer than ``after_sequence``."""

        if timeout_s < 0:
            raise ValueError("timeout_s must be non-negative")
        if max_age_s is not None and max_age_s < 0:
            raise ValueError("max_age_s must be non-negative")
        deadline = self._monotonic_clock() + timeout_s
        with self._condition:
            while True:
                frame = self._latest
                is_new = frame is not None and (
                    after_sequence is None or frame.sequence > after_sequence
                )
                is_fresh = frame is not None and (
                    max_age_s is None
                    or self._monotonic_clock() - frame.received_monotonic_s <= max_age_s
                )
                if is_new and is_fresh:
                    return frame  # type: ignore[return-value]

                remaining = deadline - self._monotonic_clock()
                if remaining <= 0:
                    detail = self._last_error or "no frame received"
                    raise FrameTimeout(
                        "{} camera timed out: {}".format(self.role.value, detail)
                    )
                self._condition.wait(remaining)

    def frames(
        self, *, timeout_s: float = 2.0, max_age_s: Optional[float] = None
    ) -> Iterator[CameraFrame]:
        """Yield each newest frame once until capture stops or times out."""

        sequence: Optional[int] = None
        while not self._stop_event.is_set():
            frame = self.wait_for_frame(
                after_sequence=sequence,
                timeout_s=timeout_s,
                max_age_s=max_age_s,
            )
            sequence = frame.sequence
            yield frame

    def mjpeg_chunks(
        self,
        *,
        encoder: Optional[Callable[[Any], bytes]] = None,
        boundary: bytes = b"frame",
        timeout_s: float = 2.0,
    ) -> Iterator[bytes]:
        """Yield multipart MJPEG chunks suitable for an HTTP response."""

        encode = encoder or encode_jpeg
        for frame in self.frames(timeout_s=timeout_s):
            # All clients viewing this camera share one encode per captured
            # sequence. The cache lock prevents simultaneous viewers from
            # multiplying JPEG work on the Raspberry Pi.
            with self._mjpeg_cache_lock:
                if (
                    self._mjpeg_cache_sequence == frame.sequence
                    and self._mjpeg_cache_encoder is encode
                    and self._mjpeg_cache_bytes is not None
                ):
                    jpeg = self._mjpeg_cache_bytes
                else:
                    jpeg = encode(frame.image)
                    if not isinstance(jpeg, bytes):
                        raise TypeError("JPEG encoder must return bytes")
                    self._mjpeg_cache_sequence = frame.sequence
                    self._mjpeg_cache_encoder = encode
                    self._mjpeg_cache_bytes = jpeg
            if not isinstance(jpeg, bytes):
                raise TypeError("JPEG encoder must return bytes")
            yield (
                b"--"
                + boundary
                + b"\r\nContent-Type: image/jpeg\r\nContent-Length: "
                + str(len(jpeg)).encode("ascii")
                + b"\r\n\r\n"
                + jpeg
                + b"\r\n"
            )

    def stats(self) -> StreamStats:
        with self._condition:
            return StreamStats(
                state=self._state,
                frames_captured=self._frames_captured,
                reconnects=self._reconnects,
                consecutive_errors=self._consecutive_errors,
                last_error=self._last_error,
                last_frame_at_s=self._last_frame_at_s,
            )

    def public_status(self, *, max_age_s: float) -> Mapping[str, Any]:
        """Return one bounded, versioned, credential-free camera snapshot.

        All mutable stream values are copied under the same condition lock so
        callers never combine a sequence from one frame with freshness or
        counters from another. Device identifiers that do not match the
        closed public grammar are withheld instead of being stringified into a
        diagnostics surface.
        """

        if (
            isinstance(max_age_s, bool)
            or not isinstance(max_age_s, (int, float))
            or not math.isfinite(float(max_age_s))
            or not 0.0 <= float(max_age_s) <= _MAX_PUBLIC_SAMPLE_AGE_SECONDS
        ):
            raise ValueError("max_age_s must be finite and between 0 and 86400")
        now = float(self._monotonic_clock())
        if not math.isfinite(now):
            raise ValueError("monotonic clock returned a non-finite value")
        with self._condition:
            frame = self._latest
            state = self._state
            frames_captured = self._frames_captured
            reconnects = self._reconnects
            consecutive_errors = self._consecutive_errors

        age: Optional[float] = None
        sequence = 0
        fresh = False
        if frame is not None:
            sequence = _bounded_camera_counter(frame.sequence)
            raw_age = now - float(frame.received_monotonic_s)
            if math.isfinite(raw_age) and raw_age >= 0.0:
                age = min(raw_age, _MAX_PUBLIC_SAMPLE_AGE_SECONDS)
                fresh = state is StreamState.RUNNING and raw_age <= float(max_age_s)

        backend = str(getattr(self.source, "backend", "unknown")).lower()
        if backend not in _PUBLIC_CAMERA_BACKENDS:
            backend = "unknown"
        source_id = _public_camera_identifier(self.source_id)
        mxid = ""
        if backend == "depthai":
            mxid = _public_camera_identifier(
                getattr(self.source, "device_id", "") or ""
            )
        return {
            "schema_version": CAMERA_STATUS_SCHEMA_VERSION,
            "role": self.role.value,
            "backend": backend,
            "source_id": source_id,
            "mxid": mxid,
            "state": state.value,
            "fresh": fresh,
            "sample_age_seconds": age,
            "sequence": sequence,
            "frames_captured": _bounded_camera_counter(frames_captured),
            "reconnects": _bounded_camera_counter(reconnects),
            "consecutive_errors": _bounded_camera_counter(consecutive_errors),
        }

    def _run(self) -> None:
        opened_once = False
        while not self._stop_event.is_set():
            try:
                self.source.open()
                if opened_once:
                    with self._condition:
                        self._reconnects += 1
                opened_once = True
                empty_reads = 0
                with self._condition:
                    self._state = StreamState.RUNNING
                    self._consecutive_errors = 0
                    self._last_error = None
                    self._condition.notify_all()

                while not self._stop_event.is_set():
                    captured = self.source.read()
                    if captured is None:
                        empty_reads += 1
                        if empty_reads >= self.max_consecutive_empty_reads:
                            raise CameraUnavailable(
                                "camera returned no frames for {} reads".format(
                                    empty_reads
                                )
                            )
                        self._stop_event.wait(self.idle_delay_s)
                        continue

                    empty_reads = 0
                    now_mono = self._monotonic_clock()
                    captured_at = (
                        self._wall_clock()
                        if captured.captured_at_s is None
                        else float(captured.captured_at_s)
                    )
                    with self._condition:
                        self._sequence += 1
                        metadata = dict(captured.metadata)
                        # This generation is owned by CameraStream, not by the
                        # driver.  A reconnect therefore invalidates any
                        # point-in-time depth evidence even if frame sequence
                        # numbers and the configured source ID still look
                        # plausible.
                        metadata["stream_generation"] = self._reconnects
                        metadata["camera_mxid"] = _public_camera_identifier(
                            getattr(self.source, "device_id", "") or ""
                        )
                        self._latest = CameraFrame(
                            source_id=self.source_id,
                            role=self.role,
                            sequence=self._sequence,
                            captured_at_s=captured_at,
                            received_monotonic_s=now_mono,
                            image=captured.image,
                            metadata=metadata,
                        )
                        self._frames_captured += 1
                        self._consecutive_errors = 0
                        self._last_error = None
                        self._last_frame_at_s = captured_at
                        self._condition.notify_all()
            except Exception as exc:
                if not self._stop_event.is_set():
                    with self._condition:
                        self._state = StreamState.DEGRADED
                        self._consecutive_errors += 1
                        self._last_error = "{}: {}".format(type(exc).__name__, str(exc))
                        self._condition.notify_all()
                    logger.warning(
                        "Camera %s (%s) failed; reconnecting: %s",
                        self.source_id,
                        self.role.value,
                        exc,
                    )
            finally:
                try:
                    self.source.close()
                except Exception:
                    logger.debug(
                        "Error releasing camera %s", self.source_id, exc_info=True
                    )

            if not self._stop_event.is_set():
                self._stop_event.wait(self.reconnect_delay_s)

        with self._condition:
            self._state = StreamState.STOPPED
            self._condition.notify_all()


class DualCameraRig:
    """Named access to tool-mounted and static workspace camera streams."""

    REQUIRED_ROLES = (CameraRole.TOOL, CameraRole.WORKSPACE)

    def __init__(self, streams: Sequence[CameraStream]) -> None:
        by_role: Dict[CameraRole, CameraStream] = {}
        for stream in streams:
            if stream.role in by_role:
                raise ValueError("duplicate camera role: {}".format(stream.role.value))
            by_role[stream.role] = stream
        missing = [role.value for role in self.REQUIRED_ROLES if role not in by_role]
        if missing:
            raise ValueError("missing camera roles: {}".format(", ".join(missing)))
        self._streams = by_role

    def stream(self, role: CameraRole) -> CameraStream:
        try:
            return self._streams[CameraRole(role)]
        except (KeyError, ValueError) as exc:
            raise CameraUnavailable("unknown camera role: {!r}".format(role)) from exc

    def start(self) -> None:
        for stream in self._streams.values():
            stream.start()

    def stop(self) -> None:
        for stream in self._streams.values():
            stream.stop()

    def __enter__(self) -> "DualCameraRig":
        self.start()
        return self

    def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
        self.stop()

    def capture_pair(
        self,
        *,
        after_sequences: Optional[Mapping[Any, int]] = None,
        timeout_s: float = 2.0,
        max_age_s: float = 0.5,
        max_skew_s: float = 0.15,
    ) -> Mapping[CameraRole, CameraFrame]:
        """Return a recent, close-timestamp pair newer than role checkpoints."""

        if timeout_s < 0 or max_age_s < 0 or max_skew_s < 0:
            raise ValueError("capture timing limits must be non-negative")
        checkpoints: Dict[CameraRole, int] = {}
        for raw_role, raw_sequence in dict(after_sequences or {}).items():
            try:
                role = CameraRole(getattr(raw_role, "value", raw_role))
            except (TypeError, ValueError) as exc:
                raise ValueError("capture checkpoint has an unknown role") from exc
            if (
                isinstance(raw_sequence, bool)
                or not isinstance(raw_sequence, int)
                or raw_sequence < 0
            ):
                raise ValueError(
                    "capture checkpoint sequences must be non-negative integers"
                )
            if role in checkpoints:
                raise ValueError("capture checkpoint contains a duplicate role")
            checkpoints[role] = raw_sequence
        if after_sequences is not None and set(checkpoints) not in (
            set(),
            set(self.REQUIRED_ROLES),
        ):
            raise ValueError(
                "capture checkpoints must be empty or contain both camera roles"
            )
        if after_sequences is not None:
            # A scheduled sample must be newer than both the prior dataset
            # sample and the newest frame already present at this timepoint.
            # This prevents accepting an image captured just before the
            # one-second boundary merely because it is still fresh.
            for role in self.REQUIRED_ROLES:
                latest = self.stream(role).latest(max_age_s=max_age_s)
                if latest is not None:
                    checkpoints[role] = max(
                        checkpoints.get(role, 0),
                        int(latest.sequence),
                    )
        deadline = time.monotonic() + timeout_s
        tool = self.stream(CameraRole.TOOL).wait_for_frame(
            after_sequence=checkpoints.get(CameraRole.TOOL),
            timeout_s=timeout_s,
            max_age_s=max_age_s,
        )
        remaining = max(0.0, deadline - time.monotonic())
        workspace = self.stream(CameraRole.WORKSPACE).wait_for_frame(
            after_sequence=checkpoints.get(CameraRole.WORKSPACE),
            timeout_s=remaining,
            max_age_s=max_age_s,
        )

        while abs(tool.captured_at_s - workspace.captured_at_s) > max_skew_s:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise FrameTimeout(
                    "camera pair did not converge within {:.3f}s skew".format(
                        max_skew_s
                    )
                )
            if tool.captured_at_s < workspace.captured_at_s:
                tool = self.stream(CameraRole.TOOL).wait_for_frame(
                    after_sequence=tool.sequence,
                    timeout_s=remaining,
                    max_age_s=max_age_s,
                )
            else:
                workspace = self.stream(CameraRole.WORKSPACE).wait_for_frame(
                    after_sequence=workspace.sequence,
                    timeout_s=remaining,
                    max_age_s=max_age_s,
                )
        return {CameraRole.TOOL: tool, CameraRole.WORKSPACE: workspace}

    def capture_pair_for_run(
        self,
        *,
        after_sequences: Mapping[str, int],
        timeout_seconds: float,
        max_age_seconds: float,
        max_skew_seconds: float,
    ) -> Mapping[CameraRole, CameraFrame]:
        """Adapt the rig to :class:`CaptureRunController`'s pair contract."""

        return self.capture_pair(
            after_sequences=after_sequences,
            timeout_s=timeout_seconds,
            max_age_s=max_age_seconds,
            max_skew_s=max_skew_seconds,
        )

    def health(self) -> Mapping[CameraRole, StreamStats]:
        return {role: stream.stats() for role, stream in self._streams.items()}


def encode_jpeg(image: Any, quality: int = 85) -> bytes:
    """Encode a BGR image as JPEG, importing OpenCV only when needed."""

    try:
        import cv2
    except ImportError as exc:  # pragma: no cover - depends on deployment
        raise OptionalCameraDependencyError(
            "OpenCV is required for JPEG encoding"
        ) from exc
    ok, buffer = cv2.imencode(
        ".jpg", image, [int(cv2.IMWRITE_JPEG_QUALITY), int(quality)]
    )
    if not ok:
        raise CameraError("OpenCV failed to encode JPEG")
    return buffer.tobytes()


class OpenCVCameraSource:
    """USB, RTSP, or file camera source backed by ``cv2.VideoCapture``."""

    def __init__(
        self,
        source: Any,
        *,
        source_id: str,
        role: CameraRole,
        width: Optional[int] = None,
        height: Optional[int] = None,
        fps: Optional[float] = None,
        buffer_size: int = 1,
        api_preference: Optional[int] = None,
    ) -> None:
        self.backend = "opencv"
        self.source = source
        self.source_id = source_id
        self.role = CameraRole(role)
        self.width = width
        self.height = height
        self.fps = fps
        self.buffer_size = buffer_size
        self.api_preference = api_preference
        self._capture: Any = None
        self._lock = threading.Lock()

    def open(self) -> None:
        try:
            import cv2
        except ImportError as exc:  # pragma: no cover - deployment only
            raise OptionalCameraDependencyError(
                "OpenCV is required for OpenCVCameraSource"
            ) from exc

        self.close()
        capture = (
            cv2.VideoCapture(self.source)
            if self.api_preference is None
            else cv2.VideoCapture(self.source, self.api_preference)
        )
        if not capture.isOpened():
            capture.release()
            raise CameraUnavailable("could not open camera {!r}".format(self.source))
        if self.width is not None:
            capture.set(cv2.CAP_PROP_FRAME_WIDTH, float(self.width))
        if self.height is not None:
            capture.set(cv2.CAP_PROP_FRAME_HEIGHT, float(self.height))
        if self.fps is not None:
            capture.set(cv2.CAP_PROP_FPS, float(self.fps))
        capture.set(cv2.CAP_PROP_BUFFERSIZE, float(self.buffer_size))
        with self._lock:
            self._capture = capture

    def read(self) -> Optional[CapturedImage]:
        with self._lock:
            capture = self._capture
        if capture is None:
            raise CameraUnavailable("camera is not open")
        ok, image = capture.read()
        if not ok or image is None:
            raise CameraUnavailable("OpenCV returned an empty frame")
        height, width = image.shape[:2]
        return CapturedImage(
            image=image,
            metadata={"width": int(width), "height": int(height), "format": "bgr"},
        )

    def close(self) -> None:
        with self._lock:
            capture, self._capture = self._capture, None
        if capture is not None:
            capture.release()


class DepthAICameraSource:
    """Luxonis RGB source used for the tool-mounted camera.

    When aligned depth is enabled, RGB and depth packets are matched by their
    DepthAI sequence number before either is exposed.  The callback receives
    the same wall-clock timestamp that is attached to the RGB
    :class:`CapturedImage`, which lets the eye-in-hand mapper resolve the
    corresponding robot pose without leaking a large depth ndarray into frame
    metadata or telemetry.
    """

    def __init__(
        self,
        *,
        source_id: str = "tool-oak",
        role: CameraRole = CameraRole.TOOL,
        width: int = 1280,
        height: int = 720,
        fps: Optional[float] = None,
        queue_size: int = 2,
        device_id: Optional[str] = None,
        aligned_depth_callback: Optional[Callable[[float, Any], None]] = None,
        aligned_depth_packet_callback: Optional[
            Callable[[AlignedDepthPacket], None]
        ] = None,
        enable_aligned_depth: bool = False,
        wall_clock: Callable[[], float] = time.time,
        monotonic_clock: Callable[[], float] = time.monotonic,
    ) -> None:
        if queue_size < 1:
            raise ValueError("queue_size must be positive")
        if fps is not None and float(fps) <= 0:
            raise ValueError("fps must be positive")
        if (
            enable_aligned_depth
            and aligned_depth_callback is None
            and aligned_depth_packet_callback is None
        ):
            raise ValueError(
                "aligned_depth_callback or aligned_depth_packet_callback is required "
                "when depth is enabled"
            )
        self.backend = "depthai"
        self.source_id = source_id
        self.role = CameraRole(role)
        self.width = width
        self.height = height
        self.fps = float(fps) if fps is not None else None
        self.queue_size = queue_size
        self.device_id = device_id
        self.aligned_depth_callback = aligned_depth_callback
        self.aligned_depth_packet_callback = aligned_depth_packet_callback
        self.enable_aligned_depth = bool(enable_aligned_depth)
        self.wall_clock = wall_clock
        self.monotonic_clock = monotonic_clock
        self._device: Any = None
        self._queue: Any = None
        self._depth_queue: Any = None
        self._pending_rgb: Dict[int, Any] = {}
        self._pending_depth: Dict[int, Any] = {}
        self._timestamp_offset_s: Optional[float] = None
        self._lock = threading.Lock()

    def open(self) -> None:
        try:
            import depthai as dai
        except ImportError as exc:  # pragma: no cover - deployment only
            raise OptionalCameraDependencyError(
                "depthai is required for DepthAICameraSource"
            ) from exc

        self.close()
        pipeline = dai.Pipeline()
        camera = pipeline.createColorCamera()
        output = pipeline.createXLinkOut()
        output.setStreamName("rgb")
        camera.setPreviewSize(self.width, self.height)
        if self.fps is not None:
            camera.setFps(self.fps)
        camera.setInterleaved(False)
        camera.setColorOrder(dai.ColorCameraProperties.ColorOrder.BGR)
        camera.preview.link(output.input)

        if self.enable_aligned_depth:
            left = pipeline.createMonoCamera()
            right = pipeline.createMonoCamera()
            stereo = pipeline.createStereoDepth()
            depth_output = pipeline.createXLinkOut()
            depth_output.setStreamName("depth")
            left.setResolution(dai.MonoCameraProperties.SensorResolution.THE_400_P)
            right.setResolution(dai.MonoCameraProperties.SensorResolution.THE_400_P)
            left.setBoardSocket(dai.CameraBoardSocket.LEFT)
            right.setBoardSocket(dai.CameraBoardSocket.RIGHT)
            stereo.setDepthAlign(dai.CameraBoardSocket.RGB)
            stereo.setOutputSize(self.width, self.height)
            left.out.link(stereo.left)
            right.out.link(stereo.right)
            stereo.depth.link(depth_output.input)

        if self.device_id:
            device_info = dai.DeviceInfo(self.device_id)
            device = dai.Device(pipeline, device_info)
        else:
            device = dai.Device(pipeline)
        queue = device.getOutputQueue(
            name="rgb", maxSize=self.queue_size, blocking=False
        )
        depth_queue = None
        if self.enable_aligned_depth:
            depth_queue = device.getOutputQueue(
                name="depth", maxSize=self.queue_size, blocking=False
            )
        with self._lock:
            self._device = device
            self._queue = queue
            self._depth_queue = depth_queue
            self._pending_rgb.clear()
            self._pending_depth.clear()
            self._timestamp_offset_s = None

    def read(self) -> Optional[CapturedImage]:
        with self._lock:
            queue = self._queue
            depth_queue = self._depth_queue
        if queue is None:
            raise CameraUnavailable("DepthAI camera is not open")

        if not self.enable_aligned_depth:
            packet = queue.tryGet()
            if packet is None:
                return None
            return self._captured_rgb(packet)

        if depth_queue is None:
            raise CameraUnavailable("DepthAI aligned-depth queue is not open")
        rgb_packet = queue.tryGet()
        depth_packet = depth_queue.tryGet()
        with self._lock:
            if rgb_packet is not None:
                self._pending_rgb[self._packet_sequence(rgb_packet)] = rgb_packet
            if depth_packet is not None:
                self._pending_depth[self._packet_sequence(depth_packet)] = depth_packet
            common = set(self._pending_rgb).intersection(self._pending_depth)
            if not common:
                self._trim_pending_packets()
                return None
            sequence = min(common)
            packet = self._pending_rgb.pop(sequence)
            matching_depth = self._pending_depth.pop(sequence)
            self._discard_older_packets(sequence)

        image = packet.getCvFrame()
        if image is None:
            return None
        depth_mm = matching_depth.getFrame()
        if depth_mm is None:
            return None
        captured_at = self._capture_wall_timestamp(packet)
        callback = self.aligned_depth_callback
        packet_callback = self.aligned_depth_packet_callback
        if callback is None and packet_callback is None:
            # Defensive: constructor validation should prevent this.
            raise CameraUnavailable("DepthAI aligned-depth callback is unavailable")
        if packet_callback is not None:
            packet_callback(
                AlignedDepthPacket(
                    source_id=self.source_id,
                    role=self.role,
                    sequence=sequence,
                    captured_at_s=captured_at,
                    received_monotonic_s=float(self.monotonic_clock()),
                    depth_mm=depth_mm,
                )
            )
        if callback is not None:
            callback(captured_at, depth_mm)
        return CapturedImage(
            image=image,
            captured_at_s=captured_at,
            metadata={
                "width": self.width,
                "height": self.height,
                "format": "bgr",
                "depth_aligned": True,
                "depth_sequence": sequence,
            },
        )

    def _captured_rgb(self, packet: Any) -> Optional[CapturedImage]:
        image = packet.getCvFrame()
        if image is None:
            return None
        return CapturedImage(
            image=image,
            metadata={
                "width": self.width,
                "height": self.height,
                "format": "bgr",
            },
        )

    @staticmethod
    def _packet_sequence(packet: Any) -> int:
        try:
            return int(packet.getSequenceNum())
        except (AttributeError, TypeError, ValueError) as exc:
            raise CameraUnavailable(
                "DepthAI packet has no valid sequence number"
            ) from exc

    def _capture_wall_timestamp(self, packet: Any) -> float:
        """Convert DepthAI's host-synchronized steady timestamp to wall time."""

        wall_now = float(self.wall_clock())
        try:
            timestamp = packet.getTimestamp()
            packet_seconds = (
                float(timestamp.total_seconds())
                if hasattr(timestamp, "total_seconds")
                else float(timestamp)
            )
        except (AttributeError, TypeError, ValueError, OverflowError):
            return wall_now

        monotonic_now = float(self.monotonic_clock())
        if abs(monotonic_now - packet_seconds) <= 3600.0:
            # DepthAI getTimestamp() is synchronized to the host steady clock.
            return wall_now - (monotonic_now - packet_seconds)

        # Retain a stable offset for SDK/platform builds exposing another
        # monotonic epoch. Re-anchor only after a discontinuity.
        with self._lock:
            offset = self._timestamp_offset_s
            if offset is None or abs((packet_seconds + offset) - wall_now) > 5.0:
                offset = wall_now - packet_seconds
                self._timestamp_offset_s = offset
        return packet_seconds + offset

    def _trim_pending_packets(self) -> None:
        limit = max(2, self.queue_size * 2)
        for pending in (self._pending_rgb, self._pending_depth):
            for sequence in sorted(pending)[:-limit]:
                pending.pop(sequence, None)

    def _discard_older_packets(self, matched_sequence: int) -> None:
        for pending in (self._pending_rgb, self._pending_depth):
            for sequence in tuple(pending):
                if sequence <= matched_sequence:
                    pending.pop(sequence, None)
        self._trim_pending_packets()

    def close(self) -> None:
        with self._lock:
            device = self._device
            self._device = None
            self._queue = None
            self._depth_queue = None
            self._pending_rgb.clear()
            self._pending_depth.clear()
            self._timestamp_offset_s = None
        if device is not None:
            device.close()
