Python enhancement

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 a complete file-processing example, follow the Python guide.

import typing
import numpy as np
import numpy.typing as npt
from aic_sdk import Model, Processor, ProcessorAsync, ProcessorConfig, ProcessorContext, ProcessorParameter, OtelConfig

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

Processor

class

Stateful mono audio enhancement. Use one processor per independent audio stream. Initialization and processing on the same synchronous instance must not overlap. A context can control parameters from another thread. Native resources are released when Python releases the object; explicit session termination is permanent for that instance. Example

model = Model.from_file("/path/to/model.aicmodel")
processor = Processor(model, license_key)
config = ProcessorConfig.optimal(model)
processor.initialize(config)
audio = np.zeros(config.block_size, dtype=np.float32)
enhanced = processor.process(audio)

Processor() constructor

Processor(
    model: Model,
    license_key: str,
    config: ProcessorConfig | None = None,
    otel_config: OtelConfig | None = None,
) -> Processor

Creates a new audio enhancement processor instance. Use separate processors for concurrent streams or different enhancement models. If a config is provided, the processor will be initialized immediately. Otherwise, you must call initialize() before processing audio. Parameters

Model

required

The loaded enhancement or bypass model instance. See Model.

str

required

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

ProcessorConfig | None

default:"None"

Optional audio processing configuration. If provided, the processor will be initialized immediately with this configuration. See ProcessorConfig.

OtelConfig | None

default:"None"

Per-instance OpenTelemetry settings. See OtelConfig.

Raises

SDK exceptions: Invalid credentials, unsupported model types or native creation failures.
AudioConfigUnsupportedError: If the supplied config is unsupported.

Example

# Create processor without initialization
processor = Processor(model, license_key)
processor.initialize(config)

# Or create and initialize in one step
config = ProcessorConfig.optimal(model)
processor = Processor(model, license_key, config)

Processor.initialize()

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

Configures the processor for specific audio settings. This function must be called before processing any audio. For the lowest delay use the sample rate and block size returned by Model.get_optimal_sample_rate() and Model.get_optimal_block_size(). Parameters

ProcessorConfig

required

Audio processing configuration. See ProcessorConfig.

Raises

AudioConfigUnsupportedError: If the audio configuration is unsupported.

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

Example

config = ProcessorConfig.optimal(model)
processor.initialize(config)

Processor.process()

def process(self, audio: npt.NDArray[np.float32]) -> npt.NDArray[np.float32]

Accepts a one-dimensional NumPy float32 array and returns a new one-dimensional float32 array with the same number of samples. The wrapper copies the input before native processing, so the input remains unchanged, including for strided views. It releases the Python GIL during native work. Pass exactly config.block_size samples, or 1 through config.block_size with variable_block_size=True. Values represent normalized audio, conventionally -1.0 to 1.0; the wrapper does not normalize integers or mix channels. Wrong dimensionality or dtype raises TypeError before processing. On an SDK error, Python raises an exception and returns no processed array. Implement an explicit fallback if your application must keep delivering audio. Raises

NotInitializedError: If the processor has not been initialized.
AudioConfigMismatchError: If the block size does not match the configuration.
ProcessingNotAllowedError: If processing is not authorized.

See also Block size and Real-time safety.

Processor.get_context()

def get_context(self) -> ProcessorContext

Creates a ProcessorContext instance. This can be used to control all parameters and other settings of the processor. Returns

ProcessorContext: A new ProcessorContext instance.

Example

processor_context = processor.get_context()

Processor.terminate_session()

def terminate_session(self) -> None

Terminates the processor’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.

ProcessorAsync

class

Async wrapper for Processor that offloads work to background threads. Awaitable initialization, processing and termination use a shared background processing pool. Set AIC_NUM_THREADS before the first pool use to override its default of available CPU parallelism. AIC_NUM_RUNTIME_THREADS separately controls the async runtime and defaults to one. The constructor is synchronous, including initialization when config is supplied. get_context() is also synchronous and can wait for in-flight work. To avoid synchronous initialization, omit config and await initialize_async(config). Calls on one instance serialize internally; await each block in stream order and bound work across streams. Do not infer an unbounded queue or cancellation guarantee from the async API. Example

model = Model.from_file("/path/to/model.aicmodel")
processor = ProcessorAsync(model, license_key)
config = ProcessorConfig.optimal(model)
await processor.initialize_async(config)
audio = np.zeros(config.block_size, dtype=np.float32)
enhanced = await processor.process_async(audio)

ProcessorAsync() constructor

ProcessorAsync(
    model: Model,
    license_key: str,
    config: ProcessorConfig | None = None,
    otel_config: OtelConfig | None = None,
) -> ProcessorAsync

