Video analysis

VideoAnalyzer runs the global passes (transcription + scene detection), then per detected scene runs the scene VLM, the audio classifier, and the per-shot face tracker. The result is one serializable, scene-first VideoAnalysis.

from videopython.ai import VideoAnalysis, VideoAnalyzer

analysis = VideoAnalyzer().analyze_path("video.mp4")

print(analysis.source.title)
for outcome in analysis.run_info.analyzer_outcomes:
    print(outcome.analyzer, outcome.status, outcome.reason)
if analysis.scenes:
    sample = analysis.scenes.samples[0]
    if sample.scene_description:
        print(sample.scene_description.caption, sample.scene_description.shot_type)
    for track in (sample.faces or []):
        print(f"track #{track.track_id}: {track.length} frames")

analysis.save("video_analysis.json")
loaded = VideoAnalysis.load("video_analysis.json")

VideoAnalysis and its nested result types are Pydantic models, so model_dump(), model_dump_json(), model_validate() and model_validate_json() work throughout the result tree. save() / load() wrap the JSON pair with UTF-8 and parent-directory creation. Use loaded.verify_source() to compare the recorded source digest with the current file; load() alone does not read the media.

Saved identity and migration

Every result requires an AnalysisProvenance object at analysis.provenance:

Field	Contract
format_version	Required integer 1; other versions are rejected.
source_sha256	SHA256 of the source file, or null for unbound in-memory input.
sampling	The low, medium, or high preset used for the run.
models	Analyzer ID to a model-ID/revision map, or null when provenance is unknown. A model revision can also be null.

Model identities are recorded during analysis. Hugging Face models use repository revisions (AST records its requested pin, or null for an unpinned model); Ollama uses the server's resolved tag and digest. Bundled Silero and TransNetV2 weights use a package:<version> revision, which identifies the package release rather than a weight-file hash. A disabled stage, an early load failure, or an unavailable identity can leave unknown provenance. Check stage outcomes separately: known identity is not a successful-analysis flag.

analyze_path() records a resolved absolute source path and hashes the file with bounded reads. analyze(video, ...) leaves source_sha256=null, even if a source-path label was supplied, because the in-memory frames are not verified against that file. Those unbound results can be serialized but cannot be imported into MCP.

verify_source() requires a recorded file identity, compares its digest, and returns the resolved source path. It starts no models and does not replace saved settings or unknown provenance with values from the current environment. MCP import/export performs this check once per call. Source files must remain unchanged while a cached analysis is in use.

Migration: regenerate older saved analyses with VideoAnalyzer.analyze_path(). Files without provenance are rejected; there is no legacy loader. Do not add the currently installed models as if they had produced an old result. See reuse across MCP sessions.

AnalysisProvenance

Bases: BaseModel

Source identity and model revisions recorded during analysis.

Source code in src/videopython/ai/video_analysis/models.py

class AnalysisProvenance(BaseModel):
    """Source identity and model revisions recorded during analysis."""

    model_config = ConfigDict(extra="forbid")

    format_version: Literal[1]
    source_sha256: str | None = Field(pattern=r"^[0-9a-f]{64}$")
    sampling: Literal["low", "medium", "high"]
    models: dict[str, dict[str, str | None] | None]

Configuration

from videopython.ai import VideoAnalysisConfig, VideoAnalyzer

config = VideoAnalysisConfig(
    enabled_analyzers={"audio_to_text", "semantic_scene_detector", "scene_vlm", "face_tracker"},
    analyzer_params={
        "scene_vlm": {"model": "qwen3.6:27b"},
        "audio_to_text": {"model_name": "large", "vocabulary": ["Klarna", "Allegro"]},
    },
)
analysis = VideoAnalyzer(config=config, sampling="medium").analyze_path("video.mp4")

VideoAnalysisConfig.for_profile("full") enables every analyzer (audio_to_text, audio_classifier, semantic_scene_detector, scene_vlm, face_tracker) and is equivalent to a bare VideoAnalysisConfig().

Sampling presets

