video_helper package

Subpackages

Submodules

Module contents

video_helper

Multi-backend frame extraction (VidGear / PyAV / ffmpeg-pipe), video conversion, subtitle muxing, and lightweight image-ops glue for the AI Helpers suite.

Multi-surface exposure

Every public function below is reachable from four surfaces:

  • Python library import (this module).

  • Argparse CLI: video-helper (stdlib-only, always installed).

  • Click CLI: video-helper-click (needs the [cli] extra).

  • FastAPI HTTP: video_helper.api (needs the [api] extra), which also serves a minimal browser “video bench” GUI at GET /gui.

Usage Example

>>> import video_helper as vh
>>> for frame in vh.extract_frames("clip.mp4", frame_interval=1.0):
...     # frame.shape == (H, W, 3) — BGR uint8 (OpenCV convention)
...     do_something(frame)

See EXAMPLES.md at the repo root for the full cookbook (sparse access, torch / pil destinations, batched yields, hwaccel, http_headers for yt-dlp-resolved sources, scale-fit-and-pad, …).

Author

Warith Harchaoui, Ph.D. — https://linkedin.com/in/warith-harchaoui/

video_helper.black_video(duration, width, height, output_video, frame_rate=30)[source]

Generate a silent solid-black video of duration seconds.

Parameters:

  • duration (float) – Output duration in seconds.

  • width (int) – Output frame width in pixels (rounded down to even — H.264 yuv420p requires even dimensions).

  • height (int) – Output frame height in pixels (rounded down to even).

  • output_video (str) – Path to the output video file (.mp4 recommended).

  • frame_rate (int, optional) – Output frame rate (default 30).

Return type:

None

Notes

Useful as a “buffer” / breathing clip between two visuals in a montage, or as a placeholder when an asset is missing. Encoded as H.264 yuv420p with no audio track.

Examples

>>> black_video(0.5, 1920, 1080, "buffer.mp4")
video_helper.burn_subtitles(input_video, subtitles_file, output_video, force_style=None)[source]

Burn subtitles from an .srt / .vtt / .ass file into the video frames.

Parameters:

  • input_video (str) – Path to the input video file.

  • subtitles_file (str) –

    Path to a subtitles file in one of the formats libass understands:

    • .srt — plain SubRip. Renders in the libass default style; <font color="…"> tags are honored.

    • .vtt — WebVTT. Cue-class colors (<c.red>…</c>) and any inline ::cue rules from a STYLE block are honored. The companion srt2vtt() writes both pieces in one shot.

    • .ass / .ssa — Advanced SubStation Alpha. All per-cue formatting (font, color, outline, position) is honored as authored.

  • output_video (str) – Path to the output video file (.mp4 recommended).

  • force_style (str, optional) – ASS-style override forwarded to the subtitles filter’s force_style argument — e.g. "FontName=Helvetica,FontSize=24,PrimaryColour=&H00FFFFFF&". Useful for SRT (which has no native styling) or to override a global property of a VTT/ASS file without editing it. Per-cue colors from VTT/ASS still win against force_style keys they explicitly set.

Return type:

None

Notes

A single backend (libass through ffmpeg’s subtitles filter) handles all three formats, so there is no need for separate burn_srt / burn_vtt / burn_ass functions. The filter mounts the file by path; we escape : to \: and ' to \' so absolute paths on macOS / Windows behave. Video is re-encoded (the filter rewrites every frame), audio is copied if present.

Examples

Plain SRT, default style:

>>> burn_subtitles("clip.mp4", "subs.srt", "captioned.mp4")

Colored WebVTT (cue classes carry their own colors):

>>> burn_subtitles("clip.mp4", "subs.vtt", "captioned.mp4")

Force a font + size on top of any source format:

>>> burn_subtitles("clip.mp4", "subs.vtt", "captioned.mp4",
...                force_style="FontName=Helvetica,FontSize=28,Outline=2")
video_helper.compress_video(input_video, output_video=None, *, target_size_mb=97.0, audio_bitrate='128k', vcodec='libx265', min_video_bitrate_kbps=200, overwrite=True)[source]

Compress a video to a target file size via two-pass encoding.

Solves for the video bitrate that makes video + audio together fit inside target_size_mb, given the source duration, then runs a standard ffmpeg two-pass encode at that bitrate. Defaults to HEVC (libx265), tagged hvc1 (ffmpeg’s default HEVC tag, hev1, is not recognized by QuickTime / Apple players), and moves the moov atom to the front of the file (+faststart) so playback can start before the download finishes. Built for “the compressed file that gets embedded in a web video player”, not an archival master.

