space-tapePublic

Open-source X Spaces transcription.

Star

196 lines · 6.1 KB

python
1"""Parse Hydra ID3 tags muxed into an X Spaces replay.
2
3X's audio stack writes ID3v2 TXXX frames about once a second:
4
5- JSONMetadata / TIT3 — NTP clock
6- HydraParticipants — people on stage (not listeners, not the host)
7- HydraAudioLevel — int array, one level per on-stage slot
8
9Index mapping that works in practice:
10
11- HydraAudioLevel[0] = host
12- HydraAudioLevel[i + 1] = HydraParticipants[i]
13- level 0 = muted / silence
14- level ≳ 8 = unmuted and making sound
15
16Parse the **raw** yt-dlp file. ffmpeg remux to .aac / .wav drops every tag.
17"""
18
19from __future__ import annotations
20
21import json
22from pathlib import Path
23from typing import Iterable
24
25LEVEL_ON = 8
26KNOWN_DESC = {"JSONMetadata", "HydraParticipants", "HydraAudioLevel"}
27
28
29def _decode_txxx_payload(payload: bytes) -> tuple[str, str] | None:
30 if not payload:
31 return None
32 body = payload[1:] if payload[0] in (0, 1, 2, 3) else payload
33 enc = payload[0] if payload[0] in (0, 1, 2, 3) else 0
34 if enc in (1, 2):
35 sep = b"\x00\x00"
36 idx = body.find(sep)
37 if idx < 0:
38 return None
39 raw_desc, raw_val = body[:idx], body[idx + 2 :]
40 encoding = "utf-16" if enc == 1 else "utf-16-be"
41 try:
42 return (
43 raw_desc.decode(encoding, "replace").rstrip("\x00"),
44 raw_val.decode(encoding, "replace").rstrip("\x00"),
45 )
46 except Exception:
47 return None
48 desc, _, value = body.partition(b"\x00")
49 try:
50 return (
51 desc.decode("utf-8", "replace"),
52 value.rstrip(b"\x00").decode("utf-8", "replace"),
53 )
54 except Exception:
55 return None
56
57
58def _synchsafe(raw: bytes) -> int:
59 n = 0
60 for b in raw:
61 n = (n << 7) | (b & 0x7F)
62 return n
63
64
65def _frame_sizes(raw: bytes) -> list[int]:
66 """ID3v2.4 sizes are synchsafe (no high bits). High bits ⇒ v2.3 big-endian."""
67 be = int.from_bytes(raw, "big")
68 if any(b & 0x80 for b in raw):
69 return [be]
70 ss = _synchsafe(raw)
71 return [ss] if ss == be else [ss, be]
72
73
74def iter_txxx(data: bytes) -> Iterable[tuple[str, str]]:
75 """Yield (description, value) for every TXXX frame that decodes."""
76 i = 0
77 n = len(data)
78 while True:
79 i = data.find(b"TXXX", i)
80 if i < 0 or i + 10 > n:
81 return
82 sizes = [s for s in _frame_sizes(data[i + 4 : i + 8]) if 1 <= s <= min(n - (i + 10), 2_000_000)]
83 chosen: tuple[int, str, str] | None = None
84 fallback: tuple[int, str, str] | None = None
85 for size in sizes:
86 payload = data[i + 10 : i + 10 + size]
87 decoded = _decode_txxx_payload(payload)
88 if not decoded:
89 continue
90 desc, value = decoded
91 if not (desc or value):
92 continue
93 hit = (size, desc, value)
94 if desc in KNOWN_DESC:
95 chosen = hit
96 break
97 if fallback is None:
98 fallback = hit
99 picked = chosen or fallback
100 if picked is None:
101 i += 4
102 continue
103 size, desc, value = picked
104 yield desc, value
105 i += 10 + size
106
107
108def hydra_series(path: str | Path, host: str | None = None) -> list[dict]:
109 """Return ~1 Hz samples of who was unmuted.
110
111 Each row: {t, lv, guests, speaker, active}.
112 `host` replaces the `"host"` token (handle without @).
113 """
114 data = Path(path).read_bytes()
115 t0 = None
116 participants: list[dict] = []
117 rows: list[dict] = []
118 last_ntp = None
119 host_name = (host or "host").lstrip("@") or "host"
120
121 for desc, value in iter_txxx(data):
122 if desc == "JSONMetadata":
123 try:
124 ntp = float(json.loads(value)["ntp"])
125 except Exception:
126 continue
127 if t0 is None:
128 t0 = ntp
129 last_ntp = ntp
130 elif desc == "HydraParticipants":
131 try:
132 parsed = json.loads(value)
133 except Exception:
134 parsed = []
135 participants = parsed if isinstance(parsed, list) else []
136 elif desc == "HydraAudioLevel":
137 try:
138 levels = json.loads(value)
139 except Exception:
140 continue
141 if not isinstance(levels, list) or last_ntp is None or t0 is None:
142 continue
143 t = round(last_ntp - t0, 3)
144 active: list[str] = []
145 if levels and _as_level(levels[0]) >= LEVEL_ON:
146 active.append(host_name)
147 for idx, person in enumerate(participants):
148 lv = _as_level(levels[idx + 1]) if idx + 1 < len(levels) else 0
149 if lv >= LEVEL_ON:
150 active.append(_person_name(person, idx))
151 speaker = None
152 if len(active) == 1:
153 speaker = active[0]
154 elif len(active) == 2:
155 speaker = "both"
156 elif len(active) > 2:
157 speaker = "overlap"
158 rows.append(
159 {
160 "t": t,
161 "lv": levels,
162 "guests": participants,
163 "speaker": speaker,
164 "active": active,
165 }
166 )
167 return rows
168
169
170def _as_level(value: object) -> float:
171 try:
172 return float(value) # type: ignore[arg-type]
173 except (TypeError, ValueError):
174 return 0.0
175
176
177def _person_name(person: object, idx: int) -> str:
178 if isinstance(person, dict):
179 name = person.get("UserName") or person.get("UserId")
180 if isinstance(name, str) and name.strip():
181 return name.lstrip("@")
182 return f"guest-{idx}"
183
184
185def regions(series: list[dict], gap: float = 1.5) -> list[dict]:
186 """Collapse ~1 Hz samples into [start, end, speaker] runs."""
187 out: list[dict] = []
188 for row in series:
189 sp = row.get("speaker") or "silence"
190 if out and out[-1]["speaker"] == sp and row["t"] - out[-1]["end"] <= gap:
191 out[-1]["end"] = row["t"]
192 else:
193 start = out[-1]["end"] if out else row["t"]
194 out.append({"start": start, "end": row["t"], "speaker": sp})
195 return out