Real-time analysis with Tyto

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

import argparse
import json
import math
import os
import queue
import threading
import time

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

MODEL = "tyto-1.2-l-16khz"
WINDOW_SECONDS = 5


def main(path, interval):
    if not math.isfinite(interval) or interval <= 0:
        raise ValueError("Analysis interval must be finite and positive")
    key = os.environ["AIC_SDK_LICENSE"]
    audio, rate = sf.read(path, 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")
    if len(audio) < WINDOW_SECONDS * rate:
        raise ValueError("Use at least 5 s to fill this example's analysis window")
    model = aic.Model.from_file(aic.Model.download(MODEL, "./models"))
    collector, analyzer = aic.analyzer_pair(model, key)
    config = aic.ProcessorConfig(rate, model.get_optimal_block_size(rate))
    collector.initialize(config)
    ready, stop = threading.Event(), threading.Event()
    first_pass_done = threading.Event()
    failures = queue.SimpleQueue()
    scores = []

    def analyze():
        while not ready.wait(0.1):
            if stop.is_set():
                return
        while not stop.is_set():
            started = time.monotonic()
            try:
                result = analyzer.analyze_buffered()
                record = {"risk_score": result.risk_score, "noise": result.noise}
                scores.append(record)
                print(json.dumps(record))
            except Exception as error:
                failures.put(type(error).__name__)
                stop.set()
                return
            finally:
                first_pass_done.set()
            elapsed = time.monotonic() - started
            if elapsed > interval:
                print("Analysis exceeded its interval; reduce frequency or concurrency")
            stop.wait(max(0, interval - elapsed))

    worker = threading.Thread(target=analyze)
    worker.start()
    try:
        started = time.monotonic()
        buffered = 0
        for offset in range(0, len(audio), config.block_size):
            if stop.is_set():
                break
            chunk = audio[offset:offset + config.block_size]
            block = np.zeros(config.block_size, dtype=np.float32)
            block[:len(chunk)] = chunk
            collector.buffer(block)
            buffered += len(chunk)
            if buffered >= WINDOW_SECONDS * rate:
                ready.set()
            stop.wait(max(0, started + buffered / rate - time.monotonic()))
        # A full window must get one analysis attempt before shutdown.
        if ready.is_set() and not stop.is_set():
            first_pass_done.wait()
    finally:
        stop.set()
        worker.join()
        analyzer.terminate_session()
    if not failures.empty():
        raise RuntimeError(f"Analysis failed: {failures.get()}")
    if not scores:
        raise RuntimeError("No score was produced; use a longer recording")
    print(f"Completed {len(scores)} analysis pass(es)")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("input")
    parser.add_argument("--interval", type=float, default=5.0)
    args = parser.parse_args()
    main(args.input, args.interval)

python analyze_stream.py input.wav --interval 5

collector.buffer(audio)
vad.process(audio)
enhanced = processor.process(audio)