Python audio analysis

Package: aic-sdk==3.2.0. Core SDK: 0.24.0. Source: Python wrapper 3.2.0. The fragments below use these imports. Supply license_key from your approved secret source and use the loaded model and initialized objects described in each section. For complete examples, follow batch call analysis with FileAnalyzer or real-time analysis with analyzer_pair.

import numpy as np
import numpy.typing as npt
import aic_sdk as aic
from aic_sdk import Model, ProcessorConfig, Collector, Analyzer, FileAnalyzer, AnalysisResult

audio denotes a one-dimensional NumPy float32 array. Async fragments run inside an async function.

analyzer_pair()

function

def analyzer_pair(model: Model, license_key: str) -> tuple[Collector, Analyzer]

For a complete collector/worker example, follow real-time analysis with Tyto. Creates a Collector/Analyzer pair for non-real-time analysis. Buffer audio in the capture path and run analysis on a separate thread. The analyzer safely reads the collector across threads. Native buffering releases the GIL, but interpreter calls and strided-array copies do not guarantee hard real-time execution. The collector retains a span of audio determined by the analysis model. As more samples get collected, old audio is discarded. Parameters

Model

required

The loaded model instance. See Model.

str

required

SDK key or JWT for the ai-coustics SDK (generate your key at developers.ai-coustics.com).

Returns

tuple[Collector, Analyzer]: A tuple of (Collector, Analyzer).

Raises

LicenseFormatInvalidError: If the license key string contains null bytes.
SDK exceptions: Invalid credentials, an unsupported model type or native creation failures.

Example

collector, analyzer = aic.analyzer_pair(model, license_key)
config = aic.ProcessorConfig.optimal(model)
collector.initialize(config)

Collector

class

Buffers audio for later analysis. The collector is designed to be placed in the audio thread, buffering audio chunks for the Analyzer to analyze later. Created via analyzer_pair().

Collector.initialize()

def initialize(self, config: ProcessorConfig) -> None

Configures the collector for specific audio settings. This function must be called before buffering any audio. Using the sample rate and block size returned by Model.get_optimal_sample_rate() and Model.get_optimal_block_size() avoids internal resampling and rebuffering. Parameters

ProcessorConfig

required

Audio buffering configuration. See ProcessorConfig.

Raises

AudioConfigUnsupportedError: If the audio configuration is unsupported.

Do not call from audio processing threads as this allocates memory.

Example

config = aic.ProcessorConfig.optimal(model)
collector.initialize(config)

Collector.buffer()

def buffer(self, buffer: npt.NDArray[np.float32]) -> None

Buffers a one-dimensional NumPy float32 array of mono samples without changing it. Contiguous arrays are read directly and strided arrays are copied. Do not mutate input concurrently. Native work releases the GIL. Supply the configured fixed length or a length up to block_size when variable blocks are enabled. Wrong dtype or dimensionality raises TypeError. Parameters

npt.NDArray[np.float32]

required

1D NumPy array of mono float32 samples to be buffered.

Raises

NotInitializedError: If the collector has not been initialized.
AudioConfigMismatchError: If the buffer shape doesn’t match the configured audio settings.

Example

audio = np.zeros(config.block_size, dtype=np.float32)
collector.buffer(audio)

Analyzer

class

Runs an analysis model over the audio buffered by a Collector. Run analysis outside the audio thread. The analyzer safely reads audio from a collector on another thread. Created via analyzer_pair().

Analyzer.reset()

def reset(self) -> None

Requests a reset of analyzer and collector state and buffers. The collector applies it on the next collection pass. Until then, analysis reads a zero-filled snapshot instead of the previously published audio. Call this when the audio stream is interrupted or when seeking to prevent mispredictions from previous audio content. This operates on both the analyzer and its collector. The collector stays initialized to the configured settings.

Concurrency. Reset requests clear shared analyzer and collector state. Serialize reset with your application’s stream boundaries; Python calls are not a hard real-time guarantee.

Example

analyzer.reset()

Analyzer.analyze_buffered()

def analyze_buffered(self) -> AnalysisResult

Analyzes the buffered signal, releasing the GIL during native work. Calls on the same analyzer must be serialized. ProcessingNotAllowedError means no AnalysisResult is returned; other native failures use the SDK exception mapping. Analyzes a fixed-length window determined by the model. If the collector has buffered less audio, the input tail is padded with silence. Returns

AnalysisResult: An AnalysisResult.

This function is not real-time safe. Avoid calling it from audio threads.

Example

result = analyzer.analyze_buffered()
print(result.risk_score)

Analyzer.terminate_session()

def terminate_session(self) -> None

Terminates the analyzer’s telemetry session. Stop submitting audio and treat this session as closed once you request termination. Processing becomes disallowed when the native lifecycle task handles the signal. The call can return before that handling completes when other sessions remain alive; it is not proof of remote usage acknowledgment. The session is also stopped when the object is destroyed.

This method may block and is not real-time safe.

Analyzer.update_bearer_token()

def update_bearer_token(self, token: str) -> None

Replaces the bearer token on the analyzer. Use this when your license key is a JWT and needs to be refreshed before it expires. The replacement is used for subsequent authentication. A successful update does not prove backend acceptance or uninterrupted processing; continue handling processing errors. Both the original key and the new token must be JWTs; otherwise a TokenUnsupportedError is raised and the existing token stays in use. Parameters

str

required

The new JWT to install.

Raises

