AI understanding

Transcribe audio, classify sounds, detect scenes and objects, describe shots, and track faces. For one aggregate object across all of these, see Video analysis.

Class	Local model family
AudioToText	faster-whisper float32 (+ pyannote for diarization)
AudioClassifier	AST
SemanticSceneDetector	TransNetV2
SceneVLM	Ollama vision model
FaceShotTracker / FaceSmoothingTracker	OpenCV YuNet
ObjectDetector	D-FINE (COCO)

AudioToText

from videopython.ai import AudioToText

transcription = AudioToText().transcribe(video)

Model sizes: tiny, base, small, medium, large, turbo (default). Diarization is opt-in with enable_diarization=True. VAD-gated language detection runs by default; enable_vad=False detects from the leading audio instead.

Diarization skips speaker embeddings for chunk/speaker pairs with no speech. Compatible pyannote embedding backends also share frame extraction across speakers in each chunk, then apply each speaker's original mask separately. Segmentation overlap, precision, and clustering settings are unchanged. Other embedding models keep the existing extraction path. When frame extraction is shared, the backend's embedding_batch_size counts audio chunks rather than chunk/speaker pairs. Pooling materializes one frame row per active pair in addition to the shared chunk frames. Thus, a batch of 32 chunks with three active speakers per chunk can hold 32 shared plus 96 pooled frame rows. This uses more frame-tensor memory than the upstream batch of 32 pairs; it is not a fourfold estimate of total GPU memory. The measured optimization retains this batching strategy.

Measured timing and comparison limits are in the diarization verification record.

Downstream speaker identification

Speaker labels such as SPEAKER_00 identify speakers within one recording. They are not person identities and are not stable across recordings. Downstream applications can use Transcription.speakers and each segment's speaker, start, and end to select audio with Audio.slice(), then run their own identity embedding and matching models. Model selection, enrollment, and similarity thresholds belong to that consumer. Videopython does not expose its internal diarization embeddings as identity vectors.

ASR and diarization can also run as separate jobs: call AudioToText().transcribe(audio) for timed words, then AudioToText(enable_diarization=True).diarize_transcription(audio, transcription) to attach speaker labels without loading Whisper again.

Anti-hallucination knobs

Three Whisper decoder kwargs are surfaced for noisy or sparse-speech audio. Defaults: condition_on_previous_text=False (the cascading-hallucination fix), no_speech_threshold=0.6, logprob_threshold=-1.0.

AudioToText(no_speech_threshold=0.4)       # lower the no-speech probability cutoff
AudioToText(condition_on_previous_text=True)  # Whisper's upstream default; helps on clean podcasts

Brand-name vocabulary biasing

Biases Whisper's first-window decoder toward supplied proper nouns via the native initial_prompt channel, recovering near-mishears (Klarna → "carna", InPost → "in post") with no extra model dependency.

transcriber = AudioToText(vocabulary=["Klarna", "Allegro", "InPost"])   # instance default
result = transcriber.transcribe(video, vocabulary=["Pyszne", "Wolt"])   # per-call override

The list is normalized at construction: whitespace stripped, case-insensitive dedup, the casing of the first occurrence preserved. Whisper reserves ~224 tokens for the prompt; longer lists are trimmed from the tail with one WARNING log line naming the dropped count. It recovers names Whisper almost heard — it will not catch zero-prior names.

VideoDubber and LocalDubbingPipeline take the same vocabulary kwarg. Inside VideoAnalyzer, pass it through analyzer_params:

VideoAnalysisConfig(analyzer_params={"audio_to_text": {"vocabulary": ["Klarna"]}})

Per-segment confidence

TranscriptionSegment carries avg_logprob, no_speech_prob and compression_ratio from the raw Whisper output. They are None when unavailable — for example on the diarization-only path that builds segments from words without an overlap match, or on transcripts loaded from formats that do not carry the metadata.

These feed the dubbing transcript-quality gate and the translator's confidence-aware prompt, and are useful for dropping low-quality segments downstream.

for segment in result.segments:
    if segment.avg_logprob is not None and segment.avg_logprob < -1.0:
        print(f"low confidence: {segment.text!r}")

AudioToText

Bases: ManagedPredictor

Transcription service for audio and video using local Whisper models.

Uses faster-whisper in float32 for transcription (with word-level timestamps) and pyannote-audio for optional speaker diarization. By default, Silero VAD runs before Whisper to gate language detection on a 30s window built from voiced regions only — fixes Whisper's tendency to lock onto the wrong language when the file opens with silence, music, or non-vocal credits. Set enable_vad=False to detect language from the leading audio without voice-activity gating.

Three Whisper decoder kwargs are surfaced for anti-hallucination tuning:

