Scene Detection

Lightweight scene detection using histogram comparison (no AI/ML dependencies).

SceneDetector

Detects scene changes in videos by comparing color histograms of consecutive frames. When the histogram difference exceeds a threshold, a scene boundary is detected.

Basic Usage (In-Memory)

from videopython.base import Video, SceneDetector

video = Video.from_path("video.mp4")

# Create detector with custom settings
detector = SceneDetector(
    threshold=0.3,        # 0.0-1.0, lower = more sensitive
    min_scene_length=0.5  # minimum scene duration in seconds
)

# Detect scenes
scenes = detector.detect(video)

for scene in scenes:
    print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s ({scene.duration:.2f}s)")
    print(f"  Frames: {scene.start_frame} - {scene.end_frame}")

Parallel Processing (Recommended for Long Videos)

For long videos, use detect_parallel() which processes the video using multiple CPU cores. This provides ~3.5x speedup on 8-core machines.

from videopython.base import SceneDetector

detector = SceneDetector(threshold=0.3, min_scene_length=0.5)

# Process video file directly with parallel workers
scenes = detector.detect_parallel("long_video.mp4", num_workers=8)

# Or let it auto-detect CPU count
scenes = detector.detect_parallel("long_video.mp4")

Memory-Efficient Streaming

For memory-constrained environments, use detect_streaming() which processes frames one at a time with O(1) memory usage.

from videopython.base import SceneDetector

detector = SceneDetector(threshold=0.3, min_scene_length=0.5)

# Stream frames from file - only 2 frames in memory at any time
scenes = detector.detect_streaming("very_long_video.mp4")

Method Comparison

Method	Memory	Speed	Best For
`detect(video)`	O(all frames)	Fast	Short videos already in memory
`detect_parallel(path)`	O(workers)	Fastest	Long videos, multi-core systems
`detect_streaming(path)`	O(1)	Slower	Memory-constrained environments

How It Works

Converts each frame to HSV color space
Calculates normalized histograms for Hue, Saturation, and Value channels
Compares consecutive frames using histogram correlation
Marks boundaries where difference exceeds threshold
Merges scenes shorter than min_scene_length

Parameters

threshold (float, default=0.3): Sensitivity for scene change detection. Range 0.0 to 1.0. Lower values detect more scene changes.
min_scene_length (float, default=0.5): Minimum scene duration in seconds. Scenes shorter than this are merged with adjacent scenes.

SceneDetector

Detects scene changes in videos using histogram comparison.

Scene changes are detected by comparing the color histograms of consecutive frames. When the histogram difference exceeds a threshold, a scene boundary is detected.

This is a lightweight implementation using only OpenCV, suitable for the base module.

Example

from videopython.base import Video, SceneDetector video = Video.from_path("video.mp4") detector = SceneDetector(threshold=0.3, min_scene_length=0.5) scenes = detector.detect(video) for scene in scenes: ... print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s")

Source code in src/videopython/base/scene.py

