I made a VideoMobject

[kmwadman]

I sat down and created a VideoMobject that seems to work pretty well. I had made a video_handler class myself a while ago and spent some time with Claude 4 today to create a proper VideoMobject. I have confirmed that the video behaves the way I would expect under a lot of animation scenarios, but I would suspect that there are some things that don't work and some choices that don't really match the quality of code that the team is looking for. It works for my use case, but I'm not sure this is ready for a PR. I figured I would share it out regardless, and maybe someone with a little bit more experience with the project can determine if it's worth incorporating something like this into the repo itself.

`
"""Mobjects representing video content."""

from future import annotations

all = ["VideoMobject"]

import pathlib
from typing import TYPE_CHECKING, Any

import cv2
import numpy as np
from PIL import Image
from PIL.Image import Resampling

from manim import logger
from manim.constants import QUALITIES, DEFAULT_QUALITY
from manim.mobject.types.image_mobject import AbstractImageMobject
from manim.utils.images import change_to_rgba_array

if TYPE_CHECKING:
from typing_extensions import Self
import numpy.typing as npt

from manim.typing import StrPath

class VideoState:
"""Shared state object for video playback that can be synchronized across copies.

This class manages the current frame position and time accumulator for video
playback, ensuring that multiple VideoMobject instances sharing the same state
advance frames in a synchronized manner.

Parameters
----------
current_frame : int, optional
    The initial frame number. Default is 0.
time_accumulator : float, optional
    The initial time accumulator in seconds. Default is 0.0.
"""

def __init__(self, current_frame: int = 0, time_accumulator: float = 0.0) -> None:
    self.current_frame: int = current_frame
    self.time_accumulator: float = time_accumulator
    self._paused: bool = False
    
    # For synchronization of state advancement across multiple VideoMobject instances
    # sharing this VideoState. We use a simple counter that increments each time
    # the state is advanced, and each instance tracks the last counter value it saw.
    self._update_counter: int = 0
    
    # Cache for raw frame data to avoid redundant disk/cv2 reads by multiple instances.
    self._cached_frame_data: npt.NDArray[np.uint8] | None = None
    self._cached_frame_number: int = -1
    
def copy(self) -> Self:
    """Create a copy of the video state.
    
    Returns
    -------
    VideoState
        A new VideoState instance with the same frame and time values.
        
    Notes
    -----
    Synchronization fields are not copied - the new state starts fresh.
    """
    new_state = VideoState(self.current_frame, self.time_accumulator)
    new_state._paused = self._paused
    # Don't copy synchronization fields - new state starts fresh
    return new_state

def advance_state_if_needed(
    self, 
    dt_from_updater: float, 
    last_seen_counter: int,
    fps: float, 
    start_frame: int, 
    end_frame: int, 
    loop: bool
) -> tuple[bool, int, int]:
    """Advance the video state if this is the first call in the current update cycle.

    Parameters
    ----------
    dt_from_updater : float
        Time delta from the updater function.
    last_seen_counter : int
        The last update counter value seen by the calling instance.
    fps : float
        Frames per second for the video.
    start_frame : int
        Starting frame number for playback range.
    end_frame : int
        Ending frame number for playback range.
    loop : bool
        Whether to loop when reaching the end.

    Returns
    -------
    tuple[bool, int, int]
        A tuple containing:
        - did_frame_change: Whether the frame number changed
        - current_frame_to_load: The frame number to load
        - new_counter: The updated counter value
    """
    if self._paused:
        return False, self.current_frame, self._update_counter

    # Only advance if this is the first updater call in this cycle
    if last_seen_counter < self._update_counter:
        # This updater has already seen the current state, don't advance again
        return False, self.current_frame, self._update_counter
    
    # This is the first updater call for this cycle, advance the state
    self._update_counter += 1
    
    if fps <= 0:
        return False, self.current_frame, self._update_counter
        
    self.time_accumulator += dt_from_updater
    frame_duration = 1.0 / fps
    
    if frame_duration <= 0:
        return False, self.current_frame, self._update_counter

    frames_to_advance = 0
    if self.time_accumulator >= frame_duration:
        frames_to_advance = int(self.time_accumulator / frame_duration)

    if frames_to_advance > 0:
        self.time_accumulator -= frames_to_advance * frame_duration
        
        new_frame = self.current_frame + frames_to_advance
        
        # Handle looping and bounds
        if new_frame > end_frame:
            if loop:
                frame_range = end_frame - start_frame + 1
                if frame_range > 0:
                    new_frame = start_frame + ((new_frame - start_frame) % frame_range)
                else:
                    new_frame = start_frame 
            else:
                new_frame = end_frame
        elif new_frame < start_frame:
            if loop:
                frame_range = end_frame - start_frame + 1
                if frame_range > 0:
                    new_frame = end_frame - ((start_frame - new_frame - 1) % frame_range)
                else:
                    new_frame = start_frame
            else:
                new_frame = start_frame

        if new_frame != self.current_frame:
            self.current_frame = new_frame
            return True, self.current_frame, self._update_counter
    
    return False, self.current_frame, self._update_counter

class VideoMobject(AbstractImageMobject):
"""A mobject that displays video frames, updating at a specified frame rate.

This class provides a clean interface for displaying videos in Manim animations,
with support for both video files and image sequences. The video automatically
plays during animations and can be controlled with various playback options.

Parameters
----------
video_source : StrPath
    Path to video file or directory containing image sequence.
fps : float, optional
    Frame rate for video playback. If None, uses the video's native fps.
start_frame : int, optional
    Starting frame number (0-based index). Default is 0.
end_frame : int, optional
    Ending frame number. If None, plays to the end.
loop : bool, optional
    Whether to loop the video when it reaches the end. Default is True.
scale_to_resolution : int, optional
    Resolution for pixel-perfect scaling. Default uses scene quality.
image_mode : str, optional
    Image mode for PIL (RGBA, RGB, etc.). Default is "RGBA".
pixel_array_dtype : str, optional
    Data type for pixel arrays. Default is "uint8".
resampling_algorithm : Resampling, optional
    PIL resampling algorithm for scaling. Default is BICUBIC.
shared_state : VideoState, optional
    Shared state object for synchronized playback across copies.
**kwargs
    Additional keyword arguments passed to AbstractImageMobject.
    
Examples
--------
Basic usage::

    video = VideoMobject("path/to/video.mp4")
    self.add(video)
    self.wait(5)  # Video will play for 5 seconds
    
With custom settings::

    video = VideoMobject(
        "path/to/video.mp4",
        fps=24,
        start_frame=100,
        end_frame=500,
        loop=True
    ).scale(0.5).move_to(RIGHT * 2)
    
Notes
-----
The video automatically starts playing when added to a scene. Multiple copies
of the same VideoMobject will share playback state by default, ensuring
synchronized playback during animations.
"""

def __init__(
    self,
    video_source: StrPath,
    fps: float | None = None,
    start_frame: int = 0,
    end_frame: int | None = None,
    loop: bool = True,
    scale_to_resolution: int = QUALITIES[DEFAULT_QUALITY]["pixel_height"],
    image_mode: str = "RGBA",
    pixel_array_dtype: str = "uint8",
    resampling_algorithm: Resampling = Resampling.BICUBIC,
    shared_state: VideoState | None = None,
    **kwargs: Any,
) -> None:
    self.video_source = pathlib.Path(video_source)
    self.image_mode = image_mode
    
    # Store the string version of dtype for AbstractImageMobject
    self.pixel_array_dtype_str: str = pixel_array_dtype
    # Store the actual numpy dtype for internal use
    self.numpy_pixel_array_dtype: np.dtype = np.dtype(pixel_array_dtype)

    # Initialize video capture and get metadata
    self._init_video_capture()
    
    # Set up frame range
    self.start_frame = max(0, start_frame)
    if end_frame is None:
        self.end_frame = self.total_frames - 1
    else:
        self.end_frame = min(end_frame, self.total_frames - 1)
    
    # Validate frame range
    self._validate_frame_range()
    
    self.loop = loop
    
    # Set playback FPS
    if fps is None:
        self.fps = self.native_fps
    else:
        self.fps = fps
        
    # Initialize or use shared video state
    if shared_state is not None:
        self._video_state = shared_state
    else:
        initial_frame = max(self.start_frame, min(0, self.end_frame)) if self.total_frames > 0 else 0
        self._video_state = VideoState(initial_frame, 0.0)
    
    # Initialize opacity values
    self.fill_opacity: float = 1.0
    self.stroke_opacity: float = 1.0
    
    # Initialize frame data storage
    self._original_frame_data: npt.NDArray[np.uint8] | None = None
    self.pixel_array: npt.NDArray[np.uint8] | None = None
    
    # Track which frame number is currently loaded in this instance's pixel_array
    self._instance_pixel_array_frame_num: int = -1 
    # Track the last update counter this instance has seen
    self._last_seen_update_counter: int = -1

    # Load initial frame
    self._load_frame(self._video_state.current_frame)
    
    # Fallback if initial frame loading failed
    if self.pixel_array is None:
        self._create_fallback_pixel_array()

    # Initialize parent class
    super().__init__(
        scale_to_resolution=scale_to_resolution,
        pixel_array_dtype=self.pixel_array_dtype_str,
        resampling_algorithm=resampling_algorithm,
        **kwargs
    )
    
    # Add the frame update function
    self.add_updater(self._update_frame)

def _validate_frame_range(self) -> None:
    """Validate that the frame range is sensible."""
    if self.start_frame > self.end_frame and self.total_frames > 0:
        raise ValueError(
            f"Start frame {self.start_frame} cannot be greater than "
            f"end frame {self.end_frame}"
        )
    if self.start_frame >= self.total_frames and self.total_frames > 0:
        raise ValueError(
            f"Start frame {self.start_frame} exceeds video length of "
            f"{self.total_frames} frames"
        )

def _create_fallback_pixel_array(self) -> None:
    """Create a fallback pixel array if frame loading fails."""
    width, height = 640, 480  # Default dimensions
    
    if hasattr(self, 'cap') and self.cap is not None and self.cap.isOpened():
        _width = self.cap.get(cv2.CAP_PROP_FRAME_WIDTH)
        _height = self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
        if _width > 0 and _height > 0:
            width, height = int(_width), int(_height)
    elif self.is_image_sequence and self.frame_files:
        try:
            img = Image.open(self.frame_files[0])
            width, height = img.width, img.height
        except Exception:
            logger.debug(
                f"Could not read dimensions from first image of sequence "
                f"{self.video_source}"
            )

    logger.warning(
        f"Failed to load initial video frame for {self.video_source}. "
        f"Using fallback {width}x{height} transparent pixel array."
    )
    self.pixel_array = np.zeros((height, width, 4), dtype=self.numpy_pixel_array_dtype)
    if self._original_frame_data is None:
        self._original_frame_data = self.pixel_array.copy()
    self._instance_pixel_array_frame_num = self._video_state.current_frame

@property
def current_frame(self) -> int:
    """Get the current frame number from the shared video state.
    
    Returns
    -------
    int
        The current frame number.
    """
    return self._video_state.current_frame
    
@current_frame.setter
def current_frame(self, value: int) -> None:
    """Set the current frame number. This seeks the video.
    
    Parameters
    ----------
    value : int
        The frame number to seek to.
    """
    self.seek_to_frame(value)
    
def _init_video_capture(self) -> None:
    """Initialize video capture based on source type."""
    if self.video_source.is_dir():
        self._init_image_sequence()
    else:
        self._init_video_file()
        
def _init_image_sequence(self) -> None:
    """Initialize for image sequence directory."""
    image_extensions = {'.png', '.jpg', '.jpeg', '.bmp', '.tiff', '.tif'}
    self.frame_files = sorted([
        f for f in self.video_source.iterdir()
        if f.suffix.lower() in image_extensions and f.is_file()
    ])
    
    if not self.frame_files:
        raise ValueError(f"No image files found in directory {self.video_source}")
        
    self.total_frames = len(self.frame_files)
    self.is_image_sequence = True
    
    # Try to read FPS from metadata file
    metadata_path = self.video_source / 'metadata.txt'
    self.native_fps = 30.0  # Default FPS for image sequences
    if metadata_path.exists():
        try:
            with open(metadata_path, 'r') as f:
                for line in f:
                    if ':' in line:
                        key, value = line.strip().split(':', 1)
                        if key.lower().strip() == 'fps':
                            self.native_fps = float(value.strip())
                            break 
        except Exception as e:
            logger.warning(f"Could not read FPS from metadata {metadata_path}: {e}")
            
def _init_video_file(self) -> None:
    """Initialize for video file."""
    self.cap = cv2.VideoCapture(str(self.video_source))
    if not self.cap.isOpened():
        raise ValueError(f"Could not open video file: {self.video_source}")
        
    self.total_frames = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT))
    self.native_fps = self.cap.get(cv2.CAP_PROP_FPS)
    if self.native_fps <= 0:
        logger.warning(
            f"Video file {self.video_source} reported FPS <= 0 ({self.native_fps}). "
            f"Defaulting to 30 FPS."
        )
        self.native_fps = 30.0
    self.is_image_sequence = False

def _fetch_raw_frame_data(self, frame_number: int) -> npt.NDArray[np.uint8] | None:
    """Fetch raw RGBA frame data for the given frame number.
    
    Uses VideoState cache to avoid redundant reads from source.
    
    Parameters
    ----------
    frame_number : int
        The frame number to fetch.
        
    Returns
    -------
    npt.NDArray[np.uint8] | None
        The RGBA frame data, or None if loading failed.
    """
    # Check cache first
    if (self._video_state._cached_frame_number == frame_number and 
        self._video_state._cached_frame_data is not None):
        return self._video_state._cached_frame_data.copy()

    raw_data: npt.NDArray[np.uint8] | None = None
    
    if self.is_image_sequence:
        raw_data = self._fetch_from_image_sequence(frame_number)
    else:
        raw_data = self._fetch_from_video_file(frame_number)
    
    # Update cache
    if raw_data is not None:
        self._video_state._cached_frame_data = raw_data.copy() 
        self._video_state._cached_frame_number = frame_number
        return raw_data.copy()
    else:
        self._video_state._cached_frame_data = None 
        self._video_state._cached_frame_number = -1
        return None

def _fetch_from_image_sequence(self, frame_number: int) -> npt.NDArray[np.uint8] | None:
    """Fetch frame data from image sequence."""
    if 0 <= frame_number < len(self.frame_files):
        frame_path = self.frame_files[frame_number]
        try:
            image = Image.open(frame_path).convert(self.image_mode)
            frame_data_arr = np.array(image, dtype=self.numpy_pixel_array_dtype)
            return change_to_rgba_array(frame_data_arr, self.numpy_pixel_array_dtype)
        except Exception as e:
            logger.error(f"Could not load frame {frame_number} from {frame_path}: {e}")
    else:
        logger.warning(
            f"Frame number {frame_number} out of bounds for image sequence "
            f"(0-{len(self.frame_files)-1})."
        )
    return None

def _fetch_from_video_file(self, frame_number: int) -> npt.NDArray[np.uint8] | None:
    """Fetch frame data from video file."""
    if not hasattr(self, 'cap') or self.cap is None or not self.cap.isOpened():
        logger.warning(
            f"Video capture not initialized or closed for frame {frame_number}. "
            f"Attempting to re-initialize."
        )
        self._init_video_file()
        if not hasattr(self, 'cap') or self.cap is None or not self.cap.isOpened():
            logger.error(f"Failed to re-initialize video capture for {self.video_source}.")
            return None 
    
    self.cap.set(cv2.CAP_PROP_POS_FRAMES, frame_number)
    ret, frame = self.cap.read()
    if ret:
        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        image = Image.fromarray(frame_rgb).convert(self.image_mode)
        frame_data_arr = np.array(image, dtype=self.numpy_pixel_array_dtype)
        return change_to_rgba_array(frame_data_arr, self.numpy_pixel_array_dtype)
    else:
        current_pos = self.cap.get(cv2.CAP_PROP_POS_FRAMES)
        total_cap_frames = self.cap.get(cv2.CAP_PROP_FRAME_COUNT)
        if frame_number >= total_cap_frames:
            logger.debug(
                f"Attempted to read frame {frame_number} at/beyond total frames "
                f"{total_cap_frames} for {self.video_source}."
            )
        else:
            logger.error(
                f"Could not read frame {frame_number} from video {self.video_source}. "
                f"Current position: {current_pos}."
            )
    return None

def _load_frame(self, frame_number: int) -> None:
    """Load a specific frame into this VideoMobject instance's pixel_array.
    
    Parameters
    ----------
    frame_number : int
        The frame number to load.
    """
    if self.total_frames == 0:
        if self.pixel_array is None:
            self.pixel_array = np.zeros((1, 1, 4), dtype=self.numpy_pixel_array_dtype)
            self._original_frame_data = self.pixel_array.copy()
        return

    # Clamp frame_number to the actual available frames
    clamped_frame_number = max(0, min(frame_number, self.total_frames - 1))

    raw_frame_data = self._fetch_raw_frame_data(clamped_frame_number)

    if raw_frame_data is not None:
        # Check if this instance's pixel_array needs updating
        if (self.pixel_array is None or 
            self._instance_pixel_array_frame_num != clamped_frame_number or 
            self.pixel_array.shape != raw_frame_data.shape):
            
            # Initialize or re-shape pixel_array and _original_frame_data if necessary
            if self.pixel_array is None or self.pixel_array.shape != raw_frame_data.shape:
                self.pixel_array = np.empty_like(raw_frame_data)
            if self._original_frame_data is None or self._original_frame_data.shape != raw_frame_data.shape:
                self._original_frame_data = np.empty_like(raw_frame_data)

            np.copyto(self._original_frame_data, raw_frame_data)
            np.copyto(self.pixel_array, raw_frame_data)
            
            self._instance_pixel_array_frame_num = clamped_frame_number
        
        # Always apply instance-specific opacity
        self._apply_current_opacity()
    else:
        logger.warning(
            f"Failed to load frame {clamped_frame_number} for {self.video_source}. "
            f"Instance's display not updated."
        )
        
def _apply_current_opacity(self) -> None:
    """Apply the current fill_opacity to pixel_array based on _original_frame_data."""
    if self._original_frame_data is not None and self.pixel_array is not None:
        if (self.pixel_array.ndim == 3 and self._original_frame_data.ndim == 3 and 
            self.pixel_array.shape[:-1] == self._original_frame_data.shape[:-1] and 
            self.pixel_array.shape[-1] >= 4 and self._original_frame_data.shape[-1] >= 4):
            
            alpha_channel = (self._original_frame_data[:, :, 3] * self.fill_opacity)
            self.pixel_array[:, :, 3] = alpha_channel.astype(self.numpy_pixel_array_dtype)
        else:
            logger.debug("Cannot apply opacity: Incompatible shapes or no alpha channel.")

def _update_frame(self, mobject, dt: float) -> None:
    """Update function called by Manim's animation system.
    
    Parameters
    ----------
    mobject
        The mobject being updated (should be self).
    dt : float
        Time delta since last update.
    """
    if self.fps <= 0 or self.total_frames == 0:
        return
        
    # Advance the shared video state if this is the first updater call in this cycle
    frame_advanced, target_frame_to_load, new_counter = self._video_state.advance_state_if_needed(
        dt, 
        self._last_seen_update_counter,
        self.fps,
        self.start_frame,
        self.end_frame,
        self.loop
    )
    
    # Update our counter
    self._last_seen_update_counter = new_counter
    
    # Load the target frame into this instance's pixel_array
    self._load_frame(target_frame_to_load)

def get_pixel_array(self) -> npt.NDArray[np.uint8]:
    """Return the current pixel array.
    
    Returns
    -------
    npt.NDArray[np.uint8]
        The current pixel array containing RGBA image data.
    """
    if self.pixel_array is None:
        logger.error(f"get_pixel_array called when pixel_array is None for {self.video_source.name}.")
        return np.zeros((1, 1, 4), dtype=self.numpy_pixel_array_dtype)
    return self.pixel_array
    
def set_color(self, color, alpha=None, family=True):
    """Set the color of the video mobject.
    
    Parameters
    ----------
    color
        The color to set.
    alpha : float, optional
        The alpha (opacity) value.
    family : bool, optional
        Whether to apply to submobjects as well.
        
    Returns
    -------
    Self
        This VideoMobject for method chaining.
    """
    from manim.utils.color import color_to_int_rgb
    rgb = color_to_int_rgb(color)
    self.pixel_array[:, :, :3] = rgb
    if alpha is not None:
        self.pixel_array[:, :, 3] = int(255 * alpha)
        self.fill_opacity = alpha
        self.stroke_opacity = alpha
    if family:
        for submob in self.submobjects:
            submob.set_color(color, alpha, family)
    self.color = color
    return self

def set_opacity(self, alpha: float, family: bool = True) -> Self:
    """Set the opacity of the video mobject.
    
    Parameters
    ----------
    alpha : float
        The opacity value between 0 (transparent) and 1 (opaque).
    family : bool, optional
        Whether to apply to submobjects as well. Default is True.
        
    Returns
    -------
    Self
        This VideoMobject for method chaining.
    """
    self.fill_opacity = np.clip(alpha, 0.0, 1.0)
    self.stroke_opacity = self.fill_opacity
    
    if self.pixel_array is not None and self._original_frame_data is not None:
        self._apply_current_opacity()

    if family:
        for submobject in self.submobjects:
            submobject.set_opacity(alpha, family=family)
    return self

def fade(self, darkness: float = 0.5, family: bool = True) -> Self:
    """Set the video's opacity using a 1 - alpha relationship.
    
    Parameters
    ----------
    darkness : float, optional
        The darkness value where 0 is opaque and 1 is transparent. Default is 0.5.
    family : bool, optional
        Whether to apply to submobjects as well. Default is True.
        
    Returns
    -------
    Self
        This VideoMobject for method chaining.
    """
    self.set_opacity(1 - darkness, family)
    return self

def interpolate_color(
    self, mobject1: VideoMobject, mobject2: VideoMobject, alpha: float
) -> None:
    """Interpolate the pixel color values between two VideoMobjects.
    
    Parameters
    ----------
    mobject1 : VideoMobject
        The first VideoMobject to interpolate from.
    mobject2 : VideoMobject
        The second VideoMobject to interpolate to.
    alpha : float
        The interpolation parameter between 0 and 1.
        
    Raises
    ------
    AssertionError
        If the pixel arrays have incompatible shapes.
    """
    from manim.utils.bezier import interpolate
    
    assert mobject1.pixel_array.shape == mobject2.pixel_array.shape, (
        f"Mobject pixel array shapes incompatible for interpolation.\n"
        f"Mobject 1 ({mobject1}) : {mobject1.pixel_array.shape}\n"
        f"Mobject 2 ({mobject2}) : {mobject2.pixel_array.shape}"
    )
    self.fill_opacity = interpolate(
        mobject1.fill_opacity,
        mobject2.fill_opacity,
        alpha,
    )
    self.stroke_opacity = interpolate(
        mobject1.stroke_opacity,
        mobject2.stroke_opacity,
        alpha,
    )
    self.pixel_array = interpolate(
        mobject1.pixel_array,
        mobject2.pixel_array,
        alpha,
    ).astype(self.pixel_array_dtype)

def get_style(self) -> dict[str, Any]:
    """Get the style dictionary for this VideoMobject.
    
    Returns
    -------
    dict[str, Any]
        A dictionary containing style information.
    """
    from manim.utils.color import ManimColor
    return {
        "fill_color": ManimColor(self.color.get_rgb()).to_hex(),
        "fill_opacity": self.fill_opacity,
    }
    
# Video-specific control methods
    
def set_fps(self, fps: float) -> Self:
    """Set the playback frame rate.
    
    Parameters
    ----------
    fps : float
        The new frames per second value.
        
    Returns
    -------
    Self
        This VideoMobject for method chaining.
    """
    self.fps = fps
    return self
    
def set_frame_range(self, start_frame: int, end_frame: int | None = None) -> Self:
    """Set the frame range for playback.
    
    Parameters
    ----------
    start_frame : int
        The starting frame number.
    end_frame : int, optional
        The ending frame number. If None, plays to the end.
        
    Returns
    -------
    Self
        This VideoMobject for method chaining.
    """
    self.start_frame = max(0, start_frame)
    if end_frame is not None:
        self.end_frame = min(end_frame, self.total_frames - 1)
    else:
        self.end_frame = self.total_frames - 1
        
    # Reset to start frame
    self._video_state.current_frame = self.start_frame
    self._load_frame(self._video_state.current_frame)
    return self
    
def seek_to_frame(self, frame_number: int) -> Self:
    """Seek to a specific frame number.
    
    Parameters
    ----------
    frame_number : int
        The frame number to seek to.
        
    Returns
    -------
    Self
        This VideoMobject for method chaining.
    """
    if self.total_frames == 0:
        return self

    # Clamp frame_number to the mobject's defined playback range
    clamped_to_range = max(self.start_frame, min(frame_number, self.end_frame))
    final_frame_number = max(0, min(clamped_to_range, self.total_frames - 1))

    self._video_state.current_frame = final_frame_number
    self._video_state.time_accumulator = 0.0
    
    self._load_frame(self._video_state.current_frame)
    return self
    
def seek_to_time(self, time_seconds: float) -> Self:
    """Seek to a specific time in seconds.
    
    Parameters
    ----------
    time_seconds : float
        The time in seconds to seek to.
        
    Returns
    -------
    Self
        This VideoMobject for method chaining.
    """
    frame_number = int(time_seconds * self.native_fps) + self.start_frame
    return self.seek_to_frame(frame_number)
    
def set_loop(self, loop: bool) -> Self:
    """Set whether the video should loop.
    
    Parameters
    ----------
    loop : bool
        Whether to enable looping.
        
    Returns
    -------
    Self
        This VideoMobject for method chaining.
    """
    self.loop = loop
    return self
    
def pause(self) -> Self:
    """Pause video playback.
    
    Returns
    -------
    Self
        This VideoMobject for method chaining.
    """
    self._video_state._paused = True
    return self
    
def resume(self, fps: float | None = None) -> Self:
    """Resume video playback.
    
    Parameters
    ----------
    fps : float, optional
        New FPS to set when resuming. If None, keeps current FPS.
        
    Returns
    -------
    Self
        This VideoMobject for method chaining.
    """
    self._video_state._paused = False
    if fps is not None:
        self.fps = fps
    return self
    
def is_paused(self) -> bool:
    """Check if video playback is paused.
    
    Returns
    -------
    bool
        True if the video is paused, False otherwise.
    """
    return self._video_state._paused
    
def get_duration(self) -> float:
    """Get the duration of the video in seconds.
    
    Returns
    -------
    float
        The duration in seconds based on the frame range and native FPS.
    """
    frame_count = self.end_frame - self.start_frame + 1
    return frame_count / self.native_fps
    
def get_current_time(self) -> float:
    """Get the current playback time in seconds.
    
    Returns
    -------
    float
        The current time in seconds relative to the start frame.
    """
    return (self._video_state.current_frame - self.start_frame) / self.native_fps
    
def copy(self) -> Self:
    """Create a copy of the VideoMobject that shares the VideoState.
    
    Returns
    -------
    Self
        A new VideoMobject instance that shares playback state with this one.
        
    Notes
    -----
    The copy shares the same VideoState object, ensuring synchronized playback
    during animations. However, each copy maintains its own pixel arrays and
    can have different visual properties (opacity, position, etc.).
    """
    new_video = object.__new__(self.__class__)

    # Essential attributes for re-initialization
    new_video.video_source = self.video_source
    new_video.pixel_array_dtype_str = self.pixel_array_dtype_str
    new_video.numpy_pixel_array_dtype = self.numpy_pixel_array_dtype

    # Copy video attributes
    new_video.start_frame = self.start_frame
    new_video.end_frame = self.end_frame
    new_video.loop = self.loop
    new_video.image_mode = self.image_mode
    new_video.fps = self.fps
    new_video.total_frames = self.total_frames
    new_video.native_fps = self.native_fps
    new_video.is_image_sequence = self.is_image_sequence
    new_video.scale_to_resolution = getattr(self, 'scale_to_resolution', QUALITIES[DEFAULT_QUALITY]["pixel_height"])
    new_video.resampling_algorithm = getattr(self, 'resampling_algorithm', Resampling.BICUBIC)
    
    # Share the critical VideoState object
    new_video._video_state = self._video_state
    
    # Create separate capture objects for thread safety
    if not self.is_image_sequence:
        if hasattr(self, 'cap') and self.cap is not None:
            new_video.cap = cv2.VideoCapture(str(self.video_source))
            if not new_video.cap.isOpened():
                logger.warning(f"Could not open video file for copy of {self.video_source.name}")
                new_video.cap = None
        else:
            new_video.cap = None
    elif hasattr(self, 'frame_files'):
        new_video.frame_files = self.frame_files

    new_video.fill_opacity = self.fill_opacity
    new_video.stroke_opacity = self.stroke_opacity
    
    # Copy pixel arrays
    if self.pixel_array is not None:
        new_video.pixel_array = self.pixel_array.copy()
    else:
        new_video.pixel_array = None
        
    if self._original_frame_data is not None:
        new_video._original_frame_data = self._original_frame_data.copy()
    else:
        new_video._original_frame_data = None

    new_video._instance_pixel_array_frame_num = self._instance_pixel_array_frame_num
    # Initialize the copy's counter to the current state counter to ensure proper synchronization
    # This prevents the copy from thinking it needs to advance the state when it's first created
    new_video._last_seen_update_counter = self._video_state._update_counter

    # Initialize parent class
    AbstractImageMobject.__init__(
        new_video,
        scale_to_resolution=new_video.scale_to_resolution,
        pixel_array_dtype=new_video.pixel_array_dtype_str,
        resampling_algorithm=new_video.resampling_algorithm,
    )
    
    # Copy mobject properties
    if hasattr(self, 'points') and self.points is not None:
        new_video.points = self.points.copy()
    if hasattr(self, 'color'):
        new_video.color = self.color

    # Add updater to the new copy
    new_video.add_updater(new_video._update_frame)
    
    return new_video
    
def __del__(self):
    """Clean up resources when the object is destroyed."""
    if hasattr(self, 'cap') and self.cap is not None:
        try:
            self.cap.release()
            self.cap = None
        except Exception as e:
            logger.debug(f"Exception during cv2.VideoCapture release for {self.video_source.name}: {e}") 

`

[uwezi]

this looks much more comprehensive than my quick and dirty approach: https://gist.github.com/uwezi/faec101ed5d7c20222b33eee4b6c7d63#file-20240309_opencv-py

[kmwadman]

Biggest challenge was getting the updaters to play nice-- you have to be really careful about the pixel array getting handed around without the frame being advanced multiple times on things during animations that copy the mobject (FadeIn, etc).