# Sanas SDK — Python

This archive contains everything you need to run the Sanas SDK from Python:

```bash
sanas-<version>-<pytag>-<os>-<arch>/
├── examples/                 # runnable examples + shared helpers
├── sanas-<version>-…​.whl    # the Python wheel (one, matching <pytag>)
└── USAGE.md                  # this file
```

---

## 1. What is this SDK?

The Sanas SDK processes speech audio in real time. You feed it audio frames and it
returns processed audio frames — for example noise-cancelled, speech-enhanced, or
accent-converted speech, or speech translated into another language.

The Python package ships as a **self-contained wheel**: everything the SDK needs is
bundled inside it, so there is no separate native library to install or manage. You
just create a virtual environment and `pip install` the wheel.

You work with **interleaved float32 PCM** audio: push frames in with `process_frame`
and read processed frames back. The SDK does not open microphones or speakers for
you — you bring your own audio (from a file, a socket, a media stream, etc.).

## 2. What can it do?

- **Audio processing** — run a model (noise cancellation, speech enhancement, accent
  conversion, …) over your audio, selected by a model key.
- **Language Translation (LT)** — translate speech from one language to another and
  receive both translated audio and transcript segments.
- **Pipeline lifecycle** — an `audio_pipeline_state_notify` callback reports the
  processing pipeline state (`PipelineState`); wait for `RUNNING` before feeding
  frames.

Key types: `InitParams`, `create_sdk`, `Sdk`, `AudioAttributes`,
`ProcessorAttributes`, `AudioProcessor`, `AudioFrame`, `LanguageTranslationConfig`,
`CloudInferencingParams`, `PipelineState`, `SdkResult`.

## 3. Install (create a venv, install the wheel)

You only need a supported CPython and `pip` — no extra tooling required. The wheel
in this archive targets the Python version named in its filename:

| Python version   | Wheel tag       | Notes                                   |
|------------------|-----------------|-----------------------------------------|
| 3.10             | `cp310-cp310`   | version-specific                        |
| 3.11             | `cp311-cp311`   | version-specific                        |
| 3.12, 3.13, 3.14 | `cp312-abi3`    | one Stable-ABI wheel covers 3.12+       |

> Pick the archive whose `<pytag>` matches your interpreter (e.g. `cp310` for Python
> 3.10; `cp312` for Python 3.12 **or newer**).

### macOS / Linux

```bash
# from inside this extracted archive directory

# With install/builtin python
python3.10 -m venv .venv            # use the python that matches the wheel tag

# With UV
uv venv .venv --seed --python 3.10  # use uv to manage python version and --seed adds pip

. .venv/bin/activate
pip install --upgrade pip
pip install sanas-*.whl

python -c "import sanas; print('sanas', sanas.__version__)"
```

### Windows (PowerShell)

```powershell
py -3.10 -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install (Get-ChildItem sanas-*.whl).FullName

python -c "import sanas; print('sanas', sanas.__version__)"
```

The `examples/` folder is self-contained (standard-library helpers only, no numpy).
Run an example from inside the extracted archive with your venv active.

```bash
# macOS / Linux
export SANAS_API_KEY="your-key"
export SANAS_STORAGE_DIR=./storage     # optional; SDK data + logs (default: ./storage)

# Windows (PowerShell)
$env:SANAS_API_KEY = "your-key"
$env:SANAS_STORAGE_DIR = "./storage"   # optional; SDK data + logs (default: ./storage)

# Then run any example — the same commands work on every platform:
# model + input WAV are CLI args (defaults: --model VI_G_SE, --input test_input.wav)
python examples/sdk_example.py --model VI_G_SE --input test_input.wav  # audio processing
python examples/language_translation_example.py --input test_input.wav  # language translation
python examples/multi_stream_example.py --streams 4  # concurrency benchmark
```

`SANAS_STORAGE_DIR` is where the SDK keeps its data and writes logs (under
`storage_dir/logs`).

For load/soak testing, `multi_stream_example.py` runs N processors concurrently on
one SDK and reports success rate, peak concurrency, and (with `psutil` installed)
process memory growth — useful for spotting leaks. Key flags: `--streams`,
`--concurrency`, `--rate` (ramp stream starts), `--max-throughput`, `--pcm16`, and
`--save-wav-out-period N` (save every Nth stream's processed audio to
`stream_<idx>.wav`; default 0 = don't save, keeping the run allocation-free for an
accurate memory reading). Example:

```bash
python examples/multi_stream_example.py --streams 64 --concurrency 8 --rate 5
```

## 4. Example snippets

Both snippets assume a valid API key and a model key (see 4a for available keys).

Two things worth understanding before you read the code:

- **The pipeline initializes asynchronously.** After you create a processor, wait for
  `PipelineState.RUNNING` (via `audio_pipeline_state_notify`) before feeding frames.
