open-standards/wai/live-conformance/ertmp_verify.py

#!/usr/bin/env python3
"""Second implementation of WAI's Enhanced RTMP ingest (extensions/live-ingest.md §5, §7 and
§12), written from that text and from Enhanced RTMP v2 (v2-2026-01-31-r2) and the texts it
cites. It shares no code with the reference (joule-moq-ingest).

For each session of live-conformance/ertmp/vectors.json (a grant's powers and profile, and
RTMP audio, video, data and aggregate messages with their timestamps and arrivals) it reads
every message body (§12.4), binds each RTMP track to a room track (§12.3), builds each
frame's payload in its capability's framing (§12.4, §5.1), maps it to the room clock (§12.5,
§7.1), groups it (§5.2, §5.3), and requires every object of every room track the corpus
states — its group, object, `live` member (t, rtp, rate, key, cto), capability and payload —
and the counts of what was read and not carried.

    python3 -I ertmp_verify.py [--self-test] <live-conformance/ertmp>

With --self-test it then requires every mutated copy of the corpus (an object's time, media
time, group, object id, payload byte or key flag, a count, the messages' arrivals) to fail.

Exit status 0 only if every case agrees. Python 3.8 or later; standard library only.
"""
import json
import os
import struct
import sys

HERE = os.path.dirname(os.path.abspath(__file__))

CAP = {b"avc1": "wai.video.avc", b"hvc1": "wai.video.hevc", b"av01": "wai.video.av1", b"vp09": "wai.video.vp9",
       b"vp08": "wai.video.vp8", b"Opus": "wai.audio.opus.packet", b"fLaC": "wai.audio.flac", b"mp4a": "wai.audio.aac"}
FAMILY = {b"avc1": "avc1", b"hvc1": "hvc1", b"av01": "av01", b"vp09": "vp09", b"vp08": "vp8", b"Opus": "opus", b"fLaC": "flac", b"mp4a": "mp4a"}
VIDEO = {b"avc1", b"hvc1", b"av01", b"vp09", b"vp08"}
AAC_RATES = [96000, 88200, 64000, 48000, 44100, 32000, 24000, 22050, 16000, 12000, 11025, 8000, 7350]


class Bad(Exception):
    pass


# ── AMF0 (E-RTMP's [AMF0]) ───────────────────────────────────────────────────────

def amf0(b):
    p = [0]

    def take(n):
        if p[0] + n > len(b):
            raise Bad("amf0")
        s = b[p[0]:p[0] + n]
        p[0] += n
        return s

    def pairs():
        out = []
        while True:
            n = struct.unpack(">H", take(2))[0]
            if n == 0:
                if take(1) != b"\x09":
                    raise Bad("amf0 end")
                return out
            k = take(n).decode()
            out.append((k, value()))

    def value():
        m = take(1)[0]
        if m == 0:
            return struct.unpack(">d", take(8))[0]
        if m == 1:
            return take(1)[0] != 0
        if m == 2:
            return take(struct.unpack(">H", take(2))[0]).decode()
        if m == 3:
            return dict(pairs())
        if m in (5, 6):
            return None
        if m == 8:
            take(4)
            return dict(pairs())
        if m == 10:
            return [value() for _ in range(struct.unpack(">I", take(4))[0])]
        if m == 12:
            return take(struct.unpack(">I", take(4))[0]).decode()
        raise Bad("amf0 marker")

    out = []
    while p[0] < len(b):
        out.append(value())
    return out


def track_meta(o, video):
    def num(*ks):
        for k in ks:
            v = o.get(k)
            if isinstance(v, float) and not isinstance(v, bool) and v >= 0:
                return v
        return None
    if video:
        return {"width": int(num("width") or 0), "height": int(num("height") or 0), "fps": num("framerate", "videoframerate")}
    ch = num("audiochannels", "channels")
    if ch is None:
        st = o.get("stereo")
        ch = 2 if st is True else 1 if st is False else 0
    return {"rate": int(num("audiosamplerate", "samplerate") or 0), "channels": int(ch)}