sampling sizes the per-scene SceneVLM frame budget: the frame cap, the log-curve scale/base used for short scenes, and the threshold below which adjacent short scenes are merged into one VLM call.

sampling	Per-scene frame cap	Adjacent-merge threshold	Typical use
"low"	8	20 s	Quick previews, long videos
"medium" (default)	30	10 s	Balanced
"high"	60	4 s	Rich analysis, talking-head depth

sampling and the VLM model are orthogonal: one sizes the frame budget, the other picks the captioning model.

Output shape

• analysis.audio.transcription — the full Whisper transcription.
• analysis.scenes.samples — one SceneAnalysisSample per scene, each carrying:
  • scene timing (start_second, end_second, start_frame, end_frame);
  • scene_description: SceneDescription | None — caption, subjects, shot_type. None when the VLM was disabled or its forward pass failed;
  • audio_classification: AudioClassification | None — events and clip-level predictions for the scene window;
  • faces: list[FaceTrack] | None — per-shot IoU-associated tracks, each with its own frame indices and boxes.
• analysis.run_info.stage_durations_seconds — wall-clock per stage (whisper, scene_detection, scene_vlm, face_tracker, audio_classification, plus whisper_and_scene_detection_parallel when those two run together).
• analysis.run_info.analyzer_outcomes — one record for every analyzer. status is completed, skipped, or failed. A skipped analyzer has reason disabled; a failed analyzer has reason initialization_failed or execution_failed.

Classes

VideoAnalysisConfig

Bases: BaseModel

Execution config for scene-first analysis runs.

analyzer_params lets you forward keyword arguments to each predictor constructor keyed by analyzer id. For example::

VideoAnalysisConfig(
    analyzer_params={
        "audio_to_text": {"model_name": "large"},
        "scene_vlm": {"model": "qwen3.6:27b"},
    }
)

Source code in src/videopython/ai/video_analysis/models.py

class VideoAnalysisConfig(BaseModel):
    """Execution config for scene-first analysis runs.

    ``analyzer_params`` lets you forward keyword arguments to each predictor
    constructor keyed by analyzer id.  For example::

        VideoAnalysisConfig(
            analyzer_params={
                "audio_to_text": {"model_name": "large"},
                "scene_vlm": {"model": "qwen3.6:27b"},
            }
        )
    """

    enabled_analyzers: set[str] = Field(default_factory=lambda: {str(analyzer) for analyzer in ALL_ANALYZER_IDS})
    analyzer_params: dict[str, dict[str, Any]] = Field(default_factory=dict)

    @model_validator(mode="after")
    def _reject_unknown_analyzer_ids(self) -> VideoAnalysisConfig:
        unknown_enabled = sorted(set(self.enabled_analyzers) - set(ALL_ANALYZER_IDS))
        if unknown_enabled:
            raise ValueError(f"Unknown analyzer ids in enabled_analyzers: {unknown_enabled}")
        unknown_params = sorted(set(self.analyzer_params) - set(ALL_ANALYZER_IDS))
        if unknown_params:
            raise ValueError(f"Unknown analyzer ids in analyzer_params: {unknown_params}")
        return self

    def get_params(self, analyzer_id: str) -> dict[str, Any]:
        """Return kwargs dict for the given analyzer, defaulting to empty."""
        return dict(self.analyzer_params.get(analyzer_id, {}))

    @classmethod
    def for_profile(cls, profile: str, *, faces: bool = True) -> VideoAnalysisConfig:
        """Config for an analysis profile: 'full' (all analyzers) or 'editing' (catalog-only, no audio classifier)."""
        if profile == "full":
            return cls()
        if profile == "editing":
            enabled = {SEMANTIC_SCENE_DETECTOR, SCENE_VLM, AUDIO_TO_TEXT}
            if faces:
                enabled.add(FACE_TRACKER)
            return cls(enabled_analyzers=enabled)
        raise ValueError(f"Unknown profile: {profile!r} (expected 'full' or 'editing')")

get_params

get_params(analyzer_id: str) -> dict[str, Any]

