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
99 lines
2.7 KiB
Python
99 lines
2.7 KiB
Python
#!/opt/homebrew/bin/python3.11
|
|
"""Quick test to verify ASR fix works."""
|
|
|
|
import subprocess
|
|
import tempfile
|
|
import os
|
|
import time
|
|
import sys
|
|
|
|
|
|
def test_quick():
|
|
test_video = "/Users/accusys/test_video/BigBuckBunny_320x180.mp4"
|
|
if not os.path.exists(test_video):
|
|
print(f"Test video not found: {test_video}")
|
|
return
|
|
|
|
print(f"Quick test with small video...")
|
|
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
|
output_path = f.name
|
|
|
|
try:
|
|
cmd = [
|
|
"/opt/homebrew/bin/python3.11",
|
|
"scripts/asr_processor.py",
|
|
test_video,
|
|
output_path,
|
|
"--chunk-duration",
|
|
"600",
|
|
]
|
|
|
|
env = os.environ.copy()
|
|
env["MOMENTRY_DISABLE_REDIS"] = "1"
|
|
|
|
print(f"Running: {' '.join(cmd[:3])}... {cmd[3]} ...")
|
|
|
|
start = time.time()
|
|
result = subprocess.run(
|
|
cmd,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
timeout=120, # 2 minutes max
|
|
env=env,
|
|
)
|
|
elapsed = time.time() - start
|
|
|
|
print(f"\nResults:")
|
|
print(f" Time: {elapsed:.1f}s")
|
|
print(f" Return code: {result.returncode}")
|
|
|
|
if result.returncode == 0:
|
|
print(f" ✅ Success")
|
|
|
|
# Check stderr for any issues
|
|
errors = [
|
|
l
|
|
for l in result.stderr.split("\n")
|
|
if "error" in l.lower() and l.strip()
|
|
]
|
|
if errors:
|
|
print(f" ⚠️ Warnings/Errors: {len(errors)}")
|
|
for err in errors[:3]:
|
|
print(f" {err}")
|
|
else:
|
|
print(f" ✓ No errors in output")
|
|
|
|
# Quick check of output
|
|
if os.path.exists(output_path):
|
|
import json
|
|
|
|
with open(output_path, "r") as f:
|
|
data = json.load(f)
|
|
segments = data.get("segments", [])
|
|
print(f" Segments: {len(segments)}")
|
|
if segments:
|
|
print(
|
|
f" First segment text: {segments[0].get('text', '')[:50]}..."
|
|
)
|
|
else:
|
|
print(f" ❌ Failed")
|
|
print(f" stderr (last 5 lines):")
|
|
for line in result.stderr.split("\n")[-5:]:
|
|
if line.strip():
|
|
print(f" {line}")
|
|
|
|
except subprocess.TimeoutExpired:
|
|
print(f"\n❌ Timeout after 120 seconds")
|
|
except Exception as e:
|
|
print(f"\n❌ Error: {e}")
|
|
finally:
|
|
if os.path.exists(output_path):
|
|
os.unlink(output_path)
|
|
print(f" ✓ Cleaned up")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_quick()
|