Creates a new async audio enhancement processor instance. Use separate processors for concurrent streams or different enhancement models. If a config is provided, the processor will be initialized immediately. Otherwise, you must call initialize_async() before processing audio. Parameters

Model

required

The loaded enhancement or bypass model instance. See Model.

str

required

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

ProcessorConfig | None

default:"None"

Optional audio processing configuration. If provided, the processor will be initialized immediately with this configuration. See ProcessorConfig.

OtelConfig | None

default:"None"

Per-instance OpenTelemetry settings. See OtelConfig.

Raises

SDK exceptions: Invalid credentials, unsupported model types or native creation failures.
AudioConfigUnsupportedError: If the supplied config is unsupported.

Example

# Create processor without initialization
processor = ProcessorAsync(model, license_key)
await processor.initialize_async(config)

# Or create and initialize in one step
config = ProcessorConfig.optimal(model)
processor = ProcessorAsync(model, license_key, config)

ProcessorAsync.initialize_async()

async

def initialize_async(self, config: ProcessorConfig) -> typing.Awaitable[None]

Configures the processor asynchronously for specific audio settings. This function must be called before processing any audio. For the lowest delay use the sample rate and block size returned by Model.get_optimal_sample_rate() and Model.get_optimal_block_size(). Parameters

ProcessorConfig

required

Audio processing configuration. See ProcessorConfig.

Raises

AudioConfigUnsupportedError: If the audio configuration is unsupported.

Example

config = ProcessorConfig.optimal(model)
await processor.initialize_async(config)

ProcessorAsync.process_async()

async

def process_async(self, audio: npt.NDArray[np.float32]) -> typing.Awaitable[npt.NDArray[np.float32]]

Copies a one-dimensional NumPy float32 input on the calling thread, then returns an awaitable resolving to a new float32 array. Native processing runs in the background; input remains unchanged. The length, dtype and error rules of Processor.process() apply. Raises

NotInitializedError: If the processor has not been initialized.
AudioConfigMismatchError: If the block size does not match the configuration.
ProcessingNotAllowedError: If processing is not authorized.

ProcessorAsync.get_context()

def get_context(self) -> ProcessorContext

Returns a ProcessorContext for real-time parameter control. Returns

ProcessorContext: A new ProcessorContext instance.

Example

processor_context = processor.get_context()

ProcessorAsync.terminate_session_async()

async

def terminate_session_async(self) -> typing.Awaitable[None]

Terminates the processor’s telemetry session asynchronously. Await the returned awaitable and stop submitting audio. Treat the session as closed immediately; native processing becomes disallowed when the lifecycle task handles the signal. Other live sessions can allow termination handling to continue after the awaitable resolves. Await completion does not prove remote usage acknowledgment.

ProcessorContext

class

Shared control handle for processor state and parameters. There is no public ProcessorContext() constructor. Contexts from the same processor share control state. Retaining a context does not create a new processing stream or resume a terminated session. Created via Processor.get_context().

ProcessorContext.reset()

def reset(self) -> None

Requests a reset of enhancement state and buffers. The processing path applies the reset on its next processing call. Call this when the audio stream is interrupted or when seeking to prevent artifacts from previous audio content. The processor stays initialized to the configured settings.

Concurrency. The context can request a reset from another thread. Python calls still involve the interpreter; this is not a hard real-time guarantee for a Python callback.

Example

processor_context.reset()

ProcessorContext.set_parameter()

def set_parameter(self, parameter: ProcessorParameter, value: float) -> None

Modifies a processor parameter. Parameters can be changed through a context while audio is processed elsewhere. VoiceGain is a deprecated no-op: setting it emits DeprecationWarning and returns None; reading it emits the warning and returns 1.0. Parameters

ProcessorParameter

required

Parameter to modify. See ProcessorParameter.

float

required

New parameter value. See parameter documentation for ranges.

Raises

ParameterOutOfRangeError: If the parameter value is out of range.

Example

processor_context.set_parameter(ProcessorParameter.EnhancementLevel, 0.8)

ProcessorContext.get_parameter()

def get_parameter(self, parameter: ProcessorParameter) -> float

Retrieves the current value of a parameter. This function can be called from any thread. Parameters

ProcessorParameter

required

Parameter to query. See ProcessorParameter.

Returns

float: The current parameter value.

Example

level = processor_context.get_parameter(ProcessorParameter.EnhancementLevel)
print(f"Current enhancement level: {level}")

ProcessorContext.parameter()

deprecated

Deprecated. Use get_parameter() instead.

def parameter(self, parameter: ProcessorParameter) -> float

ProcessorContext.get_audio_delay()

def get_audio_delay(self) -> int

