xpark.dataset.VideoOpticalFlow#

class xpark.dataset.VideoOpticalFlow(*, output_dir: str | None = None, fps: float = 4.0, num_frames: int | None = 12, window_size: int = 8, max_pixels: int | None = None, min_pixels: int | None = None, backend: type[~xpark.dataset.processors.video_optical_flow.OpticalFlowBackend] = <class 'xpark.dataset.processors.video_optical_flow.FarnebackBackend'>, backend_kwargs: dict[str, ~typing.Any] | None = None, score_fn: ~typing.Callable[[~numpy.ndarray], ~numpy.ndarray] = <function flow_map_magnitude_norm>)#

Dense optical flow for a video: a per-pair score, plus the persisted field.

Samples frames, feeds each adjacent pair through a backend, and emits a {uri, score} struct — one score per adjacent pair, produced by score_fn. With output_dir the field is also persisted as {output_dir}/{basename}-{digest}/flow.npy, where digest is over the whole source string so same-named sources in different directories stay apart.

score_fn receives the whole (P, 2, H, W) field — everything it needs (field width, pair count) is derivable from the field’s shape — and returns one score per pair. The built-in flow_map_magnitude_norm() (default) returns the mean vector length divided by the field width: motion as a fraction of frame width, roughly resolution-independent, so one threshold covers several resolutions / backends.

All backends are OpticalFlowBackend subclasses, instantiated on the worker: FarnebackBackend (default, cv2 Farneback), RaftBackend (torchvision RAFT), MemfofBackend (multi-frame RAFT-family model), or a custom subclass.

The operator only measures — keep/drop is left to Dataset.filter(); reduce the per-pair list (e.g. max) before comparing to a threshold:

ds = ds.with_column(
    "flow",
    VideoOpticalFlow().options(num_workers={"CPU": 8}).with_column(col("video")),
)

Input accepts a local path, a remote URI (cos:// / s3:// / http(s)://) or raw bytes; output_dir takes the same protocols. An undecodable video returns an empty score list instead of failing the batch.

Parity with ffmpeg-sampled pipelines: frames are decoded by timestamp (pts <= ts, floor semantics), while ffmpeg’s fps filter picks the nearest frame — expect a ±1-frame sampling difference. Compare thresholds on the default normalized score. The resize kernel is not this operator’s concern — feed frames through a shared resizer operator when the sampling must match another pipeline’s resize exactly.

Parameters:

  • output_dir – Directory the flow fields are written to (field mode). None (default) keeps the operator scalar-only.

  • fps – Sampling rate in frames per second. Defaults to 4.0.

  • num_frames – Upper bound on sampled frames. Defaults to 12; None samples the whole clip.

  • window_size – Frames decoded per pass; consecutive windows share their boundary frame, so pairs stay continuous. Defaults to 8.

  • max_pixels – Upper bound on each frame’s pixel count; larger frames are downscaled. None (default) applies the smart_resize budget default of 65536 (~256x256).

  • min_pixels – Lower bound on each frame’s pixel count; smaller frames are upscaled. None (default) applies the smart_resize budget default of 16.

  • backend – An OpticalFlowBackend subclass, instantiated on the worker with backend_kwargs. Defaults to FarnebackBackend.

  • backend_kwargs – Keyword arguments for the backend class constructor.

  • score_fn – Field-to-score function, score_fn(flow field) -> float, applied to each (P, 2, H, W) flow field; Defaults to flow_map_magnitude_norm().

Methods

__call__(videos)

Process a batch of videos, returning one {uri, score} struct per row.

options(**kwargs)

with_column(videos)

Process a batch of videos, returning one {uri, score} struct per row.
__call__(videos: pa.ChunkedArray) → pa.Array#

Process a batch of videos, returning one {uri, score} struct per row.

options(**kwargs: Unpack[ExprUDFOptions]) → Self#
with_column(videos: pa.ChunkedArray) → pa.Array#

Process a batch of videos, returning one {uri, score} struct per row.