def metadata(values):
    """§12.2: @setDataFrame onMetaData, or onMetaData; None for another call."""
    if values and values[0] == "@setDataFrame":
        values = values[1:]
    if len(values) < 2 or values[0] != "onMetaData" or not isinstance(values[1], dict):
        return None
    o = values[1]
    m = {(True, 0): track_meta(o, True), (False, 0): track_meta(o, False)}
    for key, video in (("videoTrackIdInfoMap", True), ("audioTrackIdInfoMap", False)):
        for k, e in (o.get(key) or {}).items() if isinstance(o.get(key), dict) else []:
            if k.isdigit() and 0 < int(k) < 256 and isinstance(e, dict):
                m[(video, int(k))] = track_meta(e, video)
    return m


# ── FLV bodies (E-RTMP Enhanced Video, Enhanced Audio) ────────────────────────────

class R:
    def __init__(self, b):
        self.b, self.p = b, 0

    def take(self, n):
        if self.p + n > len(self.b):
            raise Bad("tag_truncated")
        s = self.b[self.p:self.p + n]
        self.p += n
        return s

    def u(self, n):
        return int.from_bytes(self.take(n), "big")

    def rest(self):
        s = self.b[self.p:]
        self.p = len(self.b)
        return s


def modex(r, pt):
    nano = 0
    while pt == 7:
        size = r.u(1) + 1
        if size == 256:
            size = r.u(2) + 1
        data = r.take(size)
        x = r.u(1)
        if x >> 4 == 0:
            if len(data) < 3:
                raise Bad("tag_modex")
            nano = int.from_bytes(data[:3], "big")
            if nano > 999_999:
                raise Bad("tag_modex")
        pt = x & 0x0F
    return pt, nano


def multitrack(r, mt, shared):
    out = []
    while True:
        f = shared if shared is not None else r.take(4)
        tid = r.u(1)
        if mt == 0:
            out.append((f, tid, r.rest()))
            return out
        out.append((f, tid, r.take(r.u(3))))
        if r.p >= len(r.b):
            return out


def unit(video, tid, f, kind, key, cts, nano, data):
    return {"video": video, "track": tid, "fourcc": f, "kind": kind, "key": key, "cts": cts, "nano": nano, "data": data}


def video_body(b):
    r = R(b)
    b0 = r.u(1)
    if not b0 & 0x80:
        ft, codec = b0 >> 4, b0 & 0x0F
        if ft == 5:
            return [unit(True, 0, bytes([0, 0, 0, codec]), "other", False, 0, 0, b"")]
        if codec != 7:
            return [unit(True, 0, bytes([0, 0, 0, codec]), "frames", ft == 1, 0, 0, r.rest())]
        pt = r.u(1)
        kind = {0: "start", 1: "frames", 2: "end"}.get(pt)
        if kind is None:
            raise Bad("tag_packet_type")
        cts = r.u(3)
        cts = cts - (1 << 24) if cts & 0x800000 else cts
        return [unit(True, 0, b"avc1", kind, ft == 1, cts if kind == "frames" else 0, 0, r.rest())]
    ft = (b0 >> 4) & 7
    pt, nano = modex(r, b0 & 0x0F)
    if pt != 4 and ft == 5:
        r.u(1)
        return [unit(True, 0, bytes(4), "other", False, 0, nano, b"")]
    if pt == 6:
        x = r.u(1)
        mt, pt = x >> 4, x & 0x0F
        if pt == 6 or mt > 2:
            raise Bad("tag_multitrack")
        tracks = multitrack(r, mt, r.take(4) if mt != 2 else None)
    else:
        tracks = [(r.take(4), 0, r.rest())]
    out = []
    for f, tid, body in tracks:
        t = R(body)
        if pt == 0:
            kind, cts = "start", 0
        elif pt == 1 and f in (b"avc1", b"hvc1", b"vvc1"):
            kind, cts = "frames", t.u(3)
            cts = cts - (1 << 24) if cts & 0x800000 else cts
        elif pt in (1, 3):
            kind, cts = "frames", 0
        elif pt == 2:
            kind, cts = "end", 0
        elif pt in (4, 5):
            kind, cts = "other", 0
        else:
            raise Bad("tag_packet_type")
        out.append(unit(True, tid, f, kind, ft == 1, cts, nano, t.rest()))
    return out