Parameters:

  • input_video (str) – Path to the source video.

  • output_video (str, optional) – Path to write the compressed video to. Defaults to <input>-compressed.mp4 next to the source.

  • target_size_mb (float, optional) – Target output file size in megabytes (default 97 — comfortably under the 100 MB caps common on chat apps and code-hosting attachments).

  • audio_bitrate (str, optional) – Audio bitrate passed to the AAC encoder, ffmpeg syntax (default "128k"). Subtracted from the size budget before solving for the video bitrate.

  • vcodec (str, optional) – Video codec (default "libx265" — HEVC compresses noticeably better than H.264 at the same target size). Pass "libx264" for maximum legacy-player compatibility at a larger file for the same quality; the hvc1 tag is only applied when encoding HEVC.

  • min_video_bitrate_kbps (int, optional) – Floor on the solved video bitrate (default 200) — guards against a near-zero bitrate when a long source is squeezed into a small target size, which would otherwise silently produce an unwatchable file instead of a clear error.

  • overwrite (bool, optional) – Overwrite output_video if it already exists (default True); when False and the file already exists, that path is returned as-is with no re-encode.

Returns:

Path to the compressed video (output_video).

Return type:

str

Raises:

  • AssertionError – If input_video is not a valid video file.

  • ffmpeg.Error – If either encoding pass fails.

Examples

>>> compress_video("meeting.mp4", "meeting-compressed.mp4", target_size_mb=97)
'meeting-compressed.mp4'
video_helper.concat_videos(input_videos, output_video, reencode=True, frame_rate=None)[source]

Concatenate input_videos end-to-end into output_video.

Parameters:

  • input_videos (List[str]) – Ordered list of input video paths.

  • output_video (str) – Path to the output video file (.mp4 recommended).

  • reencode (bool, optional) – Whether to re-encode (libx264). Default True — strongly recommended when the inputs come from different sources, since the concat demuxer’s stream-copy path requires identical codec, timebase, frame rate and resolution; mismatched inputs produce audio/video drift or hard ffmpeg errors. Set False only when the inputs are guaranteed bit-identical containers.

  • frame_rate (int, optional) – Force this output frame rate (only used when reencode=True).

Return type:

None

Notes

Uses the ffmpeg concat demuxer (text manifest) which is the only correct way to concatenate variable-length clips end-to-end without re-timing artefacts. The temporary manifest is written to a process-temp file and removed automatically.

Examples

>>> concat_videos(["intro.mp4", "body.mp4", "outro.mp4"], "final.mp4")
video_helper.dump_frames(frames_list, output_movie, fps=30)[source]

Save frames to a video file.

Parameters:

  • frames_list (List[np.ndarray]) – A list of frames as numpy arrays.

  • output_movie (str) – Path to the output video file.

  • fps (int, optional) – Frame rate of the output video file. Defaults to 30.

Return type:

None

Notes

The function uses VidGear to write the frames to a video file.

Usage

>>> frames = [frame1, frame2, frame3]
>>> dump_frames(frames, "output.mp4")
video_helper.extract_audio_track(input_video, output_audio, sample_rate=44100, channels=2, encoding='pcm_s16le')[source]

Extract the audio track of a video file into a standalone audio file.

Parameters:

  • input_video (str) – Path to the input video file (any container ffmpeg can read).

  • output_audio (str) – Path to the output audio file. The extension picks the container; .wav pairs naturally with encoding="pcm_s16le" for a lossless extract.

  • sample_rate (int, optional) – Output sample rate in Hz (default 44100).

  • channels (int, optional) – Output channel count (default 2 — stereo). Use 1 for mono.

  • encoding (str, optional) – Audio codec (default "pcm_s16le"). For non-WAV outputs use a codec compatible with the container (e.g. "aac" for .m4a, "libmp3lame" for .mp3).

Return type:

None

Notes

Source-of-truth companion to audio_helper.sound_converter for the case where the input is a video — sound_converter rejects video extensions in its input-validation pass, hence the dedicated function here. Drops the video stream (-vn) and re-encodes only the audio.

Examples

>>> extract_audio_track("interview.mp4", "interview.wav")
>>> extract_audio_track("clip.mov", "clip.mp3",
...                     encoding="libmp3lame", sample_rate=22050)
video_helper.extract_frames(video_path, start_index=None, end_index=None, start_instant=None, end_instant=None, stabilize=False, frame_step=1, frame_interval=None, frame_indices=None, frame_times=None, backend='auto', hwaccel=None, http_headers=None, output_width=None, output_height=None, pad_color='black', destination='numpy', device='cpu', batch_size=None, layout='image')[source]

Extract frames from a video, dispatching to the best available backend.

The function picks a backend automatically based on the requested access pattern and what’s installed locally; pass backend=... to override. Frames are yielded in the user’s preferred form via destination (numpy array or torch tensor on a chosen device), optionally batched to amortize the host→device transfer.

