Batch Call Analysis with Tyto

# /// script
# requires-python = ">=3.14"
# dependencies = [
#     "aic-sdk",
#     "numpy>=2.3.5",
#     "soundfile>=0.13.1",
# ]
# ///
"""Batch-analyze a folder of call recordings with Tyto and write a JSON
file for the call-analysis dashboard (call-analysis.ai-coustics.com).

Usage:
    uv run analyze_calls.py <folder> [output.json]
"""

import json
import os
import sys
from pathlib import Path

import numpy as np
import soundfile as sf

import aic_sdk as aic

MODEL = "tyto-l-16khz"
WINDOW_SECONDS = 5  # Tyto's fixed analysis window
STEP_SECONDS = 1  # hop between windows; 1 s gives smooth dashboard timelines
AUDIO_EXTENSIONS = {".wav", ".flac", ".mp3", ".ogg"}
DIMENSIONS = (
    "risk_score",
    "speaker_reverb",
    "speaker_loudness",
    "interfering_speech",
    "media_speech",
    "noise",
    "packet_loss",
)


def load_mono_audio(path: Path) -> tuple[np.ndarray, int]:
    """Load an audio file and mix it down to a mono float32 array."""
    audio, sample_rate = sf.read(path, dtype="float32")

    # audio is (frames,) for mono or (frames, channels) for multi-channel.
    if audio.ndim > 1:
        audio = audio.mean(axis=1)

    return np.ascontiguousarray(audio, dtype=np.float32), sample_rate


def analyze_file(analyzer: aic.FileAnalyzer, path: Path) -> dict | None:
    """Analyze one recording and return a dashboard entry for it."""
    samples, sample_rate = load_mono_audio(path)

    results = analyzer.analyze(samples, sample_rate, sample_rate * STEP_SECONDS)
    if not results:
        return None  # Shorter than one analysis window.

    return {
        "file": path.name,
        "duration_sec": round(len(samples) / sample_rate, 2),
        "frames": {
            dim: [round(getattr(r, dim), 4) for r in results] for dim in DIMENSIONS
        },
    }


def main():
    if len(sys.argv) < 2:
        sys.exit("usage: uv run analyze_calls.py <folder> [output.json]")

    folder = Path(sys.argv[1])
    output_path = Path(sys.argv[2]) if len(sys.argv) > 2 else Path("analysis.json")
    license_key = os.environ["AIC_SDK_LICENSE"]

    audio_files = sorted(
        p for p in folder.iterdir() if p.suffix.lower() in AUDIO_EXTENSIONS
    )
    if not audio_files:
        sys.exit(f"No audio files found in {folder}")

    # Download and load the analysis model, then reuse one analyzer for all files.
    model_path = aic.Model.download(MODEL, Path("./models"))
    model = aic.Model.from_file(model_path)
    analyzer = aic.FileAnalyzer(model, license_key)
    print(f"Model loaded from {model_path}")

    calls = []
    for index, path in enumerate(audio_files, start=1):
        try:
            call = analyze_file(analyzer, path)
        except Exception as error:
            print(f"[{index}/{len(audio_files)}] {path.name}: skipped ({error})")
            continue

        if call is None:
            print(
                f"[{index}/{len(audio_files)}] {path.name}: skipped "
                f"(shorter than one {WINDOW_SECONDS} s window)"
            )
            continue

        risk = call["frames"]["risk_score"]
        print(
            f"[{index}/{len(audio_files)}] {path.name}: "
            f"{len(risk)} window(s), mean risk {sum(risk) / len(risk):.2f}"
        )
        calls.append(call)

    output_path.write_text(json.dumps({"model": "Tyto", "calls": calls}, indent=2))
    print(f"\nWrote {len(calls)} call(s) to {output_path}")
    print("Upload it at https://call-analysis.ai-coustics.com/")


if __name__ == "__main__":
    main()

Model loaded from models/tyto_l_16khz_yhlek4hc_v43.aicmodel
[1/4] rec_0001.wav: 18 window(s), mean risk 0.23
[2/4] rec_0002.wav: 2 window(s), mean risk 0.60
[3/4] rec_0003.wav: 3 window(s), mean risk 0.20
[4/4] rec_0004.wav: 13 window(s), mean risk 0.30

Wrote 4 call(s) to analysis.json
Upload it at https://call-analysis.ai-coustics.com/