Return kwargs dict for the given analyzer, defaulting to empty.

Source code in src/videopython/ai/video_analysis/models.py

def get_params(self, analyzer_id: str) -> dict[str, Any]:
    """Return kwargs dict for the given analyzer, defaulting to empty."""
    return dict(self.analyzer_params.get(analyzer_id, {}))

for_profile classmethod

for_profile(
    profile: str, *, faces: bool = True
) -> VideoAnalysisConfig

Config for an analysis profile: 'full' (all analyzers) or 'editing' (catalog-only, no audio classifier).

Source code in src/videopython/ai/video_analysis/models.py

@classmethod
def for_profile(cls, profile: str, *, faces: bool = True) -> VideoAnalysisConfig:
    """Config for an analysis profile: 'full' (all analyzers) or 'editing' (catalog-only, no audio classifier)."""
    if profile == "full":
        return cls()
    if profile == "editing":
        enabled = {SEMANTIC_SCENE_DETECTOR, SCENE_VLM, AUDIO_TO_TEXT}
        if faces:
            enabled.add(FACE_TRACKER)
        return cls(enabled_analyzers=enabled)
    raise ValueError(f"Unknown profile: {profile!r} (expected 'full' or 'editing')")

VideoAnalyzer

Orchestrates scene-first analyzers and builds VideoAnalysis output.

sampling controls how aggressively the SceneVLM samples frames per scene. low is a fast preview pass for long videos, high keeps talking-head depth, medium is the previous default. The preset tunes the per-scene frame cap, the log-curve scale/base used to size short scenes, and the threshold below which adjacent scenes get merged into one VLM call.

sampling and the SceneVLM tier are orthogonal: small models can't make use of dense sampling, but the user owns that tradeoff.

Source code in src/videopython/ai/video_analysis/analyzer.py

