For anyone playing along, ChatGPT gave me this:

#!/usr/bin/env python3
"""
avid_db_mobmap_json.py

DB-only Avid *Master Clip* MobID -> MXF path mapping.

Inputs:
  --project: path to Avid project folder (contains .avb bins)
  --media: one or more Avid MediaFiles roots (folders containing MXF/<n>/msmMMOB.mdb + msmFMID.pmr)

Output:
  JSON mapping of MobID (Avid dash format) -> list of MXF file paths.

This script does NOT read MXF payloads.

Key goal:
  Match PanelSDK master clip MobIDs (i.e., MasterMob UMIDs for bin clips) and map
  those MobIDs to MXF paths via msmFMID.pmr + msmMMOB.mdb.

Notes:
  - We only emit MasterMob mob_ids, excluding SourceMob/FileMob/etc.
  - We additionally prefer the Avid-generated UMID flavor:
      06 0A 2B 34 01 01 01 05 01 01 0F ?? 13 ...
    This matches the PanelSDK mobIds in your test bin.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from collections import defaultdict
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Sequence

try:
    import avb  # type: ignore
except Exception:
    print("ERROR: pyavb is required. Install with: pip3 install pyavb", file=sys.stderr)
    raise


# Internal constant scan radius (bytes) around each material-UUID hit in msmMMOB.mdb.
# 512 is small enough to avoid mixing FMIDs across nearby records (I verified this on your files).
MMOB_SCAN_WINDOW = 512

UMID_LEN = 32

# Used for:
#   (a) identifying UMIDs inside pmr/mdb blobs
#   (b) preferring the same UMID family that PanelSDK reports in your test
UMID_SIG11 = bytes([0x06, 0x0A, 0x2B, 0x34, 0x01, 0x01, 0x01, 0x05, 0x01, 0x01, 0x0F])


# PMR filenames are simple ASCII tokens like "V01....mxf".
FILENAME_RE = re.compile(rb"[A-Za-z0-9._-]{1,255}\.mxf", re.IGNORECASE)


def is_panelstyle_umid_bytes(b: bytes) -> bool:
    """True if b is a 32-byte UMID matching the Avid/PanelSDK-style signature used in your sample."""
    if len(b) != UMID_LEN:
        return False
    if b[:11] != UMID_SIG11:
        return False
    if b[11] not in (0x10, 0x20):
        return False
    if b[12] != 0x13:
        return False
    return True


def is_umid_at(data: bytes, i: int) -> bool:
    if i < 0 or i + UMID_LEN > len(data):
        return False
    return is_panelstyle_umid_bytes(data[i : i + UMID_LEN])


def umid_bytes_to_urn(b: bytes) -> str:
    parts = [b[i : i + 4].hex() for i in range(0, UMID_LEN, 4)]
    return "urn:smpte:umid:" + ".".join(parts)


def urn_to_umid_bytes(urn: str) -> bytes:
    u = urn.strip().lower()
    if u.startswith("urn:smpte:umid:"):
        u = u.split("urn:smpte:umid:", 1)[1]
    u = u.replace(".", "")
    return bytes.fromhex(u)


def umid_bytes_to_dash(b: bytes) -> str:
    """Convert 32-byte UMID into Avid-style dash MobID."""
    if len(b) != UMID_LEN:
        raise ValueError("UMID must be 32 bytes")
    return "{}-{}-{}-{}-{}".format(
        b[0:13].hex(),
        b[13:16].hex(),
        b[16:24].hex(),
        b[24:30].hex(),
        b[30:32].hex(),
    )


def dash_to_umid_bytes(s: str) -> bytes:
    parts = s.strip().lower().split("-")
    if len(parts) != 5:
        raise ValueError(f"Not a dash MobID: {s}")
    return bytes.fromhex("".join(parts))


def material_uuid_le_from_umid(umid: bytes) -> bytes:
    """Derive the 16-byte 'material' UUID key in little-endian form for MMOB lookup."""
    if len(umid) != UMID_LEN:
        raise ValueError("UMID must be 32 bytes")
    m = umid[16:32]
    return m[0:4][::-1] + m[4:6][::-1] + m[6:8][::-1] + m[8:16]


def find_avb_files(project_root: Path) -> List[Path]:
    avbs: List[Path] = []
    for p in project_root.rglob("*.avb"):
        if p.is_file() and not p.name.startswith("._"):
            avbs.append(p)
    return sorted(avbs)


def collect_panel_mastermob_dash_ids(project_root: Path) -> List[str]:
    """
    Collect MasterMob dash MobIDs from bin items (PanelSDK-like).

    In your .avb, BinItem.name is None; the visible clip names live on BinItem.mob.name.
    We therefore iterate bin.items and take only item.mob where mob_type == 'MasterMob'.
    We also prefer the PanelSDK-style UMID family used in your sample; if that yields none,
    we fall back to all MasterMobs.
    """
    ordered: List[str] = []
    seen = set()

    for avb_path in find_avb_files(project_root):
        try:
            with avb.open(str(avb_path)) as f:
                bin_obj = f.content
                items = getattr(bin_obj, "items", None)
                if not items:
                    continue

                master_candidates: List[bytes] = []
                panelstyle_candidates: List[bytes] = []

                for it in items:
                    mob = getattr(it, "mob", None)
                    if mob is None:
                        continue
                    if getattr(mob, "mob_type", None) != "MasterMob":
                        continue

                    mob_id = getattr(mob, "mob_id", None)
                    urn = getattr(mob_id, "urn", "")
                    if not urn:
                        continue

                    try:
                        umid = urn_to_umid_bytes(urn)
                    except Exception:
                        continue

                    master_candidates.append(umid)
                    if is_panelstyle_umid_bytes(umid):
                        panelstyle_candidates.append(umid)

                picked = panelstyle_candidates if panelstyle_candidates else master_candidates

                for umid in picked:
                    dash = umid_bytes_to_dash(umid)
                    if dash not in seen:
                        seen.add(dash)
                        ordered.append(dash)

        except Exception as e:
            print(f"WARN: failed to parse bin {avb_path}: {e}", file=sys.stderr)

    return ordered


def parse_pmr_file(pmr_path: Path) -> Dict[str, str]:
    """
    Parse msmFMID.pmr to map UMID URN -> MXF full path.

    Observed record pattern:
      UMID (32 bytes)
      name_len (uint16 little endian)
      name bytes (ASCII/UTF-8)
    """
    data = pmr_path.read_bytes()
    out: Dict[str, str] = {}

    i = 0
    n = len(data)
    while i + UMID_LEN + 2 <= n:
        if not is_umid_at(data, i):
            i += 1
            continue

        umid = data[i : i + UMID_LEN]
        name_len = int.from_bytes(data[i + UMID_LEN : i + UMID_LEN + 2], "little", signed=False)

        name_start = i + UMID_LEN + 2
        name_end = name_start + name_len
        if name_len <= 0 or name_len > 512 or name_end > n:
            i += 1
            continue

        name_bytes = data[name_start:name_end]
        nb = name_bytes.lstrip(b"\x00")
        if not nb or b"\x00" in nb:
            i += 1
            continue
        if not FILENAME_RE.fullmatch(nb):
            i += 1
            continue

        fname = nb.decode("ascii", errors="strict")
        urn = umid_bytes_to_urn(umid)
        out[urn] = str((pmr_path.parent / fname).resolve())

        i = name_end

    return out


def build_pmr_map(media_roots: Sequence[Path]) -> Dict[str, str]:
    """UMID URN -> full MXF path"""
    m: Dict[str, str] = {}
    for root in media_roots:
        for pmr in root.rglob("msmFMID.pmr"):
            if not pmr.is_file():
                continue
            try:
                sub = parse_pmr_file(pmr)
            except Exception as e:
                print(f"WARN: failed to parse PMR {pmr}: {e}", file=sys.stderr)
                continue
            for k, v in sub.items():
                m.setdefault(k, v)
    return m


def iter_mdb_files(media_roots: Sequence[Path]) -> Iterable[Path]:
    for root in media_roots:
        for p in root.rglob("msmMMOB.mdb"):
            if p.is_file():
                yield p


def find_all(data: bytes, needle: bytes) -> List[int]:
    out: List[int] = []
    start = 0
    while True:
        j = data.find(needle, start)
        if j == -1:
            break
        out.append(j)
        start = j + 1
    return out


def map_mob_to_mxf_paths(
    mob_umid_bytes: bytes,
    mdb_paths: Sequence[Path],
    pmr_map: Dict[str, str],
) -> List[str]:
    """
    Return MXF paths associated with this MobID using only MDB+PMR.

    - compute material uuid LE key
    - for each occurrence in each MDB, scan +/- MMOB_SCAN_WINDOW for UMIDs
    - keep only UMIDs that exist in pmr_map (FMIDs)
    - rank by (frequency desc, min distance asc, path)
    """
    key16 = material_uuid_le_from_umid(mob_umid_bytes)

    counts: Dict[str, int] = defaultdict(int)
    mindist: Dict[str, int] = {}

    for mdb in mdb_paths:
        try:
            data = mdb.read_bytes()
        except Exception as e:
            print(f"WARN: failed to read MDB {mdb}: {e}", file=sys.stderr)
            continue

        hits = find_all(data, key16)
        if not hits:
            continue

        for h in hits:
            lo = max(0, h - MMOB_SCAN_WINDOW)
            hi = min(len(data), h + MMOB_SCAN_WINDOW)
            s = data[lo:hi]

            for off in range(0, len(s) - UMID_LEN + 1):
                if not is_umid_at(s, off):
                    continue
                urn = umid_bytes_to_urn(s[off : off + UMID_LEN])
                if urn not in pmr_map:
                    continue

                counts[urn] += 1
                d = abs((lo + off) - h)
                prev = mindist.get(urn)
                if prev is None or d < prev:
                    mindist[urn] = d

    if not counts:
        return []

    ranked_urns = sorted(
        counts.keys(),
        key=lambda u: (-counts[u], mindist.get(u, 10**9), pmr_map[u]),
    )

    # De-dupe paths preserving rank order
    out_paths: List[str] = []
    seen = set()
    for u in ranked_urns:
        p = pmr_map[u]
        if p not in seen:
            seen.add(p)
            out_paths.append(p)

    return out_paths


def main(argv: Optional[Sequence[str]] = None) -> int:
    ap = argparse.ArgumentParser(
        description="DB-only Avid MasterClip MobID -> MXF path mapping (bins + msmMMOB.mdb + msmFMID.pmr)."
    )
    ap.add_argument("--project", required=True, help="Avid Project Folder path (contains .avb bins).")
    ap.add_argument(
        "--media",
        action="append",
        required=True,
        help="Path to an Avid MediaFiles folder (repeatable).",
    )
    ap.add_argument("--out", required=True, help="Output JSON file path.")
    ap.add_argument("--stdout", action="store_true", help="Print JSON to stdout instead of writing to --out.")

    args = ap.parse_args(argv)

    project_root = Path(args.project)
    media_roots = [Path(p) for p in args.media]
    out_path = Path(args.out)

    if not project_root.is_dir():
        print(f"ERROR: --project is not a directory: {project_root}", file=sys.stderr)
        return 2
    for m in media_roots:
        if not m.is_dir():
            print(f"ERROR: --media is not a directory: {m}", file=sys.stderr)
            return 2

    # 1) PanelSDK-like master clip mobIds from bins
    master_dash_ids = collect_panel_mastermob_dash_ids(project_root)
    if not master_dash_ids:
        print("ERROR: no candidate MasterClip MobIDs found in bins", file=sys.stderr)
        return 2

    # 2) FMID -> MXF path from PMR
    pmr_map = build_pmr_map(media_roots)
    if not pmr_map:
        print("ERROR: no PMR entries found (msmFMID.pmr)", file=sys.stderr)
        return 2

    # 3) MDB files
    mdb_paths = list(iter_mdb_files(media_roots))
    if not mdb_paths:
        print("ERROR: no msmMMOB.mdb files found", file=sys.stderr)
        return 2

    result: Dict[str, List[str]] = {}

    for dash in master_dash_ids:
        try:
            umid_bytes = dash_to_umid_bytes(dash)
        except Exception:
            continue

        paths = map_mob_to_mxf_paths(
            mob_umid_bytes=umid_bytes,
            mdb_paths=mdb_paths,
            pmr_map=pmr_map,
        )

        if paths:
            # In your test data this is exactly one MXF per master mob.
            result[dash] = paths

    payload = json.dumps(result, indent=2, ensure_ascii=False)

    if args.stdout:
        print(payload)
    else:
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(payload, encoding="utf-8")
        print(str(out_path.resolve()))

    return 0


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

How to get the Media File Path(s) of a Master Clip? #26

Description

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions