"""Offline route experiment. All writes stay in this run's lab directory.""" import argparse import hashlib import importlib.metadata import inspect import json import os from pathlib import Path import platform import re import shutil import subprocess import time from datetime import date def main(): parser = argparse.ArgumentParser() parser.add_argument('--ffmpeg', required=True) args = parser.parse_args() lab = Path(__file__).resolve().parent corpus = lab / 'corpus' logs = lab / 'ffmpeg' logs.mkdir(exist_ok=True) commands = [] def run(label, tail): cmd = [args.ffmpeg, '-hide_banner', '-y', *tail] start = time.perf_counter() cp = subprocess.run(cmd, text=True, capture_output=True, timeout=90) duration = round(time.perf_counter() - start, 4) (logs / (label + '.txt')).write_text(cp.stderr + cp.stdout) commands.append({'label': label, 'command': cmd, 'returncode': cp.returncode, 'elapsed_s': duration, 'stderr': 'ffmpeg/' + label + '.txt'}) return cp, duration base = corpus / 'screen-recording.mp4' original_truth = json.loads((corpus / 'screen-recording.truth.json').read_text()) sentences = re.split(r'(?<=\.)\s+', original_truth['paragraphs'][0]) caption_text = ('1\n00:00:00,000 --> 00:00:08,000\n' + ' '.join(sentences[:2]) + '\n\n2\n00:00:08,000 --> 00:00:18,000\n' + ' '.join(sentences[2:5]) + '\n\n3\n00:00:18,000 --> 00:00:24,000\n' + ' '.join(sentences[5:]) + '\n') (corpus / 'caption-script.txt').write_text(caption_text) cp, _ = run('build-captioned', ['-i', str(base), '-f', 'srt', '-i', str(corpus / 'caption-script.txt'), '-map', '0:v:0', '-map', '0:a:0', '-map', '1:0', '-c:v', 'copy', '-c:a', 'copy', '-c:s', 'mov_text', str(corpus / 'captioned-recording.mp4')]) if cp.returncode: raise RuntimeError(cp.stderr) caption_truth = {**original_truth, 'classification': 'Authored captions from synthetic source script; manually assigned cue times, not ASR alignment', 'derived_from': 'screen-recording.mp4'} (corpus / 'captioned-recording.truth.json').write_text(json.dumps(caption_truth, indent=2)) cp, _ = run('build-silent-visual', ['-i', str(base), '-map', '0:v:0', '-an', '-vf', "drawtext=fontfile=/System/Library/Fonts/Helvetica.ttc:text='Build target staging':fontcolor=black:box=1:boxcolor=white:fontsize=36:x=100:y=500", '-c:v', 'libx264', '-preset', 'ultrafast', str(corpus / 'silent-visual.mp4')]) if cp.returncode: raise RuntimeError(cp.stderr) (corpus / 'silent-visual.truth.json').write_text(json.dumps({ 'kind': 'prose', 'format': 'mp4', 'paragraphs': ['Build target staging'], 'classification': 'Visual-only text control; no audio or subtitle track. Prose sentence metric is inapplicable (under four words).', 'derived_from': 'screen-recording.mp4'}, indent=2)) cp, _ = run('visual-frame', ['-i', str(corpus / 'silent-visual.mp4'), '-frames:v', '1', str(logs / 'silent-visual.png')]) if cp.returncode: raise RuntimeError(cp.stderr) import markitdown from markitdown.converters import _transcribe_audio from markitdown.converters import _audio_converter sources = lab / 'source' sources.mkdir(exist_ok=True) for mod, name in [(_audio_converter, 'markitdown-audio-converter.py'), (_transcribe_audio, 'markitdown-transcribe-audio.py')]: shutil.copyfile(inspect.getfile(mod), sources / name) old_path = lab / 'runs.json' upstream = lab / 'runs-upstream.json' if not upstream.exists(): shutil.copyfile(old_path, upstream) upstream_data = json.loads(upstream.read_text()) results = [x for x in upstream_data['results'] if x['tool'] == 'markovo' and x['sample'] in ('screen-recording', 'speech')] for r in results: r['elapsed_s'] = None r['receipt_reused'] = True r['timing_note'] = 'Cached upstream result: zero in upstream runner is not runtime' dependencies = {name: importlib.util.find_spec(name) is not None for name in ('speech_recognition', 'pydub')} tool_dir = lab / 'markitdown' tool_dir.mkdir(exist_ok=True) for sample in ('screen-recording', 'captioned-recording', 'silent-visual', 'speech'): path = corpus / (sample + ('.mp3' if sample == 'speech' else '.mp4')) start = time.perf_counter() error = None try: content = markitdown.MarkItDown(enable_plugins=False).convert(str(path)).text_content except Exception as exc: content = '' error = type(exc).__name__ + ': ' + str(exc) elapsed = round(time.perf_counter() - start, 4) (tool_dir / (sample + '.md')).write_text(content) diagnostic = '\n'.join([ 'Conversion returned without an exception.' if error is None else 'Conversion raised an exception.', 'Returned Markdown was empty.' if not content.strip() else 'Returned Markdown contained text.', 'No audio transcript heading was present.' if '### Audio Transcript:' not in content else 'An audio transcript heading was present.' ]) + '\n' (tool_dir / (sample + '-diagnostic.md')).write_text(diagnostic) (tool_dir / (sample + '.json')).write_text(json.dumps({ 'version': importlib.metadata.version('markitdown'), 'python': platform.python_version(), 'sample': sample, 'elapsed_s': elapsed, 'error': error, 'returned_characters': len(content), 'dependencies': dependencies, 'network_requested': False, 'diagnostic_is_runner_output_not_converter_markdown': True}, indent=2)) results.append({'tool': 'markitdown', 'sample': sample, 'output': 'markitdown/' + sample + '.md', 'elapsed_s': elapsed, 'error': error}) # A subtitle extractor has a deliberately different contract from ASR. for sample in ('screen-recording', 'captioned-recording', 'silent-visual'): out = logs / (sample + '.md') cp, elapsed = run(sample, ['-i', str(corpus / (sample + '.mp4')), '-map', '0:s:0', '-c:s', 'srt', '-f', 'srt', str(out)]) if not out.exists(): out.write_text('') error = None if cp.returncode == 0 else next( (line for line in cp.stderr.splitlines() if 'matches no streams' in line), 'FFmpeg returned ' + str(cp.returncode)) results.append({'tool': 'ffmpeg-subtitles', 'sample': sample, 'output': 'ffmpeg/' + sample + '.md', 'elapsed_s': elapsed, 'error': error}) run('extract-audio', ['-i', str(base), '-map', '0:a:0', '-vn', '-c:a', 'pcm_s16le', str(logs / 'extracted.wav')]) version = subprocess.run([args.ffmpeg, '-version'], text=True, capture_output=True, check=True).stdout (logs / 'version.txt').write_text(version) source_records = [] for p in (base, corpus / 'captioned-recording.mp4', corpus / 'silent-visual.mp4'): cp, _ = run(p.stem + '-inspect', ['-i', str(p), '-f', 'null', '-']) types = [re.sub(r'^.*?: ', '', line).split(':', 1)[0] for line in cp.stderr.splitlines() if 'Stream #0:' in line and ('Video:' in line or 'Audio:' in line or 'Subtitle:' in line)] source_records.append({'file': 'corpus/' + p.name, 'sha256': hashlib.sha256(p.read_bytes()).hexdigest(), 'bytes': p.stat().st_size, 'stream_lines': [x.strip() for x in cp.stderr.splitlines() if 'Stream #0:' in x]}) (lab / 'media-experiment.json').write_text(json.dumps({ 'date': date.today().isoformat(), 'python': platform.python_version(), 'ffmpeg': version.splitlines()[0], 'commands': commands, 'dependencies': dependencies, 'sources': source_records, 'boundary': 'One synthetic spoken fixture, authored captions, and a visual-only control. No WER, independent ASR engine, timing alignment, noisy or multi-speaker benchmark.'}, indent=2)) (lab / 'runs.json').write_text(json.dumps({ 'date': date.today().isoformat(), 'python': platform.python_version(), 'tools': {'markovo': {**upstream_data['tools']['markovo'], 'elapsed_s': None, 'note': 'Reused original public API receipts dated 2026-10-27/28'}, 'markitdown': {'version': importlib.metadata.version('markitdown'), 'dependencies': dependencies}, 'ffmpeg-subtitles': {'version': '7.1', 'contract': 'existing text subtitles, not speech recognition'}}, 'results': results}, indent=2)) print(json.dumps({'samples': len(source_records), 'results': len(results), 'dependencies': dependencies})) if __name__ == '__main__': main()