Backends

  • vidgear — OpenCV+VidGear with a producer thread. Fastest path for full sequential decode up to ~720p on macOS (uses AVFoundation under the hood) and the only backend that supports stabilize=True. Decodes from t=0 with no real seek.

  • pyav — direct ffmpeg libav bindings. Best default for windowed sequential, sparse reads, and any “torch on GPU” destination thanks to keyframe seek + hwaccel support.

  • ffmpeg-pipe — ffmpeg subprocess fallback. Useful when PyAV isn’t installed. Sequential only; honors hwaccel.

param video_path:

Path to the input video file.

type video_path:

str

param start_index:

Inclusive frame-index bounds. If None, defaults to start-of-file and end-of-file respectively.

type start_index:

int, optional

param end_index:

Inclusive frame-index bounds. If None, defaults to start-of-file and end-of-file respectively.

type end_index:

int, optional

param start_instant:

Same bounds expressed in seconds. When provided, they override the index form.

type start_instant:

float, optional

param end_instant:

Same bounds expressed in seconds. When provided, they override the index form.

type end_instant:

float, optional

param stabilize:

If True, runs VidGear’s software stabilizer. Forces backend="vidgear".

type stabilize:

bool, optional

param frame_step:

Sampling stride within the range (every Nth frame). Defaults to 1.

type frame_step:

int, optional

param frame_interval:

Sampling period in seconds. Overrides frame_step when given.

type frame_interval:

float, optional

param frame_indices:

Explicit set of frame indices to read (sparse / random access). When provided, range parameters are ignored.

type frame_indices:

list[int], optional

param frame_times:

Same as frame_indices but in seconds; converted internally.

type frame_times:

list[float], optional

param backend:

"auto" (default), "vidgear", "pyav", or "ffmpeg-pipe".

type backend:

str, optional

param hwaccel:

Hardware-accelerated decoder. Default None. Pass "auto" to enable platform-appropriate accel ("videotoolbox" on macOS, "cuda" on Linux+NVIDIA), or an explicit value. Honored only by pyav and ffmpeg-pipe. For destination="torch" with a GPU device, "auto" is enabled by default since the wall-time penalty observed on numpy-destination cells doesn’t apply (the frames go through one numpy stack and then host→device in one shot — see SPEED_ANALYSIS.md).

type hwaccel:

str, optional

param http_headers:

HTTP headers forwarded to the underlying decoder. Required for URLs that need a specific User-Agent / Referer / Cookie / Authorization — e.g. yt-dlp-resolved YouTube live streams, members-only / age-gated content, Vimeo private videos, Twitch. Joined into ffmpeg’s -headers CRLF string under the hood. Honored by pyav and ffmpeg-pipe; the vidgear backend logs a warning and ignores them (OpenCV’s HTTP layer doesn’t surface headers cleanly).

type http_headers:

dict[str, str], optional

param output_width:

Exact output frame size in pixels. Behavior:

  • Both set → scale-fit (aspect-preserving) then pad with pad_color so the output is exactly output_width × output_height. Typical for ML pipelines that need a fixed input shape.

  • Only one set → scale to that dimension preserving aspect ratio; the other dimension is derived. No padding.

  • Neither set (default) → frame keeps its native dimensions.

The transform runs in numpy via cv2.resize + cv2.copyMakeBorder post-decode. Same behavior across all backends (vidgear / pyav / ffmpeg-pipe).

type output_width:

int, optional

param output_height:

Exact output frame size in pixels. Behavior:

  • Both set → scale-fit (aspect-preserving) then pad with pad_color so the output is exactly output_width × output_height. Typical for ML pipelines that need a fixed input shape.

  • Only one set → scale to that dimension preserving aspect ratio; the other dimension is derived. No padding.

  • Neither set (default) → frame keeps its native dimensions.

The transform runs in numpy via cv2.resize + cv2.copyMakeBorder post-decode. Same behavior across all backends (vidgear / pyav / ffmpeg-pipe).

type output_height:

int, optional

param pad_color:

Padding color when scale-fit-and-pad applies (i.e. both output_width and output_height are set, and the source’s aspect ratio differs from the target). Accepts:

  • common names: "black" (default), "white", "red", "green", "blue", "yellow", "cyan", "magenta", "gray" / "grey"

  • "#RRGGBB" hex

  • "transparent" raises ValueError: it would require 4-channel BGRA output, breaking the (H, W, 3) contract; not implemented.

type pad_color:

str, optional

param destination:

