video_helper.flow module

video_helper.flow

Optional dense-optical-flow generator that wraps any BGR frame stream.

Module summary

Exposes a single public generator, iter_frame_optical_flow(), which takes any Iterator[numpy.ndarray] of (H, W, 3) BGR uint8 frames — the same contract already produced by video_helper.extract_frames() and by capture_helper.iter_camera_frames (live camera) — and re-yields each frame with 2 extra channels: dense optical flow vx/vy relative to the previous frame. Taking a generic frame iterator rather than a video path is the deliberate composability point: this module works identically for a video file or a live camera, without a hard dependency on either. Two output layouts, picked with grayscale: (H, W, 5) BGR + flow (default) or (H, W, 3) grayscale intensity + flow (smaller, motion-focused).

Three interchangeable backends, from zero-dependency classical to optional deep learning:

  • method="dis" (default) — cv2.DISOpticalFlow, the standard best CPU real-time speed/quality trade-off. No new dependency: opencv-python is already a core dependency of video-helper.

  • method="farneback" — cv2.calcOpticalFlowFarneback. Also core cv2, no new dependency. Denser/smoother field, a bit slower than DIS.

  • method="raft" — torchvision.models.optical_flow.{raft_small,raft_large}, a real deep-learning optical-flow network. Needs the [flow] extra (torch + torchvision). Quality-first, GPU-recommended; not expected to be real-time on CPU.

Usage Example

>>> 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"):
...     # flow_frame.shape == (H, W, 5), float32
...     bgr = flow_frame[..., :3].astype("uint8")
...     vx, vy = flow_frame[..., 3], flow_frame[..., 4]

Author

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

video_helper.flow.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.flow.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.flow.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.