xpark.dataset.VideoQualityCheck#

class xpark.dataset.VideoQualityCheck(*, detectors: list[BlackFrameSpec | StaticSceneSpec | OverexposedSpec | LowLightSpec | BlurSpec | MissingFrameSpec], segment_policy: SegmentPolicy | None = None, output_per_frame: bool = False, stream_format: str | None = None, pts_fallback_fps: float = 30.0, segment_hard_cap: int = 100000, raises_on_error: bool = True, factories: Mapping[type, Callable[[Any, SegmentPolicy, int], Detector]] | None = None)#

Exact (full-decode) video quality measurement operator.

Detects capture-level defects (black frames, freezes, over/under exposure, blur, frame-count completeness) and emits segment-level intervals + episode-level statistics. The operator is a pure measurement stage; accept / reject verdicts are left to downstream filters.

Prefer this operator when holding a video source at modest scale, or when segment-level [start, end) localization, missing-frame reconciliation, MCAP embedded byte-streams or transient single-frame defects are needed. Already holding a pre-extracted frames column (e.g. from a frame-train pipeline)? VideoQualityCheckFromFrames reuses it and skips the decode entirely.

Corruption coverage: hard decoder errors (av.error.InvalidDataError / av.error.EOFError from corrupt data or truncated streams) are always caught and counted into report.decode_error_count; report.decoded_frame_count counts successfully decoded frames only, so family frame_ratio denominators are never diluted by failed slots. PyAV’s per-frame is_corrupt codec flag is not consumed; pixel-domain corruption (macroblocking) is not covered.

Detector families are declared via the detectors list, a tagged union of frozen dataclasses (BlackFrameSpec, StaticSceneSpec, OverexposedSpec, LowLightSpec, BlurSpec, MissingFrameSpec). Each Spec type may appear at most once. Segment-aggregation behaviour (per-family min_duration_frames, blur sample_rate_fps) is configured via SegmentPolicy.

Parameters:

  • detectors – List of Spec instances declaring which detectors to run. Custom detectors registered via register_detector() may be mixed in – they ride the same decode pass and extend the output struct with their family fields.

  • factories – Explicit {SpecCls: factory} map for custom Specs, taking precedence over register_detector(); travels with the constructor through cloudpickle (see register_detector() for the three factory resolution paths).

  • segment_policy – Per-family aggregation policy. Defaults to SegmentPolicy field defaults.

  • output_per_frame – When true, the per_frame field of the result struct is populated with a row per decoded frame (heavy; intended for calibration / debugging). Default false.

  • stream_format – PyAV demuxer name (e.g. "hevc" / "h264") for byte-stream inputs. Required when the videos column carries raw MCAP-embedded bytes; ignored for URI / local-path inputs.

  • pts_fallback_fps – Frame rate assumed when the container carries no per-frame PTS (typically raw MCAP bytes). Only affects segment *_ns timestamps; frame numbers are unaffected. A warning is logged whenever this fallback is used.

  • segment_hard_cap – Maximum segments retained per family before truncation kicks in and report.is_truncated flips to true.

  • raises_on_error – When true (default), a data-level failure (corrupt container, no video stream, fatal decode error) raises and fails the batch. When false, such rows degrade to an all-null result – report is null marks “not checked”; healthy rows always carry a report. Infrastructure errors (download / IO) always raise.

Raises:

ValueError – At construction time – empty detectors, duplicate Spec types, out-of-range Spec/Policy thresholds, invalid operator params (pts_fallback_fps / segment_hard_cap), factories covering a built-in Spec, unknown Spec types, or custom family names colliding with built-ins. At run time (raises_on_error=True only) – containers without a video stream or fatal decode failures.

Examples

from xpark.dataset import from_items
from xpark.dataset.expressions import col
from xpark.dataset.processors.video_quality_check import (
    VideoQualityCheck, SegmentPolicy, StaticPolicy,
    BlackFrameSpec, StaticSceneSpec, OverexposedSpec,
    LowLightSpec, BlurSpec, MissingFrameSpec,
)

ds = from_items([{"video": "cos://bucket/videos/001.mp4"}])
ds = ds.with_column(
    "qc_result",
    VideoQualityCheck(
        detectors=[
            BlackFrameSpec(),
            StaticSceneSpec(),
            OverexposedSpec(),
            LowLightSpec(),
            BlurSpec(),
            MissingFrameSpec(),
        ],
        segment_policy=SegmentPolicy(
            static=StaticPolicy(min_duration_frames=60),
        ),
    )
    .options(num_workers={"CPU": (2, 8)})
    .with_column(col("video")),
)

Methods

__call__(videos)

Process a batch of videos and emit qc_result structs.

options(**kwargs)

with_column(videos)

Process a batch of videos and emit qc_result structs.
__call__(videos: pa.ChunkedArray) → pa.Array#

Process a batch of videos and emit qc_result structs.

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

Process a batch of videos and emit qc_result structs.