def audio_body(b):
    if not b:
        return [unit(False, 0, bytes(4), "other", False, 0, 0, b"")]
    r = R(b)
    b0 = r.u(1)
    fmt = b0 >> 4
    if fmt != 9:
        if fmt != 10:
            return [unit(False, 0, bytes([0, 0, 0, fmt]), "frames", False, 0, 0, r.rest())]
        kind = {0: "start", 1: "frames"}.get(r.u(1))
        if kind is None:
            raise Bad("tag_packet_type")
        return [unit(False, 0, b"mp4a", kind, False, 0, 0, r.rest())]
    pt, nano = modex(r, b0 & 0x0F)
    if pt == 5:
        x = r.u(1)
        mt, pt = x >> 4, x & 0x0F
        if pt == 5 or mt > 2:
            raise Bad("tag_multitrack")
        tracks = multitrack(r, mt, r.take(4) if mt != 2 else None)
    else:
        tracks = [(r.take(4), 0, r.rest())]
    kind = {0: "start", 1: "frames", 2: "end", 4: "other"}.get(pt)
    if kind is None:
        raise Bad("tag_packet_type")
    return [unit(False, tid, f, kind, False, 0, nano, body) for f, tid, body in tracks]


# ── Configurations and payloads (§12.4) ─────────────────────────────────────────

def config(f, b):
    try:
        if f == b"avc1":
            if b[0] != 1:
                return None
            ls = (b[4] & 3) + 1
            r = R(b)
            r.take(5)
            sets = [r.take(r.u(2)) for _ in range(r.u(1) & 0x1F)]
            sets += [r.take(r.u(2)) for _ in range(r.u(1))]
            return None if ls == 3 else {"ls": ls, "sets": sets}
        if f == b"hvc1":
            if b[0] != 1:
                return None
            r = R(b)
            r.take(21)
            ls = (r.u(1) & 3) + 1
            sets = []
            for _ in range(r.u(1)):
                r.u(1)
                for _ in range(r.u(2)):
                    sets.append(r.take(r.u(2)))
            return None if ls == 3 else {"ls": ls, "sets": sets}
        if f == b"av01":
            if len(b) < 4 or b[0] != 0x81:
                return None
            obus = av1_obus(b[4:])
            sh = next((whole for t, whole, _ in (obus or []) if t == 1), None)
            return {"seq": sh}
        if f in (b"vp09", b"vp08"):
            return {}
        if f == b"Opus":
            if not b:
                return {"streams": 1}
            if len(b) < 19 or b[:8] != b"OpusHead":
                return None
            return {"streams": 1 if b[18] == 0 else b[19]}
        if f == b"fLaC":
            if len(b) == 34:
                b = b"fLaC\x80\x00\x00\x22" + b
            if len(b) < 42 or b[:4] != b"fLaC" or b[4] & 0x7F != 0 or int.from_bytes(b[5:8], "big") != 34:
                return None
            h = bytearray(b[:42])
            h[4] |= 0x80
            return {"header": bytes(h)}
        if f == b"mp4a":
            bits = Bits(b)
            def aot():
                a = bits.f(5)
                return 32 + bits.f(6) if a == 31 else a
            obj = aot()
            sfi = bits.f(4)
            if sfi == 15:
                return None
            ch = bits.f(4)
            if obj in (5, 29):
                if bits.f(4) == 15:
                    bits.f(24)
                obj = aot()
            if not 1 <= obj <= 4 or sfi > 12 or not 1 <= ch <= 7:
                return None
            return {"profile": obj - 1, "sfi": sfi, "ch": ch}
    except (Bad, IndexError):
        return None
    return None


