otio_diff.py

"""
otio-diff — structural editorial diff between two timelines.

HANDOFF NOTES (read first)
==========================
This is a v1 scaffold. The HARD design decisions are already made and documented
inline. The full acceptance suite (test_otio_diff.py, 8 tests) PASSES against
this scaffold on opentimelineio==0.18.1 — including nested-stack flatten and
duplicate-clip multiset. Earlier drafts of these notes overclaimed what was
unfinished; the recursion and multiset logic below are VERIFIED WORKING, not
stubs. Remaining work is polish (frame-accurate output, collection handling,
real-file validation), tracked in HANDOFF.md. See `# TODO(handoff):` markers, but
note some are "verify against reality," not "build from scratch."

Design decisions locked in (do not relitigate without reason):
  1. A cut is NOT positionally diffable. Inserting one clip at the head shifts
     every downstream timecode, so we MATCH clips by identity first, then
     classify. See `clip_key()`.
  2. Identity key = (media target_url, source_range.start_time). Duration is
     compared as an ATTRIBUTE after matching, so an out-point trim reads as
     "retimed" instead of removed+added. (Revised 2026-07-02 with a concrete
     failing case — see clip_key docstring.) Name is deliberately NOT in the key
     when a media URL exists (editors rename freely; media+in-point is the stable
     identity), but is the best available fallback for offline media. Clips that
     merely slid on the timeline (ripple from an upstream edit) are reported as
     "shifted", separate from "retimed".
  3. Scope is STRUCTURAL EDITORIAL ONLY: clips, timing, order. We do NOT diff
     effects/transitions/retime curves. OTIO's own docs are explicit that those
     serialize in proprietary, tool-specific ways and don't round-trip. Diffing
     them is the rabbit hole that turns this from a weekend tool into a
     maintenance sink. Keep it out of v1.
  4. Multi-format is FREE. read_from_file() auto-detects .otio/.edl/.fcpxml/.aaf
     via installed adapters. We never write a parser.

Maintenance profile: pin opentimelineio; the core schema is stable/versioned.
Expect ~one touch/year on version bump.

Tested against: opentimelineio (pin exact version in requirements.txt at handoff).
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass, asdict
from typing import Optional

import opentimelineio as otio


# ---------------------------------------------------------------------------
# Flatten: walk the canonical Timeline -> Stack(tracks) -> Track -> items tree
# into a flat list of clip records. Handles nesting recursively.
# ---------------------------------------------------------------------------

@dataclass
class ClipRecord:
    name: str
    media_url: Optional[str]          # target_url of the media reference, or None
    src_start: Optional[float]        # source_range.start_time in seconds, or None
    src_duration: Optional[float]     # source_range.duration in seconds, or None
    timeline_start: Optional[float]   # position on the timeline in seconds, or None
    track_index: int
    position_index: int               # ordinal within its track (for move detection)
    rate: Optional[float] = None      # clip frame rate, for seconds->frames in output


def _seconds(rt: Optional[otio.opentime.RationalTime]) -> Optional[float]:
    """RationalTime -> float seconds, or None. RationalTime is value*(1/rate)."""
    if rt is None:
        return None
    # to_seconds() exists on modern OTIO; fall back to manual if a pin lacks it.
    try:
        return rt.to_seconds()
    except AttributeError:
        return rt.value / rt.rate if rt.rate else None


def _media_url(clip: otio.schema.Clip) -> Optional[str]:
    """Extract the media target_url, tolerating MissingReference."""
    ref = clip.media_reference
    if ref is None:
        return None
    # ExternalReference has target_url; MissingReference does not.
    return getattr(ref, "target_url", None)


def flatten_timeline(tl: otio.schema.Timeline) -> list[ClipRecord]:
    """
    Produce a flat, ordered list of ClipRecords from a Timeline.

    Implementation: explicit tracks→items walk (walk() below), recursing into
    nested Stack/Track. VERIFIED: passes test_nested_stack_flattens. An earlier
    note framed this as an unfilled stub — it is not; the recursion is complete.

    # TODO(handoff): the only open refinement is track/position semantics for
    # DEEPLY nested compositions (current policy: nested clips keep the parent
    # track_index, positions increment inline). Confirm this reads correctly for
    # your use cases against a real nested export; adjust only if it doesn't.
    """
    records: list[ClipRecord] = []

    def walk(container, track_index: int) -> None:
        pos = 0
        for item in container:
            if isinstance(item, otio.schema.Clip):
                src = item.source_range
                # timeline position within this container
                try:
                    tl_range = container.trimmed_range_of_child(item)
                    tl_start = _seconds(tl_range.start_time) if tl_range else None
                except Exception:
                    tl_start = None
                records.append(ClipRecord(
                    name=item.name or "",
                    media_url=_media_url(item),
                    src_start=_seconds(src.start_time) if src else None,
                    src_duration=_seconds(src.duration) if src else None,
                    timeline_start=tl_start,
                    track_index=track_index,
                    position_index=pos,
                    rate=(src.duration.rate if src and src.duration.rate else None),
                ))
                pos += 1
            elif isinstance(item, (otio.schema.Stack, otio.schema.Track)):
                # TODO(handoff): recurse into nested compositions. Decide whether
                # nested clips inherit the parent track_index or get a synthetic
                # one. Recommend: keep parent track_index, keep incrementing pos,
                # so a nested stack reads as inline for diff purposes.
                walk(item, track_index)
            elif isinstance(item, otio.schema.Gap):
                # Gaps affect downstream timecode but are not clips. Skip them
                # entirely: position_index counts CLIPS only, so a gap appearing
                # or resizing (e.g. a lift-style trim in a real EDL) does not
                # read as every downstream clip having "moved". Timing effects
                # of gaps are already captured via timeline_start.
                # (Real-file finding, 2026-07-02: counting gaps as position
                # slots produced 6 phantom moves from one 12-frame trim.)
                pass
            # Transitions: intentionally ignored (out of v1 scope).

    for t_idx, track in enumerate(tl.tracks):
        walk(track, t_idx)

    return records


# ---------------------------------------------------------------------------
# Match + classify. This is the heart of the tool.
# ---------------------------------------------------------------------------

def clip_key(rec: ClipRecord) -> Optional[tuple]:
    """
    Identity key for matching a clip across two timelines. See design note #2.

    Identity is (media_url, src_start) when a media URL is available. Offline
    clips fall back to (name, src_start), since MissingReference has no durable
    media identifier. An unnamed offline clip returns None and is deliberately
    left unmatched: reporting add/remove is safer than silently treating two
    ambiguous records as unchanged.

    Duration is deliberately NOT part of identity. It's compared as an
    attribute after matching, so an out-point trim reads as "retimed" rather
    than removed+added. (Verified failing case that forced this: with duration
    in the key, a shortened clip fell out of its own identity and test_retimed
    only passed because a downstream clip's knock-on timeline shift populated
    `retimed`.)

    Known limitation: a head trim changes src_start and therefore identity, so it
    reads as removed+added. Loosening further (url-only + nearest-match pairing)
    is deferred until a real export shows it's needed.

    Rounding src times to frame-ish granularity avoids float drift between
    adapters. # TODO(handoff): make rounding rate-aware if false-mismatches show
    up in AAF<->FCPXML tests (adapters can differ in sub-frame representation).
    """
    def r(x): return round(x, 4) if x is not None else None
    if rec.media_url:
        # Preserve the established key shape for normal, online media.
        return (rec.media_url, r(rec.src_start))
    if rec.name:
        return ("name", rec.name, r(rec.src_start))
    return None


@dataclass
class DiffResult:
    added: list[dict]      # in B, not A
    removed: list[dict]    # in A, not B
    retimed: list[dict]    # same identity, different trimmed duration
    moved: list[dict]      # same identity, different ordinal position
    shifted: list[dict]    # same identity/content, only slid on the timeline (ripple)
    unchanged_count: int


def diff(a: list[ClipRecord], b: list[ClipRecord]) -> DiffResult:
    """
    Match-then-classify. Build key -> record maps for both sides.

    Edge case (VERIFIED WORKING on synthetic fixtures): the SAME identity key can
    appear more than once if an editor uses the same source+range twice. Policy:
    treat duplicate keys as a multiset — pair them up in order, surplus on either
    side becomes added/removed. The surplus-pairing below implements this and
    passes test_duplicate_clip_multiset. (An earlier note called this a
    ship-blocker requiring rework — that was wrong; it works.)
    # TODO(handoff): the only remaining risk is ORDERING within a duplicate group
    # under real adapter output — in-index pairing assumes stable clip order. If a
    # real AAF/FCPXML export reorders duplicates, refine the pairing to match on
    # nearest timeline position. Verify against a real file before worrying about it.
    """
    from collections import defaultdict

    a_by_key: dict[tuple, list[ClipRecord]] = defaultdict(list)
    b_by_key: dict[tuple, list[ClipRecord]] = defaultdict(list)
    unmatched_a: list[ClipRecord] = []
    unmatched_b: list[ClipRecord] = []
    for r in a:
        key = clip_key(r)
        if key is None:
            unmatched_a.append(r)
        else:
            a_by_key[key].append(r)
    for r in b:
        key = clip_key(r)
        if key is None:
            unmatched_b.append(r)
        else:
            b_by_key[key].append(r)

    added = [asdict(r) for r in unmatched_b]
    removed = [asdict(r) for r in unmatched_a]
    retimed, moved, shifted = [], [], []

    # Phase 1: pair matched clips; classify surplus as added/removed.
    matched: list[tuple[tuple, ClipRecord, ClipRecord]] = []
    all_keys = set(a_by_key) | set(b_by_key)
    for key in all_keys:
        a_recs = a_by_key.get(key, [])
        b_recs = b_by_key.get(key, [])

        # Multiset pairing: match duplicates in order, surplus -> added/removed.
        # Verified on synthetic fixtures; see TODO in docstring re: real-file ordering.
        pairs = min(len(a_recs), len(b_recs))
        for i in range(pairs):
            matched.append((key, a_recs[i], b_recs[i]))

        # surplus on A = removed, surplus on B = added
        for r in a_recs[pairs:]:
            removed.append(asdict(r))
        for r in b_recs[pairs:]:
            added.append(asdict(r))

    # Phase 2: moved = RELATIVE order of matched clips changed, not absolute
    # ordinal. Absolute position_index breaks on real files: a lift-trim inserts
    # a Gap, a removal deletes a slot, and either way every downstream index
    # changes while nothing actually reordered. (Real-EDL finding, 2026-07-02:
    # one 12-frame trim read as 6 phantom moves; one removal read as 4.)
    # Method: order matched pairs by their A-side and B-side timeline positions;
    # the longest increasing subsequence of A-ranks in B order is the set of
    # clips that kept their relative order — everything outside it moved. A
    # track change is a move unconditionally.
    moved_idx: set[int] = set()
    same_track = [i for i, (_, ra, rb) in enumerate(matched)
                  if ra.track_index == rb.track_index]
    moved_idx.update(i for i in range(len(matched)) if i not in same_track)

    order_a = sorted(same_track,
                     key=lambda i: (matched[i][1].track_index, matched[i][1].position_index))
    rank_a = {idx: n for n, idx in enumerate(order_a)}
    order_b = sorted(same_track,
                     key=lambda i: (matched[i][2].track_index, matched[i][2].position_index))
    seq = [rank_a[i] for i in order_b]

    # Longest increasing subsequence (patience sorting, O(n log n)); indices of
    # seq elements in the LIS kept their relative order.
    import bisect
    tails: list[int] = []          # tails[k] = smallest seq value ending a LIS of length k+1
    tails_pos: list[int] = []      # position in seq of that tail
    prev: list[int] = [-1] * len(seq)
    for pos, val in enumerate(seq):
        k = bisect.bisect_left(tails, val)
        if k == len(tails):
            tails.append(val)
            tails_pos.append(pos)
        else:
            tails[k] = val
            tails_pos[k] = pos
        prev[pos] = tails_pos[k - 1] if k > 0 else -1
    in_lis: set[int] = set()
    if tails_pos:
        p = tails_pos[-1]
        while p != -1:
            in_lis.add(p)
            p = prev[p]
    moved_idx.update(order_b[pos] for pos in range(len(seq)) if pos not in in_lis)

    # Phase 3: classify each matched pair.
    unchanged = 0
    for i, (key, ra, rb) in enumerate(matched):
        changed = False
        # retimed: the clip's own content timing changed (trimmed duration)
        if ra.src_duration != rb.src_duration:
            retimed.append({"key": key, "before": asdict(ra), "after": asdict(rb)})
            changed = True
        if i in moved_idx:
            moved.append({"key": key, "before": asdict(ra), "after": asdict(rb)})
            changed = True
        # shifted: content and order untouched, but the clip slid on the
        # timeline — the ripple effect of an upstream edit. Kept separate so
        # one trim doesn't read as N retimes downstream.
        if not changed and ra.timeline_start != rb.timeline_start:
            shifted.append({"key": key, "before": asdict(ra), "after": asdict(rb)})
            changed = True
        if not changed:
            unchanged += 1

    return DiffResult(added, removed, retimed, moved, shifted, unchanged)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def load(path: str, rate: Optional[float] = None) -> list[ClipRecord]:
    """
    read_from_file auto-detects .otio/.edl/.fcpxml/.aaf via adapters.

    `rate` is forwarded to the EDL adapter: CMX 3600 EDLs don't carry their
    frame rate, and the adapter assumes 24fps — a 25fps EDL fails to parse
    without the hint. (Real-file finding, 2026-07-02.) Only .edl accepts it.

    Some adapters return a SerializableCollection rather than a Timeline
    (multi-sequence exports). Policy: diff the FIRST Timeline found and warn on
    stderr if there were more — diffing one pair of cuts is the tool's job;
    element-wise multi-timeline diffing is out of v1 scope.
    """
    kwargs = {}
    if rate is not None and path.lower().endswith(".edl"):
        kwargs["rate"] = rate
    tl = otio.adapters.read_from_file(path, **kwargs)
    if isinstance(tl, otio.schema.SerializableCollection):
        timelines = [t for t in tl if isinstance(t, otio.schema.Timeline)]
        if not timelines:
            raise ValueError(f"{path}: collection contains no Timeline")
        if len(timelines) > 1:
            print(
                f"warning: {path} contains {len(timelines)} timelines; "
                f"diffing the first ({timelines[0].name or 'unnamed'})",
                file=sys.stderr,
            )
        tl = timelines[0]
    return flatten_timeline(tl)


def _label(rec: dict) -> str:
    """Editor-facing clip label: name, falling back to the media file basename."""
    if rec.get("name"):
        return rec["name"]
    url = rec.get("media_url")
    return url.rsplit("/", 1)[-1] if url else "<unnamed>"


def _fmt(seconds: Optional[float], rate: Optional[float]) -> str:
    """Duration/offset for editors: frames when rate is known, else seconds."""
    if seconds is None:
        return "?"
    if rate:
        return f"{round(seconds * rate)}f"
    return f"{seconds:.3f}s"


def human(result: DiffResult) -> str:
    counts = []
    if result.added:
        counts.append(f"{len(result.added)} added")
    if result.removed:
        counts.append(f"{len(result.removed)} removed")
    if result.retimed:
        counts.append(f"{len(result.retimed)} retimed")
    if result.moved:
        counts.append(f"{len(result.moved)} moved")
    if result.shifted:
        counts.append(f"{len(result.shifted)} shifted")
    if not counts:
        return "No structural changes."

    lines = [", ".join(counts) + f" ({result.unchanged_count} unchanged)"]

    for rec in result.added:
        lines.append(f"  + {_label(rec)} added ({_fmt(rec['src_duration'], rec.get('rate'))})")
    for rec in result.removed:
        lines.append(f"  - {_label(rec)} removed ({_fmt(rec['src_duration'], rec.get('rate'))})")
    for item in result.retimed:
        ra, rb = item["before"], item["after"]
        rate = rb.get("rate") or ra.get("rate")
        da, db = ra["src_duration"], rb["src_duration"]
        if da is not None and db is not None:
            verb = "shortened" if db < da else "lengthened"
            delta = _fmt(abs(db - da), rate)
            lines.append(
                f"  ~ {_label(rb)} {verb} by {delta} "
                f"({_fmt(da, rate)} -> {_fmt(db, rate)})"
            )
        else:
            lines.append(f"  ~ {_label(rb)} retimed")
    for item in result.moved:
        ra, rb = item["before"], item["after"]
        lines.append(
            f"  > {_label(rb)} moved "
            f"(track {ra['track_index']} pos {ra['position_index']} -> "
            f"track {rb['track_index']} pos {rb['position_index']})"
        )
    for item in result.shifted:
        ra, rb = item["before"], item["after"]
        rate = rb.get("rate") or ra.get("rate")
        ta, tb = ra["timeline_start"], rb["timeline_start"]
        if ta is not None and tb is not None:
            direction = "earlier" if tb < ta else "later"
            lines.append(f"  . {_label(rb)} shifted {_fmt(abs(tb - ta), rate)} {direction}")
        else:
            lines.append(f"  . {_label(rb)} shifted")

    return "\n".join(lines)


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description="Structural editorial diff between two timelines.")
    p.add_argument("a", help="baseline timeline (.otio/.edl/.fcpxml/.aaf)")
    p.add_argument("b", help="revised timeline (.otio/.edl/.fcpxml/.aaf)")
    p.add_argument("--json", action="store_true", help="emit JSON instead of human summary")
    p.add_argument("--rate", type=float, default=None,
                   help="frame rate hint for EDL inputs (EDLs don't carry it; default 24)")
    args = p.parse_args(argv)

    try:
        a = load(args.a, rate=args.rate)
        b = load(args.b, rate=args.rate)
    except Exception as e:  # adapters raise varied errors; surface cleanly
        print(f"error: could not read timelines: {e}", file=sys.stderr)
        return 2

    result = diff(a, b)
    if args.json:
        print(json.dumps(asdict(result), indent=2))
    else:
        print(human(result))
    # diff(1) convention, agent-friendly: 0 = no structural changes,
    # 1 = changes found, 2 = error (returned above on read failure).
    has_changes = bool(result.added or result.removed or result.retimed
                       or result.moved or result.shifted)
    return 1 if has_changes else 0


if __name__ == "__main__":
    raise SystemExit(main())