xpark.dataset.VideoShotDetect#

class xpark.dataset.VideoShotDetect(*, method: Literal['scenedetect'] = 'scenedetect', output_dir: str | None = None, detector: SceneDetector = 'adaptive', threshold: float | None = None, min_scene_len: int = 15, **kwargs)#
class xpark.dataset.VideoShotDetect(*, method: Literal['transnetv2'], output_dir: str | None = None, threshold: float = 0.5, inference_batch_size: int = 4)

Video shot boundary detection and segmentation.

Two detection methods via method parameter:

  • "scenedetect" (default): Traditional methods based on PySceneDetect. CPU only, fast, suitable for large-scale processing.

  • "transnetv2": Deep learning method based on 3D CNN. GPU recommended.

Two output modes determined by output_dir:

  • output_dir set: detect + cut + persist, returns list[str] (URI list).

  • output_dir unset (default): detect only, returns list[struct(frame_num, seconds)] for F1 evaluation or custom cutting.

Input supports any fsspec protocol transparently (local / cos / s3 / http / hf).

Parameters:

  • method – Detection method, "scenedetect" or "transnetv2".

  • output_dir – Cut result output directory. When set, segments are written here and URIs are returned; when None (default), only boundary positions are returned.

[scenedetect only]

detector: Detector type: "adaptive" (default) / "content" / "threshold". threshold: Detection threshold, None uses detector defaults:

adaptive=3.0, content=27.0, threshold=12.0.

min_scene_len: Minimum scene length in frames. **kwargs: Detector-specific parameters passed through to PySceneDetect.

[transnetv2 only]

threshold: Transition probability threshold, default 0.5. inference_batch_size: Number of 100-frame sliding windows per GPU forward

call, default 4. Controls VRAM peak per inference call and is independent of Ray’s .options(batch_size=...) which controls videos per task. Default 4 fits ~0.5GB VRAM on T4; increase to 16-32 on 24GB+ cards for higher throughput.

Examples

from xpark.dataset import from_items
from xpark.dataset.expressions import col
from xpark.dataset.processors.video_shot_detect import VideoShotDetect

ds = from_items([{"video": "cos://bucket/video.mp4"}])

# Cut + persist mode (production): segments written to output_dir
ds = ds.with_column(
    "shots",
    VideoShotDetect(method="transnetv2", output_dir="cos://bucket/shots/")
    .options(num_workers={"GPU": 1})
    .with_column(col("video")),
)
# shots: ["cos://bucket/shots/video/0000.mp4", ...]

# Detect-only mode (evaluation / custom cutting)
ds = ds.with_column(
    "boundaries",
    VideoShotDetect(method="transnetv2")
    .options(num_workers={"GPU": 1})
    .with_column(col("video")),
)
# boundaries: [{"frame_num": 156, "seconds": 5.2}, ...]

# SceneDetect adaptive (default, CPU)
ds = ds.with_column(
    "shots",
    VideoShotDetect(output_dir="/data/shots/")
    .options(num_workers={"CPU": 2})
    .with_column(col("video")),
)

Methods

__call__(videos)

Process a batch of videos.

options(**kwargs)

with_column(videos)

Process a batch of videos.
__call__(videos: pa.ChunkedArray) → pa.Array#

Process a batch of videos.

Returns:

pa.Array of list[string], each element is a

list of segment URIs (cut + persist mode).

When output_dir is None: pa.Array of list[struct(frame_num, seconds)],

each element is a list of boundary frame positions (detect-only mode).

Return type:

When output_dir is set

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

Process a batch of videos.

Returns:

pa.Array of list[string], each element is a

list of segment URIs (cut + persist mode).

When output_dir is None: pa.Array of list[struct(frame_num, seconds)],

each element is a list of boundary frame positions (detect-only mode).

Return type:

When output_dir is set