78 lines
4.1 KiB
Python
Executable File
78 lines
4.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Run adaptive PlayerModel playback against real loopback media and offline yt-dlp metadata."""
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
|
|
from streaming import StreamingServer
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("binary", type=Path)
|
|
parser.add_argument("fixtures", type=Path)
|
|
args = parser.parse_args()
|
|
extractor = shutil.which("yt-dlp")
|
|
if extractor is None:
|
|
extractor = next((str(path) for path in (Path("/opt/homebrew/bin/yt-dlp"), Path("/usr/local/bin/yt-dlp"))
|
|
if os.access(path, os.X_OK)), None)
|
|
if extractor is None:
|
|
print("SKIP native adaptive playback: yt-dlp is not installed; controlled-clock policy checks still run")
|
|
return 0
|
|
required = ("quality-1080.mp4", "quality-720.mp4")
|
|
if any(not (args.fixtures / name).is_file() for name in required):
|
|
parser.error("Missing real quality playback fixtures")
|
|
with StreamingServer(args.fixtures) as server, tempfile.TemporaryDirectory(prefix="terebi-quality-check-") as directory:
|
|
root = Path(directory)
|
|
formats = []
|
|
for height, width in ((720, 1280), (1080, 1920)):
|
|
path = "/media/quality-720.mp4" if height == 720 else "/stall/quality-low/quality-1080.mp4?after_seconds=2"
|
|
formats.append({"format_id": str(height), "height": height, "width": width,
|
|
"url": server.base_url + path, "protocol": "http", "ext": "mp4",
|
|
"vcodec": "mpeg4", "acodec": "aac", "dynamic_range": "SDR",
|
|
"http_headers": {"X-Terebi-Fixture": "combined"}})
|
|
info = {"id": "aqz-KE-bpKQ", "title": "Automatic quality playback fixture", "duration": 120,
|
|
"is_live": False, "webpage_url": "https://www.youtube.com/watch?v=aqz-KE-bpKQ",
|
|
"extractor": "youtube", "extractor_key": "Youtube", "formats": formats}
|
|
metadata = root / "info.json"
|
|
metadata.write_text(json.dumps(info))
|
|
wrapper = root / "yt-dlp"
|
|
control = root / "control"
|
|
control.mkdir()
|
|
wrapper.write_text("#!/bin/bash\nset -eu\nargs=(\"$@\")\n"
|
|
+ "control=" + shlex.quote(str(control)) + "\n"
|
|
+ 'if [[ "$*" == *"[height=1080]"* ]]; then\n'
|
|
+ ' if [[ -f "$control/fail-1080" ]]; then\n'
|
|
+ ' echo "Fixture extraction deliberately failed" >&2; exit 1\n'
|
|
+ ' fi\n'
|
|
+ ' if [[ -f "$control/gate-1080" ]]; then\n'
|
|
+ " " + shlex.quote(extractor) + " --load-info-json " + shlex.quote(str(metadata))
|
|
+ ' "${args[@]:0:${#args[@]}-2}" > "$control/output-1080.json"\n'
|
|
+ ' trap \': > "$control/finished-1080"; exit 0\' TERM\n'
|
|
+ ' : > "$control/entered-1080"\n'
|
|
+ ' while [[ ! -f "$control/release-1080" ]]; do sleep 0.01; done\n'
|
|
+ ' cat "$control/output-1080.json" || true\n'
|
|
+ ' : > "$control/finished-1080"\n'
|
|
+ ' exit 0\n'
|
|
+ ' fi\n'
|
|
+ 'fi\nexec ' + shlex.quote(extractor)
|
|
+ " --load-info-json " + shlex.quote(str(metadata))
|
|
+ ' "${args[@]:0:${#args[@]}-2}"\n')
|
|
wrapper.chmod(0o755)
|
|
environment = dict(os.environ, PATH=str(root) + os.pathsep + os.environ.get("PATH", ""),
|
|
TEREBI_QUALITY_CONTROL=str(control))
|
|
try:
|
|
result = subprocess.run([str(args.binary.resolve())], env=environment, timeout=90, check=False)
|
|
except subprocess.TimeoutExpired:
|
|
raise SystemExit("FAIL: adaptive playback exceeded its infrastructure deadline")
|
|
return result.returncode
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|