video_in.py

import glob
import fnmatch
import logging
import math
import os
import re
from threading import Condition, Event, Thread
from time import time_ns, sleep
from typing import Any
from urllib.parse import urlparse

import cv2

try:
    import boto3
    from botocore.exceptions import ClientError, NoCredentialsError
    HAS_BOTO3 = True
except ImportError:
    HAS_BOTO3 = False

from openfilter.filter_runtime.utils import json_getval, dict_without, split_commas_maybe, hide_uri_users_and_pwds, resolve_override_source_uri, Deque

__all__ = ['is_video', 'is_video_file', 'is_video_webcam', 'is_video_stream', 'VideoReader', 'MultiVideoReader']

logger = logging.getLogger(__name__)

VIDEO_IN_BGR      = bool(json_getval((os.getenv('VIDEO_IN_BGR') or os.getenv('FILTER_BGR') or 'true').lower()))
VIDEO_IN_SYNC     = bool(json_getval((os.getenv('VIDEO_IN_SYNC') or os.getenv('FILTER_SYNC') or 'false').lower()))
VIDEO_IN_LOOP     = _ if isinstance(_ := json_getval((os.getenv('VIDEO_IN_LOOP') or os.getenv('FILTER_LOOP') or 'false').lower()), bool) else int(_)
VIDEO_IN_MAXFPS   = None if (_ := json_getval((os.getenv('VIDEO_IN_MAXFPS') or os.getenv('FILTER_MAXFPS') or 'null').lower())) is None else float(_)
VIDEO_IN_MAXFPS_BY_INDEX = bool(json_getval((os.getenv('VIDEO_IN_MAXFPS_BY_INDEX') or os.getenv('FILTER_MAXFPS_BY_INDEX') or 'false').lower()))
VIDEO_IN_MAXSIZE  = os.getenv('VIDEO_IN_MAXSIZE') or os.getenv('FILTER_MAXSIZE') or None
VIDEO_IN_RESIZE   = os.getenv('VIDEO_IN_RESIZE') or os.getenv('FILTER_RESIZE') or None
VIDEO_IN_PATTERN  = os.getenv('VIDEO_IN_PATTERN') or os.getenv('FILTER_PATTERN') or None
VIDEO_IN_RECURSIVE = bool(json_getval((os.getenv('VIDEO_IN_RECURSIVE') or os.getenv('FILTER_RECURSIVE') or 'false').lower()))

# OpenCV's ffmpeg backend reports a sentinel ~1000 fps (the 1 ms MKV/webm container
# timebase, not a real rate) for some VFR files. Any CAP_PROP_FPS at or above this
# ceiling - or non-finite - is implausible for real footage and is treated as no
# reported rate, so pts falls through to the guarded POS_MSEC branch instead of
# emitting frame_n / 1000 (~1 ms/frame) offsets that are simply wrong.
FPS_SANE_CEILING  = 1000.0

re_video          = re.compile(r'^(rtsp|rtmp|http|https|file|webcam|s3)://')
re_video_stream   = re.compile(r'^(rtsp|rtmp|http|https)://')

is_video          = lambda name: bool(re_video.match(name))
is_video_file     = lambda name: name.startswith('file://')
is_video_webcam   = lambda name: name.startswith('webcam://')
is_video_stream   = lambda name: bool(re_video_stream.match(name))
is_video_s3       = lambda name: name.startswith('s3://')

VIDEO_EXTENSIONS = {'mp4', 'mkv', 'avi', 'mov', 'webm', 'flv', 'wmv', 'm4v', '3gp', 'mpg', 'mpeg'}

def is_video_file_ext(path: str) -> bool:
    """Check if the given file path has a valid video file extension."""
    ext = path.split('.')[-1].lower() if '.' in path else ''
    return ext in VIDEO_EXTENSIONS

def matches_pattern(path: str, pattern: str) -> bool:
    """Check if path matches the given pattern (glob or regex)."""
    if not pattern:
        return True
    if '*' in pattern or '?' in pattern:
        return fnmatch.fnmatch(os.path.basename(path), pattern)
    try:
        return bool(re.search(pattern, path))
    except re.error:
        return pattern in path


def scan_dir_videos(path: str, pattern: str | None = None, recursive: bool | None = None) -> list[str]:
    """List video files in a directory honoring `pattern` and `recursive`, sorted alphabetically.

    Shared by VideoReader.__init__ and VideoReader.get_info so both agree on which file in a
    directory is "first", instead of get_info silently reimplementing (and diverging from) the scan.
    Also resolves the VIDEO_IN_PATTERN/VIDEO_IN_RECURSIVE env var defaults here (rather than in each
    caller) so both callers apply them identically.
    """
    pattern   = VIDEO_IN_PATTERN if pattern is None else pattern
    recursive = VIDEO_IN_RECURSIVE if recursive is None else recursive

    if recursive:
        candidates = glob.glob(os.path.join(path, '**', '*'), recursive=True)
    else:
        candidates = glob.glob(os.path.join(path, '*'))

    return [p for p in sorted(candidates) if os.path.isfile(p) and is_video_file_ext(p) and matches_pattern(p, pattern)]


re_size = re.compile(r'^\s* (\d+) \s* ([x+]) \s* (\d+) \s* (n(?:ear)? | l(?:in)? | c(?:ub)?)? \s*$', re.VERBOSE | re.IGNORECASE)

def parse_size(s: str):
    if not (m := re_size.match(s)):
        raise ValueError(f'invalid size {s!r}')

    return m.groups()


def parse_s3_uri(s3_uri: str):
    """Parse S3 URI into bucket and key components.
    
    Args:
        s3_uri: S3 URI in format s3://bucket/key
        
    Returns:
        tuple: (bucket, key)
        
    Raises:
        ValueError: If URI format is invalid
    """
    if not s3_uri.startswith('s3://'):
        raise ValueError(f'Invalid S3 URI: {s3_uri}')
    
    parsed = urlparse(s3_uri)
    if not parsed.netloc:
        raise ValueError(f'Invalid S3 URI: {s3_uri} (missing bucket)')
    
    bucket = parsed.netloc
    key = parsed.path.lstrip('/')
    
    if not key:
        raise ValueError(f'Invalid S3 URI: {s3_uri} (missing key)')
    
    return bucket, key