Where frames land. Default "numpy".

  • "numpy" — BGR uint8 np.ndarray in OpenCV’s channels-last layout.

  • "torch" — RGB uint8 torch.Tensor in PyTorch’s channels-first layout. PyTorch imported lazily.

  • "pil" — PIL.Image.Image (mode "RGB", size=(W, H) per Pillow convention). Pillow imported lazily. batch_size not supported (Pillow has no batched type).

See the layout table for exact shapes.

type destination:

str, optional

param device:

Target device when destination="torch". "cpu" (default), "mps" (Apple Silicon), "cuda" (NVIDIA), or "auto" (cuda > mps > cpu). Ignored when destination="numpy".

type device:

str, optional

param batch_size:

If provided, yield a batched tensor / array per batch instead of one frame at a time. The last batch may be smaller. Strongly recommended with destination="torch" + GPU device: one host→device transfer per batch instead of one per frame (typical 5-20× win).

type batch_size:

int, optional

param layout:

Axis convention for batched yields (ignored when batch_size is None and for destination="pil"). "image" (default) — each batch is a stack of independent images; "video" — each batch is a video clip with a time axis.

Concrete shapes per (destination, layout, batch_size):

destination

layout

batch_size

yield

"numpy"

any

None

(H, W, 3) HWC, BGR uint8

"numpy"

"image"

N

(N, H, W, 3) NHWC, BGR uint8

"numpy"

"video"

N

(N, H, W, 3) THWC, BGR uint8 (same mem; T == N)

"torch"

any

None

(3, H, W) CHW, RGB uint8

"torch"

"image"

N

(N, 3, H, W) NCHW, RGB uint8 (batch of images)

"torch"

"video"

N

(3, N, H, W) CTHW, RGB uint8 (video clip; T == N)

"pil"

n/a

forbidden

PIL.Image mode=``”RGB”, size=``(W, H)
type layout:

str, optional

Yields:

numpy.ndarray – Successive frames as (H, W, 3) BGR uint8 arrays — same convention as OpenCV and the previous VidGear-only implementation.

Examples

>>> # Sequential time range — dispatcher picks PyAV (windowed)
>>> for frame in extract_frames("clip.mp4", start_instant=10, end_instant=20, frame_step=5):
...     process(frame)

>>> # Sparse access at specific times — routed to PyAV
>>> list(extract_frames("clip.mp4", frame_times=[1.5, 12.0, 47.0]))

>>> # Stream as torch tensors on Apple Silicon, batched for one transfer per 32 frames
>>> for batch in extract_frames("clip.mp4",
...                             destination="torch", device="mps", batch_size=32):
...     # batch.shape == (N, H, W, 3); N == 32 for all but the last batch
...     model(batch)
Parameters:

  • video_path (str)

  • start_index (int | None)

  • end_index (int | None)

  • start_instant (float | None)

  • end_instant (float | None)

  • stabilize (bool)

  • frame_step (int)

  • frame_interval (float | None)

  • frame_indices (Sequence[int] | None)

  • frame_times (Sequence[float] | None)

  • backend (str)

  • hwaccel (str | None)

  • http_headers (dict | None)

  • output_width (int | None)

  • output_height (int | None)

  • pad_color (str)

  • destination (str)

  • device (str)

  • batch_size (int | None)

  • layout (str)

Return type:

Iterator

video_helper.extract_optical_flow(input_video, output_path=None, *, method='dis', dis_preset='fast', raft_variant='small', device='cpu', clip_flow=None, start_instant=None, end_instant=None, frame_step=1, frame_interval=None, fps=None, output_width=None, output_height=None, wavelet='db2', overwrite=True)[source]

Compute dense optical flow over a video file and write it to disk.

File-level convenience wrapper around iter_frame_optical_flow() for the common “just give me flow for this video” case (CLI / API surfaces need a single input path and a single output path, not a frame iterator). The output kind is inferred from output_path’s extension, mirroring how video_helper.main.video_converter() infers its container:

  • .npy — the raw flow-only array, (T, H, W, 2) float32 (vx, vy), one entry per input frame (frame 1 is all zeros).

  • anything else (default .mp4) — an HSV-color-wheel visualization video (direction -> hue, magnitude -> value, see _flow_to_rgb()), viewable directly without loading numpy.

