Processing a Single Stream - Sanas Developer Hub

Documentation Index

Fetch the complete documentation index at: /llms.txt

Use this file to discover all available pages before exploring further.

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:

1. create_sdk → activate_api_key (synchronous, returns an SdkResult)
2. create_audio_processor with an audio_pipeline_state_notify callback
3. Wait for PipelineState.RUNNING before feeding frames
4. Feed 20 ms frames at real time, then drain the buffered tail with silence
5. 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:

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()