class SceneDetector:
    """Detects scene changes in videos using histogram comparison.

    Scene changes are detected by comparing the color histograms of consecutive frames.
    When the histogram difference exceeds a threshold, a scene boundary is detected.

    This is a lightweight implementation using only OpenCV, suitable for the base module.

    Example:
        >>> from videopython.base import Video, SceneDetector
        >>> video = Video.from_path("video.mp4")
        >>> detector = SceneDetector(threshold=0.3, min_scene_length=0.5)
        >>> scenes = detector.detect(video)
        >>> for scene in scenes:
        ...     print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s")
    """

    def __init__(self, threshold: float = 0.3, min_scene_length: float = 0.5):
        """Initialize the scene detector.

        Args:
            threshold: Sensitivity for scene change detection (0.0 to 1.0).
                      Lower values detect more scene changes. Default: 0.3
            min_scene_length: Minimum scene duration in seconds. Scenes shorter than
                            this will be merged with adjacent scenes. Default: 0.5
        """
        if not 0.0 <= threshold <= 1.0:
            raise ValueError("threshold must be between 0.0 and 1.0")
        if min_scene_length < 0:
            raise ValueError("min_scene_length must be non-negative")

        self.threshold = threshold
        self.min_scene_length = min_scene_length

    def _calculate_hsv_histogram(self, hsv: np.ndarray) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
        """Calculate normalized HSV histograms for all channels.

        Args:
            hsv: HSV image as numpy array (H, W, 3)

        Returns:
            Tuple of (hue_hist, saturation_hist, value_hist), each normalized 0-1
        """
        h_hist = cv2.calcHist([hsv], [0], None, [50], [0, 180])
        s_hist = cv2.calcHist([hsv], [1], None, [60], [0, 256])
        v_hist = cv2.calcHist([hsv], [2], None, [60], [0, 256])

        cv2.normalize(h_hist, h_hist, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX)
        cv2.normalize(s_hist, s_hist, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX)
        cv2.normalize(v_hist, v_hist, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX)

        return (h_hist, s_hist, v_hist)

    def _calculate_histogram_difference(self, frame1: np.ndarray, frame2: np.ndarray) -> float:
        """Calculate histogram difference between two frames.

        Args:
            frame1: First frame (H, W, 3) in RGB format
            frame2: Second frame (H, W, 3) in RGB format

        Returns:
            Difference score between 0.0 (identical) and 1.0 (completely different)
        """
        hsv1 = cv2.cvtColor(frame1, cv2.COLOR_RGB2HSV)
        hsv2 = cv2.cvtColor(frame2, cv2.COLOR_RGB2HSV)

        hist1 = self._calculate_hsv_histogram(hsv1)
        hist2 = self._calculate_hsv_histogram(hsv2)

        correlations = [cv2.compareHist(h1, h2, cv2.HISTCMP_CORREL) for h1, h2 in zip(hist1, hist2)]
        avg_correlation = sum(correlations) / len(correlations)

        return 1.0 - avg_correlation

    def detect(self, video: Video) -> list[SceneDescription]:
        """Detect scenes in a video.

        Args:
            video: Video object to analyze

        Returns:
            List of SceneDescription objects representing detected scenes, ordered by time.
            Frame descriptions are not populated - use VideoAnalyzer for full analysis.
        """
        if len(video.frames) == 0:
            return []

        if len(video.frames) == 1:
            return [SceneDescription(start=0.0, end=video.total_seconds, start_frame=0, end_frame=1)]

        scene_boundaries = [0]

        for i in range(1, len(video.frames)):
            difference = self._calculate_histogram_difference(video.frames[i - 1], video.frames[i])

            if difference > self.threshold:
                scene_boundaries.append(i)

        scene_boundaries.append(len(video.frames))

        scenes = []
        for i in range(len(scene_boundaries) - 1):
            start_frame = scene_boundaries[i]
            end_frame = scene_boundaries[i + 1]

            start_time = start_frame / video.fps
            end_time = end_frame / video.fps

            scenes.append(
                SceneDescription(
                    start=start_time,
                    end=end_time,
                    start_frame=start_frame,
                    end_frame=end_frame,
                )
            )

        if self.min_scene_length > 0:
            scenes = self._merge_short_scenes(scenes)

        return scenes

    def detect_streaming(
        self,
        path: str | Path,
        start_second: float | None = None,
        end_second: float | None = None,
    ) -> list[SceneDescription]:
        """Detect scenes by streaming frames from file.

        Memory usage is O(1) - only 2 frames in memory at any time.
        This is suitable for processing very long videos that would not
        fit in memory when loaded entirely.

        Args:
            path: Path to video file
            start_second: Optional start time for analysis
            end_second: Optional end time for analysis

        Returns:
            List of SceneDescription objects representing detected scenes.

        Example:
            >>> detector = SceneDetector(threshold=0.3)
            >>> scenes = detector.detect_streaming("long_video.mp4")
            >>> for scene in scenes:
            ...     print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s")
        """
        from videopython.base.video import FrameIterator, VideoMetadata

        metadata = VideoMetadata.from_path(path)

        scene_boundaries: list[int] = []
        prev_frame: np.ndarray | None = None
        first_frame_idx: int | None = None
        last_frame_idx: int = 0

        with FrameIterator(path, start_second, end_second) as frames:
            for frame_idx, frame in frames:
                if first_frame_idx is None:
                    first_frame_idx = frame_idx
                    scene_boundaries.append(frame_idx)

                if prev_frame is not None:
                    difference = self._calculate_histogram_difference(prev_frame, frame)
                    if difference > self.threshold:
                        scene_boundaries.append(frame_idx)

                prev_frame = frame
                last_frame_idx = frame_idx

        # Handle empty video case
        if first_frame_idx is None:
            return []

        # Add end boundary (one past the last frame)
        scene_boundaries.append(last_frame_idx + 1)

        # Build SceneDescription objects
        scenes = []
        for i in range(len(scene_boundaries) - 1):
            start_frame = scene_boundaries[i]
            end_frame = scene_boundaries[i + 1]

            start_time = start_frame / metadata.fps
            end_time = end_frame / metadata.fps

            scenes.append(
                SceneDescription(
                    start=start_time,
                    end=end_time,
                    start_frame=start_frame,
                    end_frame=end_frame,
                )
            )

        if self.min_scene_length > 0:
            scenes = self._merge_short_scenes(scenes)

        return scenes

    def detect_parallel(
        self,
        path: str | Path,
        num_workers: int | None = None,
        start_second: float | None = None,
        end_second: float | None = None,
    ) -> list[SceneDescription]:
        """Detect scenes using parallel processing.

        Splits video into segments and processes them in parallel using
        multiple CPU cores. Most efficient for long videos on multi-core systems.

        Memory usage: O(workers * chunk_size) - each worker processes independently.

        Args:
            path: Path to video file
            num_workers: Number of parallel workers (default: CPU count)
            start_second: Optional start time for analysis
            end_second: Optional end time for analysis

        Returns:
            List of SceneDescription objects representing detected scenes.

        Example:
            >>> detector = SceneDetector(threshold=0.3)
            >>> scenes = detector.detect_parallel("long_video.mp4", num_workers=4)
            >>> for scene in scenes:
            ...     print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s")
        """
        import multiprocessing
        from concurrent.futures import ProcessPoolExecutor
        from functools import partial

        from videopython.base.video import VideoMetadata

        metadata = VideoMetadata.from_path(path)

        if num_workers is None:
            num_workers = multiprocessing.cpu_count()

        # Determine time range
        actual_start = start_second if start_second is not None else 0.0
        actual_end = end_second if end_second is not None else metadata.total_seconds

        total_duration = actual_end - actual_start

        # For short videos, just use streaming
        if total_duration < 10 or num_workers <= 1:
            return self.detect_streaming(path, start_second, end_second)

        # Split into segments for parallel processing
        segment_duration = total_duration / num_workers
        segments: list[tuple[float, float]] = []

        for i in range(num_workers):
            seg_start = actual_start + i * segment_duration
            seg_end = actual_start + (i + 1) * segment_duration
            if i == num_workers - 1:
                seg_end = actual_end  # Ensure last segment covers to end
            segments.append((seg_start, seg_end))

        # Process segments in parallel
        worker_func = partial(
            _detect_segment_worker,
            path=str(path),
            threshold=self.threshold,
        )

        with ProcessPoolExecutor(max_workers=num_workers) as executor:
            results = list(executor.map(worker_func, segments))

        # Merge results from all segments
        all_boundaries: list[int] = []
        boundary_frames_at_edges: list[tuple[int, int]] = []  # (last_frame_of_seg, first_frame_of_next)

        for i, (boundaries, first_frame, last_frame) in enumerate(results):
            if i == 0:
                all_boundaries.extend(boundaries)
            else:
                # Check boundary between segments
                prev_last_frame = results[i - 1][2]
                # We need to check if there's a scene change at segment boundary
                # For now, include boundaries from this segment (excluding first which was start)
                if boundaries:
                    # First boundary of non-first segment is the segment start
                    # We may need to check for scene change here
                    all_boundaries.extend(boundaries[1:] if len(boundaries) > 1 else [])
                boundary_frames_at_edges.append((prev_last_frame, first_frame))

        # Handle empty case
        if not all_boundaries:
            return []

        # Add end boundary
        last_frame_idx = results[-1][2]
        all_boundaries.append(last_frame_idx + 1)

        # Deduplicate and sort
        all_boundaries = sorted(set(all_boundaries))

        # Build SceneDescription objects
        scenes = []
        for i in range(len(all_boundaries) - 1):
            start_frame = all_boundaries[i]
            end_frame = all_boundaries[i + 1]

            start_time = start_frame / metadata.fps
            end_time = end_frame / metadata.fps

            scenes.append(
                SceneDescription(
                    start=start_time,
                    end=end_time,
                    start_frame=start_frame,
                    end_frame=end_frame,
                )
            )

        if self.min_scene_length > 0:
            scenes = self._merge_short_scenes(scenes)

        return scenes

    @classmethod
    def detect_from_path(
        cls,
        path: str | Path,
        threshold: float = 0.3,
        min_scene_length: float = 0.5,
    ) -> list[SceneDescription]:
        """Convenience method for one-shot streaming scene detection.

        Creates a SceneDetector instance and runs streaming detection.
        Memory usage is O(1), suitable for very long videos.

        Args:
            path: Path to video file
            threshold: Scene change threshold (0.0-1.0). Lower values
                      detect more scene changes. Default: 0.3
            min_scene_length: Minimum scene duration in seconds. Default: 0.5

        Returns:
            List of SceneDescription objects representing detected scenes.

        Example:
            >>> scenes = SceneDetector.detect_from_path("video.mp4", threshold=0.3)
            >>> print(f"Found {len(scenes)} scenes")
        """
        detector = cls(threshold=threshold, min_scene_length=min_scene_length)
        return detector.detect_streaming(path)

    def _merge_short_scenes(self, scenes: list[SceneDescription]) -> list[SceneDescription]:
        """Merge scenes that are shorter than min_scene_length.

        Args:
            scenes: List of scenes to process

        Returns:
            List of scenes with short scenes merged into adjacent ones
        """
        if not scenes:
            return scenes

        merged = [scenes[0]]

        for scene in scenes[1:]:
            last_scene = merged[-1]

            if last_scene.duration < self.min_scene_length:
                merged[-1] = SceneDescription(
                    start=last_scene.start,
                    end=scene.end,
                    start_frame=last_scene.start_frame,
                    end_frame=scene.end_frame,
                )
            else:
                merged.append(scene)

        if len(merged) > 1 and merged[-1].duration < self.min_scene_length:
            second_last = merged[-2]
            last = merged[-1]
            merged[-2] = SceneDescription(
                start=second_last.start,
                end=last.end,
                start_frame=second_last.start_frame,
                end_frame=last.end_frame,
            )
            merged.pop()

        return merged