class Bits:
    def __init__(self, b):
        self.b, self.i = b, 0

    def f(self, n):
        v = 0
        for _ in range(n):
            if self.i // 8 >= len(self.b):
                raise Bad("bits")
            v = (v << 1) | ((self.b[self.i // 8] >> (7 - self.i % 8)) & 1)
            self.i += 1
        return v

    def uvlc(self):
        z = 0
        while self.f(1) == 0:
            z += 1
            if z >= 32:
                return (1 << 32) - 1
        return self.f(z) + (1 << z) - 1


def leb128(b, p):
    v = 0
    for i in range(8):
        if p + i >= len(b):
            return None
        v |= (b[p + i] & 0x7F) << (7 * i)
        if not b[p + i] & 0x80:
            return v, i + 1
    return None


def av1_obus(b):
    out, p = [], 0
    while p < len(b):
        s, h = p, b[p]
        p += 1 + (1 if h & 4 else 0)
        if not h & 2 or p > len(b):
            return None
        x = leb128(b, p)
        if x is None:
            return None
        p += x[1]
        if p + x[0] > len(b):
            return None
        out.append(((h >> 3) & 0x0F, b[s:p + x[0]], b[p:p + x[0]]))
        p += x[0]
    return out


def annexb(data, ls, sets, key, is_set):
    nals, p = [], 0
    while p < len(data):
        if p + ls > len(data):
            return None
        n = int.from_bytes(data[p:p + ls], "big")
        p += ls
        if p + n > len(data):
            return None
        if n:
            nals.append(data[p:p + n])
        p += n
    if not nals:
        return None
    o = b""
    if key and not any(is_set(n) for n in nals):
        o += b"".join(b"\x00\x00\x00\x01" + s for s in sets)
    return o + b"".join(b"\x00\x00\x00\x01" + n for n in nals)


def frame_payload(f, cfg, key, data):
    """(payload, None) or (None, reason)."""
    if f in (b"avc1", b"hvc1", b"mp4a", b"fLaC") and cfg is None:
        return None, "no_sequence_start"
    if f == b"avc1":
        x = annexb(data, cfg["ls"], cfg["sets"], key, lambda n: (n[0] & 0x1F) in (7, 8))
        return (x, None) if x else (None, "frame_unreadable")
    if f == b"hvc1":
        x = annexb(data, cfg["ls"], cfg["sets"], key, lambda n: 32 <= (n[0] >> 1) & 0x3F <= 34)
        return (x, None) if x else (None, "frame_unreadable")
    if f == b"av01":
        obus = av1_obus(data)
        if obus is None:
            return None, "frame_unreadable"
        o = b""
        if key and not any(t == 1 for t, _, _ in obus) and cfg and cfg.get("seq"):
            o += cfg["seq"]
        o += b"".join(w for t, w, _ in obus if t != 2)
        return (o, None) if o else (None, "frame_unreadable")
    if f in (b"vp09", b"vp08"):
        return (data, None) if data else (None, "frame_unreadable")
    if f == b"Opus":
        if cfg and cfg["streams"] > 1:
            return None, "frame_not_carried"
        return (data, None) if data else (None, "frame_unreadable")
    if f == b"mp4a":
        n = len(data) + 7
        if n > 0x1FFF:
            return None, "frame_not_carried"
        ch = cfg["ch"]
        h = bytes([0xFF, 0xF1, (cfg["profile"] << 6) | (cfg["sfi"] << 2) | (ch >> 2), ((ch & 3) << 6) | ((n >> 11) & 3), (n >> 3) & 0xFF, ((n & 7) << 5) | 0x1F, 0xFC])
        return h + data, None
    if f == b"fLaC":
        if len(data) < 2 or data[0] != 0xFF or data[1] & 0xFC != 0xF8:
            return None, "frame_unreadable"
        return cfg["header"] + data, None
    return None, "frame_unreadable"


def is_idr(au):
    return any(n and n[0] & 0x1F == 5 for n in nals_of(au))


def is_irap(au):
    return any(n and 16 <= (n[0] >> 1) & 0x3F <= 23 for n in nals_of(au))


def nals_of(au):
    """The NAL units of an Annex B unit this verifier built: each after 00 00 00 01."""
    parts = au.split(b"\x00\x00\x00\x01")
    return [p for p in parts[1:]]


def vp9_size(fr):
    b = Bits(fr)
    try:
        if b.f(2) != 2:
            return None
        lo, hi = b.f(1), b.f(1)
        prof = (hi << 1) | lo
        if prof == 3:
            b.f(1)
        if b.f(1):
            return None
        ftype = b.f(1)
        b.f(1)
        b.f(1)
        if ftype != 0 or b.f(24) != 0x498342:
            return None
        if prof >= 2:
            b.f(1)
        cs = b.f(3)
        if cs != 7:
            b.f(1)
            if prof in (1, 3):
                b.f(3)
        elif prof in (1, 3):
            b.f(1)
        return b.f(16) + 1, b.f(16) + 1
    except Bad:
        return None


def vp8_size(fr):
    if len(fr) < 10 or fr[0] & 1 or fr[3:6] != b"\x9d\x01\x2a":
        return None
    return struct.unpack("<H", fr[6:8])[0] & 0x3FFF, struct.unpack("<H", fr[8:10])[0] & 0x3FFF


def av1_size(tu):
    obus = av1_obus(tu)
    sh = next((p for t, _, p in (obus or []) if t == 1), None)
    if sh is None:
        return None
    b = Bits(sh)
    try:
        b.f(3)
        b.f(1)
        if b.f(1):
            b.f(5)
        else:
            dm, bdl = 0, 0
            if b.f(1):
                b.f(32)
                b.f(32)
                if b.f(1):
                    b.uvlc()
                dm = b.f(1)
                if dm:
                    bdl = b.f(5) + 1
                    b.f(32)
                    b.f(5)
                    b.f(5)
            idd = b.f(1)
            for _ in range(b.f(5) + 1):
                b.f(12)
                if b.f(5) > 7:
                    b.f(1)
                if dm and b.f(1):
                    b.f(bdl)
                    b.f(bdl)
                    b.f(1)
                if idd and b.f(1):
                    b.f(4)
        wb, hb = b.f(4) + 1, b.f(4) + 1
        return b.f(wb) + 1, b.f(hb) + 1
    except Bad:
        return None


def framing(cap, data, size, ticks, delta):
    """§5.1's framings: IVF for VP8 and VP9, SPEC §5's AV1 framing; the others are their payload."""
    if cap in ("wai.video.vp8", "wai.video.vp9"):
        w, h = size
        four = b"VP80" if cap == "wai.video.vp8" else b"VP90"
        return b"DKIF" + struct.pack("<HH", 0, 32) + four + struct.pack("<HHIIII", min(w, 65535), min(h, 65535), 90000, 1, 1, 0) + struct.pack("<IQ", len(data), ticks) + data
    if cap == "wai.video.av1":
        w, h = size
        tu = b"\x12\x00" + data
        return struct.pack("<IIIII", 1, h, w, 90000, max(1, min(delta, 0xFFFFFFFF))) + struct.pack("<I", len(tu)) + tu
    return data


# ── A track (§5.2, §5.3, §7.1) ───────────────────────────────────────────────────

class Track:
    def __init__(self, media, layer, prm):
        self.media, self.layer, self.prm = media, layer, prm
        self.anchor, self.last, self.group, self.next_id, self.last_ticks = None, 0, None, None, None
        self.size, self.cap = None, None
        self.objects = []

    def map(self, r, arrival):
        """§7.1: anchored at the first frame; re-anchored past the skew bound, or below R0."""
        skew = self.prm["skew_us"]
        if self.anchor is None or r < self.anchor[1]:
            self.anchor = (arrival, r)
            t = arrival
        else:
            t = self.anchor[0] + (r - self.anchor[1]) * 1_000_000 // 1_000_000
            if abs(t - arrival) > skew:
                self.anchor = (arrival, r)
                t = arrival
        self.last = max(self.last, t)
        return self.last

    def open(self, start):
        gid = self.next_id if self.next_id is not None else start
        self.next_id = gid + 1
        self.group = {"id": gid, "start": start, "n": 0}

    def frame(self, cap, payload, r, arrival, key, cto, counts, hint_size):
        video = self.media == "video"
        if cap != self.cap:
            self.group = None
            self.cap = cap
            self.size = None
        t = self.map(r, arrival)
        if video:
            if cap == "wai.video.avc":
                key = key or is_idr(payload)
            elif cap == "wai.video.hevc":
                key = key or is_irap(payload)
            if key:
                sz = {"wai.video.vp8": vp8_size, "wai.video.vp9": vp9_size, "wai.video.av1": av1_size}.get(cap)
                if sz is not None:
                    got = sz(payload)
                    if got is None and cap == "wai.video.av1":
                        got = self.size
                    if got is None:
                        counts["dropped_unreadable"] = counts.get("dropped_unreadable", 0) + 1
                        return
                    self.size = got
                self.open(t)
            elif self.group is None:
                counts["dropped_before_keyframe"] = counts.get("dropped_before_keyframe", 0) + 1
                return
        else:
            win = self.prm["audio_group_us"]
            ends = self.group["start"] + win if self.group else None
            if ends is None or t >= ends:
                start = ends if ends is not None and t < ends + win else t
                self.open(start)
        ticks = t * 90000 // 1_000_000
        delta = 3000 if self.last_ticks is None else max(0, ticks - self.last_ticks)
        self.last_ticks = ticks
        g = self.group
        live_key = key if video else None
        o = {"group": g["id"], "object": g["n"], "t": t, "rtp": r, "rate": 1_000_000, "key": live_key,
             "cto": cto if video and cto else None, "capability": cap,
             "payload": framing(cap, payload, self.size or (0, 0), ticks, delta).hex()}
        g["n"] += 1
        self.objects.append(o)
        if video and t - g["start"] > self.prm["max_video_group_us"]:
            counts["group_long"] = counts.get("group_long", 0) + 1


def run(s):
    g = s["grant"]
    prof = g["video_profile"]
    layout = []
    if g["audio"]:
        layout.append(("audio", 0))
    if g["video"]:
        layout += [("video", l) for l in range(len(prof["layers"]) if prof else 1)]
    tracks = {k: Track(k[0], k[1], s["params"]) for k in layout}
    meta, binds, counts = {}, {}, {}

    def count(k):
        counts[k] = counts.get(k, 0) + 1

    def bind(u):
        f = u["fourcc"]
        if f not in CAP:
            return "codec_not_carried"
        media = "video" if u["video"] else "audio"
        if not g[media]:
            return "publish_not_permitted"
        p = prof if u["video"] else None
        if p is not None and FAMILY[f] not in p["families"]:
            return "codec_not_permitted"
        if not u["video"]:
            if u["track"] != 0:
                return "audio_track_not_default"
            layer = 0
        else:
            tm = meta.get((True, u["track"]))
            if p is not None and tm and tm["width"] and tm["height"] and tm["fps"] is not None:
                fps = int(tm["fps"] + 0.5)
                hit = [i for i, (w, h, fr) in enumerate(p["layers"]) if (w, h, fr) == (tm["width"], tm["height"], fps)]
                if not hit:
                    return "track_no_layer"
                layer = hit[0]
            elif u["track"] == 0:
                layer = 0
            else:
                return "track_no_layer"
        if (media, layer) not in tracks:
            return "track_no_layer"
        if any(isinstance(b, dict) and (b["media"], b["layer"]) == (media, layer) for b in binds.values()):
            return "track_layer_taken"
        return {"media": media, "layer": layer, "fourcc": f, "config": None, "high": 0, "last": None}

    def on_unit(u, ts, arrival):
        k = (u["video"], u["track"])
        if k not in binds:
            binds[k] = bind(u)
        b = binds[k]
        if isinstance(b, str):
            return count(b)
        tr = tracks[(b["media"], b["layer"])]
        if u["kind"] == "start":
            c = config(u["fourcc"], u["data"])
            if c is None:
                return count("sequence_start_unreadable")
            b["config"], b["fourcc"] = c, u["fourcc"]
            return
        if u["kind"] == "end":
            tr.group = None
            return
        if u["kind"] == "other" or not u["data"]:
            return count("packet_not_carried")
        if u["fourcc"] != b["fourcc"]:
            return count("codec_changed_without_sequence_start")
        payload, why = frame_payload(u["fourcc"], b["config"], u["key"], u["data"])
        if why:
            return count(why)
        if b["last"] is not None and ts < b["last"] and b["last"] - ts > 1 << 31:
            b["high"] += 1 << 32
        b["last"] = ts
        r = (b["high"] + ts) * 1000 + u["nano"] // 1000
        tr.frame(CAP[u["fourcc"]], payload, r, arrival, u["key"], u["cts"] * 1000 if u["video"] else None, counts, None)

    def on_media(video, ts, body, arrival):
        try:
            units = video_body(body) if video else audio_body(body)
        except Bad as e:
            return count(str(e) if str(e).startswith("tag_") else "tag_truncated")
        for u in units:
            on_unit(u, ts, arrival)

    for m in s["messages"]:
        body = bytes.fromhex(m["payload"])
        if m["type"] in (8, 9):
            on_media(m["type"] == 9, m["ts"], body, m["arrival_us"])
        elif m["type"] == 18:
            md = metadata(amf0(body))
            if md is not None:
                meta.clear()
                meta.update(md)
                count("metadata")
        elif m["type"] == 22:
            p, first = 0, None
            while p + 11 <= len(body):
                ty = body[p] & 0x1F
                size = int.from_bytes(body[p + 1:p + 4], "big")
                sub = (body[p + 7] << 24) | int.from_bytes(body[p + 4:p + 7], "big")
                first = sub if first is None else first
                at = (m["ts"] + sub - first) & 0xFFFFFFFF
                if ty in (8, 9):
                    on_media(ty == 9, at, body[p + 11:p + 11 + size], m["arrival_us"])
                elif ty == 18:
                    md = metadata(amf0(body[p + 11:p + 11 + size]))
                    if md is not None:
                        meta.clear()
                        meta.update(md)
                        count("metadata")
                p += 11 + size + 4
    out = []
    for (media, layer), tr in tracks.items():
        if tr.objects:
            out.append({"media": media, "layer": layer, "objects": tr.objects})
    return out, counts


def check(v):
    """Every check of the corpus `v`: the failures."""
    fails = []
    for s in v["sessions"]:
        try:
            tracks, counts = run(s)
        except Exception as e:  # a stated value that breaks the reading is a failure too
            fails.append(f"{s['name']}: {e!r}")
            continue
        want = {(t["media"], t["layer"]): t["objects"] for t in s["tracks"]}
        got = {(t["media"], t["layer"]): t["objects"] for t in tracks}
        if sorted(want) != sorted(got):
            fails.append(f"{s['name']}: tracks {sorted(got)} against {sorted(want)}")
        for k in want:
            for i, (a, b) in enumerate(zip(got.get(k, []), want[k])):
                if a != b:
                    diff = sorted(x for x in b if a.get(x) != b.get(x))
                    fails.append(f"{s['name']} {k} object {i}: {diff}")
                    break
            if len(got.get(k, [])) != len(want[k]):
                fails.append(f"{s['name']} {k}: {len(got.get(k, []))} objects against {len(want[k])}")
        if counts != s["counts"]:
            fails.append(f"{s['name']}: counts {counts} against {s['counts']}")
    return fails


def mutations(v):
    """Copies of the corpus, each with one stated or sent value changed: each must fail."""
    import copy
    for i, s in enumerate(v["sessions"]):
        for j, t in enumerate(s["tracks"]):
            for field in ("t", "rtp", "group", "object"):
                m = copy.deepcopy(v)
                m["sessions"][i]["tracks"][j]["objects"][-1][field] += 1
                yield f"{s['name']} {t['media']}/{t['layer']}: the last object's {field}", m
            m = copy.deepcopy(v)
            o = m["sessions"][i]["tracks"][j]["objects"][0]
            p = bytearray(bytes.fromhex(o["payload"]))
            p[len(p) // 2] ^= 0x01
            o["payload"] = p.hex()
            yield f"{s['name']} {t['media']}/{t['layer']}: a payload byte", m
            if t["media"] == "video":
                m = copy.deepcopy(v)
                m["sessions"][i]["tracks"][j]["objects"][0]["key"] = not t["objects"][0]["key"]
                yield f"{s['name']} video/{t['layer']}: the key flag", m
        if s["counts"]:
            m = copy.deepcopy(v)
            k = sorted(s["counts"])[0]
            m["sessions"][i]["counts"][k] += 1
            yield f"{s['name']}: the count {k}", m
        if s["tracks"] and len(s["messages"]) > 1:
            # Every message after the first arrives 600 ms later: past the skew bound, so
            # the mapping re-anchors and the objects' times move.
            m = copy.deepcopy(v)
            for msg in m["sessions"][i]["messages"][1:]:
                msg["arrival_us"] += 600_000
            yield f"{s['name']}: the arrivals", m


def main(args):
    self_test = "--self-test" in args
    args = [a for a in args if a != "--self-test"]
    d = args[0] if args else os.path.join(HERE, "ertmp")
    v = json.load(open(os.path.join(d, "vectors.json")))
    fails = check(v)
    for f in fails:
        print("FAIL", f)
    objs = sum(len(t["objects"]) for s in v["sessions"] for t in s["tracks"])
    print(f"ertmp: {len(v['sessions'])} sessions, {objs} objects; {'every one agrees' if not fails else str(len(fails)) + ' disagree'}")
    if fails:
        return 1
    if self_test:
        survived, total = [], 0
        for what, m in mutations(v):
            total += 1
            if not check(m):
                survived.append(what)
        for w in survived:
            print("SURVIVED", w)
        print(f"ertmp self-test: {total - len(survived)} of {total} mutated corpora refused")
        return 1 if survived else 0
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))