Metadata-Version: 2.4
Name: datalabrotterdam-nova-sdk
Version: 1.6.0.dev3
Summary: Python runtime client for the Nova AI API
Author: Datalab Rotterdam
License-Expression: MIT
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# Python SDK

Python client for the Nova AI API.

The package is intentionally small:

- sync methods
- no external dependencies
- typed request and response structures
- simple streaming support for chat completions with a normal `for` loop
- audio helpers for transcription, translation, speaker diarization, and speech generation

## Install

```bash
pip install datalabrotterdam-nova-sdk
```

## Create A Client

```python
from nova_ai import NovaAIClient

client = NovaAIClient(
    api_key="sk_live_..."
)
```

If you do not pass `base_url`, the client uses `https://api.nova.datalabrotterdam.nl/v1`.

## Example: Normal Chat

```python
from nova_ai import NovaAIClient

client = NovaAIClient(
    api_key="sk_live_..."
)

completion = client.chat.completions.create({
    "model": "llama3.2",
    "messages": [
        {"role": "system", "content": "You are helpful."},
        {"role": "user", "content": "Explain embeddings in one sentence."}
    ],
    "max_tokens": 256,
    "temperature": 0.7,
})

print(completion["choices"][0]["message"]["content"])
print(completion.get("usage_source"))
print(completion.get("metrics", {}).get("tokens_per_second"))
```

`completion["usage_source"]` is:
- `"provider"` when the upstream provider reported usage
- `"gateway_estimated"` when Nova AI estimated usage to keep the contract consistent

## Example: Streaming Chat

```python
from nova_ai import NovaAIClient, collect_chat_text

client = NovaAIClient(
    api_key="sk_live_..."
)

stream = client.chat.completions.stream({
    "model": "llama3.2",
    "messages": [{"role": "user", "content": "Write three short tips."}]
})

for event in stream:
    if event["type"] == "chunk":
        choices = event["data"].get("choices", [])
        if choices:
            print(choices[0].get("delta", {}).get("content", ""), end="")

print()

print(collect_chat_text(client.chat.completions.stream({
    "model": "llama3.2",
    "messages": [{"role": "user", "content": "Summarize vectors in one line."}]
})))
```

## Example: Embeddings

```python
from nova_ai import NovaAIClient

client = NovaAIClient(
    api_key="sk_live_..."
)

embedding = client.embeddings.create({
    "model": "nomic-embed-text",
    "input": "Nova AI"
})

print(len(embedding["data"][0]["embedding"]))
```

## Example: Audio Transcription

```python
from nova_ai import NovaAIClient

client = NovaAIClient(
    api_key="sk_live_..."
)

transcript = client.audio.transcriptions.create(
    file_path="./sample.wav",
    model="whisper-1",
    language="nl",
)

print(transcript["text"])
```

## Example: Speaker Diarization

Returns who spoke when (time segments with a speaker label), not the words. Combine it with a `verbose_json` transcription to get a transcript per speaker.

```python
from nova_ai import NovaAIClient

client = NovaAIClient(
    api_key="sk_live_..."
)

result = client.audio.diarizations.create(
    file_path="./meeting.wav",
    model="nemotron-3-diarization",
)

for segment in result["segments"]:
    print(f'{segment["speaker"]}: {segment["start"]}s - {segment["end"]}s')
```

Labels such as `speaker_0` only mean something within one response or stream.

## Example: Live Diarization

`client.audio.diarizations.stream_url(model)` builds the `wss://` URL. The SDK has no WebSocket dependency, so bring your own client (here [`websockets`](https://pypi.org/project/websockets/)), stream 16 kHz mono PCM16 binary frames, and send `{"type": "end"}` when done. The full event protocol is in the gateway's API docs (`WS /v1/audio/diarization/stream`).

```python
import json
import threading

from websockets.sync.client import connect

# Raw 16 kHz mono PCM16, e.g. `ffmpeg -i meeting.mp3 -f s16le -ac 1 -ar 16000 meeting.pcm`
with open("meeting.pcm", "rb") as source:
    pcm = source.read()

url = client.audio.diarizations.stream_url("nemotron-3-diarization")
with connect(url, additional_headers={"Authorization": "Bearer sk_live_..."}) as ws:
    assert json.loads(ws.recv())["type"] == "ready"

    def send_audio():
        for offset in range(0, len(pcm), 3200):  # ~100 ms per frame
            ws.send(pcm[offset:offset + 3200])
        ws.send(json.dumps({"type": "end"}))

    threading.Thread(target=send_audio, daemon=True).start()
    for message in ws:
        event = json.loads(message)
        if event["type"] == "segment":
            print(event["speaker"], event["start"], event["end"])
        elif event["type"] in ("done", "error"):
            break
```

## Example: Speech Synthesis

```python
from nova_ai import NovaAIClient

client = NovaAIClient(
    api_key="sk_live_..."
)

audio = client.audio.speech.create({
    "model": "tts-1",
    "input": "Hello from Nova AI",
    "voice": "alloy",
    "response_format": "wav",
})

with open("speech.wav", "wb") as target:
    target.write(audio)
```

List the voices a model offers (values for `voice`):

```python
voices = client.audio.voices.list("kokoro")["voices"]
```

## Included Methods

- `client.providers.list()`
- `client.models.list(limit=None, after=None)`
- `client.chat.completions.create(payload, provider_configuration_id=None, request_id=None)`
- `client.chat.completions.stream(payload, provider_configuration_id=None, request_id=None)`
- `client.embeddings.create(payload, provider_configuration_id=None, request_id=None)`
- `client.audio.transcriptions.create(file_path=..., model=..., ...)`
- `client.audio.translations.create(file_path=..., model=..., ...)`
- `client.audio.diarizations.create(file_path=..., model=..., provider_configuration_id=None, request_id=None)`
- `client.audio.diarizations.stream_url(model)`
- `client.audio.speech.create(payload, provider_configuration_id=None, request_id=None)`
- `client.audio.voices.list(model, provider_configuration_id=None, request_id=None)`

## Audio Notes

- `client.audio.transcriptions.create()`, `client.audio.translations.create()` and `client.audio.diarizations.create()` upload audio as `multipart/form-data`.
- `client.audio.speech.create()` returns raw bytes.
- The gateway enforces runtime scopes:
  - `audio:transcribe`
  - `audio:translate`
  - `audio:diarize` (upload and live stream)
  - `audio:synthesis` (speech and voice listing)

## Why streaming uses `for`, not `async for`

For this Python SDK, a normal iterator is the simplest way to support streaming without extra dependencies. That means you can stream like this:

```python
for event in client.chat.completions.stream({...}):
    ...
```

Using `async for` is also possible, but it usually means introducing an async HTTP client such as `httpx` or `aiohttp`, which makes the first version of the SDK heavier.