init

__init__(
    threshold: float = 0.3, min_scene_length: float = 0.5
)

Initialize the scene detector.

Parameters:

Name	Type	Description	Default
`threshold`	`float`	Sensitivity for scene change detection (0.0 to 1.0). Lower values detect more scene changes. Default: 0.3	`0.3`
`min_scene_length`	`float`	Minimum scene duration in seconds. Scenes shorter than this will be merged with adjacent scenes. Default: 0.5	`0.5`

Source code in src/videopython/base/scene.py

def __init__(self, threshold: float = 0.3, min_scene_length: float = 0.5):
    """Initialize the scene detector.

    Args:
        threshold: Sensitivity for scene change detection (0.0 to 1.0).
                  Lower values detect more scene changes. Default: 0.3
        min_scene_length: Minimum scene duration in seconds. Scenes shorter than
                        this will be merged with adjacent scenes. Default: 0.5
    """
    if not 0.0 <= threshold <= 1.0:
        raise ValueError("threshold must be between 0.0 and 1.0")
    if min_scene_length < 0:
        raise ValueError("min_scene_length must be non-negative")

    self.threshold = threshold
    self.min_scene_length = min_scene_length

detect

detect(video: Video) -> list[SceneDescription]

Detect scenes in a video.

Parameters:

Name	Type	Description	Default
`video`	`Video`	Video object to analyze	required

