gps-denied-onboard/src/gps_denied_onboard/frame_source/video_file.py

"""``VideoFileFrameSource`` — replay-only file decoder (AZ-398).

Streams an MP4 / MKV / AVI file frame-by-frame via OpenCV's
:class:`cv2.VideoCapture`. Each emitted :class:`NavCameraFrame`
carries:

- ``frame_id`` — a strictly-increasing counter starting at 0.
- ``timestamp`` — UTC wall-clock at decode time (from the injected
  :class:`Clock`); the file's own pts is NOT used for this field
  because replay deterministically remaps it.
- ``image`` — the decoded BGR ``numpy.ndarray`` (OpenCV native order).
- ``metadata["monotonic_ns"]`` — the injected :class:`Clock`'s
  ``monotonic_ns()`` at decode time. This is the value AC-2 asserts
  non-decreasing.
- ``metadata["source_pts_ns"]`` — the file's per-frame PTS in ns (the
  ``CAP_PROP_POS_MSEC`` reading × 1e6) for downstream determinism.

Gated by ``BUILD_VIDEO_FILE_FRAME_SOURCE`` (Invariant 9). The check is
performed at constructor entry — a Tier-0 build that imports this
module by accident still raises ``FrameSourceConfigError`` cleanly
without attempting an OpenCV import.
"""

from __future__ import annotations

import logging
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING, Any

from gps_denied_onboard.frame_source.errors import (
    FrameSourceConfigError,
    FrameSourceError,
)

if TYPE_CHECKING:
    from gps_denied_onboard._types.nav import NavCameraFrame
    from gps_denied_onboard.clock import Clock


_BUILD_FLAG = "BUILD_VIDEO_FILE_FRAME_SOURCE"

_logger = logging.getLogger(__name__)


def _build_flag_on() -> bool:
    """``ON`` / ``1`` / ``true`` / ``yes`` (case-insensitive) → ``True``."""
    raw = os.environ.get(_BUILD_FLAG, "")
    return raw.strip().lower() in {"on", "1", "true", "yes"}


class VideoFileFrameSource:
    """Replay :class:`FrameSource` strategy backed by ``cv2.VideoCapture``.

    Stream-decodes a video file; per-frame decode is amortised by
    OpenCV's internal buffer. The strategy preserves the file's frame
    order — there is no seek, no random-access path; this keeps
    replay deterministic (Invariant 10).

    Constructor parameters:

    - ``path`` — filesystem path to an MP4/MKV/AVI (existence checked
      at construction).
    - ``camera_calibration_id`` — string identifier propagated into
      every emitted :class:`NavCameraFrame` so downstream consumers
      (C1, C3, C4) load the correct intrinsics for the recording.
    - ``clock`` — injected :class:`Clock`; the strategy reads
      ``clock.monotonic_ns()`` per emitted frame for the
      ``metadata["monotonic_ns"]`` ordering field.
    """

    __slots__ = (
        "_path",
        "_camera_calibration_id",
        "_clock",
        "_capture",
        "_frame_counter",
        "_last_monotonic_ns",
        "_closed",
        "_eos_returned",
    )

    def __init__(
        self,
        *,
        path: Path | str,
        camera_calibration_id: str,
        clock: "Clock",
    ) -> None:
        if not _build_flag_on():
            raise FrameSourceConfigError(
                f"{_BUILD_FLAG} is OFF in this binary; "
                "VideoFileFrameSource is unavailable. Rebuild with the "
                "flag set to ON in the replay binary's Dockerfile."
            )
        resolved = Path(path)
        if not resolved.is_file():
            raise FrameSourceConfigError(
                f"VideoFileFrameSource: path {resolved!s} does not exist "
                "or is not a regular file."
            )
        try:
            import cv2 as _cv2
        except ImportError as exc:
            raise FrameSourceConfigError(
                "VideoFileFrameSource requires opencv-python; not "
                "importable in this binary."
            ) from exc
        capture = _cv2.VideoCapture(str(resolved))
        if not capture.isOpened():
            capture.release()
            raise FrameSourceConfigError(
                f"VideoFileFrameSource: cv2.VideoCapture could not open "
                f"{resolved!s} (unsupported codec or corrupt header)."
            )
        self._path = resolved
        self._camera_calibration_id = camera_calibration_id
        self._clock = clock
        self._capture = capture
        self._frame_counter = 0
        self._last_monotonic_ns = -1
        self._closed = False
        self._eos_returned = False

    def next_frame(self) -> "NavCameraFrame | None":
        from gps_denied_onboard._types.nav import NavCameraFrame

        if self._closed or self._eos_returned:
            return None
        try:
            import cv2 as _cv2
        except ImportError as exc:  # pragma: no cover — established at __init__.
            raise FrameSourceError(
                "VideoFileFrameSource: opencv-python disappeared between "
                "construction and next_frame()"
            ) from exc
        ok, image = self._capture.read()
        if not ok:
            self._eos_returned = True
            return None
        if image is None:
            # OpenCV's read() returning ok=True with image=None signals a
            # decoder-internal failure for the current frame; treat as a
            # transient error per Invariant 4 rather than silently
            # advancing.
            raise FrameSourceError(
                f"VideoFileFrameSource: video decode failed at frame "
                f"{self._frame_counter} (cv2.VideoCapture.read returned "
                "ok=True with image=None)"
            )
        monotonic_ns = self._clock.monotonic_ns()
        if monotonic_ns < self._last_monotonic_ns:
            raise FrameSourceError(
                f"VideoFileFrameSource: clock went backwards at frame "
                f"{self._frame_counter}: {monotonic_ns} ns followed "
                f"{self._last_monotonic_ns} ns (Invariant 3)"
            )
        pos_msec = float(self._capture.get(_cv2.CAP_PROP_POS_MSEC))
        source_pts_ns = int(pos_msec * 1_000_000)
        timestamp = datetime.fromtimestamp(
            self._clock.time_ns() / 1e9, tz=timezone.utc
        )
        metadata: dict[str, Any] = {
            "monotonic_ns": monotonic_ns,
            "source_pts_ns": source_pts_ns,
            "source": "video_file",
        }
        frame = NavCameraFrame(
            frame_id=self._frame_counter,
            timestamp=timestamp,
            image=image,
            camera_calibration_id=self._camera_calibration_id,
            metadata=metadata,
        )
        self._frame_counter += 1
        self._last_monotonic_ns = monotonic_ns
        return frame

    def close(self) -> None:
        if self._closed:
            _logger.debug(
                "VideoFileFrameSource(%s) close called twice; no-op",
                self._path,
            )
            return
        self._closed = True
        try:
            self._capture.release()
        except Exception:  # pragma: no cover — defensive.
            # cv2.VideoCapture.release should never raise; if it does on
            # an exotic backend, we still want to flag the source as
            # closed so a second close() stays a no-op.
            _logger.exception(
                "VideoFileFrameSource(%s) cv2.release() raised", self._path
            )


__all__ = ["VideoFileFrameSource"]