gabriel / muse public
_otio_bridge.py python
635 lines 21.4 KB
Raw
sha256:9a3bb190de5a18742792038aa709072a582ada8b223d5dfa4f72c70831ec3ba3 docs: revert migrate hub-scoping/domain-integers rows from … Sonnet 5 6 hours ago
1 """Pure-JSON OTIO bridge — no compiled opentimelineio (VID-1 §5)."""
2
3 from __future__ import annotations
4
5 import json
6 import pathlib
7 from typing import Literal
8
9 from muse.core.types import JsonObject, JsonValue, load_json_file
10
11 from muse.plugins.timeline.entity import (
12 AudioLane,
13 Caption,
14 Clip,
15 Effect,
16 ExternalReference,
17 Gap,
18 Keyframe,
19 Marker,
20 Metadata,
21 Project,
22 Sequence,
23 SourceMedia,
24 Track,
25 TrackItem,
26 Transition,
27 media_content_id,
28 meta_get,
29 stringify_metadata_value,
30 )
31 from muse.plugins.timeline.rational_time import RationalRate, RationalTime, TimeRange
32
33 _VOLATILE_METADATA_KEYS: frozenset[str] = frozenset(
34 {"generator", "otio_schema_version", "created", "modified", "timestamp"}
35 )
36
37 _KNOWN_SCHEMA_PREFIXES: frozenset[str] = frozenset(
38 {
39 "Timeline.",
40 "Stack.",
41 "Track.",
42 "Clip.",
43 "Gap.",
44 "Transition.",
45 "ExternalReference.",
46 "MissingReference.",
47 "RationalTime.",
48 "TimeRange.",
49 "SerializableCollection.",
50 "Effect.",
51 "Marker.",
52 }
53 )
54
55
56 class BridgeError(Exception):
57 """Raised when OTIO JSON cannot be bridged to the canonical IR."""
58
59
60 def load_otio_json(path: pathlib.Path) -> JsonObject:
61 raw = load_json_file(path)
62 if raw is None or not isinstance(raw, dict):
63 raise BridgeError(f"invalid OTIO JSON: {path}")
64 return raw
65
66
67 def _schema_version(obj: JsonObject) -> tuple[str, int]:
68 schema = obj.get("OTIO_SCHEMA")
69 if not isinstance(schema, str) or "." not in schema:
70 raise BridgeError("missing OTIO_SCHEMA")
71 name, ver_str = schema.rsplit(".", 1)
72 if not ver_str.isdigit():
73 raise BridgeError(f"invalid OTIO_SCHEMA version: {schema}")
74 known = any(schema.startswith(p) for p in _KNOWN_SCHEMA_PREFIXES)
75 if not known:
76 raise BridgeError(f"unknown OTIO_SCHEMA family: {schema}")
77 return name, int(ver_str)
78
79
80 def _check_schema(obj: JsonObject, family: str, max_version: int = 99) -> None:
81 name, ver = _schema_version(obj)
82 if name != family:
83 raise BridgeError(f"expected {family}, got {name}")
84 if ver > max_version:
85 raise BridgeError(f"unsupported {family} version {ver}")
86
87
88 def _parse_rate(raw: JsonValue) -> RationalRate:
89 if isinstance(raw, (int, float)):
90 return RationalRate.from_float(float(raw))
91 if isinstance(raw, dict):
92 val = raw.get("value", raw.get("num"))
93 den = raw.get("rate", raw.get("den", 1))
94 if isinstance(val, (int, float)) and isinstance(den, (int, float)):
95 if float(den) == 1.0:
96 return RationalRate.from_float(float(val))
97 num_i = int(val)
98 den_i = int(den)
99 return RationalRate(num=num_i, den=den_i)
100 raise BridgeError("invalid rate in RationalTime")
101
102
103 def _exact_int(value: JsonValue, field: str) -> int:
104 if isinstance(value, bool):
105 raise BridgeError(f"non-integral {field}")
106 if isinstance(value, int):
107 return value
108 if isinstance(value, float):
109 rounded = round(value)
110 if abs(value - rounded) > 1e-9:
111 raise BridgeError(f"non-integral {field}: {value}")
112 return int(rounded)
113 raise BridgeError(f"invalid {field}")
114
115
116 def _parse_rational_time(obj: JsonObject) -> RationalTime:
117 _check_schema(obj, "RationalTime")
118 rate = _parse_rate(obj.get("rate", 1))
119 value = _exact_int(obj["value"], "RationalTime.value")
120 return RationalTime(value=value, rate=rate)
121
122
123 def _parse_time_range(obj: JsonObject | None) -> TimeRange | None:
124 if obj is None:
125 return None
126 _check_schema(obj, "TimeRange")
127 start_raw = obj.get("start_time")
128 dur_raw = obj.get("duration")
129 if not isinstance(start_raw, dict) or not isinstance(dur_raw, dict):
130 raise BridgeError("invalid TimeRange")
131 return TimeRange(
132 start_time=_parse_rational_time(start_raw),
133 duration=_parse_rational_time(dur_raw),
134 )
135
136
137 def _canonicalize_metadata(raw: JsonValue) -> Metadata:
138 if not isinstance(raw, dict):
139 return ()
140 pairs: list[tuple[str, str]] = []
141 for key in sorted(raw.keys()):
142 if key in _VOLATILE_METADATA_KEYS or key.endswith("_abspath"):
143 continue
144 pairs.append((key, stringify_metadata_value(raw[key])))
145 return tuple(pairs)
146
147
148 def _relativise_url(
149 target_url: str,
150 otio_path: pathlib.Path,
151 repo_root: pathlib.Path | None,
152 ) -> str:
153 if "://" in target_url:
154 return target_url
155 if ".." in pathlib.PurePosixPath(target_url.replace("\\", "/")).parts:
156 raise BridgeError(f"path traversal in target_url: {target_url!r}")
157 if repo_root is None:
158 return target_url.replace("\\", "/")
159 if target_url.startswith("file://"):
160 path_part = target_url[7:]
161 else:
162 path_part = target_url
163 candidate = pathlib.Path(path_part)
164 if not candidate.is_absolute():
165 candidate = (otio_path.parent / candidate).resolve()
166 try:
167 rel = candidate.relative_to(repo_root.resolve())
168 except ValueError:
169 return target_url.replace("\\", "/")
170 return rel.as_posix()
171
172
173 def _parse_external_reference(
174 obj: JsonObject,
175 otio_path: pathlib.Path,
176 repo_root: pathlib.Path | None,
177 *,
178 force_offline: bool = False,
179 ) -> ExternalReference:
180 _check_schema(obj, "ExternalReference")
181 raw_url = obj.get("target_url", "")
182 if not isinstance(raw_url, str):
183 raise BridgeError("ExternalReference.target_url must be string")
184 url = _relativise_url(raw_url, otio_path, repo_root)
185 is_offline = force_offline
186 if repo_root is not None and not force_offline and "://" not in url:
187 resolved = repo_root / url
188 is_offline = not resolved.exists()
189 md = _canonicalize_metadata(obj.get("metadata"))
190 essence = meta_get(md, "essence_hash")
191 return ExternalReference(
192 target_url=url,
193 available_range=_parse_time_range(
194 obj.get("available_range") # type: ignore[arg-type]
195 if isinstance(obj.get("available_range"), dict)
196 else None
197 ),
198 essence_hash=essence,
199 is_offline=is_offline,
200 metadata=md,
201 )
202
203
204 def _parse_media_reference(
205 obj: JsonObject | None,
206 otio_path: pathlib.Path,
207 repo_root: pathlib.Path | None,
208 ) -> ExternalReference:
209 if obj is None:
210 raise BridgeError("clip missing media_reference")
211 schema = obj.get("OTIO_SCHEMA", "")
212 if isinstance(schema, str) and schema.startswith("MissingReference"):
213 _schema_version(obj)
214 return ExternalReference(
215 target_url="",
216 available_range=None,
217 essence_hash=None,
218 is_offline=True,
219 metadata=_canonicalize_metadata(obj.get("metadata")),
220 )
221 return _parse_external_reference(obj, otio_path, repo_root)
222
223
224 def _parse_keyframe(obj: JsonObject) -> Keyframe:
225 time_obj = obj.get("time")
226 if not isinstance(time_obj, dict):
227 raise BridgeError("invalid keyframe time")
228 interp_raw = obj.get("interpolation", "linear")
229 interp = str(interp_raw) if isinstance(interp_raw, str) else "linear"
230 if interp not in ("hold", "linear", "bezier", "constant"):
231 interp = "linear"
232 return Keyframe(
233 time=_parse_rational_time(time_obj),
234 value=stringify_metadata_value(obj.get("value", "")),
235 interpolation=interp, # type: ignore[arg-type]
236 metadata=_canonicalize_metadata(obj.get("metadata")),
237 )
238
239
240 def _parse_effect(obj: JsonObject) -> Effect:
241 _check_schema(obj, "Effect")
242 params_raw = obj.get("parameters")
243 params: list[tuple[str, str]] = []
244 if isinstance(params_raw, dict):
245 for k in sorted(params_raw):
246 if isinstance(k, str):
247 params.append((k, stringify_metadata_value(params_raw[k])))
248 kf_raw = obj.get("keyframes")
249 keyframes: list[Keyframe] = []
250 if isinstance(kf_raw, list):
251 for item in kf_raw:
252 if isinstance(item, dict):
253 keyframes.append(_parse_keyframe(item))
254 keyframes.sort(key=lambda k: (k.time, k.value))
255 return Effect(
256 name=str(obj.get("name", "")),
257 effect_type=str(obj.get("effect_name", obj.get("name", "Effect"))),
258 parameters=tuple(params),
259 keyframes=tuple(keyframes),
260 metadata=_canonicalize_metadata(obj.get("metadata")),
261 )
262
263
264 def _parse_marker(obj: JsonObject) -> Marker:
265 _check_schema(obj, "Marker")
266 marked = obj.get("marked_range")
267 if not isinstance(marked, dict):
268 raise BridgeError("marker missing marked_range")
269 return Marker(
270 name=str(obj.get("name", "")),
271 marked_range=_parse_time_range(marked) or TimeRange(
272 start_time=RationalTime(0, RationalRate(24, 1)),
273 duration=RationalTime(0, RationalRate(24, 1)),
274 ),
275 color=str(obj.get("color", "")),
276 comment=str(obj.get("comment", "")),
277 metadata=_canonicalize_metadata(obj.get("metadata")),
278 )
279
280
281 def _parse_caption(obj: JsonObject, default_rate: RationalRate) -> Caption:
282 marked = obj.get("marked_range")
283 if isinstance(marked, dict):
284 tr = _parse_time_range(marked)
285 else:
286 tr = None
287 if tr is None:
288 tr = TimeRange(
289 start_time=RationalTime(0, default_rate),
290 duration=RationalTime(0, default_rate),
291 )
292 return Caption(
293 text=str(obj.get("text", obj.get("name", ""))),
294 marked_range=tr,
295 style=str(obj.get("style", "")),
296 metadata=_canonicalize_metadata(obj.get("metadata")),
297 )
298
299
300 def _parse_transition(obj: JsonObject, default_rate: RationalRate) -> Transition:
301 _check_schema(obj, "Transition")
302 in_off = obj.get("in_offset")
303 out_off = obj.get("out_offset")
304 in_rt = (
305 _parse_rational_time(in_off)
306 if isinstance(in_off, dict)
307 else RationalTime(0, default_rate)
308 )
309 out_rt = (
310 _parse_rational_time(out_off)
311 if isinstance(out_off, dict)
312 else RationalTime(0, default_rate)
313 )
314 return Transition(
315 transition_type=str(obj.get("transition_type", obj.get("name", ""))),
316 in_offset=in_rt,
317 out_offset=out_rt,
318 metadata=_canonicalize_metadata(obj.get("metadata")),
319 )
320
321
322 def _parse_gap(obj: JsonObject, default_rate: RationalRate) -> Gap:
323 _check_schema(obj, "Gap")
324 sr = obj.get("source_range")
325 if isinstance(sr, dict):
326 tr = _parse_time_range(sr)
327 if tr is not None:
328 return Gap(duration=tr.duration, metadata=_canonicalize_metadata(obj.get("metadata")))
329 dur = obj.get("duration")
330 if isinstance(dur, dict):
331 return Gap(
332 duration=_parse_rational_time(dur),
333 metadata=_canonicalize_metadata(obj.get("metadata")),
334 )
335 return Gap(
336 duration=RationalTime(0, default_rate),
337 metadata=_canonicalize_metadata(obj.get("metadata")),
338 )
339
340
341 def _parse_clip(
342 obj: JsonObject,
343 media_pool: dict[str, SourceMedia],
344 otio_path: pathlib.Path,
345 repo_root: pathlib.Path | None,
346 nested_map: dict[str, str],
347 ) -> Clip:
348 _check_schema(obj, "Clip")
349 media_ref = obj.get("media_reference")
350 ext = _parse_media_reference(
351 media_ref if isinstance(media_ref, dict) else None,
352 otio_path,
353 repo_root,
354 )
355 mid = media_content_id(ext.target_url, ext.available_range, ext.essence_hash)
356 if mid not in media_pool:
357 media_pool[mid] = SourceMedia(
358 content_id=mid,
359 name=str(obj.get("name", mid)),
360 references=(("default", ext),),
361 active_reference_key="default",
362 available_range=ext.available_range,
363 color=None,
364 metadata=(),
365 )
366 sr = obj.get("source_range")
367 if not isinstance(sr, dict):
368 raise BridgeError("clip missing source_range")
369 source_range = _parse_time_range(sr)
370 if source_range is None:
371 raise BridgeError("invalid clip source_range")
372 effects_raw = obj.get("effects")
373 effects: list[Effect] = []
374 if isinstance(effects_raw, list):
375 for ef in effects_raw:
376 if isinstance(ef, dict):
377 effects.append(_parse_effect(ef))
378 effects.sort(key=lambda e: (e.effect_type, e.name))
379 markers_raw = obj.get("markers")
380 markers: list[Marker] = []
381 if isinstance(markers_raw, list):
382 for mk in markers_raw:
383 if isinstance(mk, dict):
384 markers.append(_parse_marker(mk))
385 markers.sort(key=lambda m: (m.marked_range.start_time, m.name))
386 nested_id: str | None = None
387 md = _canonicalize_metadata(obj.get("metadata"))
388 nested_meta = meta_get(md, "nested_sequence_id")
389 if nested_meta:
390 nested_id = nested_meta
391 active_ref = obj.get("active_media_reference_key")
392 if isinstance(active_ref, str) and active_ref in nested_map:
393 nested_id = nested_map[active_ref]
394 return Clip(
395 name=str(obj.get("name", "")),
396 source_media_id=mid,
397 source_range=source_range,
398 enabled=bool(obj.get("enabled", True)),
399 effects=tuple(effects),
400 markers=tuple(markers),
401 color=None,
402 nested_sequence_id=nested_id,
403 metadata=_canonicalize_metadata(obj.get("metadata")),
404 )
405
406
407 def _parse_track_items(
408 children: list[JsonValue],
409 media_pool: dict[str, SourceMedia],
410 otio_path: pathlib.Path,
411 repo_root: pathlib.Path | None,
412 nested_map: dict[str, str],
413 default_rate: RationalRate,
414 ) -> tuple[TrackItem, ...]:
415 items: list[TrackItem] = []
416 for child in children:
417 if not isinstance(child, dict):
418 continue
419 schema = child.get("OTIO_SCHEMA", "")
420 if not isinstance(schema, str):
421 continue
422 if schema.startswith("Clip."):
423 items.append(_parse_clip(child, media_pool, otio_path, repo_root, nested_map))
424 elif schema.startswith("Gap."):
425 items.append(_parse_gap(child, default_rate))
426 elif schema.startswith("Transition."):
427 items.append(_parse_transition(child, default_rate))
428 return tuple(items)
429
430
431 def _track_kind(raw: JsonValue) -> Literal["video", "subtitle"]:
432 if isinstance(raw, str) and raw.lower() in ("subtitle", "subtitles", "caption"):
433 return "subtitle"
434 return "video"
435
436
437 def _parse_track(
438 obj: JsonObject,
439 ordinal: int,
440 media_pool: dict[str, SourceMedia],
441 otio_path: pathlib.Path,
442 repo_root: pathlib.Path | None,
443 nested_map: dict[str, str],
444 default_rate: RationalRate,
445 ) -> Track:
446 _check_schema(obj, "Track")
447 kind = _track_kind(obj.get("kind"))
448 children = obj.get("children")
449 child_list = children if isinstance(children, list) else []
450 items = _parse_track_items(
451 child_list, media_pool, otio_path, repo_root, nested_map, default_rate
452 )
453 caps_raw = obj.get("metadata", {})
454 captions: list[Caption] = []
455 if isinstance(caps_raw, dict):
456 cap_list = caps_raw.get("captions")
457 if isinstance(cap_list, list):
458 for cap in cap_list:
459 if isinstance(cap, dict):
460 captions.append(_parse_caption(cap, default_rate))
461 captions.sort(key=lambda c: (c.marked_range.start_time, c.text))
462 return Track(
463 name=str(obj.get("name", f"track_{ordinal}")),
464 kind=kind,
465 ordinal=ordinal,
466 enabled=bool(obj.get("enabled", True)),
467 items=items,
468 captions=tuple(captions),
469 metadata=_canonicalize_metadata(obj.get("metadata")),
470 )
471
472
473 def _parse_audio_lane(
474 obj: JsonObject,
475 ordinal: int,
476 media_pool: dict[str, SourceMedia],
477 otio_path: pathlib.Path,
478 repo_root: pathlib.Path | None,
479 nested_map: dict[str, str],
480 default_rate: RationalRate,
481 ) -> AudioLane:
482 _check_schema(obj, "Track")
483 children = obj.get("children")
484 child_list = children if isinstance(children, list) else []
485 items = _parse_track_items(
486 child_list, media_pool, otio_path, repo_root, nested_map, default_rate
487 )
488 md = _canonicalize_metadata(obj.get("metadata"))
489 sample_rate = 48000
490 sr_raw = meta_get(md, "sample_rate")
491 if sr_raw is not None and sr_raw.isdigit():
492 sample_rate = int(sr_raw)
493 return AudioLane(
494 name=str(obj.get("name", f"audio_{ordinal}")),
495 ordinal=ordinal,
496 enabled=bool(obj.get("enabled", True)),
497 sample_rate=sample_rate,
498 channel_count=2,
499 channel_layout="stereo",
500 items=items,
501 metadata=md,
502 )
503
504
505 def _parse_sequence(
506 obj: JsonObject,
507 ordinal: int,
508 project_key: str,
509 media_pool: dict[str, SourceMedia],
510 otio_path: pathlib.Path,
511 repo_root: pathlib.Path | None,
512 nested_map: dict[str, str],
513 ) -> Sequence:
514 default_rate = RationalRate(24, 1)
515 gst = obj.get("global_start_time")
516 global_start: RationalTime | None = None
517 if isinstance(gst, dict):
518 global_start = _parse_rational_time(gst)
519 default_rate = global_start.rate
520 tracks_raw = obj.get("tracks")
521 video_tracks: list[Track] = []
522 audio_lanes: list[AudioLane] = []
523 subtitle_tracks: list[Track] = []
524 markers: list[Marker] = []
525 if isinstance(tracks_raw, dict):
526 children = tracks_raw.get("children")
527 if isinstance(children, list):
528 v_ord = 0
529 a_ord = 0
530 s_ord = 0
531 for child in children:
532 if not isinstance(child, dict):
533 continue
534 schema = child.get("OTIO_SCHEMA", "")
535 if not isinstance(schema, str) or not schema.startswith("Track."):
536 continue
537 kind_raw = child.get("kind")
538 if isinstance(kind_raw, str) and kind_raw.lower() == "audio":
539 lane = _parse_audio_lane(
540 child, a_ord, media_pool, otio_path, repo_root, nested_map, default_rate
541 )
542 audio_lanes.append(lane)
543 a_ord += 1
544 elif _track_kind(kind_raw) == "subtitle":
545 tr = _parse_track(
546 child, s_ord, media_pool, otio_path, repo_root, nested_map, default_rate
547 )
548 subtitle_tracks.append(
549 Track(
550 name=tr.name,
551 kind="subtitle",
552 ordinal=s_ord,
553 enabled=tr.enabled,
554 items=tr.items,
555 captions=tr.captions,
556 metadata=tr.metadata,
557 )
558 )
559 s_ord += 1
560 else:
561 tr = _parse_track(
562 child, v_ord, media_pool, otio_path, repo_root, nested_map, default_rate
563 )
564 video_tracks.append(tr)
565 v_ord += 1
566 markers_raw = obj.get("markers")
567 if isinstance(markers_raw, list):
568 for mk in markers_raw:
569 if isinstance(mk, dict):
570 markers.append(_parse_marker(mk))
571 markers.sort(key=lambda m: (m.marked_range.start_time, m.name))
572 return Sequence(
573 name=str(obj.get("name", project_key)),
574 ordinal=ordinal,
575 global_start_time=global_start,
576 video_tracks=tuple(sorted(video_tracks, key=lambda t: t.ordinal)),
577 audio_lanes=tuple(sorted(audio_lanes, key=lambda t: t.ordinal)),
578 subtitle_tracks=tuple(sorted(subtitle_tracks, key=lambda t: t.ordinal)),
579 markers=tuple(markers),
580 metadata=_canonicalize_metadata(obj.get("metadata")),
581 )
582
583
584 def _project_key_from(obj: JsonObject, otio_path: pathlib.Path) -> str:
585 md = obj.get("metadata")
586 if isinstance(md, dict):
587 pk = md.get("muse.project_key")
588 if isinstance(pk, str) and pk:
589 return pk
590 return otio_path.stem
591
592
593 def parse_otio_to_project(
594 raw: JsonObject,
595 otio_path: pathlib.Path,
596 repo_root: pathlib.Path | None = None,
597 ) -> Project:
598 """Map OTIO JSON → canonical :class:`Project` IR."""
599 schema = raw.get("OTIO_SCHEMA", "")
600 media_pool: dict[str, SourceMedia] = {}
601 nested_map: dict[str, str] = {}
602 if isinstance(schema, str) and schema.startswith("SerializableCollection."):
603 _check_schema(raw, "SerializableCollection")
604 project_key = _project_key_from(raw, otio_path)
605 sequences: list[Sequence] = []
606 children = raw.get("children")
607 if isinstance(children, list):
608 for idx, child in enumerate(children):
609 if isinstance(child, dict):
610 sequences.append(
611 _parse_sequence(
612 child, idx, project_key, media_pool, otio_path, repo_root, nested_map
613 )
614 )
615 return Project(
616 project_key=project_key,
617 name=str(raw.get("name", project_key)),
618 sequences=tuple(sorted(sequences, key=lambda s: s.ordinal)),
619 media_pool=tuple(sorted(media_pool.values(), key=lambda m: m.content_id)),
620 render_cache=(),
621 metadata=_canonicalize_metadata(raw.get("metadata")),
622 )
623 if isinstance(schema, str) and schema.startswith("Timeline."):
624 _check_schema(raw, "Timeline")
625 project_key = _project_key_from(raw, otio_path)
626 seq = _parse_sequence(raw, 0, project_key, media_pool, otio_path, repo_root, nested_map)
627 return Project(
628 project_key=project_key,
629 name=str(raw.get("name", project_key)),
630 sequences=(seq,),
631 media_pool=tuple(sorted(media_pool.values(), key=lambda m: m.content_id)),
632 render_cache=(),
633 metadata=_canonicalize_metadata(raw.get("metadata")),
634 )
635 raise BridgeError(f"unsupported root OTIO_SCHEMA: {schema!r}")
File History 1 commit
sha256:9a3bb190de5a18742792038aa709072a582ada8b223d5dfa4f72c70831ec3ba3 docs: revert migrate hub-scoping/domain-integers rows from … Sonnet 5 6 hours ago