Returns:

Type	Description
`list[SceneDescription]`	List of SceneDescription objects representing detected scenes, ordered by time.
`list[SceneDescription]`	Frame descriptions are not populated - use VideoAnalyzer for full analysis.

Source code in src/videopython/base/scene.py

def detect(self, video: Video) -> list[SceneDescription]:
    """Detect scenes in a video.

    Args:
        video: Video object to analyze

    Returns:
        List of SceneDescription objects representing detected scenes, ordered by time.
        Frame descriptions are not populated - use VideoAnalyzer for full analysis.
    """
    if len(video.frames) == 0:
        return []

    if len(video.frames) == 1:
        return [SceneDescription(start=0.0, end=video.total_seconds, start_frame=0, end_frame=1)]

    scene_boundaries = [0]

    for i in range(1, len(video.frames)):
        difference = self._calculate_histogram_difference(video.frames[i - 1], video.frames[i])

        if difference > self.threshold:
            scene_boundaries.append(i)

    scene_boundaries.append(len(video.frames))

    scenes = []
    for i in range(len(scene_boundaries) - 1):
        start_frame = scene_boundaries[i]
        end_frame = scene_boundaries[i + 1]

        start_time = start_frame / video.fps
        end_time = end_frame / video.fps

        scenes.append(
            SceneDescription(
                start=start_time,
                end=end_time,
                start_frame=start_frame,
                end_frame=end_frame,
            )
        )

    if self.min_scene_length > 0:
        scenes = self._merge_short_scenes(scenes)

    return scenes

