b54c2def30
- Add database migrations (006-028) for face recognition, identity, file_uuid - Add test scripts for ASR, face, search, processing - Add portal frontend (Tauri) - Add config, benchmark, and monitoring utilities - Add model checkpoints and pretrained model references
67 lines
1.8 KiB
Python
67 lines
1.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Test the updated ASR processor on a small video file."""
|
|
|
|
import sys
|
|
import os
|
|
import subprocess
|
|
import json
|
|
import tempfile
|
|
import time
|
|
|
|
VIDEO_PATH = "../test_video/The_Great_Train_Robbery.mp4"
|
|
if not os.path.exists(VIDEO_PATH):
|
|
print(f"Video not found: {VIDEO_PATH}")
|
|
sys.exit(1)
|
|
|
|
# Create temporary output file
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
|
output_path = f.name
|
|
|
|
print(f"Testing ASR processor on {VIDEO_PATH}")
|
|
print(f"Output: {output_path}")
|
|
|
|
# Run asr_processor.py with timeout
|
|
cmd = [
|
|
"/opt/homebrew/bin/python3.11",
|
|
"scripts/asr_processor.py",
|
|
VIDEO_PATH,
|
|
output_path,
|
|
"--uuid",
|
|
"test123",
|
|
]
|
|
print(f"Command: {' '.join(cmd)}")
|
|
|
|
start = time.time()
|
|
try:
|
|
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
|
|
elapsed = time.time() - start
|
|
print(f"Process completed in {elapsed:.1f}s")
|
|
print(f"Return code: {proc.returncode}")
|
|
if proc.stdout:
|
|
print(f"STDOUT:\n{proc.stdout}")
|
|
if proc.stderr:
|
|
print(f"STDERR:\n{proc.stderr}")
|
|
|
|
# Check output
|
|
if os.path.exists(output_path):
|
|
with open(output_path, "r") as f:
|
|
data = json.load(f)
|
|
print(f"Output contains {len(data.get('segments', []))} segments")
|
|
print(
|
|
f"Language: {data.get('language')} (prob {data.get('language_probability')})"
|
|
)
|
|
print(f"Processing mode: {data.get('processing_mode', 'unknown')}")
|
|
else:
|
|
print("ERROR: Output file not created")
|
|
except subprocess.TimeoutExpired:
|
|
print("ERROR: Process timed out after 300 seconds")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"ERROR: {e}")
|
|
sys.exit(1)
|
|
finally:
|
|
# Clean up
|
|
if os.path.exists(output_path):
|
|
os.unlink(output_path)
|
|
print("Cleanup done")
|