Processing a Single Stream

This tutorial walks through the core Sanas SDK flow using the sdk_example.py example. You’ll read a WAV file, run it through a model, and write the processed audio back out. Along the way you’ll use two shared helper modules — wav_utils.py (stdlib-only WAV I/O) and helpers.py (pipeline readiness and real-time pacing) — which every example depends on.

The SDK works with interleaved float32 PCM in [-1, 1]. You bring your own audio; the SDK never opens a microphone or speaker for you.

The core flow

Every audio-processing session follows the same five steps:

create_sdk → activate_api_key (synchronous, returns an SdkResult)
create_audio_processor with an audio_pipeline_state_notify callback
Wait for PipelineState.RUNNING before feeding frames
Feed 20 ms frames at real time, then drain the buffered tail with silence
Save the processed audio

process_frame is synchronous and returns the processed frame directly. The model buffers ~50–100 ms internally, so the output lags the input. After the real audio is exhausted you push silence to pull the buffered tail back out.

Prerequisites

Set your credentials and pick an input file via environment variables and CLI flags:

Variable	Required	Default	Purpose
`SANAS_API_KEY`	Yes	—	Licence key
`SANAS_STORAGE_DIR`	No	`./storage`	SDK data + logs (created if missing)

export SANAS_API_KEY="your-key"
export SANAS_STORAGE_DIR=./storage

python sdk_example.py --model VI_G_SE --input test_input.wav --output output.wav

Command-line flags: --model (model key, default VI_G_SE), --input, --output, and --pcm16 (opt into raw 16-bit PCM uplink — cloud inference only).

Step 1: Create and activate the SDK

Activation is inline and blocking. Always check the returned SdkResult before continuing.

import sanas

sdk = sanas.create_sdk(sanas.InitParams(storage_dir=storage_dir))
res = sdk.activate_api_key(api_key)   # blocks, returns SdkResult
if not res.success:
    sys.exit(f"Activation failed: {res.message}")
print(f"[activation] OK (SDK {sdk.version})")

Step 2: Load the input audio

load_samples (from helpers.py) wraps read_wav and returns the audio as an array('f') ready to slice into AudioFrames.

from helpers import load_samples

samples, sample_rate, channels = load_samples(args.input)
print(f"[input] {sample_rate} Hz, {channels} ch, {len(samples) // channels} frames")

Under the hood, wav_utils.read_wav parses the WAV header directly (Python’s wave module only handles PCM) so it accepts both 16-bit PCM and 32-bit IEEE float input, including WAVE_FORMAT_EXTENSIBLE. It returns interleaved little-endian float32 regardless of host endianness.

Step 3: Build the processor with a state callback

The pipeline initializes asynchronously, so you must wait for PipelineState.RUNNING before feeding frames. PipelineWaiter (from helpers.py) bridges the audio_pipeline_state_notify callback to a waitable value.

from helpers import PipelineWaiter

waiter = PipelineWaiter(on_state=lambda s: print(f"[pipeline] {s}"))

cloud_params = sanas.CloudInferencingParams()
cloud_params.use_pcm16 = args.pcm16

attrs = sanas.ProcessorAttributes(
    audio_attributes=sanas.AudioAttributes(
        sampling_rate=sample_rate,
        channels=channels,
        model_name=args.model,
        cloud_inferencing_params=cloud_params,
        audio_pipeline_state_notify=waiter.callback,
    )
)

Why a Condition instead of a plain Event?

A threading.Event only signals “something happened” and carries no value. PipelineWaiter uses a threading.Condition that notifies on every state change while keeping the latest state in last_state. Waiters wake on any transition and then inspect the value to decide whether to proceed (RUNNING), fail (NOT_RUNNING), or keep waiting. The state callback fires on a background thread, so publishing under a lock is required.

Step 4: Feed frames and drain the tail

Open the processor, wait for RUNNING, then hand off to feed_and_drain.

with sdk.create_audio_processor(attrs) as proc:
    print("[pipeline] waiting for RUNNING ...")
    waiter.wait_until_running(timeout=30.0)

    def _progress(pushed, total):
        pct = 100.0 * pushed / total if total else 100.0
        print(f"\r[feed] {pct:5.1f}%", end="", flush=True)

    result = feed_and_drain(proc, samples, sample_rate, channels, progress=_progress)
    print()

