video

class Video(FileScalar, name="edu.cornell.video", extensions=("mov", "mp4")):
    """A video leaf; a loaded ``Video`` is its file ``Path``.

    Plus the methods below, which read ``self`` and extract metadata or
    frames via OpenCV.
    """

    __slots__ = ()

    def kind(self) -> str:
        """This video's format: 'mov', 'mp4', or 'unknown'."""
        suffix = self.suffix.lower()
        if suffix == ".mov":
            return "mov"
        if suffix == ".mp4":
            return "mp4"
        return "unknown"

    def metadata(self) -> dict[str, Any]:
        """Read width, height, fps, frame_count, and duration via OpenCV."""
        import cv2

        cap = cv2.VideoCapture(str(self))
        try:
            if not cap.isOpened():
                raise OSError(f"Could not open video: {self}")
            width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
            height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
            fps = cap.get(cv2.CAP_PROP_FPS)
            frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
            duration = frame_count / fps if fps > 0 else 0
            return {
                "width": width,
                "height": height,
                "fps": fps,
                "frame_count": frame_count,
                "duration": duration,
            }
        finally:
            cap.release()

    def intrinsics(self) -> CameraIntrinsics | None:
        """The whole-clip camera intrinsics baked into the movie, or None if absent.

        The recorder's asset-level snapshot: one representative reading for the clip.
        For the reading at each frame, use ``frame_intrinsics``. Spec:
        ``docs/dsl/movie-metadata-format.md``.
        """
        import av

        with av.open(str(self)) as container:
            payload = container.metadata.get(_INTRINSICS_METADATA_KEY)
        return CameraIntrinsics.from_json(payload) if payload is not None else None

    def frame_intrinsics(self) -> list[tuple[float, CameraIntrinsics]]:
        """Per-frame ``(time_seconds, intrinsics)`` from the timed metadata track.

        One entry per video frame the recorder tagged, in presentation order — the true
        focal length and principal point at *every* frame, not just the whole-clip
        average from ``intrinsics``. Empty when the movie carries no per-frame track.
        """
        import av

        readings: list[tuple[float, CameraIntrinsics]] = []
        with av.open(str(self)) as container:
            tracks = [stream for stream in container.streams if stream.type == "data"]
            if not tracks:
                return readings
            track = tracks[0]
            for packet in container.demux(track):
                sample = bytes(packet)
                if not sample or packet.pts is None:
                    continue
                time = float(packet.pts * track.time_base)
                for value in _iter_timed_metadata_values(sample):
                    readings.append((time, CameraIntrinsics.from_json(value)))
        return readings

    def frame_at_index(self, index: int) -> "Image.Image":
        """Decode the ``index``-th video frame as a PIL image — exact under VFR.

        Frames are counted in presentation order via PyAV rather than estimated
        from an average fps, so this is the reader for contracts that pair the
        i-th frame with external per-frame records (``PosedVideo``'s pose and
        depth sidecars). For a cheap approximate seek (thumbnails), use
        ``frame_at``.

        Raises:
            IndexError: If ``index`` is negative or past the last frame.
        """
        import av

        if index < 0:
            raise IndexError(f"frame index {index} is negative")
        with av.open(str(self)) as container:
            stream = container.streams.video[0]
            for position, frame in enumerate(container.decode(stream)):
                if position == index:
                    return frame.to_image()
        raise IndexError(f"{self} has no frame {index}")

    def frame_at(self, t: float = 0.0) -> "Image.Image":
        """Extract the frame nearest ``t`` seconds as a PIL image (approximate).

        Seeks by ``t × average fps``, which drifts on variable-frame-rate clips —
        fine for thumbnails; use ``frame_at_index`` for exact per-frame reads.
        """
        import cv2
        from PIL import Image as PILImage

        cap = cv2.VideoCapture(str(self))
        try:
            if not cap.isOpened():
                raise OSError(f"Could not open video: {self}")
            fps = cap.get(cv2.CAP_PROP_FPS)
            cap.set(cv2.CAP_PROP_POS_FRAMES, int(t * fps))
            ret, frame = cap.read()
            if not ret:
                raise OSError(f"Could not read frame at {t}s from {self}")
            return PILImage.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
        finally:
            cap.release()

    def thumbnail(
        self,
        max_axis: int = 256,
        t: float = 0.0,
    ) -> "Image.Image":
        """A PIL thumbnail of the frame at ``t`` seconds.

        Resized so its largest dimension is ``max_axis``.
        """
        return resize(self.frame_at(t), max_axis)

    def tooltip(
        self,
        max_axis: int = 64,
        t: float = 0.0,
    ) -> str:
        """A base64 PNG data URL for the frame at ``t`` seconds (Altair/Vega tooltips).

        Resized so its largest dimension is ``max_axis``.
        """
        return data_url_png(self.thumbnail(max_axis, t))