class VideoAnalyzer:
    """Orchestrates scene-first analyzers and builds `VideoAnalysis` output.

    ``sampling`` controls how aggressively the SceneVLM samples frames per
    scene. ``low`` is a fast preview pass for long videos, ``high`` keeps
    talking-head depth, ``medium`` is the previous default. The preset
    tunes the per-scene frame cap, the log-curve scale/base used to size
    short scenes, and the threshold below which adjacent scenes get
    merged into one VLM call.

    ``sampling`` and the SceneVLM ``tier`` are orthogonal: small models
    can't make use of dense sampling, but the user owns that tradeoff.
    """

    def __init__(
        self,
        config: VideoAnalysisConfig | None = None,
        *,
        sampling: SamplingPreset = DEFAULT_SAMPLING_PRESET,
    ):
        if sampling not in SAMPLING_PRESETS:
            supported = ", ".join(SAMPLING_PRESETS)
            raise ValueError(f"sampling must be one of: {supported}")
        self.config = config or VideoAnalysisConfig()
        self.sampling: SamplingPreset = sampling
        self._sampling_profile = SAMPLING_PRESETS[sampling]

    def analyze_path(self, path: str | Path) -> VideoAnalysis:
        """Analyze a video path in scene-first mode."""
        path_obj = Path(path).resolve()
        metadata = VideoMetadata.from_path(path_obj)
        source = self._build_source(
            metadata=metadata,
            path_obj=path_obj,
            duration_seconds=metadata.total_seconds,
            title_fallback=path_obj.stem,
        )
        return self._analyze(video=None, source_path=path_obj, metadata=metadata, source=source)

    def analyze(self, video: Video, *, source_path: str | Path | None = None) -> VideoAnalysis:
        """Analyze an in-memory `Video` object."""
        path_obj = Path(source_path) if source_path else None
        metadata = VideoMetadata.from_video(video)
        source = self._build_source(
            metadata=metadata,
            path_obj=path_obj,
            duration_seconds=video.total_seconds,
            title_fallback=path_obj.stem if path_obj is not None else None,
        )
        return self._analyze(
            video=video,
            source_path=path_obj,
            metadata=metadata,
            source=source,
        )

    def _analyze(
        self,
        *,
        video: Video | None,
        source_path: Path | None,
        metadata: VideoMetadata,
        source: VideoAnalysisSource,
    ) -> VideoAnalysis:
        mode = "path" if source_path is not None else "video"
        if source_path is None and video is None:
            raise ValueError("Either `source_path` or `video` must be provided")

        t_analysis_start = time.perf_counter()
        provenance = AnalysisProvenance(
            format_version=1,
            source_sha256=source_digest(source_path) if source_path is not None and video is None else None,
            sampling=self.sampling,
            models={analyzer: None for analyzer in ALL_ANALYZER_IDS},
        )
        enabled = self.config.enabled_analyzers
        analyzer_outcomes: dict[str, AnalyzerOutcome] = {
            analyzer: AnalyzerOutcome(
                analyzer=analyzer,
                status="completed" if analyzer in enabled else "skipped",
                reason=None if analyzer in enabled else "disabled",
            )
            for analyzer in ALL_ANALYZER_IDS
        }

        run_info = AnalysisRunInfo(
            created_at=detectors.utc_now_iso(),
            mode=mode,
            library_version=detectors.library_version(),
            analyzer_outcomes=list(analyzer_outcomes.values()),
        )

        run_whisper = AUDIO_TO_TEXT in enabled
        run_scene_det = SEMANTIC_SCENE_DETECTOR in enabled

        transcription = None
        detected: list[SceneBoundary] | None = None

        # SceneVLM is loaded *after* Whisper/TransNetV2 finish (not concurrently)
        # because transformers' from_pretrained(torch_dtype="auto") mutates the
        # process-global torch.get_default_dtype() during model construction,
        # which corrupts Whisper's model weights if they're initialized at the
        # same time.
        if run_whisper and run_scene_det:
            transcription, detected = detectors.run_whisper_and_scene_detection(
                config=self.config, source_path=source_path, video=video, run_info=run_info, provenance=provenance
            )
        else:
            if run_whisper:
                with detectors.record_stage(run_info, "whisper"):
                    transcription = detectors.run_whisper(
                        config=self.config, source_path=source_path, video=video, provenance=provenance
                    )

            if run_scene_det:
                with detectors.record_stage(run_info, "scene_detection"):
                    detected = detectors.run_scene_detection(
                        config=self.config, source_path=source_path, video=video, provenance=provenance
                    )

        if run_whisper and transcription is None:
            analyzer_outcomes[AUDIO_TO_TEXT] = AnalyzerOutcome(
                analyzer=AUDIO_TO_TEXT,
                status="failed",
                reason="execution_failed",
            )
        if run_scene_det and detected is None:
            analyzer_outcomes[SEMANTIC_SCENE_DETECTOR] = AnalyzerOutcome(
                analyzer=SEMANTIC_SCENE_DETECTOR,
                status="failed",
                reason="execution_failed",
            )

        if run_scene_det:
            detectors.reset_transnetv2_torch_state()

        # Whisper and TransNetV2 are done -- free their GPU memory before
        # loading SceneVLM (~9GB). Python GC doesn't guarantee immediate
        # cleanup, so force it and release the CUDA cache.
        if run_whisper or run_scene_det:
            gc.collect()
            detectors.release_gpu_cache()

        scenes = self._default_scene_boundaries(metadata)
        if detected is not None:
            scenes = self._normalize_scene_boundaries(detected, metadata)

        if not scenes:
            scenes = self._default_scene_boundaries(metadata)

        with detectors.record_stage(run_info, "scene_analysis"):
            scene_section = self._analyze_scenes(
                source_path=source_path,
                video=video,
                metadata=metadata,
                scenes=scenes,
                run_info=run_info,
                analyzer_outcomes=analyzer_outcomes,
                provenance=provenance,
            )

        audio_section = AudioAnalysisSection(transcription=transcription) if transcription is not None else None

        run_info.total_duration_seconds = time.perf_counter() - t_analysis_start
        run_info.analyzer_outcomes = list(analyzer_outcomes.values())
        logger.info("Total analysis completed in %.2fs", run_info.total_duration_seconds)
        return VideoAnalysis(
            source=source,
            provenance=provenance,
            config=self.config,
            run_info=run_info,
            audio=audio_section,
            scenes=scene_section if scene_section.samples else None,
        )

    def _analyze_scenes(
        self,
        *,
        source_path: Path | None,
        video: Video | None,
        metadata: VideoMetadata,
        scenes: list[SceneBoundary],
        run_info: AnalysisRunInfo,
        analyzer_outcomes: dict[str, AnalyzerOutcome],
        provenance: AnalysisProvenance,
    ) -> SceneAnalysisSection:
        enabled = self.config.enabled_analyzers

        # A missing extra or model-load failure does not abort the other analyzers.
        scene_vlm = (
            source_metadata.try_init(lambda: SceneVLM(**self.config.get_params(SCENE_VLM)), "SceneVLM")
            if SCENE_VLM in enabled
            else None
        )
        audio_classifier = (
            source_metadata.try_init(
                lambda: AudioClassifier(**self.config.get_params(AUDIO_CLASSIFIER)), "AudioClassifier"
            )
            if AUDIO_CLASSIFIER in enabled
            else None
        )
        face_tracker = (
            source_metadata.try_init(lambda: FaceShotTracker(**self.config.get_params(FACE_TRACKER)), "FaceShotTracker")
            if FACE_TRACKER in enabled
            else None
        )

        for analyzer, component in (
            (SCENE_VLM, scene_vlm),
            (AUDIO_CLASSIFIER, audio_classifier),
            (FACE_TRACKER, face_tracker),
        ):
            if analyzer in enabled and component is None:
                analyzer_outcomes[analyzer] = AnalyzerOutcome(
                    analyzer=analyzer,
                    status="failed",
                    reason="initialization_failed",
                )

        path_audio: Audio | None = None
        if audio_classifier is not None and source_path is not None:
            try:
                path_audio = Audio.from_path(source_path)
            except (OSError, RuntimeError, ValueError):
                logger.warning(
                    "Failed to load audio from path, audio classification will use clip fallback",
                    exc_info=True,
                )
                path_audio = None

        descriptions: list[SceneDescription | None] = [None] * len(scenes)
        if scene_vlm is not None:
            with detectors.record_stage(run_info, "scene_vlm"):
                try:
                    descriptions = detectors.run_scene_vlm_batched(
                        scene_vlm=scene_vlm,
                        profile=self._sampling_profile,
                        sampling=self.sampling,
                        source_path=source_path,
                        video=video,
                        metadata=metadata,
                        scenes=scenes,
                    )
                except (IndexError, OSError, RuntimeError, ValueError):
                    logger.warning("Batched SceneVLM failed, skipping visual understanding", exc_info=True)
            if any(description is None for description in descriptions):
                analyzer_outcomes[SCENE_VLM] = AnalyzerOutcome(
                    analyzer=SCENE_VLM,
                    status="failed",
                    reason="execution_failed",
                )

        samples: list[SceneAnalysisSample] = []
        audio_classifier_failed = False
        face_tracker_failed = False
        audio_ctx = (
            detectors.record_stage(run_info, "audio_classification") if audio_classifier is not None else nullcontext()
        )
        face_ctx = detectors.record_stage(run_info, "face_tracker") if face_tracker is not None else nullcontext()
        with audio_ctx, face_ctx:
            for index, scene in enumerate(scenes):
                sample = SceneAnalysisSample(
                    scene_index=index,
                    start_second=float(scene.start),
                    end_second=float(scene.end),
                    start_frame=int(scene.start_frame),
                    end_frame=int(scene.end_frame),
                    scene_description=descriptions[index],
                )

                if audio_classifier is not None:
                    try:
                        scene_clip: Video | None = None
                        if path_audio is None:
                            try:
                                scene_clip = self._load_scene_video_clip(
                                    source_path=source_path,
                                    video=video,
                                    start_second=scene.start,
                                    end_second=scene.end,
                                )
                            except (OSError, RuntimeError, ValueError):
                                scene_clip = None
                        sample.audio_classification = detectors.run_scene_audio_classification(
                            audio_classifier=audio_classifier,
                            path_audio=path_audio,
                            scene_clip=scene_clip,
                            scene_start=scene.start,
                            scene_end=scene.end,
                        )
                    except (OSError, RuntimeError, ValueError):
                        audio_classifier_failed = True
                        logger.warning(
                            "AudioClassifier failed for scene %d (%.1f-%.1fs)",
                            index,
                            scene.start,
                            scene.end,
                            exc_info=True,
                        )

                if face_tracker is not None:
                    try:
                        sample.faces = detectors.run_scene_face_tracker(
                            face_tracker=face_tracker,
                            source_path=source_path,
                            video=video,
                            metadata=metadata,
                            scene=scene,
                        )
                    except (IndexError, OSError, RuntimeError, ValueError):
                        face_tracker_failed = True
                        logger.warning(
                            "FaceShotTracker failed for scene %d (%.1f-%.1fs)",
                            index,
                            scene.start,
                            scene.end,
                            exc_info=True,
                        )

                samples.append(sample)

        if audio_classifier_failed:
            analyzer_outcomes[AUDIO_CLASSIFIER] = AnalyzerOutcome(
                analyzer=AUDIO_CLASSIFIER,
                status="failed",
                reason="execution_failed",
            )
        if face_tracker_failed:
            analyzer_outcomes[FACE_TRACKER] = AnalyzerOutcome(
                analyzer=FACE_TRACKER,
                status="failed",
                reason="execution_failed",
            )

        for analyzer, component in (
            (SCENE_VLM, scene_vlm),
            (AUDIO_CLASSIFIER, audio_classifier),
            (FACE_TRACKER, face_tracker),
        ):
            if component is not None:
                provenance.models[analyzer] = component.model_provenance()
        return SceneAnalysisSection(samples=samples)

    def _load_scene_video_clip(
        self,
        *,
        source_path: Path | None,
        video: Video | None,
        start_second: float,
        end_second: float,
    ) -> Video | None:
        if end_second <= start_second:
            return None
        if source_path is not None:
            return Video.from_path(str(source_path), start_second=start_second, end_second=end_second)
        v = detectors.require_video(video)
        return v[round(start_second * v.fps) : round(end_second * v.fps)]

    def _default_scene_boundaries(self, metadata: VideoMetadata) -> list[SceneBoundary]:
        if metadata.total_seconds <= 0 or metadata.frame_count <= 0:
            return []
        return [
            SceneBoundary(
                start=0.0,
                end=float(metadata.total_seconds),
                start_frame=0,
                end_frame=int(metadata.frame_count),
            )
        ]

    def _normalize_scene_boundaries(self, scenes: list[SceneBoundary], metadata: VideoMetadata) -> list[SceneBoundary]:
        normalized: list[SceneBoundary] = []
        max_time = float(metadata.total_seconds)
        max_frame = int(metadata.frame_count)

        for item in scenes:
            start = max(0.0, min(max_time, float(item.start)))
            end = max(0.0, min(max_time, float(item.end)))
            if end <= start:
                continue

            start_frame = int(item.start_frame)
            end_frame = int(item.end_frame)
            start_frame = max(0, min(max_frame, start_frame))
            end_frame = max(0, min(max_frame, end_frame))
            if end_frame <= start_frame:
                start_frame = int(round(start * metadata.fps))
                end_frame = max(start_frame + 1, int(round(end * metadata.fps)))
                start_frame = max(0, min(max_frame, start_frame))
                end_frame = max(0, min(max_frame, end_frame))
                if end_frame <= start_frame:
                    continue

            normalized.append(
                SceneBoundary(
                    start=round(start, 6),
                    end=round(end, 6),
                    start_frame=start_frame,
                    end_frame=end_frame,
                )
            )

        normalized.sort(key=lambda scene: (scene.start, scene.end))
        return normalized

    def _build_source(
        self,
        *,
        metadata: VideoMetadata,
        path_obj: Path | None,
        duration_seconds: float,
        title_fallback: str | None,
    ) -> VideoAnalysisSource:
        tags = source_metadata.extract_source_tags(path_obj) if path_obj else {}
        creation_time = source_metadata.creation_time_from_tags(tags)
        geo = source_metadata.parse_geo_metadata(tags)
        title = tags.get("title") or title_fallback

        return VideoAnalysisSource(
            title=title,
            path=str(path_obj) if path_obj else None,
            filename=path_obj.name if path_obj else None,
            duration=duration_seconds,
            fps=metadata.fps,
            width=metadata.width,
            height=metadata.height,
            frame_count=metadata.frame_count,
            creation_time=creation_time,
            geo=geo,
            raw_tags=tags or None,
        )