def s3_to_presigned_url(s3_uri: str, expiration: int = 3600, region: str = None):
    """Convert S3 URI to presigned HTTPS URL.
    
    Args:
        s3_uri: S3 URI in format s3://bucket/key
        expiration: URL expiration time in seconds (default: 1 hour)
        region: AWS region (optional, will use default if not specified)
        
    Returns:
        str: HTTPS URL for streaming the S3 object
        
    Raises:
        ValueError: If S3 URI is invalid or boto3 not available
        ClientError: If S3 access fails
        NoCredentialsError: If AWS credentials not configured
    """
    if not HAS_BOTO3:
        raise ValueError('boto3 is required for S3 support. Install with: pip install boto3')
    
    bucket, key = parse_s3_uri(s3_uri)
    
    s3_client = boto3.client('s3', region_name=region)
    
    try:
        presigned_url = s3_client.generate_presigned_url(
            'get_object',
            Params={'Bucket': bucket, 'Key': key},
            ExpiresIn=expiration
        )
        return presigned_url
    except NoCredentialsError:
        raise ValueError(
            f'AWS credentials not found for {s3_uri}. Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY '
            'environment variables or configure AWS credentials file.'
        )
    except ClientError as e:
        raise ValueError(f'Failed to generate presigned URL for {s3_uri}: {e}')