Returns the delay applied to the audio in samples for the current audio configuration. This reports SDK signal delay, including algorithmic delay and internal buffering. It does not measure CPU execution, scheduling, transport or application queue time. The processed audio leaves Processor.process() this many samples behind its input. It does not include VAD delay; use VadContext.get_prediction_delay() for a separate VAD. Delay behavior.

Before initialization: Returns the base processing delay using the model’s optimal block size at its native sample rate
After initialization: Returns the actual delay for your specific configuration, including any additional buffering introduced by a non-optimal block size

Returns

int: The delay in samples.

After initialization, delay is expressed in samples at the configured sample rate; before initialization, use the model native rate. To convert to time units: delay_ms = (delay_samples * 1000) / sample_rate

Using a block size different from the optimal value returned by get_optimal_block_size() will increase the delay beyond the model’s base latency.

Example

delay = processor_context.get_audio_delay()
print(f"Audio delay: {delay} samples")

ProcessorContext.update_bearer_token()

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

Replaces the bearer token on the running processor. 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 error 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

processor_context.update_bearer_token(renewed_jwt)

ProcessorParameter

enum

Parameter constants for audio enhancement. Use the named constants directly. The stub presents this type as an enum; the runtime exposes PyO3 enum-like objects, so do not depend on standard-library enum.Enum iteration or .value behavior.

ProcessorParameter members

ProcessorParameter.Bypass

Controls whether audio processing is bypassed while preserving algorithmic delay. When enabled, the input audio passes through unmodified, but the output is still delayed by the same amount as during normal processing. The delay remains when switching between bypass and enhancement. Range: 0.0–1.0

0.0: Enhancement active (normal processing)
Any value greater than 0.0 up to 1.0: Bypass enabled (latency-compensated passthrough); reading the value returns 1.0

Default: 0.0

ProcessorParameter.EnhancementLevel

Tune enhancement strength for your speech-to-text (STT) engine or listening task. The exact behavior depends on the active model:

Quail models: Controls how aggressively the model suppresses noise. When used with Quail Voice Focus, it also suppresses background and competing speech.
Rook models: Controls the mixback and therefore the intensity of the enhancement.

Range: 0.0–1.0. The initial enhancement level can be model-specific; read it with get_parameter() when needed.

ProcessorParameter.VoiceGain

deprecated

This parameter has no effect and will be removed in a future version.

Retained for compatibility. Setting any value emits DeprecationWarning and has no effect; reading returns 1.0 with the same warning. See the Python API index, Python guide and troubleshooting.

SDK

API reference

Upgrades

LiveKit

AI tools

Historical migrations

Processor

Processor() constructor

Processor.initialize()

Processor.process()

Processor.get_context()

Processor.terminate_session()

ProcessorAsync

ProcessorAsync() constructor

ProcessorAsync.initialize_async()

ProcessorAsync.process_async()

ProcessorAsync.get_context()

ProcessorAsync.terminate_session_async()

ProcessorContext

ProcessorContext.reset()

ProcessorContext.set_parameter()

ProcessorContext.get_parameter()

ProcessorContext.parameter()

ProcessorContext.get_audio_delay()

ProcessorContext.update_bearer_token()

ProcessorParameter

ProcessorParameter members

ProcessorParameter.Bypass

ProcessorParameter.EnhancementLevel

ProcessorParameter.VoiceGain

​Processor

​Processor() constructor

​Processor.initialize()

​Processor.process()

​Processor.get_context()

​Processor.terminate_session()

​ProcessorAsync

​ProcessorAsync() constructor

​ProcessorAsync.initialize_async()

​ProcessorAsync.process_async()

​ProcessorAsync.get_context()

​ProcessorAsync.terminate_session_async()

​ProcessorContext

​ProcessorContext.reset()

​ProcessorContext.set_parameter()

​ProcessorContext.get_parameter()

​ProcessorContext.parameter()

​ProcessorContext.get_audio_delay()

​ProcessorContext.update_bearer_token()

​ProcessorParameter

​ProcessorParameter members

​ProcessorParameter.Bypass

​ProcessorParameter.EnhancementLevel

​ProcessorParameter.VoiceGain

Processor

Processor() constructor

Processor.initialize()

Processor.process()

Processor.get_context()

Processor.terminate_session()

ProcessorAsync

ProcessorAsync() constructor

ProcessorAsync.initialize_async()

ProcessorAsync.process_async()

ProcessorAsync.get_context()

ProcessorAsync.terminate_session_async()

ProcessorContext

ProcessorContext.reset()

ProcessorContext.set_parameter()

ProcessorContext.get_parameter()

ProcessorContext.parameter()

ProcessorContext.get_audio_delay()

ProcessorContext.update_bearer_token()

ProcessorParameter

ProcessorParameter members

ProcessorParameter.Bypass

ProcessorParameter.EnhancementLevel

ProcessorParameter.VoiceGain