detect_streaming

detect_streaming(
    path: str | Path,
    start_second: float | None = None,
    end_second: float | None = None,
) -> list[SceneDescription]

Detect scenes by streaming frames from file.

Memory usage is O(1) - only 2 frames in memory at any time. This is suitable for processing very long videos that would not fit in memory when loaded entirely.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to video file	required
`start_second`	`float \| None`	Optional start time for analysis	`None`
`end_second`	`float \| None`	Optional end time for analysis	`None`

Returns:

Type	Description
`list[SceneDescription]`	List of SceneDescription objects representing detected scenes.

Example

detector = SceneDetector(threshold=0.3) scenes = detector.detect_streaming("long_video.mp4") for scene in scenes: ... print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s")

Source code in src/videopython/base/scene.py

def detect_streaming(
    self,
    path: str | Path,
    start_second: float | None = None,
    end_second: float | None = None,
) -> list[SceneDescription]:
    """Detect scenes by streaming frames from file.

    Memory usage is O(1) - only 2 frames in memory at any time.
    This is suitable for processing very long videos that would not
    fit in memory when loaded entirely.

    Args:
        path: Path to video file
        start_second: Optional start time for analysis
        end_second: Optional end time for analysis

    Returns:
        List of SceneDescription objects representing detected scenes.

    Example:
        >>> detector = SceneDetector(threshold=0.3)
        >>> scenes = detector.detect_streaming("long_video.mp4")
        >>> for scene in scenes:
        ...     print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s")
    """
    from videopython.base.video import FrameIterator, VideoMetadata

    metadata = VideoMetadata.from_path(path)

    scene_boundaries: list[int] = []
    prev_frame: np.ndarray | None = None
    first_frame_idx: int | None = None
    last_frame_idx: int = 0

    with FrameIterator(path, start_second, end_second) as frames:
        for frame_idx, frame in frames:
            if first_frame_idx is None:
                first_frame_idx = frame_idx
                scene_boundaries.append(frame_idx)

            if prev_frame is not None:
                difference = self._calculate_histogram_difference(prev_frame, frame)
                if difference > self.threshold:
                    scene_boundaries.append(frame_idx)

            prev_frame = frame
            last_frame_idx = frame_idx

    # Handle empty video case
    if first_frame_idx is None:
        return []

    # Add end boundary (one past the last frame)
    scene_boundaries.append(last_frame_idx + 1)

    # Build SceneDescription objects
    scenes = []
    for i in range(len(scene_boundaries) - 1):
        start_frame = scene_boundaries[i]
        end_frame = scene_boundaries[i + 1]

        start_time = start_frame / metadata.fps
        end_time = end_frame / metadata.fps

        scenes.append(
            SceneDescription(
                start=start_time,
                end=end_time,
                start_frame=start_frame,
                end_frame=end_frame,
            )
        )

    if self.min_scene_length > 0:
        scenes = self._merge_short_scenes(scenes)

    return scenes

detect_parallel

detect_parallel(
    path: str | Path,
    num_workers: int | None = None,
    start_second: float | None = None,
    end_second: float | None = None,
) -> list[SceneDescription]

Detect scenes using parallel processing.

Splits video into segments and processes them in parallel using multiple CPU cores. Most efficient for long videos on multi-core systems.

Memory usage: O(workers * chunk_size) - each worker processes independently.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to video file	required
`num_workers`	`int \| None`	Number of parallel workers (default: CPU count)	`None`
`start_second`	`float \| None`	Optional start time for analysis	`None`
`end_second`	`float \| None`	Optional end time for analysis	`None`