class VideoReader:
    def __init__(self,
        source:  str,
        cond:    Condition | None = None,
        *,
        bgr:     bool | None = None,
        sync:    bool | None = None,
        loop:    bool | int = False,
        maxfps:  float | None = None,
        maxfps_by_index: bool | None = None,
        maxsize: str | None = None,
        resize:  str | None = None,
        region:  str | None = None,
        expiration: int | None = None,
        pattern:  str | None = None,
        recursive: bool | None = None,
    ):
        """Read a single video file, network stream, webcam, or video directory until the end.

        Args:
            source: Source video stream, can be file, web stream like 'rtsp://...', a webcam index starting at 0 -
                'webcam://0', or a local directory path (e.g. 'file:///path/to/dir') for sequential playback of matching video files in alphabetical order.

            cond: A threading.Condition to .notify_all() whenever a new frame is read.

            bgr: True means images in BGR mode, False means RGB. Has env var default.

            sync: Only has meaning for files/directories. If True then frames will be delivered one by one without skipping or
                waiting to maintain realtime, in this way all frames will be read. If None the the system default is
                used which comes from an environment variable (and is False if not present).

            loop: Only has meaning for files/directories. True or 0 means infinite loop, False means loop once, otherwise int loops
                through the video files. Has env var default.

            maxsize: Maximum image size to allow, above this will be resized down. Valid codes are '123x456' which will
                proportionally resize maintaining aspect ratio so that neither dimension exceeds the max, '123+456' will
                resize without maintaining aspect ratio. Optional suffixes 'near', 'lin' and 'cub' specify
                interpolation, default is 'near'est neighbor.

            resize: Straight resize always, can not be specified together with `maxsize`, it is one or the other.

            region: S3 bucket AWS region (optional).

            expiration: Presigned URL expiration in seconds (optional).

            pattern: Filter string or glob pattern (e.g., `*.mp4`) to select specific video files from directories. Has env var default.

            recursive: If True, subfolders in directories are recursively scanned for video files. Has env var default.
        """

        if not isinstance((loop := VIDEO_IN_LOOP if loop is None else loop), (bool, int)) or loop < 0:
            raise ValueError(f"invalid loop '{loop}', must be a bool or nonnegative integer")

        self.source        = hide_uri_users_and_pwds(source)
        self.cond          = cond
        self.loop          = 0 if loop is True else 1 if loop is False else loop
        self.maxfps        = maxfps = VIDEO_IN_MAXFPS if maxfps is None else maxfps
        self.maxsize       = None if (s := VIDEO_IN_MAXSIZE if maxsize is None else maxsize) is None else parse_size(s)
        self.resize        = None if (s := VIDEO_IN_RESIZE if resize is None else resize) is None else parse_size(s)
        self.state         = 0     # 0 = before start, 1 = playing, 2 = stopped / done
        self.sync_evt      = None  # file video only: set for 'sync' True, and also for 'sync' False when maxfps_by_index engages (by-index needs the back-pressure), so this being set does NOT imply sync is on
        self.ns_per_fps    = None  # this is set only for file video with 'sync' option False
        self.ns_per_maxfps = None if maxfps is None else 1_000_000_000 // maxfps
        self.index_stride  = None  # set only for file video when maxfps is applied by source index
        self.frame_i       = 0
        self.is_file       = is_file = is_video_file(source) or is_video_s3(source)
        self.as_bgr        = bool(VIDEO_IN_BGR if bgr is None else bgr)  # only validated after first frame is read (set to False if frames are grayscale)
        self.is_dir        = False
        self.dir_files     = []
        self.dir_index     = 0
        self.source_dir_path = None

        if self.maxsize and self.resize:
            raise ValueError(f"can not specify both 'maxsize' and 'resize' together in {self.source!r}")

        if is_file:
            if is_video_file(source):
                source = source[7:]  # Remove 'file://' prefix
            elif is_video_s3(source):
                # Convert S3 URI to presigned HTTPS URL
                logger.info(f"Converting S3 URI to presigned URL: {hide_uri_users_and_pwds(source)}")
                try:
                    s3_expiration = expiration or 3600  # Default 1 hour
                    source = s3_to_presigned_url(source, expiration=s3_expiration, region=region)
                    logger.debug(f"Generated presigned URL for S3 video (region={region}, expiration={s3_expiration}s)")
                except Exception as e:
                    raise ValueError(f'Failed to generate presigned URL for S3 source {self.source!r}: {e}')
            self.is_file = True

            if os.path.isdir(source):
                self.is_dir = True
                self.source_dir_path = self.source

                self.dir_files = scan_dir_videos(source, pattern, recursive)

                if not self.dir_files:
                    raise RuntimeError(f'no valid video files found in directory: {self.source!r}')
                self.dir_index = 0
                source = self.dir_files[0]
                self.source = f'file://{source}'
            else:
                self.is_dir = False

            if sync := VIDEO_IN_SYNC if sync is None else sync:
                self.sync_evt = Event()

                self.sync_evt.set()

        else:
            if sync:
                logger.warning(f"'sync' does not apply to videos which are not files in {self.source!r}")

            if is_video_webcam(source):
                source = int(source[9:])
            elif not is_video_stream(source):
                raise ValueError(f'invalid source {self.source!r}')

        self.ssource  = source  # with 'file://' stripped and 'webcam://num' converted to num
        self.extras   = {}      # decoder position of the last frame read, files only - 'frame_n' (0-based source frame index) and 'pts_s' (presentation timestamp, seconds); same tuple slot and key naming as the seekable-replay extras (PR #118) so the two contracts merge by key union
        self.stop_evt = Event()
        self.deque    = Deque(maxlen=1)
        self.thread   = Thread(target=self.thread_reader, daemon=True)
        self.cap      = cap = cv2.VideoCapture(source)
        if not cap.isOpened():
            raise RuntimeError(f'failed to open video source: {self.source!r}')
        fps           = cap.get(cv2.CAP_PROP_FPS) or None
        self.native_fps = fps  # container frame rate, kept for pts fallback (self.fps may be overwritten by maxfps)

        if is_file and not sync:
            self.ns_per_fps = 1_000_000_000 // (fps or 15)  # cv2 reads files as fast as possible, this is to keep it realtime, default to 15 if video doesn't provide fixed framerate

        self.fps = fps

        # maxfps normally means "N frames per second of WALL CLOCK": the reader either
        # sleeps (sync) or paces to the container rate (no-sync), so an hour of file costs
        # an hour however fast the machine is. For offline processing what is wanted is the
        # same selection taken from the source timeline and read as fast as the box allows.
        # Opt in with maxfps_by_index: on a sync=False file the reader keeps 1 frame in every
        # ceil(fps/maxfps) and reads flat out, both clock paths switched off. The selection is
        # unchanged in steady state, it is just no longer spread over real time.
        #
        # It is deliberately a sync=False optimisation. sync=True documents a no-skip contract
        # ("will not skip any frames"), so by-index no-ops under sync and leaves that paced
        # path untouched rather than silently subsampling it. It also no-ops when fps is the
        # >= FPS_SANE_CEILING sentinel a VFR container reports for no real rate (the value
        # _cap_read rejects as bogus): a stride derived from it would drop nearly every frame.
        by_index = VIDEO_IN_MAXFPS_BY_INDEX if maxfps_by_index is None else maxfps_by_index

        # Kept so the gate can run again per file: a directory source opens members with
        # their own rates, and each has to be gated on its own fps (see _open_dir_file).
        self._by_index = by_index
        self._sync     = bool(sync)

        if self._by_index_stride_for(fps) is not None:
            # ceil, not round: round picks the nearest stride and banker's-rounds 2.5 -> 2,
            # which lets the by-index rate exceed maxfps (e.g. 25 fps / maxfps 10 -> stride 2
            # -> 12.5 fps). ceil keeps the effective rate at or below the cap for every ratio,
            # matching the wall-clock path (fps > maxfps here already guarantees stride >= 2).
            self.index_stride  = self._by_index_stride_for(fps)
            self.ns_per_fps    = None
            self.ns_per_maxfps = None

            # We only get here under sync=False, which switches off both clock paths - and
            # those were the only back-pressure against the reader outrunning the consumer.
            # Without the sync_evt handshake the reader decodes flat out into the 1-slot deque
            # and overwrites selected frames before they are popped, so give the by-index path
            # the same handshake sync=True has (sync_evt is None here, never created above).
            self.sync_evt = Event()
            self.sync_evt.set()

            logger.info(f'maxfps by index: keeping 1 frame in every {self.index_stride} '
                        f'of {fps:.1f} fps -> {fps / self.index_stride:.2f} fps effective')

        # The two no-op paths that would otherwise be silent: by-index was asked for on a file
        # whose rate is above maxfps (so it would have engaged), but sync is on or the reported
        # rate is the >= FPS_SANE_CEILING sentinel. An operator setting FILTER_MAXFPS_BY_INDEX
        # fleet-wide gets no speedup on those pipelines, so say why. NOT logged for fps <= maxfps,
        # which is a legitimate, frequent no-op and would just be noise.
        elif by_index and is_file and maxfps and fps and fps > maxfps and (sync or fps >= FPS_SANE_CEILING):
            reason = ('sync is on, its no-skip guarantee is preserved' if sync else
                      f'the source reports no real frame rate ({fps:.0f} fps sentinel)')

            logger.info(f'maxfps by index requested but not engaged ({reason}); '
                        f'maxfps {maxfps:.1f} applied on the wall clock instead')

        if fps is None:
            logger.warning(f'video does not have fixed framerate {self.source!r}{"" if maxfps is None else ", maxfps ignored"}')

            fps_str = ''

        else:
            if maxfps is None or fps <= maxfps:
                fps_str = f'  ({fps:.1f} fps)'

            else:
                fps_str = f'  ({fps:.1f} fps at {maxfps:.1f} maxfps)'
                fps     = maxfps

                if not is_file or not sync:
                    self.fps = fps

        logger.info(f'video open: {self.source}{fps_str}')

    def __iter__(self):
        return self

    def __next__(self):
        if (item := self.read()) is None:
            raise StopIteration

        return item

    def start(self):  # idempotent and safe to call whenever
        if self.state != 0:
            return

        self.state = 1
        self.tfps  = self.tmaxfps = time_ns()

        self.thread.start()

    def stop(self):  # idempotent and safe to call whenever
        if self.state != 1:
            return

        self.state = 2

        self.stop_evt.set()
        self.cap.release()

    def _cap_read(self):
        """cap.read() capturing the decoder position of the frame being read.

        The SINGLE choke point for reads off self.cap: any new read path (e.g. a
        post-seek re-read) must go through here or its frames silently lose their
        position. CAP_PROP_POS_MSEC / CAP_PROP_POS_FRAMES refer to the frame about
        to be decoded, so they are sampled BEFORE the read and committed (into
        self.extras) only when the read succeeds. This is the frame's presentation
        timestamp — the video offset jump-to-frame needs — which wall-clock `ts`
        cannot provide. Files only: stream/webcam positions are not meaningful here."""
        if not self.is_file:
            return self.cap.read()

        pos_msec  = self.cap.get(cv2.CAP_PROP_POS_MSEC)
        pos_frame = self.cap.get(cv2.CAP_PROP_POS_FRAMES)
        ret, image = self.cap.read()

        if ret and image is not None:
            self.extras = extras = {'frame_n': (frame_n := int(pos_frame))}

            # Primary: frame index over the container frame rate. Exact for CFR
            # sources and immune to the B-frame reordering and report-zero bugs
            # POS_MSEC shows on several backends (both reproduced by the repo's
            # own test clip). This DELIBERATELY treats every fps-reporting source
            # as CFR: a true-VFR file that reports an average rate gets a nominal
            # timeline (frame_n / avg_fps) that can drift from the container pts,
            # but stays exact under the inverse mapping (offset * avg_fps) used
            # to seek by frame, so jump-to-frame round-trips to the same frame -
            # while POS_MSEC on such files hits the bugs above with no way to
            # tell a true pts from a lie. The CFR path is taken only for a
            # PLAUSIBLE reported rate: an implausible one (>= FPS_SANE_CEILING or
            # non-finite - e.g. the ffmpeg-backend ~1000 fps VFR sentinel, the 1 ms
            # container timebase rather than an average) is not a rate anything
            # drifts around, so frame_n / native_fps would just be wrong; those
            # fall through to POS_MSEC like a no-rate container. POS_MSEC is used
            # only for containers with no usable rate, and only while it looks sane
            # (nonzero past frame 0); otherwise pts_s is omitted and frame_n
            # remains the consumer's only (still exact) position.
            if self.native_fps and self.native_fps < FPS_SANE_CEILING:
                extras['pts_s'] = frame_n / self.native_fps
            elif pos_msec > 0 or frame_n == 0:
                extras['pts_s'] = pos_msec / 1000

        return ret, image

    def _by_index_stride_for(self, fps):
        """The by-index stride for a file reporting ``fps``, or None if the gate declines it.

        Shared by __init__ and _open_dir_file so that every member of a directory source is
        gated exactly the way a lone file is: the stride is derived from THAT file's rate.
        """
        maxfps = self.maxfps

        if (self._by_index and self.is_file and not self._sync and maxfps and fps
                and fps < FPS_SANE_CEILING and fps > maxfps):
            return math.ceil(fps / maxfps)

        return None

    def _open_dir_file(self, index: int):
        self.dir_index = index
        active_source = self.dir_files[index]
        self.cap.release()
        self.cap = cv2.VideoCapture(active_source)
        if not self.cap.isOpened():
            raise RuntimeError(f'failed to open video in directory: {active_source}')

        self.source = f'file://{active_source}'
        self.ssource = active_source
        fps = self.cap.get(cv2.CAP_PROP_FPS) or None
        self.native_fps = fps
        self.fps = fps
        if self.maxfps is not None and fps is not None and fps > self.maxfps:
            # keys on the sync flag, the same test __init__ uses. Keying on sync_evt made
            # meta['src_fps'] flip mid-directory: by-index creates the event on the first
            # member, so later members reported their native rate instead of maxfps.
            if not self.is_file or not self._sync:
                self.fps = self.maxfps

        # A directory's members can each report their own rate, so re-run the by-index gate
        # for this file instead of carrying dir_files[0]'s decision for the whole directory:
        # the stride is ceil(fps / maxfps) of THIS file, and a file the gate declines falls
        # back to the wall-clock pacing it would have had on its own. Without this a 30 fps
        # file followed by a 60 fps one keeps stride 6 throughout and the second file is
        # delivered at 10 fps under maxfps=5.
        #
        # The sync_evt handshake, once created, is never torn down: the consumer side holds a
        # reference to it, so swapping it back to None mid-stream would strand a reader waiting
        # on the old event. Leaving it in place only adds back-pressure, which is harmless on
        # the files the gate declines.
        if (stride := self._by_index_stride_for(fps)) is not None:
            if stride != self.index_stride:
                logger.info(f'maxfps by index: keeping 1 frame in every {stride} of '
                            f'{fps:.1f} fps -> {fps / stride:.2f} fps effective ({active_source})')

            self.index_stride  = stride
            self.ns_per_fps    = None
            self.ns_per_maxfps = None

            if self.sync_evt is None:  # by-index needs the handshake, see the __init__ gate
                self.sync_evt = Event()

                self.sync_evt.set()

        else:
            if self.index_stride is not None:
                logger.info(f'maxfps by index not engaged for {active_source} '
                            f'({"no fixed rate" if fps is None else f"{fps:.1f} fps"}); '
                            f'maxfps applied on the wall clock instead')

            self.index_stride  = None
            self.ns_per_maxfps = None if self.maxfps is None else 1_000_000_000 // self.maxfps

            # Restore this member's own pacing. The by-index branch above nulls ns_per_fps,
            # and the trailing `if not self.sync_evt:` at the end of this method cannot put it
            # back once by-index has created the event, so a declined member would otherwise
            # run with no rate of its own.
            if not self._sync:
                self.ns_per_fps = 1_000_000_000 // (fps or 15)

        # by-index counts source frames, so restart the count with every file: each member of
        # a directory selects from its own index 0, on the first pass and on every loop pass.
        # Same reason as the reset next to the VideoCapture reopen in read_one.
        self.frame_i = 0

        if not self.sync_evt:
            self.ns_per_fps = 1_000_000_000 // (self.native_fps or 15)

    def read_one(self):
        def wait(is_eof=False) -> bool:  # returns if should return frame or keep looping
            t = time_ns()

            if (stride := self.index_stride) is not None:
                # At EOF the caller wants the last real frame held for the consumer, not a
                # by-index decision. Skip the stride test (the phantom EOF index is off-stride
                # (stride-1)/stride of the time) and go straight to the handshake, otherwise the
                # (None, ...) sentinel evicts the last frame from the 1-slot deque before it is
                # popped.
                if not is_eof:
                    self.frame_i += 1

                    if (self.frame_i - 1) % stride:  # dropped: no back-pressure wait, no sleep
                        return False

                if (sync_evt := self.sync_evt) is not None:
                    sync_evt.wait()
                    sync_evt.clear()

                return True

            if self.is_file:
                # The paced, no-skip contract belongs to sync=True and must be selected on the
                # sync flag. It used to key on `sync_evt is not None`, which was equivalent
                # until by-index started creating that event on a sync=False reader: a
                # directory member the gate declines then ran sync=True semantics, handing the
                # chain every frame at maxfps instead of subsampling it.
                if self._sync:
                    if (sync_evt := self.sync_evt) is not None:
                        sync_evt.wait()
                        sync_evt.clear()

                    if (ns_per_maxfps := self.ns_per_maxfps) is not None:  # sleep until we reach maxfps
                        if (tleft := (ns_per_maxfps - ((t := time_ns()) - (tmaxfps := self.tmaxfps)))) > 0:
                            sleep(tleft / 1_000_000_000)

                            t = tmaxfps + ns_per_maxfps

                        self.tmaxfps = t

                    return True

                if (tleft := ((ns_per_fps := self.ns_per_fps) - (t - (tfps := self.tfps)))) > 0:  # cap video file to maximum of its specified fps
                    sleep(tleft / 1_000_000_000)

                    t = tfps + ns_per_fps

                self.tfps = t

            if (ns_per_maxfps := self.ns_per_maxfps) is not None:  # skip frames until we reach maxfps, keep remainder, if more than one frame over then discard extra time
                if (tdiff := t - (tmaxfps := self.tmaxfps)) < ns_per_maxfps:
                    return False

                self.tmaxfps = tmaxfps + (tdiff // ns_per_maxfps) * ns_per_maxfps

            # Back-pressure for a sync=False reader that by-index gave a handshake to. It runs
            # on the delivering path only, after the skip decision above: a dropped frame must
            # not consume a credit the consumer never posts back for it.
            if not self._sync and (sync_evt := self.sync_evt) is not None:
                sync_evt.wait()
                sync_evt.clear()

            return True

        while True:
            # With maxfps_by_index the stride already knows, before any decoding, that
            # this frame will be dropped. grab() advances the file without rebuilding
            # the image, which is the expensive half: a 1 h 30 fps clip otherwise
            # decodes 108000 frames to hand 3600 downstream, and the run becomes
            # decode-bound rather than GPU-bound.
            if (stride := self.index_stride) is not None and self.frame_i % stride:
                self.frame_i += 1

                if self.cap.grab():
                    continue

                # A failed grab() is EOF, and EOF on a skipped frame is the common
                # case: the last frame of a file lands off-stride (stride-1)/stride
                # of the time. Leaving the loop here would reach `return image`
                # with image unbound, killing the reader thread before it appends
                # the EOF sentinel, so the consumer would block forever. Falling
                # through instead keeps the directory transition, the loop restart
                # and wait(is_eof=True) on the skip path too.
                ret, image = False, None

            else:
                ret, image = self._cap_read()

            if not ret or image is None:
                if not self.is_file:
                    return None

                # We do the wait()s below in order to maintain the last frame for the same amount of time as others
                # because otherwise cv2 returns (False, None) immediately after the last frame has been read from a
                # file which does not allow enough time for the last frame to be picked up, thus losing it. The last
                # frame may be lost on a realtime video if it takes too long to query it but it will never be lost on
                # a `sync` video.

                if self.is_dir and self.dir_index + 1 < len(self.dir_files):
                    success = False
                    while self.dir_index + 1 < len(self.dir_files):
                        next_index = self.dir_index + 1
                        active_source = self.dir_files[next_index]
                        try:
                            logger.info(f'video transition: opening next file in directory: {active_source}')
                            self._open_dir_file(next_index)
                            ret, image = self._cap_read()
                            if ret and image is not None:
                                success = True
                                break
                            self.dir_index = next_index
                        except Exception as e:
                            logger.error(f'Error during video transition to {active_source}: {e}')
                            self.dir_index = next_index

                    if success:
                        if wait():
                            break
                        continue

                    # If we ran out of directory files, fall through to loop/restart logic below

                # Handle loop/restart
                if loop := self.loop:
                    self.loop = loop - 1

                    if not self.loop:
                        wait(is_eof=True)

                        return None

                try:
                    if self.is_dir:
                        logger.info(f'directory loop: {self.source_dir_path}{f"  (last loop)" if loop == 2 else f"  ({self.loop} left)" if loop else ""}')
                        success = False
                        self.dir_index = 0
                        while self.dir_index < len(self.dir_files):
                            active_source = self.dir_files[self.dir_index]
                            try:
                                self._open_dir_file(self.dir_index)
                                ret, image = self._cap_read()
                                if ret and image is not None:
                                    success = True
                                    break
                                self.dir_index += 1
                            except Exception as e:
                                logger.error(f'Error during directory loop restart for {active_source}: {e}')
                                self.dir_index += 1

                        if not success:
                            return None
                    else:
                        logger.info(f'video loop: {self.source}{f"  (last loop)" if loop == 2 else f"  ({self.loop} left)" if loop else ""}')
                        self.cap.release()
                        self.cap = cv2.VideoCapture(self.ssource)
                        self.frame_i = 0  # by-index counts source frames; restart it with the cap so each pass selects from source index 0
                        if not self.cap.isOpened():
                            raise RuntimeError(f'failed to reopen video source: {self.source!r}')
                        ret, image = self._cap_read()
                        if not ret or image is None:
                            return None

                except Exception as e:
                    logger.error(f'Error during loop restart: {e}')
                    wait(is_eof=True)

                    return None

            if wait():
                break

        return image

    def thread_reader(self):
        cond = self.cond

        if size := (maxsize := self.maxsize) or self.resize:
            width, aspect, height, interp = size

            width  = int(width)
            height = int(height)
            aspect = aspect != '+'
            interp = (
                cv2.INTER_NEAREST
                if interp is None or (interp := interp.upper()[:1]) == 'N' else
                cv2.INTER_CUBIC
                if interp == 'C' else
                cv2.INTER_LINEAR
            )

        while True:
            image  = None if self.stop_evt.is_set() else self.read_one()
            tframe = time_ns()

            if image is not None:
                shape = image.shape

                if size:
                    h, w, *_ = shape

                    if maxsize:
                        if (hgt := (oh := h) > height) + (wgt := (ow := w) > width):
                            if aspect:
                                if not hgt:
                                    h = int(h * width / w)
                                elif not wgt:
                                    w = int(w * height / h)
                                else:
                                    h = int(h * (s := min(width / w, height / h)))
                                    w = int(w * s)

                            if (newsize := (min(width, w), min(height, h))) != (ow, oh):
                                image = cv2.resize(image, newsize, interpolation=interp)

                    else:  # resize
                        if (hne := h != height) + (wne := w != width):
                            if not aspect:
                                newsize = (width, height)
                            elif not hne:
                                newsize = (width, int(h * width / w))
                            elif not wne:
                                newsize = (int(w * height / h), height)
                            else:
                                newsize = (int(w * (s := min(width / w, height / h))), int(h * s))

                            image = cv2.resize(image, newsize, interpolation=interp)

                if len(shape) != 3:
                    self.as_bgr = False  # because not validated on init
                elif not self.as_bgr:
                    image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)

            extras_copy = dict(self.extras) if self.extras else {}
            extras_copy['source'] = self.source
            extras_copy['fps'] = self.fps
            self.deque.append((image, tframe, extras_copy))

            if cond is not None:
                with cond:
                    cond.notify_all()

            if image is None:
                break

        self.state = 2

    @property
    def playing(self) -> bool:
        return self.state == 1

    @property
    def stopped(self) -> bool:
        return self.state == 2

    @property
    def frame_available(self) -> bool:
        return bool(self.deque)

    def read(self, with_tframe=False):  # -> np.ndarray | tuple[np.ndarray, int, dict] | None
        # BREAKING CHANGE (unreleased): with_tframe=True returns (image, tframe, extras) instead of (image, tframe);
        # extras always carries 'source'/'fps' (added in thread_reader) and, for file sources only, 'frame_n'/'pts_s';
        # this absorbs future additions without another arity break (same shape as the seekable-replay branch).
        # with_tframe=False (the default) is unaffected.
        if self.state == 0:
            raise RuntimeError('can not read from video before it is started')
        elif self.state == 2:
            return None

        if (image_n_tframe := self.deque.popleft()) is None:
            self.state = 2

            self.cap.release()

            return None

        if (sync_evt := self.sync_evt) is not None:
            sync_evt.set()

        return image_n_tframe if with_tframe else image_n_tframe[0]

    @staticmethod
    def get_info(source: str, pattern: str | None = None, recursive: bool | None = None) -> tuple[int, int, str, float]:  # (height, width, format, fps)
        path = source[7:] if is_video_file(source) else source
        if os.path.isdir(path):
            # First file per scan_dir_videos, so this reports the file __init__ will actually play
            dir_files = scan_dir_videos(path, pattern, recursive)

            if not dir_files:
                raise RuntimeError(f'no valid video files found in directory: {source}')

            for entry_path in dir_files:
                test_cap = cv2.VideoCapture(entry_path)
                if test_cap.isOpened():
                    ret, image = test_cap.read()
                    if ret:
                        width  = image.shape[1]
                        height = image.shape[0]
                        fps    = test_cap.get(cv2.CAP_PROP_FPS)
                        is_bgr = test_cap.get(cv2.CAP_PROP_CONVERT_RGB)
                        format = 'GRAY' if len(image.shape) == 2 else 'BGR' if is_bgr else 'RGB'
                        test_cap.release()
                        return height, width, format, fps
                    test_cap.release()
            raise RuntimeError(f'no valid video files found in directory: {source}')

        cap = cv2.VideoCapture(int(source[9:]) if is_video_webcam(source) else path)

        try:
            ret, image = cap.read()

            if not ret:
                raise RuntimeError('failed to read a frame')

            width  = image.shape[1]
            height = image.shape[0]
            fps    = cap.get(cv2.CAP_PROP_FPS)
            is_bgr = cap.get(cv2.CAP_PROP_CONVERT_RGB)
            format = 'GRAY' if len(image.shape) == 2 else 'BGR' if is_bgr else 'RGB'

            return height, width, format, fps

        finally:
            cap.release()