analyze_path

analyze_path(path: str | Path) -> VideoAnalysis

Analyze a video path in scene-first mode.

Source code in src/videopython/ai/video_analysis/analyzer.py

def analyze_path(self, path: str | Path) -> VideoAnalysis:
    """Analyze a video path in scene-first mode."""
    path_obj = Path(path).resolve()
    metadata = VideoMetadata.from_path(path_obj)
    source = self._build_source(
        metadata=metadata,
        path_obj=path_obj,
        duration_seconds=metadata.total_seconds,
        title_fallback=path_obj.stem,
    )
    return self._analyze(video=None, source_path=path_obj, metadata=metadata, source=source)

analyze

analyze(
    video: Video, *, source_path: str | Path | None = None
) -> VideoAnalysis

Analyze an in-memory Video object.

Source code in src/videopython/ai/video_analysis/analyzer.py

def analyze(self, video: Video, *, source_path: str | Path | None = None) -> VideoAnalysis:
    """Analyze an in-memory `Video` object."""
    path_obj = Path(source_path) if source_path else None
    metadata = VideoMetadata.from_video(video)
    source = self._build_source(
        metadata=metadata,
        path_obj=path_obj,
        duration_seconds=video.total_seconds,
        title_fallback=path_obj.stem if path_obj is not None else None,
    )
    return self._analyze(
        video=video,
        source_path=path_obj,
        metadata=metadata,
        source=source,
    )

