#!/usr/bin/env python3
|
# -*- coding: utf-8 -*-
|
"""CPU 压测：对单条视频重复调用 pipeline，统计耗时。"""
|
import argparse
|
import os
|
import sys
|
import time
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
from app.schemas import AnalyzeRequest
|
from app.pipeline import run_analyze, get_registry
|
|
|
def main():
|
parser = argparse.ArgumentParser()
|
parser.add_argument("video_url", help="可访问的 MP4 URL")
|
parser.add_argument("--media-id", type=int, default=1)
|
parser.add_argument("--duration", type=float, default=0)
|
parser.add_argument("--sample-fps", type=float, default=0.5)
|
parser.add_argument("--no-asr", action="store_true")
|
parser.add_argument("--repeat", type=int, default=1)
|
args = parser.parse_args()
|
|
reg = get_registry()
|
print(f"model_version={reg.version} ready={reg.ready}")
|
if not reg.ready:
|
print("警告: ONNX 模型未就绪，结果可能失败")
|
|
req = AnalyzeRequest(
|
media_id=args.media_id,
|
video_url=args.video_url,
|
sample_fps=args.sample_fps,
|
enable_asr=not args.no_asr,
|
duration_sec=args.duration if args.duration > 0 else None,
|
)
|
times = []
|
for i in range(args.repeat):
|
t0 = time.perf_counter()
|
resp = run_analyze(req)
|
elapsed = time.perf_counter() - t0
|
times.append(elapsed)
|
print(f"run {i+1}: success={resp.success} elapsed={elapsed:.1f}s "
|
f"storefront={resp.storefront.time_sec if resp.storefront else None} "
|
f"handover={resp.handover.time_sec if resp.handover else None} "
|
f"msg={resp.message or ''}")
|
if times:
|
print(f"avg={sum(times)/len(times):.1f}s min={min(times):.1f}s max={max(times):.1f}s")
|
|
|
if __name__ == "__main__":
|
main()
|
