VideoFile

def get_fragment(self, start: float, end: float) -> "VideoFragment":
    """
    Returns a video fragment from the specified time range.

    Args:
        start (float): The start time of the fragment in seconds.
        end (float): The end time of the fragment in seconds.

    Returns:
        VideoFragment: A Model representing the video fragment.
    """
    if start < 0 or end < 0 or start >= end:
        raise ValueError(
            f"Can't get video fragment for '{self.path}', "
            f"invalid time range: ({start:.3f}, {end:.3f})"
        )

    return VideoFragment(video=self, start=start, end=end)

def get_fragments(
    self,
    duration: float,
    start: float = 0,
    end: float | None = None,
) -> "Iterator[VideoFragment]":
    """
    Splits the video into multiple fragments of a specified duration.

    Args:
        duration (float): The duration of each video fragment in seconds.
        start (float): The starting time in seconds (default: 0).
        end (float, optional): The ending time in seconds. If None, the entire
                               remaining video is processed (default: None).

    Returns:
        Iterator[VideoFragment]: An iterator yielding video fragments.

    Note:
        If end is not specified, duration will be taken from the video file,
        which means video metadata needs to be read.
    """
    if duration <= 0:
        raise ValueError("duration must be a positive float")
    if start < 0:
        raise ValueError("start must be a non-negative float")

    if end is None:
        end = self.get_info().duration

    if end < 0:
        raise ValueError("end must be a non-negative float")
    if start >= end:
        raise ValueError("start must be less than end")

    while start < end:
        yield self.get_fragment(start, min(start + duration, end))
        start += duration

def get_frame(self, frame: int, video_stream_index: int = 0) -> "VideoFrame":
    """
    Returns a specific video frame by its frame number.

    This returns a frame reference without decoding or validating that the
    frame exists. The returned timestamp is estimated from FPS metadata.
    Pixel access methods decode the requested frame; use ``get_frames()``
    for sequential access and decoded frame timestamps when available.

    Args:
        frame (int): The frame number to read.
        video_stream_index: Zero-based index among video streams to read.
            Defaults to 0.

    Returns:
        VideoFrame: Video frame model.
    """
    if frame < 0:
        raise ValueError("frame must be a non-negative integer")

    from .video import video_frame

    return video_frame(self, frame, video_stream_index=video_stream_index)

def get_frames(
    self,
    start: int = 0,
    end: int | None = None,
    step: int = 1,
    video_stream_index: int = 0,
) -> "Iterator[VideoFrame]":
    """
    Returns video frames from the specified range in the video.

    Args:
        start (int): The starting frame number (default: 0).
        end (int, optional): The ending frame number (exclusive). If None,
                             frames are read until the end of the video
                             (default: None).
        step (int): The interval between frames to read (default: 1).
        video_stream_index: Zero-based index among video streams to read.
            Defaults to 0.

    Returns:
        Iterator[VideoFrame]: An iterator yielding video frames.

    Note:
        If end is not specified, number of frames will be taken from the video file,
        this means video metadata needs to be read.
    """
    from .video import validate_frame_range, video_frames

    start, end, step = validate_frame_range(
        self, start, end, step, video_stream_index=video_stream_index
    )

    yield from video_frames(
        self, start, end, step, video_stream_index=video_stream_index
    )

def get_info(self, video_stream_index: int = 0) -> "Video":
    """
    Retrieves metadata and information about the video file.

    With prefetch and cache off, reads only the parts of the file it needs.

    Args:
        video_stream_index: Zero-based index among video streams to inspect.
            Defaults to 0.

    Returns:
        Video: A Model containing video metadata such as duration,
               resolution, frame rate, and codec details.
    """
    from .video import video_info

    if video_stream_index not in self._video_info_cache:
        self._video_info_cache[video_stream_index] = video_info(
            self, video_stream_index=video_stream_index
        )
    return self._video_info_cache[video_stream_index]

def save(  # type: ignore[override]
    self,
    destination: str,
    client_config: dict | None = None,
    *,
    content_type: str | None = None,
    content_disposition: str | None = None,
    cache_control: str | None = None,
    content_encoding: str | None = None,
    metadata: dict[str, str] | None = None,
    write_options: dict[str, Any] | None = None,
) -> "VideoFile":
    """Writes its content to destination"""
    result = super().save(
        destination,
        client_config=client_config,
        content_type=content_type,
        content_disposition=content_disposition,
        cache_control=cache_control,
        content_encoding=content_encoding,
        metadata=metadata,
        write_options=write_options,
    )
    vf = VideoFile(**result.model_dump())
    vf._set_stream(self._catalog)
    return vf

def get_np(self) -> "ndarray":
    """
    Returns a video frame from the video file as a NumPy array.

    The decoded ``av.VideoFrame`` is cached in ``_decoded`` on the first
    call, so repeated calls to ``get_np()`` return the cached pixels
    without a second decode pass regardless of whether the frame came
    from ``VideoFile.get_frames()`` or ``VideoFile.get_frame()``.

    Display rotation is applied (matching FFmpeg/OpenCV), so a 90/270
    rotation swaps width and height. Channels are RGB; consumers expecting
    BGR (e.g. OpenCV, Ultralytics) must convert.

    Returns:
        ndarray: A NumPy array representing the video frame,
                 in the shape (height, width, channels).
    """
    from .video import _decode_video_frame, _frame_to_ndarray

    if self._decoded is None:
        self._decoded, _ = _decode_video_frame(
            self.video,
            self.frame,
            self.video_stream_index,
        )
    return _frame_to_ndarray(self._decoded)