- **`process_frame` is synchronous but expects real-time audio.** Feed frames at the
  rate they would arrive live (one 20 ms frame every 20 ms). The snippets below pace
  with a monotonic-clock deadline so timing does not drift.

### 4a. Audio processing

Set `model_name` to one of the model keys enabled for your account:

| Category                    | Model keys                                             |
|-----------------------------|--------------------------------------------------------|
| Agentic Speech Enhancement  | `AGENTIC_ST_SE`, `AGENTIC_VI_GT_SE`, `AGENTIC_VI_G_SE` |
| AT                          | `AT5.2`                                                |
| Speech Enhancement          | `SE1.2`, `SE2.1`, `SE2.2`, `VI_G_SE`                   |

```python
import array
import threading
import time
import sanas
from examples.wav_utils import read_wav, save_wav  # or copy these helpers

# 1. Create + activate (activation is synchronous and returns an SdkResult).
sdk = sanas.create_sdk(sanas.InitParams(storage_dir="./storage"))
res = sdk.activate_api_key("YOUR_API_KEY")
if not res.success:
    raise RuntimeError(f"activation failed ({res.error_type}): {res.message}")

# 2. Read input audio as interleaved float32.
float32_bytes, sample_rate, channels = read_wav("test_input.wav")
samples = array.array("f"); samples.frombytes(float32_bytes)

# 3. Get notified when the pipeline is ready. The state callback fires on a
#    background thread, so publish each state under a Condition and let the main
#    thread wait for a terminal state (RUNNING = go, NOT_RUNNING = failed). A
#    Condition (unlike a bare Event) lets us wake on any change and inspect which
#    state arrived.
cond = threading.Condition()
state = {"value": None}
def on_state(s):
    with cond:
        state["value"] = s
        cond.notify_all()

attrs = sanas.ProcessorAttributes(
    audio_attributes=sanas.AudioAttributes(
        sampling_rate=sample_rate,
        channels=channels,
        model_name="<your-model-key>",    # from your Sanas Developer Docs
        audio_pipeline_state_notify=on_state,
    )
)

# 4. Feed 20 ms frames at real time; process_frame returns the processed frame.
frame_len = sample_rate * channels // 50      # samples in 20 ms
frame_period = 0.020                          # seconds per frame
out_bytes = []
_terminal = (sanas.PipelineState.RUNNING, sanas.PipelineState.NOT_RUNNING)
with sdk.create_audio_processor(attrs) as proc:
    with cond:                                # wait until the pipeline settles
        cond.wait_for(lambda: state["value"] in _terminal, timeout=30)
    if state["value"] != sanas.PipelineState.RUNNING:
        raise RuntimeError(f"pipeline did not start (state: {state['value']})")

    # `deadline` is the wall-clock time the NEXT frame should be sent. After each
    # frame we advance it by one frame period and sleep until we reach it, so the
    # feed tracks real time instead of drifting with per-frame processing cost.
    deadline = time.monotonic()
    for i in range(0, len(samples) - frame_len + 1, frame_len):
        frame = proc.process_frame(sanas.AudioFrame(samples=samples[i:i + frame_len]))
        if frame.frame_count:
            out_bytes.append(bytes(frame.samples))   # memoryview is transient; copy
        deadline += frame_period
        time.sleep(max(0.0, deadline - time.monotonic()))

    # Drain the buffered tail by pushing silence (still paced at real time).
    silence = array.array("f", bytes(frame_len * 4))
    for _ in range(10):
        frame = proc.process_frame(sanas.AudioFrame(samples=silence))
        if frame.frame_count:
            out_bytes.append(bytes(frame.samples))
        deadline += frame_period
        time.sleep(max(0.0, deadline - time.monotonic()))

save_wav("output.wav", b"".join(out_bytes), sample_rate, channels)
```

> `examples/helpers.py` provides `PipelineWaiter`, `sleep_until`, and `feed_and_drain`,
> which wrap steps 3–4 above — prefer them over hand-rolling the loop.

### 4a-i. Cloud inference tuning (`CloudInferencingParams`)

When the SDK runs on **cloud / remote inference** (the default), you can pass a
`CloudInferencingParams` on `AudioAttributes.cloud_inferencing_params` to tune the
remote path. These parameters are **only meaningful for cloud inference and are
ignored for local inference**. The field is optional — omit it to keep the defaults.

Currently it exposes a single knob:

| Field        | Type   | Default | Meaning                                                                                                                                              |
|--------------|--------|---------|------------------------------------------------------------------------------------------------------------------------------------------------------|
| `use_pcm16`  | `bool` | `False` | Negotiate L16 (raw 16-bit linear PCM) at the session sample rate instead of the rate-default codec. Preserves fidelity and avoids server-side transcoding, at the cost of more bandwidth. |