Returns:

Type	Description
`list[SceneDescription]`	List of SceneDescription objects representing detected scenes.

Example

detector = SceneDetector(threshold=0.3) scenes = detector.detect_parallel("long_video.mp4", num_workers=4) for scene in scenes: ... print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s")

Source code in src/videopython/base/scene.py

def detect_parallel(
    self,
    path: str | Path,
    num_workers: int | None = None,
    start_second: float | None = None,
    end_second: float | None = None,
) -> list[SceneDescription]:
    """Detect scenes using parallel processing.

    Splits video into segments and processes them in parallel using
    multiple CPU cores. Most efficient for long videos on multi-core systems.

    Memory usage: O(workers * chunk_size) - each worker processes independently.

    Args:
        path: Path to video file
        num_workers: Number of parallel workers (default: CPU count)
        start_second: Optional start time for analysis
        end_second: Optional end time for analysis

    Returns:
        List of SceneDescription objects representing detected scenes.

    Example:
        >>> detector = SceneDetector(threshold=0.3)
        >>> scenes = detector.detect_parallel("long_video.mp4", num_workers=4)
        >>> for scene in scenes:
        ...     print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s")
    """
    import multiprocessing
    from concurrent.futures import ProcessPoolExecutor
    from functools import partial

    from videopython.base.video import VideoMetadata

    metadata = VideoMetadata.from_path(path)

    if num_workers is None:
        num_workers = multiprocessing.cpu_count()

    # Determine time range
    actual_start = start_second if start_second is not None else 0.0
    actual_end = end_second if end_second is not None else metadata.total_seconds

    total_duration = actual_end - actual_start

    # For short videos, just use streaming
    if total_duration < 10 or num_workers <= 1:
        return self.detect_streaming(path, start_second, end_second)

    # Split into segments for parallel processing
    segment_duration = total_duration / num_workers
    segments: list[tuple[float, float]] = []

    for i in range(num_workers):
        seg_start = actual_start + i * segment_duration
        seg_end = actual_start + (i + 1) * segment_duration
        if i == num_workers - 1:
            seg_end = actual_end  # Ensure last segment covers to end
        segments.append((seg_start, seg_end))

    # Process segments in parallel
    worker_func = partial(
        _detect_segment_worker,
        path=str(path),
        threshold=self.threshold,
    )

    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        results = list(executor.map(worker_func, segments))

    # Merge results from all segments
    all_boundaries: list[int] = []
    boundary_frames_at_edges: list[tuple[int, int]] = []  # (last_frame_of_seg, first_frame_of_next)

    for i, (boundaries, first_frame, last_frame) in enumerate(results):
        if i == 0:
            all_boundaries.extend(boundaries)
        else:
            # Check boundary between segments
            prev_last_frame = results[i - 1][2]
            # We need to check if there's a scene change at segment boundary
            # For now, include boundaries from this segment (excluding first which was start)
            if boundaries:
                # First boundary of non-first segment is the segment start
                # We may need to check for scene change here
                all_boundaries.extend(boundaries[1:] if len(boundaries) > 1 else [])
            boundary_frames_at_edges.append((prev_last_frame, first_frame))

    # Handle empty case
    if not all_boundaries:
        return []

    # Add end boundary
    last_frame_idx = results[-1][2]
    all_boundaries.append(last_frame_idx + 1)

    # Deduplicate and sort
    all_boundaries = sorted(set(all_boundaries))

    # Build SceneDescription objects
    scenes = []
    for i in range(len(all_boundaries) - 1):
        start_frame = all_boundaries[i]
        end_frame = all_boundaries[i + 1]

        start_time = start_frame / metadata.fps
        end_time = end_frame / metadata.fps

        scenes.append(
            SceneDescription(
                start=start_time,
                end=end_time,
                start_frame=start_frame,
                end_frame=end_frame,
            )
        )

    if self.min_scene_length > 0:
        scenes = self._merge_short_scenes(scenes)

    return scenes

detect_from_path `classmethod`

detect_from_path(
    path: str | Path,
    threshold: float = 0.3,
    min_scene_length: float = 0.5,
) -> list[SceneDescription]

Convenience method for one-shot streaming scene detection.