def read_bytes(self, format: str = "jpg") -> bytes:
    """
    Returns a video frame from the video file as image bytes.

    Uses the cached decoded frame when available (e.g. after ``get_np()``
    or for frames yielded by ``VideoFile.get_frames()``), avoiding a second
    decode pass.

    Args:
        format (str): The desired image format (e.g., 'jpg', 'png').
                      Defaults to 'jpg'.

    Returns:
        bytes: The encoded video frame as image bytes.
    """
    from PIL import Image as PilImage

    from .video import _image_format

    buf = BytesIO()
    PilImage.fromarray(self.get_np()).save(buf, format=_image_format(format))
    return buf.getvalue()

def save(
    self,
    destination: str | os.PathLike[str],
    format: str = "jpg",
    client_config: dict | None = None,
    *,
    content_type: str | None = None,
    content_disposition: str | None = None,
    cache_control: str | None = None,
    content_encoding: str | None = None,
    metadata: dict[str, str] | None = None,
    write_options: dict[str, Any] | None = None,
) -> "ImageFile":
    """
    Saves the current video frame as an image file.

    Uses the cached decoded frame when available (e.g. after ``get_np()``
    or for frames yielded by ``VideoFile.get_frames()``), avoiding a second
    decode pass.

    If ``destination`` is a remote path, the image file will be uploaded
    to remote storage.

    Args:
        destination: Output directory path or URI (e.g. ``s3://…``, ``gs://…``).
        format: Image format (e.g., 'jpg', 'png'). Defaults to 'jpg'.
        client_config: Optional client configuration (e.g. credentials).
        content_type: ``Content-Type`` to set on the written object.
        content_disposition: ``Content-Disposition`` to set on the object.
        cache_control: ``Cache-Control`` to set on the object.
        content_encoding: ``Content-Encoding`` to set on the object.
        metadata: Custom key/value metadata to attach to the object.
        write_options: Raw, backend-native write kwargs merged after the
            normalized fields above.

    Returns:
        ImageFile: A Model representing the saved image file.
    """
    catalog = self.video._catalog
    if catalog is None:
        raise RuntimeError("Cannot save video frame: catalog is not set")

    destination = stringify_path(destination)
    image_bytes = self.read_bytes(format)
    extension = format.removeprefix(".")
    output_file = posixpath.join(
        destination, f"{self.video.get_file_stem()}_{self.frame:04d}.{extension}"
    )
    client, rel_path = self.video._resolve_destination(output_file, client_config)
    result = client.upload(
        image_bytes,
        rel_path,
        write_config=WriteConfig(
            content_type=content_type,
            content_disposition=content_disposition,
            cache_control=cache_control,
            content_encoding=content_encoding,
            metadata=metadata,
            write_options=write_options,
        ),
    )
    image = ImageFile(**result.model_dump())
    image._set_stream(catalog)
    return image

def save(
    self,
    destination: str,
    format: str | None = None,
    client_config: dict | None = None,
    timeout: float | None = None,
    *,
    content_type: str | None = None,
    content_disposition: str | None = None,
    cache_control: str | None = None,
    content_encoding: str | None = None,
    metadata: dict[str, str] | None = None,
    write_options: dict[str, Any] | None = None,
) -> "VideoFile":
    """
    Saves the video fragment as a new video file.

    If ``destination`` is a remote path, the video file will be uploaded
    to remote storage.

    The output is named ``{stem}_{hash}_{start_ms}_{end_ms}.{format}``, where
    ``hash`` is derived from the source video's ``source`` and ``path``, so
    videos with the same name in different directories don't overwrite each
    other.

    Args:
        destination: Output directory path or URI (e.g. ``s3://…``, ``gs://…``).
        format: Output video format (e.g., 'mp4', 'avi').
                If None, inferred from the file extension.
        client_config: Optional client configuration (e.g. credentials).
        timeout: FFmpeg subprocess timeout in seconds. If None, a timeout is
            computed from the fragment duration. Set to 0 to disable.
        content_type: ``Content-Type`` to set on the written object.
        content_disposition: ``Content-Disposition`` to set on the object.
        cache_control: ``Cache-Control`` to set on the object.
        content_encoding: ``Content-Encoding`` to set on the object.
        metadata: Custom key/value metadata to attach to the object.
        write_options: Raw, backend-native write kwargs merged after the
            normalized fields above.

    Returns:
        VideoFile: A Model representing the saved video file.
    """
    from .video import save_video_fragment

    return save_video_fragment(
        self.video,
        self.start,
        self.end,
        destination,
        format,
        client_config=client_config,
        timeout=timeout,
        write_config=WriteConfig(
            content_type=content_type,
            content_disposition=content_disposition,
            cache_control=cache_control,
            content_encoding=content_encoding,
            metadata=metadata,
            write_options=write_options,
        ),
    )