"""Timeline sidecar manifest (mirrors ``muse.plugins.midi.manifest``).""" from __future__ import annotations import json import logging import pathlib from typing import Literal, TypedDict from muse._version import __version__ from muse.core.types import JsonObject, Manifest, load_json_file from muse.plugins.timeline.entity import IdentifiedProject logger = logging.getLogger(__name__) type _ProjectMap = dict[str, "TimelineProjectEntry"] type _SequenceMap = dict[str, "TimelineSequenceEntry"] type _TrackMap = dict[str, "TimelineTrackEntry"] class TimelineClipEntry(TypedDict): clip_id: str name: str source_media_id: str occurrence: int class TimelineTrackEntry(TypedDict): track_identity: str kind: Literal["video", "audio", "subtitle"] ordinal: int name: str clips: list[TimelineClipEntry] class TimelineSequenceEntry(TypedDict): sequence_identity: str name: str ordinal: int tracks: _TrackMap class TimelineProjectEntry(TypedDict): project_key: str name: str content_id: str sequences: _SequenceMap class TimelineManifest(TypedDict): domain: Literal["timeline"] schema_version: str snapshot_id: str files: Manifest projects: _ProjectMap def _timeline_manifests_dir(root: pathlib.Path) -> pathlib.Path: return root / ".muse" / "timeline_manifests" def build_timeline_manifest( files: Manifest, projects: dict[str, IdentifiedProject], project_hashes: dict[str, str], snapshot_id: str = "", ) -> TimelineManifest: projects_out: _ProjectMap = {} for path, identified in sorted(projects.items()): seq_map: _SequenceMap = {} for id_seq in identified.sequences: track_map: _TrackMap = {} for id_tr in id_seq.tracks: track_map[id_tr.track_identity] = TimelineTrackEntry( track_identity=id_tr.track_identity, kind=id_tr.kind, ordinal=id_tr.track.ordinal, name=id_tr.track.name, clips=[ TimelineClipEntry( clip_id=ic.clip_id, name=ic.clip.name, source_media_id=ic.clip.source_media_id, occurrence=ic.occurrence, ) for ic in id_tr.clips ], ) seq_map[id_seq.sequence_identity] = TimelineSequenceEntry( sequence_identity=id_seq.sequence_identity, name=id_seq.sequence.name, ordinal=id_seq.sequence.ordinal, tracks=track_map, ) projects_out[path] = TimelineProjectEntry( project_key=identified.project.project_key, name=identified.project.name, content_id=project_hashes[path], sequences=seq_map, ) return TimelineManifest( domain="timeline", schema_version=__version__, snapshot_id=snapshot_id, files=files, projects=projects_out, ) def write_timeline_manifest( repo_root: pathlib.Path, manifest: TimelineManifest, ) -> pathlib.Path: snapshot_id = manifest.get("snapshot_id", "") if not snapshot_id: raise ValueError("TimelineManifest.snapshot_id must be non-empty") path = _timeline_manifests_dir(repo_root) / f"{snapshot_id}.json" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(f"{json.dumps(manifest, indent=2, sort_keys=True)}\n") logger.debug("Timeline manifest written: %s", snapshot_id[:8]) return path def _manifest_files(raw: JsonObject) -> Manifest: files_in = raw.get("files") files: Manifest = {} if not isinstance(files_in, dict): return files for path_key, digest in files_in.items(): if isinstance(path_key, str) and isinstance(digest, str): files[path_key] = digest return files def read_timeline_manifest( repo_root: pathlib.Path, snapshot_id: str, ) -> TimelineManifest | None: path = _timeline_manifests_dir(repo_root) / f"{snapshot_id}.json" if not path.exists(): return None raw = load_json_file(path) if raw is None or not isinstance(raw, dict): return None if raw.get("domain") != "timeline": return None schema_version = raw.get("schema_version") snap_id = raw.get("snapshot_id") if not isinstance(schema_version, str) or not isinstance(snap_id, str): return None return TimelineManifest( domain="timeline", schema_version=schema_version, snapshot_id=snap_id, files=_manifest_files(raw), projects={}, )