Build the params object and attach it (default-construct, then set the field):

```python
cloud_params = sanas.CloudInferencingParams()
cloud_params.use_pcm16 = True          # opt into raw 16-bit linear PCM uplink

attrs = sanas.ProcessorAttributes(
    audio_attributes=sanas.AudioAttributes(
        sampling_rate=sample_rate,
        channels=channels,
        model_name="<your-model-key>",
        cloud_inferencing_params=cloud_params,   # optional; omit for defaults
        audio_pipeline_state_notify=on_state,
    )
)
```

You can also read/set it on an existing `AudioAttributes`:

```python
audio_attrs = sanas.AudioAttributes(model_name="<your-model-key>")
audio_attrs.cloud_inferencing_params.use_pcm16 = True
```

> Leave `cloud_inferencing_params` unset (or `use_pcm16=False`) unless you
> specifically need raw PCM uplink — the rate-default codec is the recommended
> default for most sessions.

> **Language Translation does not support `use_pcm16`.** L16 applies only to the
> audio-processing path (SE / NC models); leave `cloud_inferencing_params` unset
> for the LT (`lt_config`) path. Accordingly, `sdk_example.py` and
> `multi_stream_example.py` expose a `--pcm16` flag, but
> `language_translation_example.py` does not.

### 4b. Language Translation

Language Translation is selected by the **presence of `lt_config`** (not by a model
key), so leave `model_name` empty. Translated audio comes back from `process_frame`;
transcripts arrive on the `lt_config` callback.

```python
import array
import threading
import time
import sanas
from examples.wav_utils import read_wav, save_wav

sdk = sanas.create_sdk(sanas.InitParams(storage_dir="./storage"))
res = sdk.activate_api_key("YOUR_API_KEY")
if not res.success:
    raise RuntimeError(f"activation failed ({res.error_type}): {res.message}")

float32_bytes, sample_rate, channels = read_wav("test_input.wav")
samples = array.array("f"); samples.frombytes(float32_bytes)

def on_transcript(tf):
    kind = "Translation" if tf.type_ == sanas.TranscriptType.TRANSLATION else "Transcription"
    for seg in tf.transcript_data_.complete:
        print(f"[{kind}] {seg.text}")

# Publish pipeline states under a Condition and wait for a terminal one (see the
# audio-processing snippet for why a Condition is used instead of an Event).
cond = threading.Condition()
state = {"value": None}
def on_state(s):
    with cond:
        state["value"] = s
        cond.notify_all()

attrs = sanas.ProcessorAttributes(
    audio_attributes=sanas.AudioAttributes(
        sampling_rate=sample_rate,
        channels=channels,
        model_name="",                    # LT is selected by lt_config, not a key
        audio_pipeline_state_notify=on_state,
    ),
    lt_config=sanas.LanguageTranslationConfig(
        language_in="en-US",
        language_out="es-ES",
        conversation_id="",               # optional; links two-party sessions
        callback=on_transcript,
    ),
)

frame_len = sample_rate * channels // 50      # samples in 20 ms
frame_period = 0.020                          # seconds per frame
out_bytes = []
_terminal = (sanas.PipelineState.RUNNING, sanas.PipelineState.NOT_RUNNING)
with sdk.create_audio_processor(attrs) as proc:
    with cond:
        cond.wait_for(lambda: state["value"] in _terminal, timeout=30)
    if state["value"] != sanas.PipelineState.RUNNING:
        raise RuntimeError(f"pipeline did not start (state: {state['value']})")

    # Pace the feed to real time (see the audio-processing snippet for details).
    deadline = time.monotonic()
    for i in range(0, len(samples) - frame_len + 1, frame_len):
        frame = proc.process_frame(sanas.AudioFrame(samples=samples[i:i + frame_len]))
        if frame.frame_count:
            out_bytes.append(bytes(frame.samples))
        deadline += frame_period
        time.sleep(max(0.0, deadline - time.monotonic()))

    # Translation adds several seconds of latency, so keep pushing silence for a few
    # seconds after the input ends to pull back the translated tail.
    silence = array.array("f", bytes(frame_len * 4))
    for _ in range(int(5.0 / frame_period)):  # ~5 s of silence
        frame = proc.process_frame(sanas.AudioFrame(samples=silence))
        if frame.frame_count:
            out_bytes.append(bytes(frame.samples))
        deadline += frame_period
        time.sleep(max(0.0, deadline - time.monotonic()))

save_wav("translated.wav", b"".join(out_bytes), sample_rate, channels)
```

---

For the full, runnable versions of both snippets (with real-time pacing and precise
draining), see `examples/sdk_example.py` and `examples/language_translation_example.py`.
