space-tapePublic

Open-source X Spaces transcription.

Star

124 lines · 3.7 KB

python
1from __future__ import annotations
2
3import json
4import sys
5from pathlib import Path
6
7sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
8
9from space_tape.hydra import hydra_series, iter_txxx, regions
10
11
12def txxx(desc: str, value: str) -> bytes:
13 body = b"\x03" + desc.encode("utf-8") + b"\x00" + value.encode("utf-8") + b"\x00"
14 return b"TXXX" + len(body).to_bytes(4, "big") + b"\x00\x00" + body
15
16
17def tagged_clip(*, synchsafe: bool = False) -> bytes:
18 """~10s of Hydra ticks: host 0–4s, guest 4–8s, both 8–10s."""
19 guest = {
20 "ParticipantIndex": 1,
21 "UserId": "1eWKyPLBZDnQA",
22 "UserName": "djay44444",
23 "SessionId": "595fec0b-252b-48af-a2f4-f356c4855e3b",
24 "ProfileUrl": "https://pbs.twimg.com/profile_images/example.jpg",
25 }
26 t0 = 3998530786.0
27 frames = bytearray(b"ID3\x04\x00\x00\x00\x00\x00\x00")
28 for i in range(11):
29 ntp = t0 + i
30 if i < 4:
31 levels = [78]
32 elif i < 8:
33 levels = [0, 63]
34 else:
35 levels = [54, 61]
36 frames += _maybe_synchsafe(
37 txxx("JSONMetadata", json.dumps({"HydraVersion": 4, "ntp": ntp})),
38 synchsafe,
39 )
40 frames += _maybe_synchsafe(
41 txxx("HydraParticipants", json.dumps([guest])),
42 synchsafe,
43 )
44 frames += _maybe_synchsafe(
45 txxx("HydraAudioLevel", json.dumps(levels)),
46 synchsafe,
47 )
48 return bytes(frames)
49
50
51def _maybe_synchsafe(frame: bytes, synchsafe: bool) -> bytes:
52 if not synchsafe:
53 return frame
54 size = int.from_bytes(frame[4:8], "big")
55 ss = bytes(
56 [
57 (size >> 21) & 0x7F,
58 (size >> 14) & 0x7F,
59 (size >> 7) & 0x7F,
60 size & 0x7F,
61 ]
62 )
63 return frame[:4] + ss + frame[8:]
64
65
66def test_iter_txxx_roundtrip(tmp_path: Path | None = None) -> None:
67 data = tagged_clip()
68 descs = [d for d, _ in iter_txxx(data)]
69 assert descs.count("JSONMetadata") == 11
70 assert descs.count("HydraParticipants") == 11
71 assert descs.count("HydraAudioLevel") == 11
72
73
74def test_host_is_slot_zero(tmp_path: Path | None = None) -> None:
75 path = Path(__file__).parent / "fixtures"
76 path.mkdir(exist_ok=True)
77 clip = path / "tagged_clip.bin"
78 clip.write_bytes(tagged_clip())
79 series = hydra_series(clip, host="notpierce69")
80 assert series[0]["speaker"] == "notpierce69"
81 assert series[0]["active"] == ["notpierce69"]
82 assert series[5]["speaker"] == "djay44444"
83 assert series[9]["speaker"] == "both"
84 assert "notpierce69" in series[9]["active"]
85 assert "djay44444" in series[9]["active"]
86
87
88def test_regions_collapse() -> None:
89 series = hydra_series(_write_tmp(tagged_clip()), host="notpierce69")
90 runs = regions(series)
91 speakers = [r["speaker"] for r in runs]
92 assert speakers == ["notpierce69", "djay44444", "both"]
93 for run in runs:
94 assert run["end"] >= run["start"]
95
96
97def test_synchsafe_id3v24() -> None:
98 series = hydra_series(_write_tmp(tagged_clip(synchsafe=True)), host="notpierce69")
99 assert len(series) == 11
100 assert series[0]["speaker"] == "notpierce69"
101 assert series[5]["speaker"] == "djay44444"
102
103
104def test_empty_file_is_empty_series(tmp_path: Path | None = None) -> None:
105 series = hydra_series(_write_tmp(b"not an id3 file at all"))
106 assert series == []
107
108
109def _write_tmp(data: bytes) -> Path:
110 dest = Path(__file__).parent / "fixtures"
111 dest.mkdir(exist_ok=True)
112 path = dest / "tmp.bin"
113 path.write_bytes(data)
114 return path
115
116
117if __name__ == "__main__":
118 test_iter_txxx_roundtrip()
119 test_host_is_slot_zero()
120 test_regions_collapse()
121 test_synchsafe_id3v24()
122 test_empty_file_is_empty_series()
123 print("ok")