xpark.dataset.VideoWatermarkDetect#

class xpark.dataset.VideoWatermarkDetect(_local_model: str = 'finetrainers/laion-watermark-detection', num_frames: int | None = None, fps: int | None = None, keyframes_only: bool = True, start_time: float = 0.0, end_time: float | None = None, reduce_mode: Literal['max', 'avg', 'min'] = 'max')#

Video watermark detection processor based on LAION watermark classifier.

Extracts frames from video, predicts a per-frame watermark probability with the image watermark classifier, then aggregates frame probabilities into a single video-level score in [0, 1]. Higher scores mean the video is more likely to contain a watermark.

Parameters:

  • _local_model – The watermark detection model name. default: “finetrainers/laion-watermark-detection”. available models: [‘finetrainers/laion-watermark-detection’]

  • num_frames – Number of frames to extract uniformly. Default 3 (when fps is also None).

  • fps – Extract frames at this frame rate, int type (mutually exclusive with num_frames).

  • keyframes_only – Only extract keyframes (I-frames). Default True. Keyframes are inherently representative of scene changes, providing the most discriminative frames with minimal decoding overhead.

  • start_time – Start time in seconds for frame extraction. Default 0.0.

  • end_time – End time in seconds for frame extraction. Default None (end of video).

  • reduce_mode – Aggregation method for frame watermark probabilities (“max”, “avg”, “min”). Default “max” — a video is considered watermarked if any single frame is watermarked, so max matches the business semantic. Use avg for a smoother global signal.

Examples

from xpark.dataset.expressions import col
from xpark.dataset import VideoWatermarkDetect, from_items

ds = from_items([{"video": "/path/to/video.mp4"}])
ds = ds.with_column(
    "video_watermark_prob",
    VideoWatermarkDetect(num_frames=5)
    .options(num_workers={"GPU": 1}, batch_size=2)
    .with_column(col("video")),
)
print(ds.take(1))

Methods

__call__(videos)

Call self as a function.

options(**kwargs)

with_column(videos)

__call__(videos: pa.ChunkedArray) → pa.Array#

Call self as a function.

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