class MultiVideoReader:
    """Read multiple videos simultaneously returning time-synchronized frames until one of them is exhausted."""

    def __init__(self, sources: list[str], sources_kwargs: list[dict[str, Any]] | None = None):
        kwargss     = [{}] * len(sources) if sources_kwargs is None else sources_kwargs
        self.cond   = cond = Condition()
        self.videos = [VideoReader(source, cond, **kwargs) for source, kwargs in zip(sources, kwargss)]
        self.state  = 0  # 0 = before start, 1 = playing, 2 = stopped / done

    def __iter__(self):
        return self

    def __next__(self):
        if (item := self.read()) is None:
            raise StopIteration

        return item

    def start(self):  # idempotent and safe to call whenever
        if self.state != 0:
            return

        self.state = 1

        for video in self.videos:
            video.start()

    def stop(self):  # idempotent and safe to call whenever
        if self.state != 1:
            return

        self.state = 2

        for video in self.videos:
            video.stop()

    @property
    def playing(self) -> bool:
        return self.state == 1

    @property
    def stopped(self) -> bool:
        return self.state == 2

    @property
    def frame_available(self) -> bool:
        return all(vid.frame_available for vid in self.videos)

    def read(self, with_tframe=False):  # -> list[np.ndarray | tuple[np.ndarray, int, dict]] | None
        if self.state == 0:
            raise RuntimeError('can not read from videos before they are started')
        elif self.state == 2:
            return None

        cond   = self.cond
        videos = self.videos

        while not all(video.frame_available for video in videos):
            with cond:
                cond.wait()

        images = [video.read(with_tframe) for video in videos]

        if any(image is None for image in images):
            self.stop()

            return None

        return images