Parameters:

  • input_video (str) – Path to the source video.

  • output_path (str, optional) – Where to write the result. Defaults to <input>-flow.mp4 next to the source. Extension controls the output kind (see above).

  • method ({"dis", "farneback", "raft"}, default "dis") – Optical-flow backend — see iter_frame_optical_flow().

  • dis_preset ({"ultrafast", "fast", "medium"}, default "fast") – Speed/quality preset for method="dis". Ignored otherwise.

  • raft_variant ({"small", "large"}, default "small") – RAFT network variant. Only used for method="raft".

  • device (str, default "cpu") – Torch device for method="raft" only.

  • clip_flow (float or None, default None) – Symmetric pixel clip for outlier suppression — see iter_frame_optical_flow().

  • start_instant (float, optional) – Start time in seconds (forwarded to video_helper.main.extract_frames()).

  • end_instant (float, optional) – End time in seconds (forwarded to video_helper.main.extract_frames()).

  • frame_step (int, default 1) – Take every Nth frame (forwarded to video_helper.main.extract_frames()).

  • frame_interval (float, optional) – Sample one frame every N seconds (forwarded to video_helper.main.extract_frames(); mutually exclusive with frame_step there).

  • fps (float, optional) – Frame rate for the .mp4 visualization output. Defaults to the source video’s probed frame rate divided by frame_step (ignored for .npy output, and only a rough estimate when frame_interval is used instead of frame_step).

  • output_width (int, optional) – Resize the flow field (not the source frames) to this width before writing — e.g. a smaller .npy for storage, or a smaller visualization video. Must be given together with output_height. Goes through resize_flow() (wavelet-based, magnitude-rescaled, discontinuity-aware) rather than plain interpolation.

  • output_height (int, optional) – Resize the flow field to this height. See output_width.

  • wavelet (str, default "db2") – PyWavelets wavelet name forwarded to resize_flow(). Ignored unless output_width/output_height are set.

  • overwrite (bool, default True) – Overwrite output_path if it already exists; when False and the file already exists, that path is returned as-is with no recompute.

Returns:

Path to the written file (output_path).

Return type:

str

Raises:

  • AssertionError – If input_video is not a valid video file.

  • ValueError – If method, dis_preset, or raft_variant is not supported (propagated from iter_frame_optical_flow()).

  • ImportError – If method="raft" is requested but torchvision is not installed, or if output_width/output_height are requested but PyWavelets is not installed.

Examples

>>> extract_optical_flow("clip.mp4", "clip-flow.mp4", method="dis")
'clip-flow.mp4'
>>> extract_optical_flow("clip.mp4", "clip-flow.npy", method="dis")
'clip-flow.npy'
video_helper.extract_unique_colors(srt_file_path)[source]

Extract all unique hex color codes from an SRT file.

Parameters:

srt_file_path (str) – Path to the input .srt file.