• condition_on_previous_text defaults to False (Whisper's own default is True). With conditioning on, a single hallucinated filler phrase cascades through the rest of the file because each window's decoder is primed by the previous window's decoded text. Turning it off is the most commonly recommended fix for that failure mode; the cost on clean audio is small (slightly less context for ambiguous homophones across sentence boundaries).
• no_speech_threshold and logprob_threshold are forwarded with Whisper's defaults (0.6 and -1.0). Lowering no_speech_threshold makes the no-speech probability gate easier to trigger; logprob_threshold also affects whether a window is skipped.

vocabulary biases Whisper's first-window decoder toward a caller- supplied list of brand names, product names, or proper nouns via the native initial_prompt channel. Recovers near-mishears (e.g. Klarna → "carna") without new model deps; will not catch zero-prior names. Per-call override is available on :meth:transcribe.

Source code in src/videopython/ai/understanding/audio.py

class AudioToText(ManagedPredictor):
    """Transcription service for audio and video using local Whisper models.

    Uses faster-whisper in float32 for transcription (with word-level timestamps) and
    pyannote-audio for optional speaker diarization. By default, Silero VAD
    runs before Whisper to gate language detection on a 30s window built from
    voiced regions only — fixes Whisper's tendency to lock onto the wrong
    language when the file opens with silence, music, or non-vocal credits.
    Set ``enable_vad=False`` to detect language from the leading audio without
    voice-activity gating.

    Three Whisper decoder kwargs are surfaced for anti-hallucination tuning:

    - ``condition_on_previous_text`` defaults to ``False`` (Whisper's own
      default is ``True``). With conditioning on, a single hallucinated filler
      phrase cascades through the rest of the file because each window's
      decoder is primed by the previous window's decoded text. Turning it off
      is the most commonly recommended fix for that failure mode; the cost on
      clean audio is small (slightly less context for ambiguous homophones
      across sentence boundaries).
    - ``no_speech_threshold`` and ``logprob_threshold`` are forwarded with
      Whisper's defaults (``0.6`` and ``-1.0``). Lowering
      ``no_speech_threshold`` makes the no-speech probability gate easier
      to trigger; ``logprob_threshold`` also affects whether a window is skipped.

    ``vocabulary`` biases Whisper's first-window decoder toward a caller-
    supplied list of brand names, product names, or proper nouns via the
    native ``initial_prompt`` channel. Recovers near-mishears (e.g. Klarna
    → "carna") without new model deps; will not catch zero-prior names.
    Per-call override is available on :meth:`transcribe`.
    """

    PYANNOTE_DIARIZATION_MODEL = "pyannote/speaker-diarization-community-1"
    _model_attrs = ("_model", "_diarization_pipeline", "_vad_model")

    def __init__(
        self,
        model_name: Literal["tiny", "base", "small", "medium", "large", "turbo"] = "turbo",
        enable_diarization: bool = False,
        enable_vad: bool = True,
        condition_on_previous_text: bool = False,
        no_speech_threshold: float = 0.6,
        logprob_threshold: float | None = -1.0,
        vocabulary: list[str] | None = None,
        device: str | None = None,
    ):
        if model_name not in _WHISPER_MODELS:
            choices = ", ".join(_WHISPER_MODELS)
            raise ValueError(f"Unsupported Whisper model {model_name!r}. Choose one of: {choices}.")
        self._loaded_models: dict[str, str | None] = {}
        self.model_name = model_name
        self.enable_diarization = enable_diarization
        self.enable_vad = enable_vad
        self.condition_on_previous_text = condition_on_previous_text
        self.no_speech_threshold = no_speech_threshold
        self.logprob_threshold = logprob_threshold
        self.vocabulary = _normalize_vocabulary(vocabulary)
        self.device = select_device(device, mps_allowed=False)
        log_device_initialization(
            "AudioToText",
            requested_device=device,
            resolved_device=self.device,
        )
        self._model: Any = None
        self._diarization_pipeline: Any = None
        self._vad_model: Any = None

    def _transcribe_kwargs(self, language: str | None, vocabulary: list[str]) -> dict[str, Any]:
        """Kwargs threaded into faster-whisper from both call sites.

        ``initial_prompt`` is omitted entirely on the no-vocab path."""
        kwargs: dict[str, Any] = {
            "word_timestamps": True,
            "language": language,
            "beam_size": 1,
            "best_of": 5,
            "temperature": [0.0, 0.2, 0.4, 0.6, 0.8, 1.0],
            "compression_ratio_threshold": 2.4,
            "condition_on_previous_text": self.condition_on_previous_text,
            "no_speech_threshold": self.no_speech_threshold,
            "log_prob_threshold": self.logprob_threshold,
            "vad_filter": False,
        }
        prompt = _build_initial_prompt(vocabulary, self._model.hf_tokenizer)
        if prompt is not None:
            kwargs["initial_prompt"] = prompt
        return kwargs

    def model_provenance(self) -> dict[str, str | None] | None:
        return dict(self._loaded_models) if self._loaded_models else None

    def _init_local(self) -> None:
        """Initialize local Whisper model."""
        from videopython.ai._optional import require

        faster_whisper = require("faster_whisper", feature="AudioToText")
        model_id = _WHISPER_MODELS[self.model_name]
        self._model = faster_whisper.WhisperModel(
            model_id,
            revision=pinned(model_id),
            device=self.device,
            compute_type="float32",
        )
        self._loaded_models[model_id] = pinned(model_id)

    def _run_whisper(self, audio: Any, language: str | None, vocabulary: list[str]) -> dict[str, Any]:
        segments_source, info = self._model.transcribe(
            audio=audio,
            **self._transcribe_kwargs(language, vocabulary),
        )
        segments = [
            {
                "start": segment.start,
                "end": segment.end,
                "text": segment.text,
                "words": [{"start": word.start, "end": word.end, "word": word.word} for word in segment.words or []],
                "avg_logprob": segment.avg_logprob,
                "no_speech_prob": segment.no_speech_prob,
                "compression_ratio": segment.compression_ratio,
            }
            for segment in segments_source
        ]
        return {"segments": segments, "language": info.language}

    def _init_diarization(self) -> None:
        """Initialize pyannote speaker diarization pipeline."""
        import torch

        from videopython.ai._optional import require
        from videopython.ai.understanding import _pyannote_patches

        Pipeline = require("pyannote.audio", feature="AudioToText diarization").Pipeline

        self._diarization_pipeline = Pipeline.from_pretrained(
            self.PYANNOTE_DIARIZATION_MODEL, revision=pinned(self.PYANNOTE_DIARIZATION_MODEL)
        )
        self._loaded_models[self.PYANNOTE_DIARIZATION_MODEL] = pinned(self.PYANNOTE_DIARIZATION_MODEL)
        _pyannote_patches.install(self._diarization_pipeline)
        self._diarization_pipeline.to(torch.device(self.device))

    def _init_vad(self) -> None:
        """Initialize Silero VAD model.

        The model is ~2 MB and CPU-fast (~5-15s for a 90 min movie); we keep
        it on CPU regardless of ``self.device`` since dispatch overhead would
        outweigh inference cost.
        """
        from videopython.ai._optional import require

        load_silero_vad = require("silero_vad", feature="AudioToText VAD").load_silero_vad

        self._vad_model = load_silero_vad()
        self._loaded_models["silero-vad/bundled"] = package_revision("silero-vad")

    def _process_transcription_result(self, transcription_result: dict[str, Any]) -> Transcription:
        """Process raw transcription result into a Transcription object."""
        transcription_segments = []
        for segment in transcription_result["segments"]:
            transcription_words = [
                TranscriptionWord(word=word["word"], start=float(word["start"]), end=float(word["end"]))
                for word in segment.get("words", [])
            ]
            transcription_segment = TranscriptionSegment(
                start=segment["start"],
                end=segment["end"],
                text=segment["text"],
                words=transcription_words,
                avg_logprob=segment.get("avg_logprob"),
                no_speech_prob=segment.get("no_speech_prob"),
                compression_ratio=segment.get("compression_ratio"),
            )
            transcription_segments.append(transcription_segment)

        return Transcription(segments=transcription_segments, language=transcription_result.get("language"))

    @staticmethod
    def _assign_speakers_to_words(
        words: list[TranscriptionWord],
        diarization_result: Any,
    ) -> list[TranscriptionWord]:
        """Assign speakers to chronological words from exclusive diarization tracks."""
        speaker_segments: list[tuple[float, float, str]] = []
        annotation = diarization_result.exclusive_speaker_diarization
        for turn, _, speaker in annotation.itertracks(yield_label=True):
            speaker_segments.append((turn.start, turn.end, speaker))

        if not speaker_segments:
            return words

        speaker_segments.sort(key=lambda segment: (segment[0], segment[1]))
        speaker_midpoints = [(start + end) / 2.0 for start, end, _ in speaker_segments]
        result = []
        segment_index = 0
        for word in words:
            while segment_index < len(speaker_segments) and speaker_segments[segment_index][1] <= word.start:
                segment_index += 1

            best_speaker: str | None = None
            best_overlap = 0.0
            candidate_index = segment_index
            while candidate_index < len(speaker_segments) and speaker_segments[candidate_index][0] < word.end:
                seg_start, seg_end, speaker = speaker_segments[candidate_index]
                overlap = min(word.end, seg_end) - max(word.start, seg_start)
                if overlap > best_overlap:
                    best_overlap = overlap
                    best_speaker = speaker
                candidate_index += 1

            if best_speaker is None:
                word_mid = (word.start + word.end) / 2.0
                nearest_index = bisect_left(speaker_midpoints, word_mid)
                if nearest_index == len(speaker_segments):
                    nearest_index -= 1
                elif nearest_index > 0:
                    previous_distance = word_mid - speaker_midpoints[nearest_index - 1]
                    next_distance = speaker_midpoints[nearest_index] - word_mid
                    if previous_distance <= next_distance:
                        nearest_index -= 1
                best_speaker = speaker_segments[nearest_index][2]

            result.append(
                TranscriptionWord(
                    word=word.word,
                    start=word.start,
                    end=word.end,
                    speaker=best_speaker,
                )
            )
        return result

    def diarize_transcription(self, audio: Audio, transcription: Transcription) -> Transcription:
        """Attach speaker labels to a pre-computed transcription using pyannote.

        Useful when callers have a transcription (e.g. pre-computed and edited)
        but no speakers, and want per-speaker voice cloning in dubbing without
        re-running Whisper. Runs pyannote standalone on ``audio`` and overlays
        speakers onto the supplied transcription's words.

        Requires word-level timings: at least one segment must contain more
        than one word. Transcriptions loaded from SRT (one synthetic word per
        segment) will not produce useful speakers and are rejected.
        """
        import numpy as np
        import torch

        all_words = sorted(transcription.words, key=lambda word: (word.start, word.end))
        if not all_words:
            raise ValueError("Cannot diarize a transcription with no words.")

        if not any(len(seg.words) > 1 for seg in transcription.segments):
            raise ValueError(
                "Cannot diarize a transcription without word-level timings. "
                "Supplied transcription has at most one word per segment "
                "(e.g. loaded from SRT). Provide a transcription with "
                "word-level timings, or omit `transcription` to let the "
                "pipeline transcribe and diarize from scratch."
            )

        if self._diarization_pipeline is None:
            self._init_diarization()

        audio_mono = audio.to_mono().resample(_WHISPER_SAMPLE_RATE)
        waveform = torch.from_numpy(audio_mono.data.astype(np.float32)).unsqueeze(0)
        diarization_result = self._diarization_pipeline(
            {"waveform": waveform, "sample_rate": audio_mono.metadata.sample_rate}
        )

        all_words = self._assign_speakers_to_words(all_words, diarization_result)

        # Rebuilding from words regroups by speaker and drops the per-segment
        # confidence the supplied transcription carried, exactly as it does on the
        # combined path -- so re-attach it the same way. Without this, splitting
        # transcription and diarization into two calls silently loses confidence
        # that running them as one keeps.
        source_segments = sorted(transcription.segments, key=lambda segment: (segment.start, segment.end))
        rebuilt = Transcription(words=all_words, language=transcription.language)
        _attach_confidence_by_overlap(rebuilt.segments, source_segments)
        return rebuilt

    def _run_vad(self, audio_mono: Audio) -> list[tuple[float, float]]:
        """Return voiced spans in seconds using Silero VAD.

        Audio must already be mono at 16 kHz,
        which is one of Silero's two supported rates.
        """
        import numpy as np
        import torch

        if self._vad_model is None:
            self._init_vad()

        from silero_vad import get_speech_timestamps

        waveform = torch.from_numpy(audio_mono.data.astype(np.float32))
        timestamps = get_speech_timestamps(
            waveform,
            self._vad_model,
            sampling_rate=audio_mono.metadata.sample_rate,
            return_seconds=True,
        )
        return [(float(ts["start"]), float(ts["end"])) for ts in timestamps]

    def _detect_language(self, audio_mono: Audio, voiced_spans: list[tuple[float, float]]) -> str:
        """Run Whisper language detection on a 30s window of voiced audio.

        Whisper's auto-detection only inspects the first 30s of input. When
        the file opens with silence/music/credits, that window contains no
        speech and detection picks the closest-looking thing (typically
        English). Concatenating up to 30 seconds of voiced audio fixes this.
        """
        import numpy as np

        sample_rate = audio_mono.metadata.sample_rate
        chunks: list[np.ndarray] = []
        remaining = _WHISPER_LANGUAGE_SAMPLES
        for start, end in voiced_spans:
            if remaining <= 0:
                break
            chunk = audio_mono.data[int(start * sample_rate) : int(end * sample_rate)][:remaining]
            chunks.append(chunk)
            remaining -= len(chunk)

        voiced_audio = np.concatenate(chunks).astype(np.float32)
        language, _, _ = self._model.detect_language(audio=voiced_audio)
        return language

    def _transcribe_with_diarization(
        self, audio_mono: Audio, language: str | None, vocabulary: list[str]
    ) -> Transcription:
        """Transcribe with word timestamps and assign speakers via pyannote."""
        import numpy as np
        import torch

        if self._diarization_pipeline is None:
            self._init_diarization()

        audio_data = audio_mono.data
        transcription_result = self._run_whisper(audio_data, language, vocabulary)

        waveform = torch.from_numpy(audio_data.astype(np.float32)).unsqueeze(0)
        diarization_result = self._diarization_pipeline(
            {"waveform": waveform, "sample_rate": audio_mono.metadata.sample_rate}
        )

        transcription = self._process_transcription_result(transcription_result)

        # Capture original Whisper segments before flattening to words. The
        # diarization rebuild via Transcription(words=...) regroups by speaker,
        # which loses the per-segment confidence M1.3 plumbed through. We
        # re-attach by max-overlap match below so M2's confidence-aware
        # translation prompts have signal on the diarized path too.
        whisper_segments = transcription.segments

        all_words: list[TranscriptionWord] = []
        for seg in transcription.segments:
            all_words.extend(seg.words)

        if all_words:
            all_words = self._assign_speakers_to_words(all_words, diarization_result)

        rebuilt = Transcription(words=all_words, language=transcription.language)
        _attach_confidence_by_overlap(rebuilt.segments, whisper_segments)
        return rebuilt

    def _transcribe_local(self, audio: Audio, vocabulary: list[str]) -> Transcription:
        """Transcribe using local Whisper model.

        When ``enable_vad`` is True (default), Silero VAD locates voiced
        regions and a 30s voiced window is used for Whisper language
        detection -- avoiding the well-known failure where Whisper locks
        onto the wrong language because the first 30s of input is silence
        or music. The detected language is then passed into
        ``transcribe()`` so chunked decoding stays consistent. If VAD
        finds no speech, an empty Transcription is returned without
        invoking Whisper.
        """
        if self._model is None:
            self._init_local()

        audio_mono = audio.to_mono().resample(_WHISPER_SAMPLE_RATE)

        language: str | None = None
        if self.enable_vad:
            voiced_spans = self._run_vad(audio_mono)
            if not voiced_spans:
                return Transcription(segments=[])
            language = self._detect_language(audio_mono, voiced_spans)

        if self.enable_diarization:
            return self._transcribe_with_diarization(audio_mono, language, vocabulary)

        transcription_result = self._run_whisper(audio_mono.data, language, vocabulary)
        return self._process_transcription_result(transcription_result)

    def transcribe(self, media: Audio | Video, vocabulary: list[str] | None = None) -> Transcription:
        """Transcribe audio or video to text.

        ``vocabulary`` overrides the constructor default for this call only;
        a per-call list wins over the instance's vocabulary so one
        :class:`AudioToText` instance can serve multiple tenants. Pass
        ``None`` (the default) to use the constructor's list.
        """
        if isinstance(media, Video):
            if media.audio.is_silent:
                return Transcription(segments=[])
            audio = media.audio
        elif isinstance(media, Audio):
            if media.is_silent:
                return Transcription(segments=[])
            audio = media
        else:
            raise TypeError(f"Unsupported media type: {type(media)}. Expected Audio or Video.")

        effective_vocab = self.vocabulary if vocabulary is None else _normalize_vocabulary(vocabulary)
        return self._transcribe_local(audio, effective_vocab)

diarize_transcription

diarize_transcription(
    audio: Audio, transcription: Transcription
) -> Transcription

Attach speaker labels to a pre-computed transcription using pyannote.

Useful when callers have a transcription (e.g. pre-computed and edited) but no speakers, and want per-speaker voice cloning in dubbing without re-running Whisper. Runs pyannote standalone on audio and overlays speakers onto the supplied transcription's words.

Requires word-level timings: at least one segment must contain more than one word. Transcriptions loaded from SRT (one synthetic word per segment) will not produce useful speakers and are rejected.

Source code in src/videopython/ai/understanding/audio.py

def diarize_transcription(self, audio: Audio, transcription: Transcription) -> Transcription:
    """Attach speaker labels to a pre-computed transcription using pyannote.

    Useful when callers have a transcription (e.g. pre-computed and edited)
    but no speakers, and want per-speaker voice cloning in dubbing without
    re-running Whisper. Runs pyannote standalone on ``audio`` and overlays
    speakers onto the supplied transcription's words.

    Requires word-level timings: at least one segment must contain more
    than one word. Transcriptions loaded from SRT (one synthetic word per
    segment) will not produce useful speakers and are rejected.
    """
    import numpy as np
    import torch

    all_words = sorted(transcription.words, key=lambda word: (word.start, word.end))
    if not all_words:
        raise ValueError("Cannot diarize a transcription with no words.")

    if not any(len(seg.words) > 1 for seg in transcription.segments):
        raise ValueError(
            "Cannot diarize a transcription without word-level timings. "
            "Supplied transcription has at most one word per segment "
            "(e.g. loaded from SRT). Provide a transcription with "
            "word-level timings, or omit `transcription` to let the "
            "pipeline transcribe and diarize from scratch."
        )

    if self._diarization_pipeline is None:
        self._init_diarization()

    audio_mono = audio.to_mono().resample(_WHISPER_SAMPLE_RATE)
    waveform = torch.from_numpy(audio_mono.data.astype(np.float32)).unsqueeze(0)
    diarization_result = self._diarization_pipeline(
        {"waveform": waveform, "sample_rate": audio_mono.metadata.sample_rate}
    )

    all_words = self._assign_speakers_to_words(all_words, diarization_result)

    # Rebuilding from words regroups by speaker and drops the per-segment
    # confidence the supplied transcription carried, exactly as it does on the
    # combined path -- so re-attach it the same way. Without this, splitting
    # transcription and diarization into two calls silently loses confidence
    # that running them as one keeps.
    source_segments = sorted(transcription.segments, key=lambda segment: (segment.start, segment.end))
    rebuilt = Transcription(words=all_words, language=transcription.language)
    _attach_confidence_by_overlap(rebuilt.segments, source_segments)
    return rebuilt

transcribe

transcribe(
    media: Audio | Video,
    vocabulary: list[str] | None = None,
) -> Transcription

Transcribe audio or video to text.

vocabulary overrides the constructor default for this call only; a per-call list wins over the instance's vocabulary so one :class:AudioToText instance can serve multiple tenants. Pass None (the default) to use the constructor's list.

Source code in src/videopython/ai/understanding/audio.py

def transcribe(self, media: Audio | Video, vocabulary: list[str] | None = None) -> Transcription:
    """Transcribe audio or video to text.

    ``vocabulary`` overrides the constructor default for this call only;
    a per-call list wins over the instance's vocabulary so one
    :class:`AudioToText` instance can serve multiple tenants. Pass
    ``None`` (the default) to use the constructor's list.
    """
    if isinstance(media, Video):
        if media.audio.is_silent:
            return Transcription(segments=[])
        audio = media.audio
    elif isinstance(media, Audio):
        if media.is_silent:
            return Transcription(segments=[])
        audio = media
    else:
        raise TypeError(f"Unsupported media type: {type(media)}. Expected Audio or Video.")

    effective_vocab = self.vocabulary if vocabulary is None else _normalize_vocabulary(vocabulary)
    return self._transcribe_local(audio, effective_vocab)

AudioClassifier

Sound, music and audio-event classification with timestamps, using an Audio Spectrogram Transformer.

from videopython.ai import AudioClassifier

result = AudioClassifier(confidence_threshold=0.3).classify(video)

for label, confidence in result.clip_predictions.items():
    print(f"{label}: {confidence:.2f}")

for event in result.events:
    print(f"{event.start:.1f}s - {event.end:.1f}s: {event.label} ({event.confidence:.2f})")

AudioClassifier

Bases: ManagedPredictor

Audio event and sound classification using AST.

Source code in src/videopython/ai/understanding/classification.py

class AudioClassifier(ManagedPredictor):
    """Audio event and sound classification using AST."""

    _model_attrs = ("_model", "_processor")
    AST_SAMPLE_RATE: int = 16000
    AST_CHUNK_SECONDS: float = 10.0
    AST_HOP_SECONDS: float = 5.0

    def __init__(
        self,
        model_name: str = "MIT/ast-finetuned-audioset-10-10-0.4593",
        confidence_threshold: float = 0.3,
        top_k: int = 10,
        device: str | None = None,
    ):
        self._loaded_models: dict[str, str | None] | None = None
        self.model_name = model_name
        self.confidence_threshold = confidence_threshold
        self.top_k = top_k
        self.device = select_device(device, mps_allowed=True)
        log_device_initialization(
            "AudioClassifier",
            requested_device=device,
            resolved_device=self.device,
        )

        self._model: Any = None
        self._processor: Any = None
        self._labels: list[str] = []

    def model_provenance(self) -> dict[str, str | None] | None:
        return dict(self._loaded_models) if self._loaded_models is not None else None

    def _init_local(self) -> None:
        """Initialize local AST model from HuggingFace."""
        from videopython.ai._optional import require

        _transformers = require("transformers", feature="AudioClassifier")
        ASTFeatureExtractor = _transformers.ASTFeatureExtractor
        ASTForAudioClassification = _transformers.ASTForAudioClassification

        self._processor = ASTFeatureExtractor.from_pretrained(self.model_name, revision=pinned(self.model_name))
        self._model = ASTForAudioClassification.from_pretrained(self.model_name, revision=pinned(self.model_name))
        self._model.to(self.device)
        self._model.eval()
        self._loaded_models = {self.model_name: pinned(self.model_name)}

        self._labels = [self._model.config.id2label[i] for i in range(len(self._model.config.id2label))]

    def _merge_events(self, events: list[AudioEvent], gap_threshold: float = 0.5) -> list[AudioEvent]:
        """Merge consecutive events of the same class."""
        if not events:
            return []

        events_by_label: dict[str, list[AudioEvent]] = {}
        for event in events:
            if event.label not in events_by_label:
                events_by_label[event.label] = []
            events_by_label[event.label].append(event)

        merged = []
        for label, label_events in events_by_label.items():
            sorted_events = sorted(label_events, key=lambda e: e.start)
            current = sorted_events[0]

            for next_event in sorted_events[1:]:
                if next_event.start - current.end <= gap_threshold:
                    current = AudioEvent(
                        start=current.start,
                        end=next_event.end,
                        label=label,
                        confidence=max(current.confidence, next_event.confidence),
                    )
                else:
                    merged.append(current)
                    current = next_event

            merged.append(current)

        return sorted(merged, key=lambda e: e.start)

    def _classify_local(self, audio: Audio) -> AudioClassification:
        """Classify audio using local AST model with sliding window."""
        import numpy as np
        import torch

        if self._model is None:
            self._init_local()

        audio_processed = audio.to_mono().resample(self.AST_SAMPLE_RATE)
        audio_data = audio_processed.data.astype(np.float32)

        chunk_samples = int(self.AST_CHUNK_SECONDS * self.AST_SAMPLE_RATE)
        hop_samples = int(self.AST_HOP_SECONDS * self.AST_SAMPLE_RATE)
        total_samples = len(audio_data)

        all_chunk_probs = []
        chunk_times = []

        if total_samples <= chunk_samples:
            chunks = [(0, audio_data)]
        else:
            chunks = []
            start = 0
            while start < total_samples:
                end = min(start + chunk_samples, total_samples)
                chunk = audio_data[start:end]
                if len(chunk) < chunk_samples:
                    chunk = np.pad(chunk, (0, chunk_samples - len(chunk)))
                chunks.append((start, chunk))
                start += hop_samples

        for start_sample, chunk in chunks:
            start_time = start_sample / self.AST_SAMPLE_RATE

            inputs = self._processor(
                chunk,
                sampling_rate=self.AST_SAMPLE_RATE,
                return_tensors="pt",
            )
            inputs = {k: v.to(self.device) for k, v in inputs.items()}

            with torch.no_grad():
                outputs = self._model(**inputs)
                logits = outputs.logits[0]
                probs = torch.sigmoid(logits).cpu().numpy()

            all_chunk_probs.append(probs)
            chunk_times.append(start_time)

        chunk_probs_array = np.array(all_chunk_probs)

        events = []
        for start_time, probs in zip(chunk_times, chunk_probs_array):
            end_time = start_time + self.AST_CHUNK_SECONDS
            top_indices = np.argsort(probs)[-self.top_k :][::-1]

            for class_idx in top_indices:
                confidence = float(probs[class_idx])
                if confidence >= self.confidence_threshold:
                    label = self._labels[class_idx]
                    events.append(
                        AudioEvent(
                            start=start_time,
                            end=min(end_time, total_samples / self.AST_SAMPLE_RATE),
                            label=label,
                            confidence=confidence,
                        )
                    )

        merged_events = self._merge_events(events)

        clip_preds = np.mean(chunk_probs_array, axis=0)
        top_clip_indices = np.argsort(clip_preds)[-self.top_k :][::-1]
        clip_predictions = {
            self._labels[idx]: float(clip_preds[idx])
            for idx in top_clip_indices
            if clip_preds[idx] >= self.confidence_threshold
        }

        return AudioClassification(events=merged_events, clip_predictions=clip_predictions)

    def classify(self, media: Audio | Video) -> AudioClassification:
        """Classify audio events in audio or video."""
        if isinstance(media, Video):
            if media.audio.is_silent:
                return AudioClassification(events=[], clip_predictions={})
            audio = media.audio
        elif isinstance(media, Audio):
            if media.is_silent:
                return AudioClassification(events=[], clip_predictions={})
            audio = media
        else:
            raise TypeError(f"Unsupported media type: {type(media)}. Expected Audio or Video.")

        return self._classify_local(audio)

classify

classify(media: Audio | Video) -> AudioClassification

Classify audio events in audio or video.

Source code in src/videopython/ai/understanding/classification.py

def classify(self, media: Audio | Video) -> AudioClassification:
    """Classify audio events in audio or video."""
    if isinstance(media, Video):
        if media.audio.is_silent:
            return AudioClassification(events=[], clip_predictions={})
        audio = media.audio
    elif isinstance(media, Audio):
        if media.is_silent:
            return AudioClassification(events=[], clip_predictions={})
        audio = media
    else:
        raise TypeError(f"Unsupported media type: {type(media)}. Expected Audio or Video.")

    return self._classify_local(audio)

SceneVLM

Describes scenes with an Ollama vision model. Needs a running Ollama server and a vision model that supports structured output; model is any tag you have pulled (default qwen3.6:27b).

analyze_scene() and analyze_frame() return a SceneDescription: a one-sentence caption, an open-list subjects, and a closed-enum shot_type. The schema is handed to Ollama's format, so the model returns valid JSON directly.

from videopython.ai import SceneVLM

vlm = SceneVLM()
description = vlm.analyze_frame(frame_array)

description.caption      # "A man in a cap speaks into a microphone."
description.subjects     # ["man", "microphone", "cap"]
description.shot_type    # "medium"

SceneVLM.unload() clears the Ollama client, for low_memory parity.

SceneVLM

Bases: ManagedPredictor

Generates structured scene descriptions with a local Ollama vision model.

The model must be vision-capable and support Ollama's structured-output format; ollama pull <model> first. options are extra Ollama generation options merged over temperature=0.

A scene's frames are sent as one multi-image request, so the context window is sized to the frame count automatically -- Ollama's 4096-token default fits only one or two frames and fails anything larger. Pass an explicit num_ctx in options to override that sizing.

Source code in src/videopython/ai/understanding/image.py

class SceneVLM(ManagedPredictor):
    """Generates structured scene descriptions with a local Ollama vision model.

    The model must be vision-capable and support Ollama's structured-output
    ``format``; ``ollama pull <model>`` first. ``options`` are extra Ollama
    generation options merged over ``temperature=0``.

    A scene's frames are sent as one multi-image request, so the context window
    is sized to the frame count automatically -- Ollama's 4096-token default
    fits only one or two frames and *fails* anything larger. Pass an explicit
    ``num_ctx`` in ``options`` to override that sizing.
    """

    def __init__(
        self,
        model: str = DEFAULT_SCENE_VLM_MODEL,
        *,
        host: str | None = None,
        options: dict[str, Any] | None = None,
    ) -> None:
        self._client = OllamaStructuredClient(model=model, host=host, options=options)

    def analyze_frame(self, image: np.ndarray | Image.Image, prompt: str | None = None) -> SceneDescription:
        """Analyze one frame and return a structured scene description."""
        return self.analyze_scene([image], prompt=prompt)

    def analyze_scene(self, images: list[np.ndarray | Image.Image], prompt: str | None = None) -> SceneDescription:
        """Analyze a scene's frames and return a structured description."""
        if not images:
            raise ValueError("`images` must contain at least one frame")
        frames = [_to_rgb_array(image) for image in images]
        data = self._client.generate_json(
            system=_SYSTEM_PROMPT, text=prompt or _USER_PROMPT, schema=_SCENE_SCHEMA, images=frames
        )
        shot_type = data.get("shot_type")
        return SceneDescription(
            caption=str(data.get("caption", "")),
            subjects=[str(s) for s in data.get("subjects", [])],
            shot_type=shot_type if shot_type in _SHOT_TYPES else None,
        )

    def model_provenance(self) -> dict[str, str | None] | None:
        return self._client.model_provenance()

    def unload(self) -> None:
        self._client.unload()

analyze_frame

analyze_frame(
    image: ndarray | Image, prompt: str | None = None
) -> SceneDescription

Analyze one frame and return a structured scene description.

Source code in src/videopython/ai/understanding/image.py

def analyze_frame(self, image: np.ndarray | Image.Image, prompt: str | None = None) -> SceneDescription:
    """Analyze one frame and return a structured scene description."""
    return self.analyze_scene([image], prompt=prompt)

analyze_scene

analyze_scene(
    images: list[ndarray | Image], prompt: str | None = None
) -> SceneDescription

Analyze a scene's frames and return a structured description.

Source code in src/videopython/ai/understanding/image.py

def analyze_scene(self, images: list[np.ndarray | Image.Image], prompt: str | None = None) -> SceneDescription:
    """Analyze a scene's frames and return a structured description."""
    if not images:
        raise ValueError("`images` must contain at least one frame")
    frames = [_to_rgb_array(image) for image in images]
    data = self._client.generate_json(
        system=_SYSTEM_PROMPT, text=prompt or _USER_PROMPT, schema=_SCENE_SCHEMA, images=frames
    )
    shot_type = data.get("shot_type")
    return SceneDescription(
        caption=str(data.get("caption", "")),
        subjects=[str(s) for s in data.get("subjects", [])],
        shot_type=shot_type if shot_type in _SHOT_TYPES else None,
    )

SemanticSceneDetector

TransNetV2 scene-boundary detection — more accurate than histogram methods, especially on fades and dissolves.

from videopython.ai import SemanticSceneDetector

detector = SemanticSceneDetector(threshold=0.5, min_scene_length=1.0)
for scene in detector.detect_streaming("video.mp4"):
    print(f"{scene.start:.1f}s - {scene.end:.1f}s ({scene.duration:.1f}s)")

SemanticSceneDetector

Bases: ManagedPredictor

ML-based scene detection using TransNetV2.

TransNetV2 is a neural network specifically designed for shot boundary detection, providing more accurate scene boundaries than histogram-based methods, especially for gradual transitions.

Uses the transnetv2-pytorch package with pretrained weights.

Example

from videopython.ai.understanding import SemanticSceneDetector detector = SemanticSceneDetector() scenes = detector.detect_streaming("video.mp4") for scene in scenes: ... print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s")

Source code in src/videopython/ai/understanding/temporal.py

class SemanticSceneDetector(ManagedPredictor):
    """ML-based scene detection using TransNetV2.

    TransNetV2 is a neural network specifically designed for shot boundary
    detection, providing more accurate scene boundaries than histogram-based
    methods, especially for gradual transitions.

    Uses the transnetv2-pytorch package with pretrained weights.

    Example:
        >>> from videopython.ai.understanding import SemanticSceneDetector
        >>> detector = SemanticSceneDetector()
        >>> scenes = detector.detect_streaming("video.mp4")
        >>> for scene in scenes:
        ...     print(f"Scene: {scene.start:.2f}s - {scene.end:.2f}s")
    """

    def __init__(
        self,
        threshold: float = 0.5,
        min_scene_length: float = 0.5,
        device: str | None = None,
    ):
        """Initialize the semantic scene detector.

        Args:
            threshold: Confidence threshold for scene boundaries (0.0-1.0).
                Higher values = fewer, more confident boundaries.
            min_scene_length: Minimum scene duration in seconds.
            device: Device to run on ('cuda', 'mps', 'cpu', or None for auto).
                Note: MPS may have numerical inconsistencies; use 'cpu' for
                reproducible results.
        """
        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
        self.device: str | None = device
        self._loaded_models: dict[str, str | None] | None = None
        self._model: Any = None

    def model_provenance(self) -> dict[str, str | None] | None:
        return self._loaded_models

    def _init_local(self) -> None:
        """Load the TransNetV2 model with pretrained weights."""
        if self._model is not None:
            return

        from videopython.ai._optional import require

        TransNetV2 = require("transnetv2_pytorch", feature="SemanticSceneDetector").TransNetV2

        requested_device = self.device
        device = select_device(self.device, mps_allowed=True)
        log_device_initialization(
            "SemanticSceneDetector",
            requested_device=requested_device,
            resolved_device=device,
        )
        self.device = device
        self._model = TransNetV2(device=device)
        self._model.eval()
        self._loaded_models = {"transnetv2-pytorch/bundled": package_revision("transnetv2-pytorch")}

    def detect(self, video: Video) -> list[SceneBoundary]:
        """Detect scenes in a video using ML-based boundary detection.

        Note: This method requires saving video to a temporary file for
        TransNetV2 processing. For better performance, use detect_streaming()
        with a file path directly.

        Args:
            video: Video object to analyze.

        Returns:
            List of SceneBoundary objects representing detected scenes.
        """
        import tempfile

        if len(video.frames) == 0:
            return []

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

        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=True) as tmp:
            video.save(tmp.name)
            return self.detect_streaming(tmp.name)

    def detect_streaming(self, path: str | Path) -> list[SceneBoundary]:
        """Detect scenes from a video file.

        Uses TransNetV2 with pretrained weights for accurate shot boundary
        detection.

        Args:
            path: Path to video file.

        Returns:
            List of SceneBoundary objects representing detected scenes.
        """
        self._init_local()

        raw_scenes = self._model.detect_scenes(str(path), threshold=self.threshold)

        # Convert to SceneBoundary objects
        scenes = []
        for scene_data in raw_scenes:
            start_frame = scene_data["start_frame"]
            end_frame = scene_data["end_frame"]
            start_time = float(scene_data["start_time"])
            end_time = float(scene_data["end_time"])

            scenes.append(
                SceneBoundary(
                    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 _merge_short_scenes(self, scenes: list[SceneBoundary]) -> list[SceneBoundary]:
        """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] = SceneBoundary(
                    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] = SceneBoundary(
                start=second_last.start,
                end=last.end,
                start_frame=second_last.start_frame,
                end_frame=last.end_frame,
            )
            merged.pop()

        return merged

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

        Args:
            path: Path to video file.
            threshold: Scene boundary threshold (0.0-1.0).
            min_scene_length: Minimum scene duration in seconds.

        Returns:
            List of SceneBoundary objects representing detected scenes.
        """
        detector = cls(threshold=threshold, min_scene_length=min_scene_length)
        return detector.detect_streaming(path)

__init__

__init__(
    threshold: float = 0.5,
    min_scene_length: float = 0.5,
    device: str | None = None,
)

Initialize the semantic scene detector.

Parameters:

Name	Type	Description	Default
threshold	float	Confidence threshold for scene boundaries (0.0-1.0). Higher values = fewer, more confident boundaries.	0.5
min_scene_length	float	Minimum scene duration in seconds.	0.5
device	str | None	Device to run on ('cuda', 'mps', 'cpu', or None for auto). Note: MPS may have numerical inconsistencies; use 'cpu' for reproducible results.	None

Source code in src/videopython/ai/understanding/temporal.py

def __init__(
    self,
    threshold: float = 0.5,
    min_scene_length: float = 0.5,
    device: str | None = None,
):
    """Initialize the semantic scene detector.

    Args:
        threshold: Confidence threshold for scene boundaries (0.0-1.0).
            Higher values = fewer, more confident boundaries.
        min_scene_length: Minimum scene duration in seconds.
        device: Device to run on ('cuda', 'mps', 'cpu', or None for auto).
            Note: MPS may have numerical inconsistencies; use 'cpu' for
            reproducible results.
    """
    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
    self.device: str | None = device
    self._loaded_models: dict[str, str | None] | None = None
    self._model: Any = None

detect

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

Detect scenes in a video using ML-based boundary detection.

Note: This method requires saving video to a temporary file for TransNetV2 processing. For better performance, use detect_streaming() with a file path directly.

Parameters:

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

Returns:

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

Source code in src/videopython/ai/understanding/temporal.py

def detect(self, video: Video) -> list[SceneBoundary]:
    """Detect scenes in a video using ML-based boundary detection.

    Note: This method requires saving video to a temporary file for
    TransNetV2 processing. For better performance, use detect_streaming()
    with a file path directly.

    Args:
        video: Video object to analyze.

    Returns:
        List of SceneBoundary objects representing detected scenes.
    """
    import tempfile

    if len(video.frames) == 0:
        return []

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

    with tempfile.NamedTemporaryFile(suffix=".mp4", delete=True) as tmp:
        video.save(tmp.name)
        return self.detect_streaming(tmp.name)

detect_streaming

detect_streaming(path: str | Path) -> list[SceneBoundary]

Detect scenes from a video file.

Uses TransNetV2 with pretrained weights for accurate shot boundary detection.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to video file.	required

Returns:

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

Source code in src/videopython/ai/understanding/temporal.py

def detect_streaming(self, path: str | Path) -> list[SceneBoundary]:
    """Detect scenes from a video file.

    Uses TransNetV2 with pretrained weights for accurate shot boundary
    detection.

    Args:
        path: Path to video file.

    Returns:
        List of SceneBoundary objects representing detected scenes.
    """
    self._init_local()

    raw_scenes = self._model.detect_scenes(str(path), threshold=self.threshold)

    # Convert to SceneBoundary objects
    scenes = []
    for scene_data in raw_scenes:
        start_frame = scene_data["start_frame"]
        end_frame = scene_data["end_frame"]
        start_time = float(scene_data["start_time"])
        end_time = float(scene_data["end_time"])

        scenes.append(
            SceneBoundary(
                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.5,
    min_scene_length: float = 0.5,
) -> list[SceneBoundary]

Convenience method for one-shot scene detection.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to video file.	required
threshold	float	Scene boundary threshold (0.0-1.0).	0.5
min_scene_length	float	Minimum scene duration in seconds.	0.5

Returns:

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

Source code in src/videopython/ai/understanding/temporal.py

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

    Args:
        path: Path to video file.
        threshold: Scene boundary threshold (0.0-1.0).
        min_scene_length: Minimum scene duration in seconds.

    Returns:
        List of SceneBoundary objects representing detected scenes.
    """
    detector = cls(threshold=threshold, min_scene_length=min_scene_length)
    return detector.detect_streaming(path)

Face tracking

Two YuNet-based trackers share one detector, one per use case:

• FaceShotTracker.track_shot(frames, frame_indices) returns FaceTrack objects with ids that are stable within a shot, associated by IoU. There is no embedding re-identification, so a track does not survive a shot boundary. This is what VideoAnalyzer uses.
• FaceSmoothingTracker.detect_and_track(frame, frame_index) / track_video(frames) are the single-subject smoothed-position APIs behind FaceTrackingCrop.

from videopython.ai import FaceShotTracker

for track in FaceShotTracker().track_shot(frames):
    print(f"track #{track.track_id}: {track.length} frames, first {track.frame_indices[0]}")

FaceShotTracker

Bases: _FaceTrackerBase

Per-shot multi-track face association via IoU.

Detects faces on every input frame and stitches them into FaceTracks greedily by best IoU. Tracks do not survive across shot boundaries (IoU-only association; no embedding re-id). Used by the video-analysis pipeline to bind detections to subjects within one shot.

Source code in src/videopython/ai/understanding/faces.py

class FaceShotTracker(_FaceTrackerBase):
    """Per-shot multi-track face association via IoU.

    Detects faces on every input frame and stitches them into ``FaceTrack``s
    greedily by best IoU. Tracks do not survive across shot boundaries
    (IoU-only association; no embedding re-id). Used by the video-analysis
    pipeline to bind detections to subjects within one shot.
    """

    def __init__(
        self,
        min_face_size: int = 30,
        batch_size: int = 16,
        iou_match_threshold: float = DEFAULT_IOU_MATCH_THRESHOLD,
        max_missed_frames: int = DEFAULT_MAX_MISSED_FRAMES,
    ):
        """Initialize the per-shot tracker.

        Args:
            min_face_size: Minimum face size in pixels for detection.
            batch_size: Batch size for detection. Default 16.
            iou_match_threshold: Minimum IoU between consecutive detections to
                continue an existing track.
            max_missed_frames: Consecutive frames a track may go without a
                detection before it is closed.
        """
        super().__init__(min_face_size=min_face_size)
        self.batch_size = batch_size
        self.iou_match_threshold = iou_match_threshold
        self.max_missed_frames = max_missed_frames
        logger.info("FaceShotTracker initialized (min_face_size=%s)", self.min_face_size)

    def track_shot(
        self,
        frames: list[np.ndarray] | np.ndarray,
        frame_indices: list[int] | None = None,
    ) -> list[FaceTrack]:
        """Per-shot multi-track association via IoU.

        Detection is run on every input frame (caller is expected to have
        already chosen the sampling cadence -- the analysis pipeline
        passes one frame per scene-VLM sample, lip-sync passes every
        frame in the shot). Tracks are stitched together greedily by
        best IoU above ``iou_match_threshold``; tracks with no match for
        ``max_missed_frames`` consecutive frames are closed and won't
        accept future associations.

        Track ids are integers starting at 1 within this shot. They are
        **not** stable across shots — embedding re-id is deferred.

        Args:
            frames: Frames in the shot (list or stacked ndarray).
            frame_indices: Source-video frame indices. Defaults to
                ``range(len(frames))`` when omitted.

        Returns:
            List of ``FaceTrack`` objects, one per distinct subject
            tracked in the shot.
        """
        if isinstance(frames, np.ndarray):
            frame_list = [frames[i] for i in range(frames.shape[0])] if frames.ndim == 4 else [frames]
        else:
            frame_list = list(frames)

        if not frame_list:
            return []

        if frame_indices is None:
            frame_indices = list(range(len(frame_list)))
        if len(frame_indices) != len(frame_list):
            raise ValueError("frame_indices length must match frames length")

        if self._detector is None:
            self._init_detector()
            assert self._detector is not None

        per_frame_detections: list[list[DetectedFace]] = []
        for batch_start in range(0, len(frame_list), self.batch_size):
            batch = frame_list[batch_start : batch_start + self.batch_size]
            per_frame_detections.extend(self._detector.detect_batch(batch))

        active: list[_OpenTrack] = []
        finished: list[_OpenTrack] = []
        next_id = 1

        for relative_idx, faces in enumerate(per_frame_detections):
            absolute_idx = frame_indices[relative_idx]
            available = [face for face in faces if face.bounding_box is not None]
            assignments: dict[int, DetectedFace] = {}

            for track in active:
                best_face: DetectedFace | None = None
                best_iou = self.iou_match_threshold
                last_box = track.last_box
                if last_box is None:
                    continue
                for face in available:
                    if face in assignments.values() or face.bounding_box is None:
                        continue
                    iou = _bbox_iou(last_box, face.bounding_box)
                    if iou > best_iou:
                        best_iou = iou
                        best_face = face
                if best_face is not None:
                    assignments[track.track_id] = best_face

            for track in active:
                if track.track_id in assignments:
                    face = assignments[track.track_id]
                    assert face.bounding_box is not None
                    track.frame_indices.append(absolute_idx)
                    track.boxes.append(face.bounding_box)
                    track.confidences.append(face.confidence)
                    track.last_box = face.bounding_box
                    track.missed = 0
                else:
                    track.missed += 1

            for face in available:
                if face in assignments.values() or face.bounding_box is None:
                    continue
                track = _OpenTrack(track_id=next_id, last_box=face.bounding_box)
                next_id += 1
                track.frame_indices.append(absolute_idx)
                track.boxes.append(face.bounding_box)
                track.confidences.append(face.confidence)
                active.append(track)

            still_active: list[_OpenTrack] = []
            for track in active:
                if track.missed > self.max_missed_frames:
                    finished.append(track)
                else:
                    still_active.append(track)
            active = still_active

        finished.extend(active)

        return [
            FaceTrack(
                track_id=track.track_id,
                frame_indices=track.frame_indices,
                boxes=track.boxes,
                confidences=track.confidences,
            )
            for track in finished
            if track.frame_indices
        ]

__init__

__init__(
    min_face_size: int = 30,
    batch_size: int = 16,
    iou_match_threshold: float = DEFAULT_IOU_MATCH_THRESHOLD,
    max_missed_frames: int = DEFAULT_MAX_MISSED_FRAMES,
)

Initialize the per-shot tracker.

Parameters:

Name	Type	Description	Default
min_face_size	int	Minimum face size in pixels for detection.	30
batch_size	int	Batch size for detection. Default 16.	16
iou_match_threshold	float	Minimum IoU between consecutive detections to continue an existing track.	DEFAULT_IOU_MATCH_THRESHOLD
max_missed_frames	int	Consecutive frames a track may go without a detection before it is closed.	DEFAULT_MAX_MISSED_FRAMES

Source code in src/videopython/ai/understanding/faces.py

def __init__(
    self,
    min_face_size: int = 30,
    batch_size: int = 16,
    iou_match_threshold: float = DEFAULT_IOU_MATCH_THRESHOLD,
    max_missed_frames: int = DEFAULT_MAX_MISSED_FRAMES,
):
    """Initialize the per-shot tracker.

    Args:
        min_face_size: Minimum face size in pixels for detection.
        batch_size: Batch size for detection. Default 16.
        iou_match_threshold: Minimum IoU between consecutive detections to
            continue an existing track.
        max_missed_frames: Consecutive frames a track may go without a
            detection before it is closed.
    """
    super().__init__(min_face_size=min_face_size)
    self.batch_size = batch_size
    self.iou_match_threshold = iou_match_threshold
    self.max_missed_frames = max_missed_frames
    logger.info("FaceShotTracker initialized (min_face_size=%s)", self.min_face_size)

track_shot

track_shot(
    frames: list[ndarray] | ndarray,
    frame_indices: list[int] | None = None,
) -> list[FaceTrack]

Per-shot multi-track association via IoU.

Detection is run on every input frame (caller is expected to have already chosen the sampling cadence -- the analysis pipeline passes one frame per scene-VLM sample, lip-sync passes every frame in the shot). Tracks are stitched together greedily by best IoU above iou_match_threshold; tracks with no match for max_missed_frames consecutive frames are closed and won't accept future associations.

Track ids are integers starting at 1 within this shot. They are not stable across shots — embedding re-id is deferred.

Parameters:

Name	Type	Description	Default
frames	list[ndarray] | ndarray	Frames in the shot (list or stacked ndarray).	required
frame_indices	list[int] | None	Source-video frame indices. Defaults to range(len(frames)) when omitted.	None

Returns:

Type	Description
list[FaceTrack]	List of FaceTrack objects, one per distinct subject
list[FaceTrack]	tracked in the shot.

Source code in src/videopython/ai/understanding/faces.py

def track_shot(
    self,
    frames: list[np.ndarray] | np.ndarray,
    frame_indices: list[int] | None = None,
) -> list[FaceTrack]:
    """Per-shot multi-track association via IoU.

    Detection is run on every input frame (caller is expected to have
    already chosen the sampling cadence -- the analysis pipeline
    passes one frame per scene-VLM sample, lip-sync passes every
    frame in the shot). Tracks are stitched together greedily by
    best IoU above ``iou_match_threshold``; tracks with no match for
    ``max_missed_frames`` consecutive frames are closed and won't
    accept future associations.

    Track ids are integers starting at 1 within this shot. They are
    **not** stable across shots — embedding re-id is deferred.

    Args:
        frames: Frames in the shot (list or stacked ndarray).
        frame_indices: Source-video frame indices. Defaults to
            ``range(len(frames))`` when omitted.

    Returns:
        List of ``FaceTrack`` objects, one per distinct subject
        tracked in the shot.
    """
    if isinstance(frames, np.ndarray):
        frame_list = [frames[i] for i in range(frames.shape[0])] if frames.ndim == 4 else [frames]
    else:
        frame_list = list(frames)

    if not frame_list:
        return []

    if frame_indices is None:
        frame_indices = list(range(len(frame_list)))
    if len(frame_indices) != len(frame_list):
        raise ValueError("frame_indices length must match frames length")

    if self._detector is None:
        self._init_detector()
        assert self._detector is not None

    per_frame_detections: list[list[DetectedFace]] = []
    for batch_start in range(0, len(frame_list), self.batch_size):
        batch = frame_list[batch_start : batch_start + self.batch_size]
        per_frame_detections.extend(self._detector.detect_batch(batch))

    active: list[_OpenTrack] = []
    finished: list[_OpenTrack] = []
    next_id = 1

    for relative_idx, faces in enumerate(per_frame_detections):
        absolute_idx = frame_indices[relative_idx]
        available = [face for face in faces if face.bounding_box is not None]
        assignments: dict[int, DetectedFace] = {}

        for track in active:
            best_face: DetectedFace | None = None
            best_iou = self.iou_match_threshold
            last_box = track.last_box
            if last_box is None:
                continue
            for face in available:
                if face in assignments.values() or face.bounding_box is None:
                    continue
                iou = _bbox_iou(last_box, face.bounding_box)
                if iou > best_iou:
                    best_iou = iou
                    best_face = face
            if best_face is not None:
                assignments[track.track_id] = best_face

        for track in active:
            if track.track_id in assignments:
                face = assignments[track.track_id]
                assert face.bounding_box is not None
                track.frame_indices.append(absolute_idx)
                track.boxes.append(face.bounding_box)
                track.confidences.append(face.confidence)
                track.last_box = face.bounding_box
                track.missed = 0
            else:
                track.missed += 1

        for face in available:
            if face in assignments.values() or face.bounding_box is None:
                continue
            track = _OpenTrack(track_id=next_id, last_box=face.bounding_box)
            next_id += 1
            track.frame_indices.append(absolute_idx)
            track.boxes.append(face.bounding_box)
            track.confidences.append(face.confidence)
            active.append(track)

        still_active: list[_OpenTrack] = []
        for track in active:
            if track.missed > self.max_missed_frames:
                finished.append(track)
            else:
                still_active.append(track)
        active = still_active

    finished.extend(active)

    return [
        FaceTrack(
            track_id=track.track_id,
            frame_indices=track.frame_indices,
            boxes=track.boxes,
            confidences=track.confidences,
        )
        for track in finished
        if track.frame_indices
    ]

FaceSmoothingTracker

Bases: _FaceTrackerBase

Single-subject face tracker with EMA position smoothing.

Selects one face per frame (selection_strategy) and returns a smoothed (cx, cy, w, h) tuple in normalized coords via detect_and_track / track_video. Used by FaceTrackingCrop to drive a follow-the-speaker crop.

Source code in src/videopython/ai/understanding/faces.py

class FaceSmoothingTracker(_FaceTrackerBase):
    """Single-subject face tracker with EMA position smoothing.

    Selects one face per frame (``selection_strategy``) and returns a smoothed
    ``(cx, cy, w, h)`` tuple in normalized coords via ``detect_and_track`` /
    ``track_video``. Used by ``FaceTrackingCrop`` to drive a follow-the-speaker
    crop.
    """

    def __init__(
        self,
        selection_strategy: Literal["largest", "centered", "index"] = "largest",
        face_index: int = 0,
        smoothing: float = 0.8,
        detection_interval: int = 3,
        min_face_size: int = 30,
        batch_size: int = 16,
    ):
        """Initialize the smoothing tracker.

        Args:
            selection_strategy: Which face to track — "largest" (biggest box),
                "centered" (closest to frame center), or "index" (``face_index``).
            face_index: Index of face to track when using the "index" strategy.
            smoothing: Exponential moving average factor (0-1). Higher = smoother.
            detection_interval: Run detection every N frames, hold position between.
            min_face_size: Minimum face size in pixels for detection.
            batch_size: Frames per detection batch in ``track_video``. Default 16.
        """
        super().__init__(min_face_size=min_face_size)
        self.selection_strategy = selection_strategy
        self.face_index = face_index
        self.smoothing = smoothing
        self.detection_interval = detection_interval
        self.batch_size = batch_size
        self._last_position: tuple[float, float] | None = None
        self._last_size: tuple[float, float] | None = None
        self._smoothed_position: tuple[float, float] | None = None
        self._smoothed_size: tuple[float, float] | None = None
        logger.info("FaceSmoothingTracker initialized (detection_interval=%s)", self.detection_interval)

    def _select_face(
        self,
        faces: list[DetectedFace],
        frame_width: int,
        frame_height: int,
    ) -> tuple[float, float, float, float] | None:
        """Select a face based on the configured strategy.

        Args:
            faces: List of DetectedFace objects.
            frame_width: Width of the frame.
            frame_height: Height of the frame.

        Returns:
            Tuple of (center_x, center_y, width, height) in normalized coords, or None.
        """
        faces_with_box = [(f, f.bounding_box) for f in faces if f.bounding_box is not None]
        if not faces_with_box:
            return None

        if self.selection_strategy == "largest":
            _, bbox = faces_with_box[0]
        elif self.selection_strategy == "centered":
            frame_center = (0.5, 0.5)
            _, bbox = min(
                faces_with_box,
                key=lambda fb: (fb[1].center[0] - frame_center[0]) ** 2 + (fb[1].center[1] - frame_center[1]) ** 2,
            )
        elif self.selection_strategy == "index":
            idx = self.face_index if self.face_index < len(faces_with_box) else 0
            _, bbox = faces_with_box[idx]
        else:
            _, bbox = faces_with_box[0]

        return (bbox.center[0], bbox.center[1], bbox.width, bbox.height)

    def detect_and_track(
        self,
        frame: np.ndarray,
        frame_index: int,
    ) -> tuple[float, float, float, float] | None:
        """Detect face in frame and return smoothed position.

        Args:
            frame: Video frame as numpy array (H, W, 3).
            frame_index: Index of current frame.

        Returns:
            Tuple of (center_x, center_y, width, height) in normalized coords,
            or None if no face detected and no fallback available.
        """
        if self._detector is None:
            self._init_detector()
            assert self._detector is not None

        h, w = frame.shape[:2]

        if frame_index % self.detection_interval == 0:
            faces = self._detector.detect(frame)
            face_info = self._select_face(faces, w, h)
            if face_info is not None:
                self._last_position = (face_info[0], face_info[1])
                self._last_size = (face_info[2], face_info[3])
        elif self._last_position is not None and self._last_size is not None:
            face_info = (*self._last_position, *self._last_size)
        else:
            face_info = None

        return self._smooth(face_info)

    def _smooth(
        self,
        face_info: tuple[float, float, float, float] | None,
    ) -> tuple[float, float, float, float] | None:
        """Apply EMA smoothing, or replay the last smoothed value when no detection.

        Returns ``None`` when no detection has been seen yet.
        """
        if face_info is not None:
            cx, cy, fw, fh = face_info
            if self._smoothed_position is None:
                self._smoothed_position = (cx, cy)
                self._smoothed_size = (fw, fh)
            else:
                assert self._smoothed_size is not None
                alpha = 1 - self.smoothing
                self._smoothed_position = (
                    self._smoothed_position[0] * self.smoothing + cx * alpha,
                    self._smoothed_position[1] * self.smoothing + cy * alpha,
                )
                self._smoothed_size = (
                    self._smoothed_size[0] * self.smoothing + fw * alpha,
                    self._smoothed_size[1] * self.smoothing + fh * alpha,
                )
            return (*self._smoothed_position, *self._smoothed_size)

        if self._smoothed_position is not None and self._smoothed_size is not None:
            return (*self._smoothed_position, *self._smoothed_size)
        return None

    def reset(self) -> None:
        """Reset tracker state for a new video."""
        self._last_position = None
        self._last_size = None
        self._smoothed_position = None
        self._smoothed_size = None

    def track_video(
        self,
        frames: np.ndarray,
    ) -> list[tuple[float, float, float, float] | None]:
        """Track the face through a whole clip via batched per-frame detection.

        Detection runs on every frame (the YuNet detector is CPU-only), then each
        frame's selected face is EMA-smoothed.

        Args:
            frames: Video frames array of shape (N, H, W, 3).

        Returns:
            List of face positions (cx, cy, w, h) for each frame, or None where
            no face was detected and no fallback was available.
        """
        if self._detector is None:
            self._init_detector()
            assert self._detector is not None

        n_frames = len(frames)
        if n_frames == 0:
            return []

        h, w = frames[0].shape[:2]

        detections: list[list[DetectedFace]] = []
        for batch_start in range(0, n_frames, self.batch_size):
            batch = [frames[i] for i in range(batch_start, min(batch_start + self.batch_size, n_frames))]
            detections.extend(self._detector.detect_batch(batch))

        faces = [self._select_face(frame_faces, w, h) for frame_faces in detections]
        self.reset()
        return [self._smooth(face_info) for face_info in faces]

__init__

__init__(
    selection_strategy: Literal[
        "largest", "centered", "index"
    ] = "largest",
    face_index: int = 0,
    smoothing: float = 0.8,
    detection_interval: int = 3,
    min_face_size: int = 30,
    batch_size: int = 16,
)

Initialize the smoothing tracker.

Parameters:

Name	Type	Description	Default
selection_strategy	Literal['largest', 'centered', 'index']	Which face to track — "largest" (biggest box), "centered" (closest to frame center), or "index" (face_index).	'largest'
face_index	int	Index of face to track when using the "index" strategy.	0
smoothing	float	Exponential moving average factor (0-1). Higher = smoother.	0.8
detection_interval	int	Run detection every N frames, hold position between.	3
min_face_size	int	Minimum face size in pixels for detection.	30
batch_size	int	Frames per detection batch in track_video. Default 16.	16

Source code in src/videopython/ai/understanding/faces.py

def __init__(
    self,
    selection_strategy: Literal["largest", "centered", "index"] = "largest",
    face_index: int = 0,
    smoothing: float = 0.8,
    detection_interval: int = 3,
    min_face_size: int = 30,
    batch_size: int = 16,
):
    """Initialize the smoothing tracker.

    Args:
        selection_strategy: Which face to track — "largest" (biggest box),
            "centered" (closest to frame center), or "index" (``face_index``).
        face_index: Index of face to track when using the "index" strategy.
        smoothing: Exponential moving average factor (0-1). Higher = smoother.
        detection_interval: Run detection every N frames, hold position between.
        min_face_size: Minimum face size in pixels for detection.
        batch_size: Frames per detection batch in ``track_video``. Default 16.
    """
    super().__init__(min_face_size=min_face_size)
    self.selection_strategy = selection_strategy
    self.face_index = face_index
    self.smoothing = smoothing
    self.detection_interval = detection_interval
    self.batch_size = batch_size
    self._last_position: tuple[float, float] | None = None
    self._last_size: tuple[float, float] | None = None
    self._smoothed_position: tuple[float, float] | None = None
    self._smoothed_size: tuple[float, float] | None = None
    logger.info("FaceSmoothingTracker initialized (detection_interval=%s)", self.detection_interval)

detect_and_track

detect_and_track(
    frame: ndarray, frame_index: int
) -> tuple[float, float, float, float] | None

Detect face in frame and return smoothed position.

Parameters:

Name	Type	Description	Default
frame	ndarray	Video frame as numpy array (H, W, 3).	required
frame_index	int	Index of current frame.	required

Returns:

Type	Description
tuple[float, float, float, float] | None	Tuple of (center_x, center_y, width, height) in normalized coords,
tuple[float, float, float, float] | None	or None if no face detected and no fallback available.

Source code in src/videopython/ai/understanding/faces.py

def detect_and_track(
    self,
    frame: np.ndarray,
    frame_index: int,
) -> tuple[float, float, float, float] | None:
    """Detect face in frame and return smoothed position.

    Args:
        frame: Video frame as numpy array (H, W, 3).
        frame_index: Index of current frame.

    Returns:
        Tuple of (center_x, center_y, width, height) in normalized coords,
        or None if no face detected and no fallback available.
    """
    if self._detector is None:
        self._init_detector()
        assert self._detector is not None

    h, w = frame.shape[:2]

    if frame_index % self.detection_interval == 0:
        faces = self._detector.detect(frame)
        face_info = self._select_face(faces, w, h)
        if face_info is not None:
            self._last_position = (face_info[0], face_info[1])
            self._last_size = (face_info[2], face_info[3])
    elif self._last_position is not None and self._last_size is not None:
        face_info = (*self._last_position, *self._last_size)
    else:
        face_info = None

    return self._smooth(face_info)

reset

reset() -> None

Reset tracker state for a new video.

Source code in src/videopython/ai/understanding/faces.py

def reset(self) -> None:
    """Reset tracker state for a new video."""
    self._last_position = None
    self._last_size = None
    self._smoothed_position = None
    self._smoothed_size = None

track_video

track_video(
    frames: ndarray,
) -> list[tuple[float, float, float, float] | None]

Track the face through a whole clip via batched per-frame detection.

Detection runs on every frame (the YuNet detector is CPU-only), then each frame's selected face is EMA-smoothed.

Parameters:

Name	Type	Description	Default
frames	ndarray	Video frames array of shape (N, H, W, 3).	required

Returns:

Type	Description
list[tuple[float, float, float, float] | None]	List of face positions (cx, cy, w, h) for each frame, or None where
list[tuple[float, float, float, float] | None]	no face was detected and no fallback was available.

Source code in src/videopython/ai/understanding/faces.py

def track_video(
    self,
    frames: np.ndarray,
) -> list[tuple[float, float, float, float] | None]:
    """Track the face through a whole clip via batched per-frame detection.

    Detection runs on every frame (the YuNet detector is CPU-only), then each
    frame's selected face is EMA-smoothed.

    Args:
        frames: Video frames array of shape (N, H, W, 3).

    Returns:
        List of face positions (cx, cy, w, h) for each frame, or None where
        no face was detected and no fallback was available.
    """
    if self._detector is None:
        self._init_detector()
        assert self._detector is not None

    n_frames = len(frames)
    if n_frames == 0:
        return []

    h, w = frames[0].shape[:2]

    detections: list[list[DetectedFace]] = []
    for batch_start in range(0, n_frames, self.batch_size):
        batch = [frames[i] for i in range(batch_start, min(batch_start + self.batch_size, n_frames))]
        detections.extend(self._detector.detect_batch(batch))

    faces = [self._select_face(frame_faces, w, h) for frame_faces in detections]
    self.reset()
    return [self._smooth(face_info) for face_info in faces]

ObjectDetector

Runs a D-FINE COCO model and returns DetectedObject per frame, with normalized bounding boxes sorted by confidence. Weights (Apache-2.0) download from HuggingFace on first use; class names come from the model config.

class_filter accepts D-FINE's VOC-style names and their standard COCO equivalents (for example, motorbike or motorcycle, and tvmonitor or tv). Names are normalized for case and spacing; unknown names are logged after the model loads.

from videopython.ai import ObjectDetector

detector = ObjectDetector(model_name="ustc-community/dfine-nano-coco",
                          class_filter=("person", "car"))

for obj in detector.detect(video.frames[0]):
    print(f"{obj.label} {obj.confidence:.2f} @ {obj.bounding_box}")

per_frame = detector.detect_batch(video.frames[:16])

ObjectDetector

Bases: DetectorBase[DetectedObject]

Lazy D-FINE COCO object detector returning normalized detections.

The D-FINE weights (default ustc-community/dfine-nano-coco) download from HuggingFace on first real use; class names come from the model config. Detection is gated by confidence_threshold and optionally restricted to class_filter, which accepts either D-FINE's VOC-style spellings or the standard COCO ones (motorcycle, airplane, couch, potted plant, dining table, tv) -- see :func:normalize_class_names.

Source code in src/videopython/ai/understanding/objects.py

class ObjectDetector(DetectorBase[DetectedObject]):
    """Lazy D-FINE COCO object detector returning normalized detections.

    The D-FINE weights (default ``ustc-community/dfine-nano-coco``) download from
    HuggingFace on first real use; class names come from the model config.
    Detection is gated by ``confidence_threshold`` and optionally restricted to
    ``class_filter``, which accepts either D-FINE's VOC-style spellings or the
    standard COCO ones (``motorcycle``, ``airplane``, ``couch``, ``potted plant``,
    ``dining table``, ``tv``) -- see :func:`normalize_class_names`.
    """

    DEFAULT_CONFIDENCE_THRESHOLD = 0.5
    _FEATURE = "ObjectDetector"
    # Override the base sentinel/unload set: hold the model AND the processor.
    _model_attrs = ("_model", "_processor")

    def __init__(
        self,
        model_name: str = DEFAULT_MODEL,
        confidence_threshold: float = DEFAULT_CONFIDENCE_THRESHOLD,
        class_filter: tuple[str, ...] = (),
        backend: Backend = "auto",
    ):
        """Initialize the detector.

        Args:
            model_name: D-FINE COCO HuggingFace repo id (e.g.
                ``ustc-community/dfine-nano-coco``, ``...-small-coco``,
                ``...-medium-coco``, ``...-large-coco``). Downloaded on first use.
            confidence_threshold: Minimum detection confidence in ``[0, 1]``.
            class_filter: If non-empty, only these COCO class names are kept.
                Either spelling works -- ``motorcycle`` and ``motorbike`` both
                match -- and names are normalized via
                :func:`normalize_class_names`. A name the model never emits is
                logged as a warning once the model loads, since it would
                otherwise just silently match nothing.
            backend: Detection device - ``"cpu"``, ``"gpu"``, or ``"auto"``.
        """
        super().__init__(backend=backend)
        self.model_name = model_name
        self.confidence_threshold = confidence_threshold
        self.class_filter = normalize_class_names(class_filter)
        self._model: Any = None
        self._processor: Any = None
        self._class_names: dict[int, str] = {}
        logger.info("ObjectDetector initialized with model=%s backend=%s", model_name, backend)

    def _load_model(self) -> None:
        from videopython.ai._optional import require

        tf = require("transformers", feature=self._FEATURE)
        revision = pinned(self.model_name)
        self._processor = tf.AutoImageProcessor.from_pretrained(self.model_name, revision=revision, use_fast=True)
        model = tf.DFineForObjectDetection.from_pretrained(self.model_name, revision=revision)
        model.eval()
        if self._resolve_device() == "cuda":
            model = model.to("cuda")
        self._model = model
        self._class_names = {int(k): v for k, v in model.config.id2label.items()}
        self._warn_unknown_filter_classes()

    def unknown_filter_classes(self) -> tuple[str, ...]:
        """``class_filter`` names this model never emits (empty until it loads)."""
        if not self._class_names:
            return ()
        known = set(self._class_names.values())
        return tuple(name for name in self.class_filter if name not in known)

    def _warn_unknown_filter_classes(self) -> None:
        """Log filter names that cannot match, which would otherwise draw nothing."""
        unknown = self.unknown_filter_classes()
        if unknown:
            logger.warning(
                "class_filter names not emitted by %s: %s. Nothing will be detected for them; "
                "the model's classes are %s.",
                self.model_name,
                ", ".join(unknown),
                ", ".join(sorted(set(self._class_names.values()))),
            )

    def _infer(self, images: list[np.ndarray]) -> list[list[DetectedObject]]:
        import torch

        device = self._resolve_device()
        inputs = self._processor(images=images, return_tensors="pt")
        if device == "cuda":
            inputs = inputs.to("cuda")
        with torch.no_grad():
            outputs = self._model(**inputs)
        # target_sizes is (height, width) per image; D-FINE letterboxes internally
        # so post-processing needs the original sizes to de-letterbox the boxes.
        target_sizes = torch.tensor([[img.shape[0], img.shape[1]] for img in images], device=device)
        results = self._processor.post_process_object_detection(
            outputs, target_sizes=target_sizes, threshold=self.confidence_threshold
        )
        return [self._parse(result, img.shape[1], img.shape[0]) for result, img in zip(results, images)]

    def _parse(self, result: dict[str, Any], img_w: int, img_h: int) -> list[DetectedObject]:
        detected: list[DetectedObject] = []
        scores = result["scores"].tolist()
        labels = result["labels"].tolist()
        boxes = result["boxes"].tolist()
        for score, label_id, (x1, y1, x2, y2) in zip(scores, labels, boxes):
            label = self._class_names.get(int(label_id), str(int(label_id)))
            if self.class_filter and label not in self.class_filter:
                continue
            # D-FINE boxes can sit slightly outside the frame; clamp before normalizing.
            x1 = min(max(x1, 0.0), img_w)
            x2 = min(max(x2, 0.0), img_w)
            y1 = min(max(y1, 0.0), img_h)
            y2 = min(max(y2, 0.0), img_h)
            detected.append(
                DetectedObject(
                    label=label,
                    confidence=float(score),
                    bounding_box=BoundingBox(
                        x=x1 / img_w,
                        y=y1 / img_h,
                        width=(x2 - x1) / img_w,
                        height=(y2 - y1) / img_h,
                    ),
                )
            )
        detected.sort(key=lambda d: d.confidence, reverse=True)
        return detected

__init__

__init__(
    model_name: str = DEFAULT_MODEL,
    confidence_threshold: float = DEFAULT_CONFIDENCE_THRESHOLD,
    class_filter: tuple[str, ...] = (),
    backend: Backend = "auto",
)

Initialize the detector.

Parameters:

Name	Type	Description	Default
model_name	str	D-FINE COCO HuggingFace repo id (e.g. ustc-community/dfine-nano-coco, ...-small-coco, ...-medium-coco, ...-large-coco). Downloaded on first use.	DEFAULT_MODEL
confidence_threshold	float	Minimum detection confidence in [0, 1].	DEFAULT_CONFIDENCE_THRESHOLD
class_filter	tuple[str, ...]	If non-empty, only these COCO class names are kept. Either spelling works -- motorcycle and motorbike both match -- and names are normalized via :func:normalize_class_names. A name the model never emits is logged as a warning once the model loads, since it would otherwise just silently match nothing.	()
backend	Backend	Detection device - "cpu", "gpu", or "auto".	'auto'

Source code in src/videopython/ai/understanding/objects.py

def __init__(
    self,
    model_name: str = DEFAULT_MODEL,
    confidence_threshold: float = DEFAULT_CONFIDENCE_THRESHOLD,
    class_filter: tuple[str, ...] = (),
    backend: Backend = "auto",
):
    """Initialize the detector.

    Args:
        model_name: D-FINE COCO HuggingFace repo id (e.g.
            ``ustc-community/dfine-nano-coco``, ``...-small-coco``,
            ``...-medium-coco``, ``...-large-coco``). Downloaded on first use.
        confidence_threshold: Minimum detection confidence in ``[0, 1]``.
        class_filter: If non-empty, only these COCO class names are kept.
            Either spelling works -- ``motorcycle`` and ``motorbike`` both
            match -- and names are normalized via
            :func:`normalize_class_names`. A name the model never emits is
            logged as a warning once the model loads, since it would
            otherwise just silently match nothing.
        backend: Detection device - ``"cpu"``, ``"gpu"``, or ``"auto"``.
    """
    super().__init__(backend=backend)
    self.model_name = model_name
    self.confidence_threshold = confidence_threshold
    self.class_filter = normalize_class_names(class_filter)
    self._model: Any = None
    self._processor: Any = None
    self._class_names: dict[int, str] = {}
    logger.info("ObjectDetector initialized with model=%s backend=%s", model_name, backend)

unknown_filter_classes

unknown_filter_classes() -> tuple[str, ...]

class_filter names this model never emits (empty until it loads).

Source code in src/videopython/ai/understanding/objects.py

def unknown_filter_classes(self) -> tuple[str, ...]:
    """``class_filter`` names this model never emits (empty until it loads)."""
    if not self._class_names:
        return ()
    known = set(self._class_names.values())
    return tuple(name for name in self.class_filter if name not in known)

Result types

Shared, AI-free Pydantic models from videopython.base, produced by the analyzers above and consumed by videopython.editing.

SceneBoundary

Bases: _ResultModel

Timing information for a detected scene.

A lightweight structure representing scene boundaries returned by scene detectors (e.g. videopython.ai.SemanticSceneDetector). This is a backbone type — higher-level scene analysis lives in orchestration packages.

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)

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

class SceneBoundary(_ResultModel):
    """Timing information for a detected scene.

    A lightweight structure representing scene boundaries returned by
    scene detectors (e.g. ``videopython.ai.SemanticSceneDetector``). This
    is a backbone type — higher-level scene analysis lives in orchestration
    packages.

    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)
    """

    start: float
    end: float
    start_frame: int
    end_frame: int

    @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

duration property

duration: float

Duration of the scene in seconds.

frame_count property

frame_count: int

Number of frames in this scene.

SceneDescription

Bases: _ResultModel

Structured visual scene description from the SceneVLM.

The v1 schema is intentionally narrow (caption + subjects + shot_type). Wider schemas drop JSON parse rate on small models without eval data to defend the cost. Fields are added in v2 as parse-rate measurements justify them; closed enums first, open lists last.

Attributes:

Name	Type	Description
caption	str	One-sentence summary of the scene.
subjects	list[str]	Open list of named subjects visible in the frames.
shot_type	str | None	Closed enum framing the camera distance, or None when JSON parsing fell back to raw text.

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

class SceneDescription(_ResultModel):
    """Structured visual scene description from the SceneVLM.

    The v1 schema is intentionally narrow (caption + subjects + shot_type).
    Wider schemas drop JSON parse rate on small models without eval data
    to defend the cost. Fields are added in v2 as parse-rate measurements
    justify them; closed enums first, open lists last.

    Attributes:
        caption: One-sentence summary of the scene.
        subjects: Open list of named subjects visible in the frames.
        shot_type: Closed enum framing the camera distance, or None
            when JSON parsing fell back to raw text.
    """

    caption: str
    subjects: list[str] = Field(default_factory=list)
    shot_type: str | None = None

BoundingBox

Bases: _ResultModel

A bounding box for detected objects or crop regions in an image.

Coordinates are normalized to [0, 1] relative to image dimensions. It can be embedded directly into Operation fields (for example, KenBurns.start_region) and validated as part of an operation's JSON wire format.

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

class BoundingBox(_ResultModel):
    """A bounding box for detected objects or crop regions in an image.

    Coordinates are normalized to ``[0, 1]`` relative to image dimensions.
    It can be embedded directly into ``Operation`` fields (for example,
    ``KenBurns.start_region``) and validated as part of an operation's JSON
    wire format.
    """

    model_config = ConfigDict(extra="forbid", frozen=True)

    x: float = Field(description="Left edge of the box, 0=left of the image.")
    y: float = Field(description="Top edge of the box, 0=top of the image.")
    width: float = Field(description="Width of the box, normalized to image width.")
    height: float = Field(description="Height of the box, normalized to image height.")

    @property
    def center(self) -> tuple[float, float]:
        """Center point of the bounding box."""
        return (self.x + self.width / 2, self.y + self.height / 2)

    @property
    def area(self) -> float:
        """Area of the bounding box (normalized)."""
        return self.width * self.height

center property

center: tuple[float, float]

Center point of the bounding box.

area property

area: float

Area of the bounding box (normalized).

DetectedObject

Bases: _ResultModel

An object detected in a video frame.

Attributes:

Name	Type	Description
label	str	Name/class of the detected object (e.g., "person", "car", "dog")
confidence	float	Detection confidence score between 0 and 1
bounding_box	BoundingBox | None	Optional bounding box location of the object

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

class DetectedObject(_ResultModel):
    """An object detected in a video frame.

    Attributes:
        label: Name/class of the detected object (e.g., "person", "car", "dog")
        confidence: Detection confidence score between 0 and 1
        bounding_box: Optional bounding box location of the object
    """

    label: str
    confidence: float
    bounding_box: BoundingBox | None = None

DetectedFace

Bases: _ResultModel

A face detected in a video frame.

Attributes:

Name	Type	Description
bounding_box	BoundingBox | None	Bounding box location of the face (normalized 0-1 coordinates). May be None for cloud backends that only return face counts.
confidence	float	Detection confidence score between 0 and 1

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

class DetectedFace(_ResultModel):
    """A face detected in a video frame.

    Attributes:
        bounding_box: Bounding box location of the face (normalized 0-1 coordinates).
            May be None for cloud backends that only return face counts.
        confidence: Detection confidence score between 0 and 1
    """

    bounding_box: BoundingBox | None = None
    confidence: float = 1.0

    @property
    def center(self) -> tuple[float, float] | None:
        """Center point of the face bounding box, or None if no bounding box."""
        return self.bounding_box.center if self.bounding_box else None

    @property
    def area(self) -> float | None:
        """Area of the face bounding box (normalized), or None if no bounding box."""
        return self.bounding_box.area if self.bounding_box else None

center property

center: tuple[float, float] | None

Center point of the face bounding box, or None if no bounding box.

area property

area: float | None

Area of the face bounding box (normalized), or None if no bounding box.

DetectedText

Bases: _ResultModel

Text detected in a video frame.

Attributes:

Name	Type	Description
text	str	OCR text content
confidence	float	Detection confidence score between 0 and 1
bounding_box	BoundingBox | None	Optional normalized bounding box for the text region

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

class DetectedText(_ResultModel):
    """Text detected in a video frame.

    Attributes:
        text: OCR text content
        confidence: Detection confidence score between 0 and 1
        bounding_box: Optional normalized bounding box for the text region
    """

    text: str
    confidence: float
    bounding_box: BoundingBox | None = None

FaceTrack

Bases: _ResultModel

A face tracked across consecutive frames within a single shot.

Tracks are produced by IoU association — no embedding re-id, so a track does not survive across shot/scene boundaries. frame_indices and boxes are parallel lists of equal length.

Attributes:

Name	Type	Description
track_id	int	Stable id within the shot the track was produced in. Not globally unique across scenes.
frame_indices	list[int]	Source-video frame indices for each detection.
boxes	list[BoundingBox]	Per-frame bounding boxes (normalized 0-1 coords).
confidences	list[float]	Per-frame detection confidence in [0, 1].

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

class FaceTrack(_ResultModel):
    """A face tracked across consecutive frames within a single shot.

    Tracks are produced by IoU association — no embedding re-id, so a
    track does not survive across shot/scene boundaries. ``frame_indices``
    and ``boxes`` are parallel lists of equal length.

    Attributes:
        track_id: Stable id within the shot the track was produced in.
            Not globally unique across scenes.
        frame_indices: Source-video frame indices for each detection.
        boxes: Per-frame bounding boxes (normalized 0-1 coords).
        confidences: Per-frame detection confidence in [0, 1].
    """

    track_id: int
    frame_indices: list[int]
    boxes: list[BoundingBox]
    confidences: list[float] = Field(default_factory=list)

    @property
    def length(self) -> int:
        """Number of frames in this track."""
        return len(self.frame_indices)

length property

length: int

Number of frames in this track.

MotionInfo

Bases: _ResultModel

Motion characteristics between consecutive frames.

Attributes:

Name	Type	Description
motion_type	str	Classification of camera/scene motion - "static": No significant motion - "pan": Horizontal camera movement - "tilt": Vertical camera movement - "zoom": Camera zoom in/out - "complex": Mixed or irregular motion
magnitude	float	Normalized motion magnitude (0.0 = no motion, 1.0 = high motion)
raw_magnitude	float	Raw optical flow magnitude (pixels/frame)

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

class MotionInfo(_ResultModel):
    """Motion characteristics between consecutive frames.

    Attributes:
        motion_type: Classification of camera/scene motion
            - "static": No significant motion
            - "pan": Horizontal camera movement
            - "tilt": Vertical camera movement
            - "zoom": Camera zoom in/out
            - "complex": Mixed or irregular motion
        magnitude: Normalized motion magnitude (0.0 = no motion, 1.0 = high motion)
        raw_magnitude: Raw optical flow magnitude (pixels/frame)
    """

    motion_type: str
    magnitude: float
    raw_magnitude: float

    @property
    def is_static(self) -> bool:
        """Check if this frame has no significant motion."""
        return self.motion_type == "static"

    @property
    def is_dynamic(self) -> bool:
        """Check if this frame has significant motion."""
        return self.motion_type != "static"

is_static property

is_static: bool

Check if this frame has no significant motion.

is_dynamic property

is_dynamic: bool

Check if this frame has significant motion.

AudioEvent

Bases: _ResultModel

A detected audio event with timestamp.

Attributes:

Name	Type	Description
start	float	Start time in seconds
end	float	End time in seconds
label	str	Name of the detected sound (e.g., "Music", "Speech", "Dog bark")
confidence	float	Detection confidence score between 0 and 1

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

class AudioEvent(_ResultModel):
    """A detected audio event with timestamp.

    Attributes:
        start: Start time in seconds
        end: End time in seconds
        label: Name of the detected sound (e.g., "Music", "Speech", "Dog bark")
        confidence: Detection confidence score between 0 and 1
    """

    start: float
    end: float
    label: str
    confidence: float

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

duration property

duration: float

Duration of the audio event in seconds.

AudioClassification

Bases: _ResultModel

Complete audio classification results.

Attributes:

Name	Type	Description
events	list[AudioEvent]	List of detected audio events with timestamps
clip_predictions	dict[str, float]	Overall class probabilities for the entire audio clip

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

class AudioClassification(_ResultModel):
    """Complete audio classification results.

    Attributes:
        events: List of detected audio events with timestamps
        clip_predictions: Overall class probabilities for the entire audio clip
    """

    events: list[AudioEvent]
    clip_predictions: dict[str, float] = Field(default_factory=dict)