Creates a SceneDetector instance and runs streaming detection. Memory usage is O(1), suitable for very long videos.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to video file	required
`threshold`	`float`	Scene change threshold (0.0-1.0). Lower values detect more scene changes. Default: 0.3	`0.3`
`min_scene_length`	`float`	Minimum scene duration in seconds. Default: 0.5	`0.5`

Returns:

Type	Description
`list[SceneDescription]`	List of SceneDescription objects representing detected scenes.

Example

scenes = SceneDetector.detect_from_path("video.mp4", threshold=0.3) print(f"Found {len(scenes)} scenes")

Source code in src/videopython/base/scene.py

@classmethod
def detect_from_path(
    cls,
    path: str | Path,
    threshold: float = 0.3,
    min_scene_length: float = 0.5,
) -> list[SceneDescription]:
    """Convenience method for one-shot streaming scene detection.

    Creates a SceneDetector instance and runs streaming detection.
    Memory usage is O(1), suitable for very long videos.

    Args:
        path: Path to video file
        threshold: Scene change threshold (0.0-1.0). Lower values
                  detect more scene changes. Default: 0.3
        min_scene_length: Minimum scene duration in seconds. Default: 0.5

    Returns:
        List of SceneDescription objects representing detected scenes.

    Example:
        >>> scenes = SceneDetector.detect_from_path("video.mp4", threshold=0.3)
        >>> print(f"Found {len(scenes)} scenes")
    """
    detector = cls(threshold=threshold, min_scene_length=min_scene_length)
    return detector.detect_streaming(path)

SceneDescription

Returned by SceneDetector.detect(). Contains timing and frame information for each detected scene.

SceneDescription `dataclass`

A self-contained description of a video scene.

A scene is a continuous segment of video where the visual content remains relatively consistent, bounded by scene changes or transitions. This class combines timing information with visual analysis, transcription, and other metadata.

Attributes:

Name	Type	Description
`start`	`float`	Scene start time in seconds
`end`	`float`	Scene end time in seconds
`start_frame`	`int`	Index of the first frame in this scene
`end_frame`	`int`	Index of the last frame in this scene (exclusive)
`frame_descriptions`	`list[FrameDescription]`	List of descriptions for frames sampled from this scene
`transcription`	`Transcription \| None`	Optional transcription of speech within this scene
`summary`	`str \| None`	Optional LLM-generated summary of the scene
`scene_type`	`str \| None`	Optional classification (e.g., "dialogue", "action", "transition")
`detected_entities`	`list[str] \| None`	Optional list of entities/objects detected in the scene
`dominant_colors`	`list[tuple[int, int, int]] \| None`	Optional dominant colors aggregated across the scene
`audio_events`	`list[AudioEvent] \| None`	Optional list of audio events detected in this scene
`avg_motion_magnitude`	`float \| None`	Optional average motion magnitude across the scene (0.0-1.0)
`dominant_motion_type`	`str \| None`	Optional most common motion type in the scene

Source code in src/videopython/base/description.py