# --- CUT HERE ---------------------------------------------------------------------------------------------------------

from openfilter.filter_runtime.filter import is_cached_file, Frame, FilterConfig, Filter
from openfilter.filter_runtime.utils import adict, split_commas_maybe

__all__ = __all__ + ['VideoInConfig', 'VideoIn']

is_video_or_cached_file = lambda s: is_video(s) or is_cached_file(s)


class VideoInConfig(FilterConfig):
    class Source(adict):
        class Options(adict):
            bgr:        bool | None
            sync:       bool | None
            loop:       bool | int | None
            maxfps:     float | None
            maxfps_by_index: bool | None
            maxsize:    str | None
            resize:     str | None
            region:     str | None
            expiration: int | None
            pattern:    str | None
            recursive:  bool | None

        source:  str
        topic:   str | None
        options: Options | None

    sources: str | list[str | Source]

    # setting these here will make them default for all videos (overridable individually)
    bgr:     bool | None
    sync:    bool | None
    loop:    bool | int | None
    maxfps:  float | None
    maxfps_by_index: bool | None
    maxsize: str | None
    resize:  str | None
    pattern:    str | None
    recursive:  bool | None

    # Logical source URI to report as meta['src'] instead of the physical source
    # (set by an orchestrator; see resolve_override_source_uri). FILTER_OVERRIDE_SOURCE_URI
    # is the direct value, FILTER_OVERRIDE_SOURCE_URI_FILE a file to read it from.
    override_source_uri:      str | None
    override_source_uri_file: str | None