Returns:

  • Set[str] – A set of unique hex color codes found in the .srt file.

  • Usage

  • —–

  • >>> srt_file = “subtitles.srt”

  • >>> unique_colors = extract_unique_colors(srt_file)

  • >>> print(unique_colors)

  • {‘#FF0000’, ‘#00FF00’, ‘#0000FF’}

Return type:

set[str]

video_helper.extract_video_chunk(input_video, sample_start, sample_end, output_video, *, copy=False)[source]

Extract a chunk of video from the specified start to end time and save it to a new file.

Parameters:

  • input_video (str) – Path to the input video file.

  • sample_start (float) – Start time in seconds for the extraction.

  • sample_end (float) – End time in seconds for the extraction.

  • output_video (str) – Path to save the extracted video chunk.

  • copy (bool, optional) – Stream-copy the cut instead of re-encoding (default False, the safe choice for an arbitrary input). Fast and lossless, but only frame-accurate when every frame of input_video is already a keyframe — i.e. input_video came from to_editing_intermediate(). On an ordinary delivery-encoded input, copy=True silently snaps the cut to the nearest keyframe instead of the exact requested timestamp.

  • Usage

  • -----

  • extract_video_chunk("input.mp4" (>>>)

  • 10.0

  • 20.0

  • "output_chunk.mp4")

  • extract_video_chunk("intermediate.mp4" (>>>)

  • 10.0

  • 20.0

  • "chunk.mp4"

  • copy=True)

Return type:

None

video_helper.image_loop_to_video(image, duration, output_video, frame_rate=30, width=None, height=None)[source]

Loop a still image for duration seconds into a silent video.

Parameters:

  • image (str) – Path to the input still (PNG, JPG, …).

  • duration (float) – Output duration in seconds.

  • output_video (str) – Path to the output video file (.mp4 recommended).

  • frame_rate (int, optional) – Output frame rate (default 30).

  • width (int, optional) – If both provided, the image is letterboxed (scale + pad with black) to the target viewport. Width and height are rounded down to even.

  • height (int, optional) – If both provided, the image is letterboxed (scale + pad with black) to the target viewport. Width and height are rounded down to even.

Return type:

None

Notes

Common in title cards, screenshot scenes, slide-style montages. Encoded as H.264 yuv420p with no audio track.

Examples

>>> image_loop_to_video("title.png", 3.0, "title.mp4",
...                     width=1920, height=1080)
video_helper.is_valid_video_file(video_file)[source]

Check that video_file exists, has a known video extension, and contains a video stream.

Combines an extension check (against video_extensions) with an ffprobe invocation so both a fake .mp4 (no video stream) and a real video renamed to .xyz are rejected.

HTTP / HTTPS URLs short-circuit to True: the only way to truly validate a remote URL is to spend bandwidth fetching part of the stream, and ffmpeg will surface a clear error downstream if the URL is bad. Callers passing a yt-dlp-resolved direct URL (with http_headers) wouldn’t even get past this gate otherwise.

Parameters:

video_file (str) – Path to the input video file, or an HTTP / HTTPS URL.

Returns:

True iff the file exists, ffprobe finds at least one video stream, and the extension is in video_extensions — OR video_file is an HTTP / HTTPS URL.

Return type:

bool

video_helper.iter_frame_optical_flow(frames, *, method='dis', dis_preset='fast', raft_variant='small', device='cpu', clip_flow=None, grayscale=False, output_width=None, output_height=None, wavelet='db2')[source]

Re-yield a BGR frame stream with 2 extra dense-optical-flow channels.

Wraps any (H, W, 3) BGR uint8 frame iterator — video_helper.extract_frames() output, capture_helper.iter_camera_frames output, or any other source sharing that contract — and yields either (H, W, 5) float32 arrays (default: BGR frame + flow) or (H, W, 3) float32 arrays (grayscale=True: single-channel intensity + flow). In both layouts the last 2 channels are always per-pixel flow vx/vy relative to the previous frame.

Parameters:

  • frames (Iterator[numpy.ndarray]) – Source frames, each (H, W, 3) BGR uint8 (OpenCV convention).

  • method ({"dis", "farneback", "raft"}, default "dis") – Optical-flow backend. "dis" and "farneback" use only opencv-python (already a core dependency, no extra install). "raft" is a deep-learning estimator that needs the [flow] extra (pip install "video-helper[flow]") and is quality-first / GPU-recommended — CPU RAFT is not expected to run in real time.

  • dis_preset ({"ultrafast", "fast", "medium"}, default "fast") – Speed/quality preset for method="dis". Ignored otherwise.

  • raft_variant ({"small", "large"}, default "small") – raft_small (speed-favoring) or raft_large (quality-favoring). Only used for method="raft". RAFT’s correlation pyramid needs feature maps at least 16px wide after an internal 8x downsample, so frames smaller than ~128x128 raise inside torchvision — not a video-helper limitation, but worth knowing before wrapping small crops/thumbnails.

  • device (str, default "cpu") – Torch device for method="raft" only: "cpu", "mps", "cuda", or "auto" (best available). Ignored for "dis"/ "farneback", which are CPU-only OpenCV calls.

  • clip_flow (float or None, default None) – When set, symmetrically clip both vx and vy to [-clip_flow, clip_flow] pixels — suppresses rare outlier vectors (e.g. at scene cuts) without changing the array shape/dtype.

  • grayscale (bool, default False) – When True, yield (H, W, 3) arrays (grayscale intensity + flow) instead of the default (H, W, 5) (BGR + flow) — a smaller, motion-focused representation for callers that don’t need color (e.g. feeding a flow-only model). RAFT still computes flow from the full-color frame pair regardless of this flag; it only changes what gets written to the non-flow output channel(s).

  • output_width (int, optional) – Resize each yielded frame to this width. Must be given together with output_height (no aspect-preserving/padding mode here — for that, pre-resize frames itself via extract_frames(output_width=..., output_height=...) so flow is computed directly at the target resolution). This parameter is for the different case of resizing an already-computed flow field — e.g. computing flow at full quality then shrinking for storage, or computing cheaply at low resolution and upsampling for display. The image channel(s) are resized with standard bilinear interpolation (no discontinuity concern for color/intensity); the flow channels go through resize_flow() (wavelet-based, magnitude-rescaled, discontinuity-aware).

  • output_height (int, optional) – Resize each yielded frame to this height. See output_width.

  • wavelet (str, default "db2") – PyWavelets wavelet name forwarded to resize_flow(). Ignored unless output_width/output_height are set.

Yields:

numpy.ndarray – grayscale=False (default): (H, W, 5) float32 array per input frame. [..., :3] is the BGR frame cast to float32 (values 0-255 — .astype(np.uint8) recovers the plain image); [..., 3] is vx, [..., 4] is vy. grayscale=True: (H, W, 3) float32; [..., 0] is grayscale intensity, [..., 1] is vx, [..., 2] is vy. Either way flow is signed pixel displacement, and the first yielded frame has zero flow (no previous frame yet), keeping frame count 1:1 with frames.

Raises:

  • ValueError – If method, dis_preset, or raft_variant is not a supported value, or if exactly one of output_width/output_height is given without the other.

  • ImportError – If method="raft" is requested but torchvision is not installed, or if output_width/output_height are requested but PyWavelets is not installed (install either with pip install "video-helper[flow]").

Return type:

Iterator[ndarray]

Examples

>>> import video_helper as vh
>>> frames = vh.extract_frames("clip.mp4", frame_step=1)
>>> for flow_frame in vh.iter_frame_optical_flow(frames, method="dis"):
...     bgr = flow_frame[..., :3].astype("uint8")
...     vx, vy = flow_frame[..., 3], flow_frame[..., 4]
...     break
>>> for flow_frame in vh.iter_frame_optical_flow(frames, method="dis", grayscale=True):
...     gray = flow_frame[..., 0].astype("uint8")
...     vx, vy = flow_frame[..., 1], flow_frame[..., 2]
...     break

Notes

Composability is the point: this function takes a generic frame iterator rather than a video path, so it works identically wrapping video_helper.extract_frames(...) (file) or capture_helper.iter_camera_frames(...) (live camera) — both already share the same (H, W, 3) BGR uint8 contract.

video_helper.mux_audio_video(input_video, input_audio, output_video, audio_codec='aac', audio_bitrate='192k', shortest=False)[source]

Mux a separate audio track onto a (typically silent) video.

Parameters:

  • input_video (str) – Path to the video file. Any existing audio track is replaced.

  • input_audio (str) – Path to the audio file (WAV, MP3, AAC, …).

  • output_video (str) – Path to the output video file (.mp4 recommended).

  • audio_codec (str, optional) – Audio codec for the output stream (default "aac"). Use "copy" if the input audio is already in a container-compatible codec.

  • audio_bitrate (str, optional) – Audio bitrate when re-encoding (default "192k"); ignored when audio_codec="copy".

  • shortest (bool, optional) – If True, the output stops when the shorter of the two streams ends. If False (default), the output keeps the video length and the audio is padded with silence (or truncated) by the muxer.

Return type:

None

Notes

Video stream is copied — no re-encoding — so the muxing is fast and lossless on the video side. Use this after assembling a silent visuals.mp4 and a separate narration.wav track.

Examples

>>> mux_audio_video("silent.mp4", "voice.wav", "final.mp4")
video_helper.overlay_image(input_video, image, output_video, x='0', y='0', scale_width=None)[source]

Overlay a still image (PNG with alpha works) on top of a video.

Parameters:

  • input_video (str) – Path to the base video.

  • image (str) – Path to the overlay image (PNG with alpha is the typical case — cursors, watermarks, logos).

  • output_video (str) – Path to the output video file (.mp4 recommended).

  • x (str, optional) – Overlay positions. Plain integers ("10") place the image statically; ffmpeg overlay expressions ("if(lt(t,1.0),0,100)", "W/2-w/2", …) move the overlay over time. Default "0", "0" (top-left).

  • y (str, optional) – Overlay positions. Plain integers ("10") place the image statically; ffmpeg overlay expressions ("if(lt(t,1.0),0,100)", "W/2-w/2", …) move the overlay over time. Default "0", "0" (top-left).

  • scale_width (int, optional) – If provided, scale the overlay to this width keeping aspect ratio — useful for cursor PNGs that come at a different size than the target frame.

Return type:

None

Notes

Time-varying expressions are evaluated per-frame (eval=frame) so animations stay smooth at any framerate. The underlying video stream is re-encoded (libx264) and the original audio track, if any, is preserved.

Examples

>>> overlay_image("clip.mp4", "cursor.png", "out.mp4",
...               x="if(lt(t,2),100,400)", y="200",
...               scale_width=24)
video_helper.resize_flow(flow, output_width, output_height, *, wavelet='db2')[source]

Resize a dense optical-flow field, preserving motion discontinuities.

Applies _wavelet_resize_channel() to vx and vy independently, then rescales the displacement magnitudes by the same factor as the spatial resize — a flow field encodes physical pixel displacement, so “5px right” at the original resolution must become “10px right” after a 2x upsample, exactly like rescaling a velocity when changing units. Skipping this step is a common, silent correctness bug in flow-resizing code (frame-only resizing tools like cv2.resize don’t know to do it, since a plain image has no such physical meaning).

This is specifically for the 2-channel vx/vy flow itself — for the BGR/grayscale image channels, standard interpolation (e.g. extract_frames’s own output_width/output_height) is the right tool; there is no discontinuity-preservation concern for color.

Parameters:

  • flow (numpy.ndarray) – Flow field (H, W, 2) float, channels [vx, vy] — e.g. the last 2 channels of an iter_frame_optical_flow() output frame.

  • output_width (int) – Target width in pixels.

  • output_height (int) – Target height in pixels.

  • wavelet (str, default "db2") – A PyWavelets wavelet name. "db2" (Daubechies-2) is a reasonable default — smoother reconstruction than "haar" (which is blocky) without the longer-filter ringing of higher-order wavelets.

Returns:

(output_height, output_width, 2) float32, magnitude-rescaled.

Return type:

numpy.ndarray

Raises:

  • ValueError – If flow is not (H, W, 2), or output_width/output_height is not a positive integer.

  • ImportError – If PyWavelets is not installed (install with pip install "video-helper[flow]").

Examples

>>> import numpy as np
>>> flow = np.zeros((64, 64, 2), dtype=np.float32)
>>> flow[:, 32:, 0] = 5.0  # a hard motion boundary: 5px/frame rightward
>>> resized = resize_flow(flow, output_width=32, output_height=32)
>>> resized.shape
(32, 32, 2)

Notes

Only the portion of the resize that is an integer power-of-two ratio (in both axes together) goes through the wavelet transform; the remainder (always < 2x, and the whole resize for a mixed up/down-sample across axes) falls back to a single nearest-neighbor snap — see _wavelet_resize_channel() for the full rationale.

video_helper.srt2vtt(srt_file_path, vtt_file_path=None, css_file_path=None)[source]

Convert an SRT subtitle file to WebVTT, preserving font colors and emitting a companion CSS file.

Any <font color="#RRGGBB">…</font> tag in the SRT is rewritten as a WebVTT <c.<hex_lowercase>>…</c> cue class, and a stylesheet binding each class to its color is written next to the VTT.

Parameters:

  • srt_file_path (str) – Path to the input .srt file.

  • vtt_file_path (str, optional) – Path to the output .vtt file. Defaults to <srt_stem>.vtt next to the input.

  • css_file_path (str, optional) – Path to the output .css file. Defaults to <srt_stem>.css next to the input.

Return type:

None

Examples

>>> srt2vtt("subtitles.srt")
>>> srt2vtt("subtitles.srt", "out.vtt", "out.css")
video_helper.video_converter(input_video, output_video=None, frame_rate=None, width=None, height=None, without_sound=False)[source]

Convert a video file to a new format with specified options.

Parameters:

  • input_video (str) – Path to the input video file.

  • output_video (str) – Path to the output video file.

  • frame_rate (int, optional) – Frame rate of the output video file.

  • width (int, optional) – Width of the output video file. If only width is specified, aspect ratio is maintained. If width is odd, it is reduced by 1 (ffmpeg reasons).

  • height (int, optional) – Height of the output video file. If only height is specified, aspect ratio is maintained. If height is odd, it is reduced by 1 (ffmpeg reasons).

  • without_sound (bool, optional) – Remove audio from the output video file.

Return type:

None

Notes

  • The output video file will be in the same format as the input, unless an output file with a different format is specified.

Examples

>>> video_converter("input.mp4", "output.mp4", frame_rate=30, width=640, height=480)
>>> video_converter("input.mp4", "output.mp4", without_sound=True)
video_helper.video_dimensions(video_file, http_headers=None)[source]

Get the dimensions of a video file (or URL) using ffmpeg-python.

Returned keys: width, height, duration, frame_rate, has_sound.

Parameters:

  • video_file (str) – Path to the input video file, OR an HTTP / HTTPS URL (e.g. a yt-dlp-resolved direct media URL).

  • http_headers (dict[str, str], optional) – HTTP headers (User-Agent, Referer, Cookie, …) forwarded to ffprobe via -headers. Required when video_file is a URL that needs specific headers — e.g. yt-dlp-resolved YouTube live, members-only, age-gated content. Ignored when video_file is a local path.

Returns:

{"width": int, "height": int, "duration": float, "frame_rate": float, "has_sound": bool}.

Return type:

dict

Examples

>>> d = video_dimensions("video.mp4")
>>> print(d)
{'width': 1920, 'height': 1080, 'duration': 10.0, 'frame_rate': 30.0, 'has_sound': True}

Notes

Uses ffmpeg.probe() (a thin wrapper over ffprobe) for metadata extraction. http_headers are passed through ffprobe’s -headers flag so URL-protected streams probe correctly.

video_helper.video_duration(input_video)[source]

Return the duration (seconds) of a video file.

Parameters:

input_video (str) – Path to the input video file.

Returns:

Duration in seconds.

Return type:

float

Notes

Mirror of audio_helper.get_audio_duration for the video side. Uses video_dimensions under the hood, which already calls ffmpeg.probe — kept as a top-level convenience so callers don’t have to remember the dict key.

Examples

>>> video_duration("clip.mp4")
12.34