feed_and_drain splits the audio into fixed 20 ms chunks and calls process_frame on each. When realtime=True (the default), it paces the feed against a fixed monotonic schedule so timing self-corrects and doesn’t drift:

next_deadline += chunk_seconds
sleep_until(next_deadline)

The final short chunk is padded with silence so no audio is dropped. After the input is exhausted, the helper drains the buffered tail. For low-latency models (NC/SE) it stops once the pipeline goes quiet (three consecutive empty frames); for high-latency pipelines you pass drain_seconds to pump silence for a fixed window instead.

sleep_until sleeps until ~1 ms before the deadline, then busy-waits the last millisecond for tighter pacing than a bare time.sleep().

Step 5: Save the output

from wav_utils import save_wav

combined = result["output"]
if combined:
    save_wav(args.output, combined, sample_rate, channels)
    print(f"[output] {len(combined) // 4} samples -> {args.output}")
else:
    print("[output] no processed frames received.")

save_wav clamps each sample to [-1, 1] and writes a 16-bit PCM WAV.

Full example

import argparse
import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)

import sanas
from helpers import PipelineWaiter, feed_and_drain, load_samples
from wav_utils import save_wav


def _run():
    parser = argparse.ArgumentParser(
        description="Process a WAV file through the Sanas SDK (single stream)."
    )
    parser.add_argument("--model", default="VI_G_SE", help="ml model key")
    parser.add_argument("--input", default="test_input.wav", help="input WAV path")
    parser.add_argument("--output", default="output.wav", help="output WAV path")
    parser.add_argument(
        "--pcm16",
        action="store_true",
        help="negotiate L16 (raw 16-bit PCM); cloud inference only",
    )
    args = parser.parse_args()

    api_key = os.environ.get("SANAS_API_KEY")
    if not api_key:
        sys.exit("SANAS_API_KEY is not set.")
    storage_dir = os.environ.get("SANAS_STORAGE_DIR", "./storage")

    sdk = sanas.create_sdk(sanas.InitParams(storage_dir=storage_dir))
    res = sdk.activate_api_key(api_key)
    if not res.success:
        sys.exit(f"Activation failed: {res.message}")
    print(f"[activation] OK (SDK {sdk.version})")

    samples, sample_rate, channels = load_samples(args.input)
    print(f"[input] {sample_rate} Hz, {channels} ch, {len(samples) // channels} frames")

    waiter = PipelineWaiter(on_state=lambda s: print(f"[pipeline] {s}"))
    cloud_params = sanas.CloudInferencingParams()
    cloud_params.use_pcm16 = args.pcm16
    attrs = sanas.ProcessorAttributes(
        audio_attributes=sanas.AudioAttributes(
            sampling_rate=sample_rate,
            channels=channels,
            model_name=args.model,
            cloud_inferencing_params=cloud_params,
            audio_pipeline_state_notify=waiter.callback,
        )
    )

    with sdk.create_audio_processor(attrs) as proc:
        print("[pipeline] waiting for RUNNING ...")
        waiter.wait_until_running(timeout=30.0)

        def _progress(pushed, total):
            pct = 100.0 * pushed / total if total else 100.0
            print(f"\r[feed] {pct:5.1f}%", end="", flush=True)

        result = feed_and_drain(proc, samples, sample_rate, channels, progress=_progress)
        print()

    combined = result["output"]
    if combined:
        save_wav(args.output, combined, sample_rate, channels)
        print(f"[output] {len(combined) // 4} samples -> {args.output}")
    else:
        print("[output] no processed frames received.")


if __name__ == "__main__":
    _run()

Next Steps

The core flow this benchmark scales up.

API Reference

Full SDK documentation for classes, enums, and callbacks.

Next Steps

Language Translation

Translate speech between languages and receive audio plus transcripts.

Processing Multiple Streams

Run many concurrent processors on one SDK instance.

References

SDK

SDK lifecycle methods including create_sdk, activate_api_key, and create_audio_processor.

AudioProcessor

process_frame method for audio processing.

Need Help?

Email Support

support@sanas.aiResponse time: 1 business day

Support Portal

Raise a support ticket for urgent issues

Getting Started

Models

Deployment

Tutorials/Examples

Enterprise

Resources

Processing a Single Stream

The core flow

Prerequisites

Step 1: Create and activate the SDK

Step 2: Load the input audio

Step 3: Build the processor with a state callback

Step 4: Feed frames and drain the tail

Step 5: Save the output

Full example

Next Steps