Batch call analysis with Tyto

python -m venv .venv
source .venv/bin/activate
python -m pip install 'aic-sdk==3.2.0' 'numpy==2.5.3' 'soundfile==0.13.1'

import argparse
import json
import os
from pathlib import Path

import aic_sdk as aic
import numpy as np
import soundfile as sf

MODEL = "tyto-1.1-l-16khz"
WINDOW_SECONDS = 5
STEP_SECONDS = 1
DIMENSIONS = (
    "risk_score", "speaker_reverb", "speaker_loudness", "interfering_speech",
    "noise", "packet_loss", "codec_degradation",
)


def analyze_file(analyzer, path):
    samples, rate = sf.read(path, dtype="float32")
    if samples.ndim != 1:
        raise ValueError("Select the user channel and save it as mono first")
    if not np.isfinite(samples).all() or not np.any(samples):
        raise ValueError("Use finite, non-silent audio")
    if not 8000 <= rate <= 192000:
        raise ValueError("Sample rate must be between 8000 and 192000 Hz")
    if len(samples) < WINDOW_SECONDS * rate:
        return None  # This example excludes partly padded windows.
    results = analyzer.analyze(np.ascontiguousarray(samples), rate, rate * STEP_SECONDS)
    expected = 1 + (len(samples) - WINDOW_SECONDS * rate) // (rate * STEP_SECONDS)
    if len(results) != expected:
        raise RuntimeError("Unexpected analysis window count")
    frames = {name: [float(getattr(result, name)) for result in results] for name in DIMENSIONS}
    if any(not np.isfinite(values).all() for values in frames.values()):
        raise RuntimeError("Non-finite analysis result")
    return {"file": path.name, "duration_sec": len(samples) / rate, "frames": frames}


def main(folder, output):
    key = os.environ["AIC_SDK_LICENSE"]
    files = sorted(path for path in folder.iterdir() if path.is_file() and path.suffix.lower() in {".wav", ".flac"})
    if not files:
        raise ValueError("No WAV or FLAC files found")
    model = aic.Model.from_file(aic.Model.download(MODEL, "./models"))
    analyzer = aic.FileAnalyzer(model, key)
    calls = []
    try:
        for path in files:
            call = analyze_file(analyzer, path)
            if call is None:
                print(f"Skipped {path.name}: shorter than this example's 5 s window")
                continue
            calls.append(call)
            values = call["frames"]["risk_score"]
            print(f"{path.name}: {len(values)} window(s), mean risk {np.mean(values):.3f}")
    finally:
        # FileAnalyzer owns its pair; release it outside a real-time callback.
        del analyzer
    if not calls:
        raise RuntimeError("No complete recordings were analyzed; no results written")
    output.write_text(json.dumps({"model": "Tyto", "calls": calls}, indent=2), encoding="utf-8")
    print(f"Wrote {len(calls)} call(s) to {output}; step {STEP_SECONDS} s")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("folder", type=Path)
    parser.add_argument("output", type=Path, nargs="?", default=Path("analysis.json"))
    args = parser.parse_args()
    main(args.folder, args.output)

python analyze_calls.py recordings/ analysis.json