Velma Streaming

import asyncio
import json
import aiohttp

API_KEY = "YOUR_API_KEY"
AUDIO_FILE = "conversation.ogg"
CHUNK_SIZE = 8192

# Either the string "default" or a JSON-encoded BatchConfig.
CONFIG = json.dumps({
    "behaviors": ["preset:conflict_de_escalation", "preset:complaints"],
    "stt": {"speaker_diarization": True, "emotion_signal": True},
    "produce_topics": True,
    "produce_topic_sentiments": True,
    "produce_summary": True,
})

async def analyze_streaming():
    url = f"wss://platform.modulate.ai/api/velma-2-streaming?api_key={API_KEY}"

    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(url) as ws:
            # 1. Send the config frame before any audio.
            await ws.send_str(CONFIG)

            async def send_audio():
                with open(AUDIO_FILE, "rb") as f:
                    while chunk := f.read(CHUNK_SIZE):
                        await ws.send_bytes(chunk)
                        await asyncio.sleep(CHUNK_SIZE / 4000)
                await ws.send_str("")  # 2. Signal end of audio.

            send_task = asyncio.create_task(send_audio())

            try:
                async for msg in ws:
                    if msg.type == aiohttp.WSMsgType.TEXT:
                        event = json.loads(msg.data)
                        etype = event["type"]
                        if etype == "clip":
                            c = event["clip"]
                            print(f"[{c['speaker_label']}] {c['text']}")
                        elif etype == "partial_clip":
                            pc = event["partial_clip"]
                            print(f"[partial {pc['clip_uuid'][:8]}] {pc['text']}")
                        elif etype == "clip_update":
                            cu = event["clip_update"]
                            print(f"[update {cu['clip_uuid'][:8]}] emotion={cu.get('emotion')} accent={cu.get('accent')}")
                        elif etype == "conversation_type":
                            print(f"Conversation type: {event['pick']['name']}")
                        elif etype == "participant_role":
                            p = event["pick"]
                            print(f"Role for {p['speaker_label']}: {p['name']}")
                        elif etype == "behavior_detection":
                            d = event["detection"]
                            print(f"Behavior {d['behavior_name']}: detected={d['detected']}")
                        elif etype == "topics":
                            print(f"Topics: {', '.join(event['topics'])}")
                        elif etype == "topic_sentiment":
                            ts = event["topic_sentiment"]
                            print(f"Sentiment ({ts['topic']}, {ts['speaker_label']}): {ts['sentiment_label']}")
                        elif etype == "summary":
                            print(f"Summary: {event['text']}")
                        elif etype == "done":
                            print(f"Done. Duration: {event['duration_ms']}ms")
                            break
                        elif etype == "error":
                            print(f"Error: {event['error']}")
                            break
                    elif msg.type in (
                        aiohttp.WSMsgType.ERROR,
                        aiohttp.WSMsgType.CLOSE,
                        aiohttp.WSMsgType.CLOSED,
                    ):
                        break
            finally:
                if not send_task.done():
                    send_task.cancel()

asyncio.run(analyze_streaming())