TokenUnsupportedError: If either the original or new token is not a JWT.
LicenseFormatInvalidError: If the token string contains null bytes.

Example

analyzer.update_bearer_token(renewed_jwt)

FileAnalyzer

class

Analyzes complete mono audio buffers. FileAnalyzer is a convenience wrapper around a Collector and Analyzer pair for non-real-time analysis of audio that is already loaded in memory. Each call to analyze() configures the collector for mono input with the model’s optimal block size. It analyzes independent five-second windows, advancing the start of each window by step_samples. For a complete file-analysis example, follow batch call analysis with Tyto. For streaming analysis, use analyzer_pair() directly. Example

analyzer = aic.FileAnalyzer(model, license_key)
results = analyzer.analyze(audio, 16000)
print(results[0].risk_score)

FileAnalyzer() constructor

FileAnalyzer(model: Model, license_key: str) -> FileAnalyzer

Creates a new file analyzer for a Tyto analysis model. The wrapper retains a strong reference to model. There is no file-path argument: decode the file into mono float32 samples before calling analyze(). Native resources are released when the object is released. This class has no public token-update or session-termination method; use the streaming Analyzer when those controls are needed. The collector is not initialized until analyze() is called. This lets the same FileAnalyzer instance analyze mono buffers with different sample rates or step sizes. Parameters

Model

required

The loaded model instance. See Model.

str

required

SDK key or JWT for the ai-coustics SDK (generate your key at developers.ai-coustics.com).

Raises

LicenseFormatInvalidError: If the license key string contains null bytes.
SDK exceptions: Invalid credentials, an unsupported model type or native creation failures.

Example

analyzer = aic.FileAnalyzer(model, license_key)

FileAnalyzer.analyze()

def analyze(
    self,
    audio: npt.NDArray[np.float32],
    sample_rate: int,
    step_samples: int | None = None,
) -> list[AnalysisResult]

Analyzes a complete mono audio buffer. The input must be a one-dimensional NumPy float32 array of mono samples at the actual sample_rate. There is no channel mixing. Native collection handles supported non-native input rates. The wrapper reads contiguous input directly and copies strided input; native analysis releases the GIL. Do not mutate the input concurrently. Wrong dtype or dimensionality raises TypeError. The analyzer evaluates five-second windows. FileAnalyzer buffers a window starting at sample 0, runs the analyzer once, resets, then repeats with a window starting step_samples later. If audio is empty, shorter than or equal to five seconds, it is padded with silence and a single result is returned. For longer signals, only complete five-second windows are analyzed after the first window. Parameters

npt.NDArray[np.float32]

required

1D NumPy array of mono float32 samples to analyze.

int

required

Sample rate of audio in Hz.

int | None

default:"None"

Number of samples to advance between analysis results. Defaults to 5 * sample_rate samples with no overlap. Must be greater than zero. Smaller values overlap windows; larger values leave gaps.

Returns

list[AnalysisResult]: A list of AnalysisResult values, one per analysis window.

Raises

AudioConfigUnsupportedError: If the sample rate or step size is unsupported.

Example

results = analyzer.analyze(audio, 16000)
print(results[0].risk_score)

float

read-only

Measure of audio dropouts or discontinuities in the stream, e.g. from packet loss, frame erasure, jitter or CPU overload. Lower indicates less problematic audio.Range: 0.0–1.0

See the Python API index, batch call analysis, real-time analysis and troubleshooting.

AnalysisResult.repr()

def __repr__(self) -> str

Returns a diagnostic string containing the object’s current fields. It is not a serialization format.

SDK

API reference

Upgrades

LiveKit

AI tools

Historical migrations

analyzer_pair()

Collector

Collector.initialize()

Collector.buffer()

Analyzer

Analyzer.reset()

Analyzer.analyze_buffered()

Analyzer.terminate_session()

Analyzer.update_bearer_token()

FileAnalyzer

FileAnalyzer() constructor

FileAnalyzer.analyze()

AnalysisResult

AnalysisResult.risk_score

AnalysisResult.speaker_reverb

AnalysisResult.speaker_loudness

AnalysisResult.interfering_speech

AnalysisResult.noise

AnalysisResult.codec_degradation

AnalysisResult.packet_loss

AnalysisResult.repr()

​analyzer_pair()

​Collector

​Collector.initialize()

​Collector.buffer()

​Analyzer

​Analyzer.reset()

​Analyzer.analyze_buffered()

​Analyzer.terminate_session()

​Analyzer.update_bearer_token()

​FileAnalyzer

​FileAnalyzer() constructor

​FileAnalyzer.analyze()

​AnalysisResult

​AnalysisResult.risk_score

​AnalysisResult.speaker_reverb

​AnalysisResult.speaker_loudness

​AnalysisResult.interfering_speech

​AnalysisResult.noise

​AnalysisResult.codec_degradation

​AnalysisResult.packet_loss

​AnalysisResult.repr()

analyzer_pair()

Collector

Collector.initialize()

Collector.buffer()

Analyzer

Analyzer.reset()

Analyzer.analyze_buffered()

Analyzer.terminate_session()

Analyzer.update_bearer_token()

FileAnalyzer

FileAnalyzer() constructor

FileAnalyzer.analyze()

AnalysisResult

AnalysisResult.risk_score

AnalysisResult.speaker_reverb

AnalysisResult.speaker_loudness

AnalysisResult.interfering_speech

AnalysisResult.noise

AnalysisResult.codec_degradation

AnalysisResult.packet_loss

AnalysisResult.repr()