gabriel / muse public
manifest.py python
158 lines 4.7 KB
Raw
sha256:61100cca63d948098d334e1b600d8fea514568a1c26ef357bf8b6380fbd8a217 chore(timeline): remove unused RationalRate import in entity.py Human minor ⚠ breaking 8 days ago
1 """Timeline sidecar manifest (mirrors ``muse.plugins.midi.manifest``)."""
2
3 from __future__ import annotations
4
5 import json
6 import logging
7 import pathlib
8 from typing import Literal, TypedDict
9
10 from muse._version import __version__
11 from muse.core.types import JsonObject, Manifest, load_json_file
12
13 from muse.plugins.timeline.entity import IdentifiedProject
14
15 logger = logging.getLogger(__name__)
16
17 type _ProjectMap = dict[str, "TimelineProjectEntry"]
18 type _SequenceMap = dict[str, "TimelineSequenceEntry"]
19 type _TrackMap = dict[str, "TimelineTrackEntry"]
20
21
22 class TimelineClipEntry(TypedDict):
23 clip_id: str
24 name: str
25 source_media_id: str
26 occurrence: int
27
28
29 class TimelineTrackEntry(TypedDict):
30 track_identity: str
31 kind: Literal["video", "audio", "subtitle"]
32 ordinal: int
33 name: str
34 clips: list[TimelineClipEntry]
35
36
37 class TimelineSequenceEntry(TypedDict):
38 sequence_identity: str
39 name: str
40 ordinal: int
41 tracks: _TrackMap
42
43
44 class TimelineProjectEntry(TypedDict):
45 project_key: str
46 name: str
47 content_id: str
48 sequences: _SequenceMap
49
50
51 class TimelineManifest(TypedDict):
52 domain: Literal["timeline"]
53 schema_version: str
54 snapshot_id: str
55 files: Manifest
56 projects: _ProjectMap
57
58
59 def _timeline_manifests_dir(root: pathlib.Path) -> pathlib.Path:
60 return root / ".muse" / "timeline_manifests"
61
62
63 def build_timeline_manifest(
64 files: Manifest,
65 projects: dict[str, IdentifiedProject],
66 project_hashes: dict[str, str],
67 snapshot_id: str = "",
68 ) -> TimelineManifest:
69 projects_out: _ProjectMap = {}
70 for path, identified in sorted(projects.items()):
71 seq_map: _SequenceMap = {}
72 for id_seq in identified.sequences:
73 track_map: _TrackMap = {}
74 for id_tr in id_seq.tracks:
75 track_map[id_tr.track_identity] = TimelineTrackEntry(
76 track_identity=id_tr.track_identity,
77 kind=id_tr.kind,
78 ordinal=id_tr.track.ordinal,
79 name=id_tr.track.name,
80 clips=[
81 TimelineClipEntry(
82 clip_id=ic.clip_id,
83 name=ic.clip.name,
84 source_media_id=ic.clip.source_media_id,
85 occurrence=ic.occurrence,
86 )
87 for ic in id_tr.clips
88 ],
89 )
90 seq_map[id_seq.sequence_identity] = TimelineSequenceEntry(
91 sequence_identity=id_seq.sequence_identity,
92 name=id_seq.sequence.name,
93 ordinal=id_seq.sequence.ordinal,
94 tracks=track_map,
95 )
96 projects_out[path] = TimelineProjectEntry(
97 project_key=identified.project.project_key,
98 name=identified.project.name,
99 content_id=project_hashes[path],
100 sequences=seq_map,
101 )
102 return TimelineManifest(
103 domain="timeline",
104 schema_version=__version__,
105 snapshot_id=snapshot_id,
106 files=files,
107 projects=projects_out,
108 )
109
110
111 def write_timeline_manifest(
112 repo_root: pathlib.Path,
113 manifest: TimelineManifest,
114 ) -> pathlib.Path:
115 snapshot_id = manifest.get("snapshot_id", "")
116 if not snapshot_id:
117 raise ValueError("TimelineManifest.snapshot_id must be non-empty")
118 path = _timeline_manifests_dir(repo_root) / f"{snapshot_id}.json"
119 path.parent.mkdir(parents=True, exist_ok=True)
120 path.write_text(f"{json.dumps(manifest, indent=2, sort_keys=True)}\n")
121 logger.debug("Timeline manifest written: %s", snapshot_id[:8])
122 return path
123
124
125 def _manifest_files(raw: JsonObject) -> Manifest:
126 files_in = raw.get("files")
127 files: Manifest = {}
128 if not isinstance(files_in, dict):
129 return files
130 for path_key, digest in files_in.items():
131 if isinstance(path_key, str) and isinstance(digest, str):
132 files[path_key] = digest
133 return files
134
135
136 def read_timeline_manifest(
137 repo_root: pathlib.Path,
138 snapshot_id: str,
139 ) -> TimelineManifest | None:
140 path = _timeline_manifests_dir(repo_root) / f"{snapshot_id}.json"
141 if not path.exists():
142 return None
143 raw = load_json_file(path)
144 if raw is None or not isinstance(raw, dict):
145 return None
146 if raw.get("domain") != "timeline":
147 return None
148 schema_version = raw.get("schema_version")
149 snap_id = raw.get("snapshot_id")
150 if not isinstance(schema_version, str) or not isinstance(snap_id, str):
151 return None
152 return TimelineManifest(
153 domain="timeline",
154 schema_version=schema_version,
155 snapshot_id=snap_id,
156 files=_manifest_files(raw),
157 projects={},
158 )
File History 1 commit
sha256:61100cca63d948098d334e1b600d8fea514568a1c26ef357bf8b6380fbd8a217 chore(timeline): remove unused RationalRate import in entity.py Human minor ⚠ 8 days ago