class VideoIn(Filter):
    """Single or multiple video input filter. Videos are assigned to topics via the ';' mapping character in `sources`.
    The default topic mapping if nothing specified is 'main'. All video sources must have unique topics. '!' allows
    setting options directly in the source string.

    config:
        sources:
            The source(s) of the video(s), comma delimited, can be file://, rtsp:// stream, s3:// bucket object,
            a webcam:// index, or a file:// directory path for sequential playback of the video files within it.

            Examples:
                'file://a.mp4!sync!loop=3, rtsp://b.com!no-bgr;c, s3://bucket/video.mp4!region=us-west-2, webcam://0;e, file:///path/to/dir!pattern=*.mp4'

                    is the same as

                ['file://a.mp4!sync!loop=3', 'rtsp://b.com!no-bgr;c', 's3://bucket/video.mp4!region=us-west-2', 'webcam://0;e', 'file:///path/to/dir!pattern=*.mp4']

                    is the same as

                [{'source': 'file://a.mp4', 'topic': 'main', 'options': {'sync': True, 'loop': 3}},
                 {'source': 'rtsp://b.com', 'topic': 'c', 'options': {'bgr': False}},
                 {'source': 's3://bucket/video.mp4', 'topic': 'main', 'options': {'region': 'us-west-2'}},
                 {'source': 'webcam://0', 'topic': 'e', 'options': {}},
                 {'source': 'file:///path/to/dir', 'topic': 'main', 'options': {'pattern': '*.mp4'}}]

                    For 'options' see below.

            `sources` individual options (text appended after source, e.g. 'file:///myvideo.mp4!no-bgr!sync!loop=3'):
                '!bgr', '!no-bgr':
                    Set `bgr` option for this source.

                '!sync', '!no-sync':
                    Set `sync` option for this source.

                '!loop', '!no-loop', '!loop=3':
                    Set `loop` option for this source.

                '!maxfps=10':
                    Set `maxfps` option for this source.

                '!maxfps_by_index', '!no-maxfps_by_index':
                    Set `maxfps_by_index` option for this source.

                '!maxsize=1280x720', '!maxsize=1280+720C':
                    Set `maxsize` option for this source.

                '!resize=1280x720lin', '!resize=1280+720':
                    Set `resize` option for this source.

                '!region=us-west-2':
                    Set AWS region for S3 sources. Only applies to s3:// sources.

                '!expiration=7200':
                    Set presigned URL expiration time in seconds for S3 sources. Default is 3600 (1 hour).
                    Only applies to s3:// sources.

                '!pattern=*.mp4':
                    Set pattern option for directory source.

                '!recursive', '!no-recursive':
                    Set recursive option for directory source.

        bgr:
            True means images in BGR format, False means RGB. Doesn't really affect anythong other than procesing speed
            since images should always be converted to the needed format. Don't touch this unless you have an explicit
            need and understanding of why you need to change it. Set here to apply to all sources or can be set
            individually per source. Global env var default FILTER_BGR / VIDEO_IN_BGR.

        sync:
            Only has meaning for file:// sources. If True then frames will be delivered one by one without skipping or
            waiting to maintain realtime, in this way all frames will be read and presented. Set here to apply to all
            sources or can be set individually per source. Global env var default FILTER_SYNC / VIDEO_IN_SYNC.

            NOTE: When a file source is read in `sync` mode, the original fps is passed downstream so that any video
            created from this file will have the original framerate regardless of the potential slowness of the
            processing applied to it. This is intended for offline processing of all frames of a video at the maximum
            speed possible while maintaining original video framerate.

        loop:
            Only has meaning for file:// sources. True or 0 means infinite loop, False means don't loop and go through
            the video only once, otherwise an int value loops through the video that number of times. Set here to apply
            to all sources or can be set individually per source. Global env var default FILTER_LOOP / VIDEO_IN_LOOP.

        maxfps:
            Restrict video to this FPS. Works for all types of video and if playing a file:// video in `sync` mode then
            will present the individual frames at this frame rate but will not skip any frames. Set here to apply to
            all sources or can be set individually per source. Global env var default FILTER_MAXFPS / VIDEO_IN_MAXFPS.

        maxfps_by_index:
            A `sync=False` optimisation, for file:// sources with `maxfps` set below the file's own rate. Off by
            default, in which case `maxfps` behaves as it always has: a limit on frames delivered per second of WALL
            CLOCK, so an hour of file takes an hour of wall clock however fast the machine is. Turned on, the same
            selection is taken from the source timeline instead, keeping 1 frame in every ceil(fps / maxfps) and reading
            as fast as the box allows. Use it for offline processing of a recording; leave it off for anything that has
            to track real time. It only engages under `sync=False`: `sync=True` documents a no-skip guarantee ("will not
            skip any frames"), so with `sync=True` this flag is a no-op and that paced, lossless path is preserved. It
            is also a no-op on a container that reports no real frame rate (the >= 1000 fps VFR sentinel).

        maxsize:
            Maximum image size to allow, above this will be resized down. Valid codes are 'WxH' which will
            proportionally resize maintaining aspect ratio so that neither dimension exceeds the max, 'W+H' will resize
            without maintaining aspect ratio. Optional suffixes 'near', 'lin' and 'cub' specify interpolation, default
            is 'near'est neighbor. Set here to apply to all sources or can be set individually per source. Global env
            var default FILTER_MAXSIZE / VIDEO_IN_MAXSIZE.

        resize:
            Same as `maxsize` but is always applied unconditionally regardless of input size.Can not be specified
            together with `maxsize`, it is one or the other. Set here to apply to all sources or can be set individually
            per source. Global env var default FILTER_RESIZE / VIDEO_IN_RESIZE.

        pattern:
            Only has meaning for folder/directory file:// sources. A glob pattern (e.g. `*.mp4`) or regular expression
            to match and filter video files in the directory. Set here to apply to all sources or can be set individually
            per source. Global env var default FILTER_PATTERN / VIDEO_IN_PATTERN.

        recursive:
            Only has meaning for folder/directory file:// sources. If True, scans the directory and its subdirectories
            recursively for matching files. Set here to apply to all sources or can be set individually per source.
            Global env var default FILTER_RECURSIVE / VIDEO_IN_RECURSIVE.

        override_source_uri:
            Logical source URI to report as each frame's meta['src'] instead of the physical source opened by the
            reader. Lets an orchestrator preserve per-file identity when the physical path is generic (e.g. a batch
            claimer downloads every object to /ws/input). Env var FILTER_OVERRIDE_SOURCE_URI, or
            FILTER_OVERRIDE_SOURCE_URI_FILE to read it from a file.

    Emitted frame meta:
        Every frame carries meta['id'] (delivery counter, rate depends on the consuming chain), meta['ts'] (wall-clock
        seconds at read), meta['src'], meta['src_fps'], meta['width'] and meta['height'].

        meta['width'] / meta['height'] are the dimensions of the image as emitted, i.e. AFTER `maxsize` or `resize`
        has been applied, so they are the pixel space every downstream coordinate (detection boxes, keypoints, crop
        regions) is expressed in. They are NOT the source video's dimensions, and under `maxsize` they are not the
        configured cap either: `maxsize='1280x720'` preserves aspect ratio, so a 2592x1520 source is emitted at
        1227x720 and only the constrained axis reaches the cap. A consumer mapping boxes back onto the original
        video therefore cannot derive this from the source resolution or from the pipeline config - it has to read
        it off the frame. A filter that resizes mid-graph is expected to update both keys.

        File sources (file:// and s3://) additionally carry the decoder position of the delivered frame - the video
        offset that cannot be reconstructed downstream:

            meta['src_frame']   - 0-based source frame index (CAP_PROP_POS_FRAMES sampled before the read, exact).
            meta['src_seconds'] - the frame's position WITHIN the source, in seconds: src_frame / container fps
                (nominal CFR timeline; see VideoReader._cap_read for the VFR position), or CAP_PROP_POS_MSEC when the
                container reports no frame rate. This is a source offset, not the wall-clock meta['ts'] nor the raw
                container PTS. Omitted when neither is trustworthy - src_frame is then still present and exact.

        With `loop`, the cap is reopened each pass, so src_frame and src_seconds restart at 0 every loop (they are the
        position WITHIN the file) while meta['id'] keeps counting across passes: a looped timeline is non-monotonic
        in src_frame/src_seconds but monotonic in id, so consumers indexing a looped source must key off id, not src_frame.
        A directory source reproduces this same restart at every file transition even without `loop`: src_frame and
        src_seconds reset to 0 (and meta['src'] changes) each time playback moves to the next file in the directory,
        while meta['id'] keeps counting across the whole run - so consumers of a directory source must key off id too.

        Stream and webcam sources have no meaningful decoder position: both keys are absent, all other meta unchanged.

    Environment variables (FILTER_* or legacy VIDEO_IN_* prefix, legacy takes precedence):
        FILTER_BGR      / VIDEO_IN_BGR
        FILTER_SYNC     / VIDEO_IN_SYNC
        FILTER_LOOP     / VIDEO_IN_LOOP
        FILTER_MAXFPS   / VIDEO_IN_MAXFPS
        FILTER_MAXFPS_BY_INDEX / VIDEO_IN_MAXFPS_BY_INDEX
        FILTER_MAXSIZE  / VIDEO_IN_MAXSIZE
        FILTER_RESIZE   / VIDEO_IN_RESIZE
        FILTER_PATTERN  / VIDEO_IN_PATTERN
        FILTER_RECURSIVE / VIDEO_IN_RECURSIVE

    S3 Configuration:
        For s3:// sources, AWS credentials are required. Set these environment variables:
        
        AWS_ACCESS_KEY_ID - Your AWS access key ID
        AWS_SECRET_ACCESS_KEY - Your AWS secret access key  
        AWS_DEFAULT_REGION - Default AWS region (optional, can be overridden per source)
        AWS_PROFILE - AWS credentials profile to use (alternative to access keys)
        
        S3 sources are converted to presigned HTTPS URLs and streamed directly without local storage.
        
        Example S3 usage:
            openfilter run - VideoIn --sources s3://my-bucket/video.mp4!region=us-west-2 - Webvis
    """

    FILTER_TYPE = 'Input'

    @classmethod
    def normalize_config(cls, config):
        sources = split_commas_maybe(config.get('sources'))  # we do not assume how Filter will normalize sources/outputs in the future
        config  = VideoInConfig(super().normalize_config(dict_without(config, 'sources')))

        if sources is not None:
            config.sources = sources

        if not sources:
            raise ValueError('must specify at least one source')
        if not config.outputs:
            raise ValueError('must specify at least one output')

        for idx, source in enumerate(sources):
            if isinstance(source, dict):
                if not isinstance(source, VideoInConfig.Source):
                    sources[idx] = VideoInConfig.Source(source)  # because silly user might have passed in dicts

            else:
                source, topic   = Filter.parse_topics(source, 1, False)
                source, options = Filter.parse_options(source)
                sources[idx]    = VideoInConfig.Source(source=source, topic=topic and topic[0],
                    options=VideoInConfig.Source.Options(options))

        for source in sources:
            if (topic := source.topic) is None:
                source.topic = 'main'
            if not isinstance(options := source.options, VideoInConfig.Source.Options):
                source.options = options = VideoInConfig.Source.Options() if options is None else VideoInConfig.Source.Options(options)
            if any((option := o) not in ('bgr', 'sync', 'loop', 'maxfps', 'maxfps_by_index', 'maxsize', 'resize', 'region', 'expiration', 'pattern', 'recursive') for o in options):
                raise ValueError(f'unknown option {option!r} in {source!r}')

        if len(set(source.topic for source in sources)) != len(sources):
            raise ValueError(f'duplicate video topics in {sources!r}')
        if not all(is_video_or_cached_file(source.source) for source in sources):
            raise ValueError('this filter only accepts video sources')

        return config

    def init(self, config):
        super().init(FilterConfig(config, sources=None))

    def setup(self, config):
        vsources = []
        topics   = []
        optionss = []

        for source in config.sources:
            vsources.append(source.source)
            topics.append(source.topic or 'main')
            optionss.append(source.options or {})

        default_options      = {'bgr': config.bgr, 'sync': config.sync, 'loop': config.loop, 'maxfps': config.maxfps,
            'maxfps_by_index': config.maxfps_by_index, 'maxsize': config.maxsize, 'resize': config.resize,
            'pattern': config.get('pattern'), 'recursive': config.get('recursive')}
        self.mvreader        = MultiVideoReader(vsources, [{**default_options, **options} for options in optionss])
        self.tops_n_vids     = tuple(zip(topics, self.mvreader.videos))
        self.id              = -1  # frame id
        self._camera_connected = 0
        self.override_source_uri = resolve_override_source_uri(config)

        # The override identifies a SINGLE logical source (an orchestrator's batch claimer
        # downloads one file and records its real URI). With multiple videos, applying one
        # override to every frame would mislabel each with the same meta['src'], so only honor
        # it for a single source; otherwise fall back to each video's real source and warn once.
        # Mirrors the ImageIn guard.
        self._apply_override = bool(self.override_source_uri) and len(self.mvreader.videos) <= 1 and not self.mvreader.videos[0].is_dir
        if self.override_source_uri and not self._apply_override:
            reason = "a directory source" if len(self.mvreader.videos) == 1 and self.mvreader.videos[0].is_dir else f"{len(self.mvreader.videos)} sources"
            logger.warning(
                f"FILTER_OVERRIDE_SOURCE_URI[_FILE] is set but this VideoIn has {reason}; "
                "the override identifies a single source, so meta['src'] will report each video's real source instead")

        self.mvreader.start()

    def shutdown(self):
        self.mvreader.stop()

    def process(self, frames):
        def get():
            if (image_n_tframes := self.mvreader.read(True)) is None:
                self._camera_connected = 0
                self.exit('video ended')

            self._camera_connected = 1
            self.id = id = self.id + 1

            def meta(vid, tfrm, extras, img):
                src = extras.get('source', vid.source) if extras else vid.source
                src_fps = extras.get('fps', vid.fps) if extras else vid.fps
                height, width = img.shape[:2]  # AFTER maxsize/resize: the frame downstream coordinates are in
                meta = {'id': id, 'ts': tfrm / 1_000_000_000, 'src': self.override_source_uri if self._apply_override else src, 'src_fps': src_fps,
                    'width': width, 'height': height}

                if extras and 'frame_n' in extras:  # file sources: decoder position of this frame = the video offset jump-to-frame needs
                    meta['src_frame'] = extras['frame_n']

                    if (pts_s := extras.get('pts_s')) is not None:
                        meta['src_seconds'] = pts_s

                return meta

            return {topic: Frame(img,
                {'meta': meta(vid, tfrm, extras, img)},
                'GRAY' if len(img.shape) == 2 else 'BGR' if vid.as_bgr else 'RGB'
            ) for (topic, vid), (img, tfrm, extras) in zip(self.tops_n_vids, image_n_tframes)}

        return get


if __name__ == '__main__':
    VideoIn.run()