VideoAnalysis

Bases: BaseModel

Serializable aggregate scene-first analysis result for one video.

Source code in src/videopython/ai/video_analysis/models.py

class VideoAnalysis(BaseModel):
    """Serializable aggregate scene-first analysis result for one video."""

    source: VideoAnalysisSource
    provenance: AnalysisProvenance
    config: VideoAnalysisConfig
    run_info: AnalysisRunInfo
    audio: AudioAnalysisSection | None = None
    scenes: SceneAnalysisSection | None = None

    def verify_source(self) -> Path:
        """Check the recorded file digest and return its resolved path without inference."""
        if self.source.path is None or self.provenance.source_sha256 is None:
            raise ValueError("Analysis has no verified file identity; analyze the source path again")
        path = Path(self.source.path).resolve()
        if source_digest(path) != self.provenance.source_sha256:
            raise ValueError("Source content differs from the saved analysis")
        return path

    def save(self, path: str | Path, *, indent: int | None = 2) -> None:
        path_obj = Path(path)
        path_obj.parent.mkdir(parents=True, exist_ok=True)
        path_obj.write_text(self.model_dump_json(indent=indent), encoding="utf-8")

    @classmethod
    def load(cls, path: str | Path) -> VideoAnalysis:
        return cls.model_validate_json(Path(path).read_text(encoding="utf-8"))

verify_source

verify_source() -> Path

Check the recorded file digest and return its resolved path without inference.

Source code in src/videopython/ai/video_analysis/models.py

def verify_source(self) -> Path:
    """Check the recorded file digest and return its resolved path without inference."""
    if self.source.path is None or self.provenance.source_sha256 is None:
        raise ValueError("Analysis has no verified file identity; analyze the source path again")
    path = Path(self.source.path).resolve()
    if source_digest(path) != self.provenance.source_sha256:
        raise ValueError("Source content differs from the saved analysis")
    return path