import argparse
import asyncio
import hashlib
import importlib.metadata
import json
import os
import platform
import time
from pathlib import Path
import aic_sdk as aic
import numpy as np
import psutil
import soundfile as sf
MODEL = "quail-vf-2.2-l-16khz"
async def measure(model, audio, config, seconds):
processor = aic.ProcessorAsync(model, os.environ["AIC_SDK_LICENSE"], config)
size = config.block_size
period = size / config.sample_rate
frames = len(audio) // size
durations, late = [], []
try:
for i in range(10):
await processor.process_async(audio[(i % frames) * size:((i % frames) + 1) * size])
start = time.monotonic()
index = 0
while time.monotonic() - start < seconds:
due = start + index * period
await asyncio.sleep(max(0, due - time.monotonic()))
before = time.monotonic()
frame = index % frames
output = await processor.process_async(audio[frame * size:(frame + 1) * size])
after = time.monotonic()
if len(output) != size or not np.isfinite(output).all():
raise RuntimeError("Invalid processor output")
durations.append(after - before)
late.append(max(0, after - (due + period)))
index += 1
return {
"frames": index,
"call_ms_p50_p95_p99_max": np.percentile(durations, [50, 95, 99, 100]).tolist(),
"max_scheduling_lateness_ms": max(late) * 1000,
"deadline_misses": sum(value > 0 for value in late),
"max_call_frame_ratio": max(durations) / period,
}
finally:
await processor.terminate_session_async()
async def main(args):
if not os.environ.get("AIC_SDK_LICENSE"):
raise ValueError("Set AIC_SDK_LICENSE before running")
audio, rate = sf.read(args.input, dtype="float32")
if audio.ndim != 1 or not np.isfinite(audio).all() or not np.any(audio):
raise ValueError("Use finite, non-silent mono audio")
model = aic.Model.from_file(aic.Model.download(MODEL, "./models"))
config = aic.ProcessorConfig(rate, model.get_optimal_block_size(rate))
if len(audio) < config.block_size:
raise ValueError("Input must contain at least one complete frame")
process = psutil.Process()
process.cpu_percent(None)
rss_before = process.memory_info().rss
tasks = [measure(model, audio, config, args.seconds) for _ in range(args.streams)]
results = await asyncio.gather(*tasks, return_exceptions=True)
failures = [type(result).__name__ for result in results if isinstance(result, BaseException)]
if failures:
raise RuntimeError(f"Run failed: {failures}")
for result in results:
result["call_ms_p50_p95_p99_max"] = [value * 1000 for value in result["call_ms_p50_p95_p99_max"]]
print(json.dumps({
"binding": importlib.metadata.version("aic-sdk"), "core": aic.get_sdk_version(),
"model": MODEL, "platform": platform.platform(), "machine": platform.machine(),
"input_sha256": hashlib.sha256(Path(args.input).read_bytes()).hexdigest(),
"sample_rate": rate, "block_size": config.block_size,
"seconds_per_stream": args.seconds, "streams": args.streams,
"process_cpu_percent": process.cpu_percent(None),
"rss_before_bytes": rss_before, "rss_after_bytes": process.memory_info().rss,
"results": results,
}, indent=2))
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("input")
parser.add_argument("--streams", type=int, choices=[1, 2], default=1)
parser.add_argument("--seconds", type=int, choices=range(1, 121), default=10)
asyncio.run(main(parser.parse_args()))
