Metadata-Version: 2.4
Name: arctan-vi
Version: 0.7.2
Summary: Arctan Voice Isolation for voice agents
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: cryptography<49,>=48.0.1
Requires-Dist: numpy<3,>=2
Requires-Dist: sentry-sdk>=2.64.0
Provides-Extra: livekit
Requires-Dist: livekit<2,>=1.0.25; extra == "livekit"
Provides-Extra: pipecat
Requires-Dist: pipecat-ai<2,>=1.3; python_version >= "3.11" and extra == "pipecat"

# Arctan Voice Isolation

Arctan Voice Isolation is a Python SDK for real-time voice isolation in voice
agents and audio pipelines.

The SDK supports:

- standard Numpy audio processing
- LiveKit audio input processing
- Pipecat audio input filtering

An Arctan SDK key is required before audio can be processed.

## Installation

Install the base SDK:

```bash
pip install arctan-vi
```

Install with framework adapters:

```bash
pip install "arctan-vi[livekit]"
pip install "arctan-vi[pipecat]"
```

The base SDK and LiveKit adapter support Python 3.10+. The Pipecat adapter
requires Python 3.11+.

Set your SDK key in the runtime environment:

```bash
export ARCTAN_SDK_KEY="arc_sk_live_..."
```

You can also pass `license_key="arc_sk_live_..."` directly to the processor or
adapter constructors. An explicit `license_key` takes precedence over
`ARCTAN_SDK_KEY`.

## Standard Python

### Process one stream

The existing single-stream API remains supported. It creates and owns its model
internally:

```python
import numpy as np
import arctan

config = arctan.ProcessorConfig(sample_rate=24000, num_channels=1)
processor = arctan.Processor(config=config)

try:
    audio = np.zeros((config.num_channels, config.num_frames), dtype=np.float32)
    enhanced = processor.process(audio)
finally:
    processor.close()
```

### Process concurrent streams

For multiple streams in one worker process, create one shared `Model` and one
`ProcessorAsync` per call or audio stream. `ProcessorAsync` runs blocking native
inference off the event loop while keeping frames within each stream ordered.

```python
import asyncio
import numpy as np
import arctan


async def main() -> None:
    model = arctan.Model()
    configs = [arctan.ProcessorConfig.optimal(model) for _ in range(2)]
    processors = [
        arctan.ProcessorAsync(model=model, config=config)
        for config in configs
    ]

    try:
        frames = [
            np.zeros((config.num_channels, config.num_frames), dtype=np.float32)
            for config in configs
        ]
        enhanced = await asyncio.gather(
            *(
                processor.process_async(frame)
                for processor, frame in zip(processors, frames)
            )
        )
        print(len(enhanced))
    finally:
        for processor in processors:
            processor.close()
        model.close()


asyncio.run(main())
```

Use one shared `Model` per worker process and exactly one processor per call or
audio stream. Do not share one processor between independent streams or submit
one stream's frames out of order. Different processors may execute concurrently,
but calls on the same `ProcessorAsync` are serialized. Close every processor
before closing its shared model.

The number of processors is the number of active streams, not a universal
capacity guarantee. Benchmark the deployment machine using representative
audio and retain CPU and RSS headroom for transport, codecs, logging, and
application work.

`Processor.process()` accepts `float32` audio shaped as `channels x frames`.
Samples must be finite and in `[-1.0, 1.0]`. The output is a new `float32`
array with the same shape.

Mono and stereo input are supported. Stereo input is downmixed for voice
isolation and returned as dual-mono stereo. The sample rate must remain stable
for the lifetime of a processor.

If `num_frames` is omitted, initialization fills it with the native frame size
for the selected sample rate. If you set `num_frames`, it must contain a whole
number of native processing blocks.

## LiveKit

Install the LiveKit extra:

```bash
pip install "arctan-vi[livekit]"
```

Create an Arctan noise-cancellation processor and pass it to LiveKit audio input
options:

```python
from arctan.livekit import arctan_enhancer
from livekit.agents import room_io

await session.start(
    agent=Assistant(),
    room=ctx.room,
    room_options=room_io.RoomOptions(
        audio_input=room_io.AudioInputOptions(
            noise_cancellation=arctan_enhancer.noise_canceller(),
        ),
    ),
)
```

You can pass SDK options directly to Arctan:

```python
noise_cancellation = arctan_enhancer.noise_canceller(
    license_key="arc_sk_live_...",
)
```

LiveKit mono and stereo frames are supported. Stereo frames are downmixed for
voice isolation and returned as dual-mono stereo. The sample rate must remain
stable, and each input frame must contain a whole number of native processing
blocks.

Create a separate noise-canceller instance for every participant, call, or
independent audio input. Do not share one instance across concurrent streams.

## Pipecat

Install the Pipecat extra:

```bash
pip install "arctan-vi[pipecat]"
```

Create an `ArctanAudioFilter` and pass it as the input audio filter for your
transport:

```python
from arctan.pipecat import arctan_enhancer
from pipecat.transports.base_transport import TransportParams

audio_filter = arctan_enhancer.ArctanAudioFilter()

transport_params = TransportParams(
    audio_in_enabled=True,
    audio_in_filter=audio_filter,
)
```

`ArctanAudioFilter` implements Pipecat's `BaseAudioFilter`. Pipecat calls
`start(sample_rate)`, `filter(audio)`, and `stop()` through the transport
lifecycle.

The Pipecat adapter accepts mono PCM16 input. Arbitrary byte chunks are buffered
until a complete native processing block is available, so `filter()` may return
`b""`. `FilterEnableFrame` can bypass or re-enable processing; toggling clears
any incomplete buffered block.

Create a separate `ArctanAudioFilter` for every call or pipeline. Do not share
one filter across independent concurrent streams.