@dataclass
class SceneDescription:
    """A self-contained description of a video scene.

    A scene is a continuous segment of video where the visual content remains relatively consistent,
    bounded by scene changes or transitions. This class combines timing information with
    visual analysis, transcription, and other metadata.

    Attributes:
        start: Scene start time in seconds
        end: Scene end time in seconds
        start_frame: Index of the first frame in this scene
        end_frame: Index of the last frame in this scene (exclusive)
        frame_descriptions: List of descriptions for frames sampled from this scene
        transcription: Optional transcription of speech within this scene
        summary: Optional LLM-generated summary of the scene
        scene_type: Optional classification (e.g., "dialogue", "action", "transition")
        detected_entities: Optional list of entities/objects detected in the scene
        dominant_colors: Optional dominant colors aggregated across the scene
        audio_events: Optional list of audio events detected in this scene
        avg_motion_magnitude: Optional average motion magnitude across the scene (0.0-1.0)
        dominant_motion_type: Optional most common motion type in the scene
    """

    start: float
    end: float
    start_frame: int
    end_frame: int
    frame_descriptions: list[FrameDescription] = field(default_factory=list)
    transcription: Transcription | None = None
    summary: str | None = None
    scene_type: str | None = None
    detected_entities: list[str] | None = None
    dominant_colors: list[tuple[int, int, int]] | None = None
    audio_events: list[AudioEvent] | None = None
    avg_motion_magnitude: float | None = None
    dominant_motion_type: str | None = None

    @property
    def duration(self) -> float:
        """Duration of the scene in seconds."""
        return self.end - self.start

    @property
    def frame_count(self) -> int:
        """Number of frames in this scene."""
        return self.end_frame - self.start_frame

    @property
    def num_frames_described(self) -> int:
        """Number of frames that were described in this scene."""
        return len(self.frame_descriptions)

    def get_frame_indices(self, num_samples: int = 3) -> list[int]:
        """Get evenly distributed frame indices from this scene.

        Args:
            num_samples: Number of frames to sample from the scene

        Returns:
            List of frame indices evenly distributed throughout the scene
        """
        if num_samples <= 0:
            raise ValueError("num_samples must be positive")

        if num_samples == 1:
            # Return middle frame
            return [self.start_frame + self.frame_count // 2]

        # Get evenly spaced frames including start and end
        step = (self.end_frame - self.start_frame - 1) / (num_samples - 1)
        return [int(self.start_frame + i * step) for i in range(num_samples)]

    def get_description_summary(self) -> str:
        """Get a summary of all frame descriptions concatenated.

        Returns:
            Single string with all frame descriptions joined
        """
        return " ".join([fd.description for fd in self.frame_descriptions])

    def get_transcription_text(self) -> str:
        """Get the full transcription text for this scene.

        Returns:
            Concatenated transcription text, or empty string if no transcription
        """
        if not self.transcription or not self.transcription.segments:
            return ""
        return " ".join(segment.text for segment in self.transcription.segments)

duration `property`

duration: float

Duration of the scene in seconds.

frame_count `property`

frame_count: int

Number of frames in this scene.

num_frames_described `property`

num_frames_described: int

Number of frames that were described in this scene.

get_frame_indices

get_frame_indices(num_samples: int = 3) -> list[int]

Get evenly distributed frame indices from this scene.

Parameters:

Name	Type	Description	Default
`num_samples`	`int`	Number of frames to sample from the scene	`3`

Returns:

Type	Description
`list[int]`	List of frame indices evenly distributed throughout the scene

Source code in src/videopython/base/description.py

def get_frame_indices(self, num_samples: int = 3) -> list[int]:
    """Get evenly distributed frame indices from this scene.

    Args:
        num_samples: Number of frames to sample from the scene

    Returns:
        List of frame indices evenly distributed throughout the scene
    """
    if num_samples <= 0:
        raise ValueError("num_samples must be positive")

    if num_samples == 1:
        # Return middle frame
        return [self.start_frame + self.frame_count // 2]

    # Get evenly spaced frames including start and end
    step = (self.end_frame - self.start_frame - 1) / (num_samples - 1)
    return [int(self.start_frame + i * step) for i in range(num_samples)]

get_description_summary

get_description_summary() -> str

Get a summary of all frame descriptions concatenated.

Returns:

Type	Description
`str`	Single string with all frame descriptions joined

Source code in src/videopython/base/description.py

def get_description_summary(self) -> str:
    """Get a summary of all frame descriptions concatenated.

    Returns:
        Single string with all frame descriptions joined
    """
    return " ".join([fd.description for fd in self.frame_descriptions])

get_transcription_text

get_transcription_text() -> str

Get the full transcription text for this scene.

Returns:

Type	Description
`str`	Concatenated transcription text, or empty string if no transcription

Source code in src/videopython/base/description.py

def get_transcription_text(self) -> str:
    """Get the full transcription text for this scene.

    Returns:
        Concatenated transcription text, or empty string if no transcription
    """
    if not self.transcription or not self.transcription.segments:
        return ""
    return " ".join(segment.text for segment in self.transcription.segments)

Scene Detection

SceneDetector

Basic Usage (In-Memory)

Parallel Processing (Recommended for Long Videos)

Memory-Efficient Streaming

Method Comparison

How It Works

Parameters

SceneDetector

__init__

detect

detect_streaming

detect_parallel

detect_from_path classmethod

SceneDescription

SceneDescription dataclass

duration property

frame_count property

num_frames_described property

get_frame_indices

get_description_summary

get_transcription_text

init

detect_from_path `classmethod`

SceneDescription `dataclass`

duration `property`

frame_count `property`

num_frames_described `property`