zzstoatzz.io
audio streaming app
plyr.fm
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = ["httpx>=0.28.0"]
5# ///
6"""
7transcoder test matrix - tests format conversion across input/output combinations.
8
9usage:
10 # run all tests against local transcoder
11 uv run scripts/transcoder/test-matrix.py
12
13 # run against production (requires TRANSCODER_AUTH_TOKEN)
14 uv run scripts/transcoder/test-matrix.py --url https://plyr-transcoder.fly.dev
15
16 # run specific input format only
17 uv run scripts/transcoder/test-matrix.py --input-format aiff
18
19 # verbose output
20 uv run scripts/transcoder/test-matrix.py -v
21"""
22
23from __future__ import annotations
24
25import argparse
26import os
27import shutil
28import subprocess
29import sys
30import tempfile
31import time
32from dataclasses import dataclass
33from pathlib import Path
34
35import httpx
36
37# test matrix configuration
38INPUT_FORMATS = ["aiff", "flac", "wav", "mp3", "m4a"]
39OUTPUT_FORMATS = ["mp3", "m4a", "wav"]
40
41# sample generation parameters
42SAMPLE_DURATION = 2 # seconds
43SAMPLE_RATE = 44100
44CHANNELS = 2
45
46
47@dataclass
48class TestResult:
49 input_format: str
50 output_format: str
51 success: bool
52 duration_ms: float
53 input_size: int
54 output_size: int
55 error: str | None = None
56
57
58def generate_sample(output_path: Path, format: str) -> bool:
59 """generate a test audio sample using ffmpeg."""
60 cmd = [
61 sys.executable,
62 "scripts/generate_audio_sample.py",
63 str(output_path),
64 "--waveform",
65 "sine",
66 "--duration",
67 str(SAMPLE_DURATION),
68 "--sample-rate",
69 str(SAMPLE_RATE),
70 "--channels",
71 str(CHANNELS),
72 "--frequency",
73 "440",
74 "--fade-in",
75 "0.1",
76 "--fade-out",
77 "0.1",
78 "--force",
79 "--log-level",
80 "error",
81 ]
82 result = subprocess.run(cmd, capture_output=True, text=True)
83 return result.returncode == 0
84
85
86def transcode_file(
87 input_path: Path,
88 target_format: str,
89 url: str,
90 auth_token: str | None,
91 timeout: float = 60.0,
92) -> tuple[bytes | None, str | None]:
93 """send file to transcoder and return result."""
94 headers = {}
95 if auth_token:
96 headers["X-Transcoder-Key"] = auth_token
97
98 try:
99 with open(input_path, "rb") as f:
100 files = {"file": (input_path.name, f)}
101 response = httpx.post(
102 f"{url}/transcode",
103 params={"target": target_format},
104 files=files,
105 headers=headers,
106 timeout=timeout,
107 )
108
109 if response.status_code == 200:
110 return response.content, None
111 else:
112 return None, f"HTTP {response.status_code}: {response.text[:200]}"
113 except httpx.TimeoutException:
114 return None, "timeout"
115 except Exception as e:
116 return None, str(e)
117
118
119def verify_audio(file_path: Path) -> bool:
120 """verify audio file is valid using ffprobe."""
121 cmd = [
122 "ffprobe",
123 "-v",
124 "error",
125 "-show_entries",
126 "format=duration",
127 "-of",
128 "csv=p=0",
129 str(file_path),
130 ]
131 result = subprocess.run(cmd, capture_output=True, text=True)
132 if result.returncode != 0:
133 return False
134 try:
135 duration = float(result.stdout.strip())
136 # allow some tolerance for duration
137 return duration > SAMPLE_DURATION * 0.8
138 except ValueError:
139 return False
140
141
142def run_test(
143 input_format: str,
144 output_format: str,
145 samples_dir: Path,
146 url: str,
147 auth_token: str | None,
148 verbose: bool = False,
149) -> TestResult:
150 """run a single transcoding test."""
151 input_path = samples_dir / f"test.{input_format}"
152
153 if not input_path.exists():
154 return TestResult(
155 input_format=input_format,
156 output_format=output_format,
157 success=False,
158 duration_ms=0,
159 input_size=0,
160 output_size=0,
161 error=f"input file not found: {input_path}",
162 )
163
164 input_size = input_path.stat().st_size
165 start = time.perf_counter()
166
167 result_bytes, error = transcode_file(input_path, output_format, url, auth_token)
168
169 duration_ms = (time.perf_counter() - start) * 1000
170
171 if error:
172 return TestResult(
173 input_format=input_format,
174 output_format=output_format,
175 success=False,
176 duration_ms=duration_ms,
177 input_size=input_size,
178 output_size=0,
179 error=error,
180 )
181
182 # write output and verify
183 output_path = (
184 samples_dir / f"output_{input_format}_to_{output_format}.{output_format}"
185 )
186 output_path.write_bytes(result_bytes)
187 output_size = len(result_bytes)
188
189 if not verify_audio(output_path):
190 return TestResult(
191 input_format=input_format,
192 output_format=output_format,
193 success=False,
194 duration_ms=duration_ms,
195 input_size=input_size,
196 output_size=output_size,
197 error="output validation failed",
198 )
199
200 return TestResult(
201 input_format=input_format,
202 output_format=output_format,
203 success=True,
204 duration_ms=duration_ms,
205 input_size=input_size,
206 output_size=output_size,
207 )
208
209
210def print_matrix(
211 results: list[TestResult], input_formats: list[str], output_formats: list[str]
212):
213 """print results as a matrix table."""
214 # build lookup
215 lookup = {(r.input_format, r.output_format): r for r in results}
216
217 # header
218 col_width = 10
219 header = (
220 "input".ljust(col_width)
221 + " | "
222 + " | ".join(f.center(col_width) for f in output_formats)
223 )
224 print("\n" + header)
225 print("-" * len(header))
226
227 # rows
228 for inf in input_formats:
229 row = inf.ljust(col_width) + " | "
230 cells = []
231 for outf in output_formats:
232 r = lookup.get((inf, outf))
233 if r is None:
234 cells.append("skip".center(col_width))
235 elif r.success:
236 cells.append(f"{r.duration_ms:.0f}ms".center(col_width))
237 else:
238 cells.append("FAIL".center(col_width))
239 row += " | ".join(cells)
240 print(row)
241
242
243def main():
244 parser = argparse.ArgumentParser(description="Transcoder test matrix")
245 parser.add_argument(
246 "--url",
247 default="http://127.0.0.1:8082",
248 help="Transcoder URL (default: http://127.0.0.1:8082)",
249 )
250 parser.add_argument(
251 "--input-format",
252 choices=INPUT_FORMATS,
253 help="Test only this input format",
254 )
255 parser.add_argument(
256 "--output-format",
257 choices=OUTPUT_FORMATS,
258 help="Test only this output format",
259 )
260 parser.add_argument(
261 "-v",
262 "--verbose",
263 action="store_true",
264 help="Verbose output",
265 )
266 parser.add_argument(
267 "--keep-files",
268 action="store_true",
269 help="Keep generated test files (useful for debugging)",
270 )
271 args = parser.parse_args()
272
273 # get auth token from environment
274 auth_token = os.environ.get("TRANSCODER_AUTH_TOKEN")
275 if "fly.dev" in args.url and not auth_token:
276 print(
277 "error: TRANSCODER_AUTH_TOKEN required for production URL", file=sys.stderr
278 )
279 sys.exit(1)
280
281 # check transcoder is running
282 try:
283 response = httpx.get(f"{args.url}/health", timeout=5.0)
284 if response.status_code != 200:
285 print(
286 f"error: transcoder health check failed: {response.status_code}",
287 file=sys.stderr,
288 )
289 sys.exit(1)
290 except Exception as e:
291 print(f"error: cannot reach transcoder at {args.url}: {e}", file=sys.stderr)
292 print("hint: start with `just transcoder run`", file=sys.stderr)
293 sys.exit(1)
294
295 # determine formats to test
296 input_formats = [args.input_format] if args.input_format else INPUT_FORMATS
297 output_formats = [args.output_format] if args.output_format else OUTPUT_FORMATS
298
299 # create temp directory for test files
300 if args.keep_files:
301 samples_dir = Path("sandbox/transcoder-test")
302 samples_dir.mkdir(parents=True, exist_ok=True)
303 else:
304 temp_dir = tempfile.mkdtemp(prefix="transcoder-test-")
305 samples_dir = Path(temp_dir)
306
307 print(f"transcoder: {args.url}")
308 print(f"test files: {samples_dir}")
309 print(f"matrix: {len(input_formats)} inputs x {len(output_formats)} outputs")
310 print()
311
312 # generate input samples
313 print("generating test samples...")
314 for fmt in input_formats:
315 sample_path = samples_dir / f"test.{fmt}"
316 if args.verbose:
317 print(f" {fmt}...", end=" ", flush=True)
318 if generate_sample(sample_path, fmt):
319 if args.verbose:
320 print(f"{sample_path.stat().st_size} bytes")
321 else:
322 print(f"failed to generate {fmt} sample", file=sys.stderr)
323 sys.exit(1)
324
325 # run test matrix
326 print("\nrunning transcoding tests...")
327 results: list[TestResult] = []
328
329 for inf in input_formats:
330 for outf in output_formats:
331 if args.verbose:
332 print(f" {inf} -> {outf}...", end=" ", flush=True)
333
334 result = run_test(
335 inf, outf, samples_dir, args.url, auth_token, args.verbose
336 )
337 results.append(result)
338
339 if args.verbose:
340 if result.success:
341 print(
342 f"{result.duration_ms:.0f}ms ({result.input_size} -> {result.output_size} bytes)"
343 )
344 else:
345 print(f"FAILED: {result.error}")
346
347 # print matrix summary
348 print_matrix(results, input_formats, output_formats)
349
350 # print failures
351 failures = [r for r in results if not r.success]
352 if failures:
353 print(f"\n{len(failures)} failures:")
354 for r in failures:
355 print(f" {r.input_format} -> {r.output_format}: {r.error}")
356
357 # cleanup
358 if not args.keep_files:
359 shutil.rmtree(samples_dir)
360
361 # exit code
362 sys.exit(0 if not failures else 1)
363
364
365if __name__ == "__main__":
366 main()