superU Voice Assistant

# Windows
pip install pyaudio websockets superu

# macOS/Linux
pip3 install pyaudio websockets superu

import asyncio
import pyaudio
import websockets
import uuid
import base64
import json
import superu

FORMAT = pyaudio.paInt16
CHANNELS = 1
SEND_SAMPLE_RATE = 16000
CHUNK_SIZE = 1024
pya = pyaudio.PyAudio()

def get_default_input_device_index():
    for i in range(pya.get_device_count()):
        dev = pya.get_device_info_by_index(i)
        if dev['maxInputChannels'] > 0:
            print(f"Using device: {dev['name']} (Index: {i})")
            return i
    raise RuntimeError("No input device found.")

mic_index = get_default_input_device_index()

async def send_audio(stream, websocket, streamId):
    await websocket.send(json.dumps({"event": "start", "start": {"streamId": streamId}}))
    while True:
        audio_chunk = await asyncio.to_thread(stream.read, CHUNK_SIZE)
        payload = base64.b64encode(audio_chunk).decode('utf-8')
        await websocket.send(json.dumps({"event": "media", "media": {"payload": payload}}))

async def receive_messages(websocket):
    output_stream = await asyncio.to_thread(
        pya.open, format=FORMAT, channels=CHANNELS, rate=SEND_SAMPLE_RATE, output=True)
    while True:
        try:
            response = json.loads(await websocket.recv())
            if response.get("event") == "playAudio":
                audio_data = base64.b64decode(response["media"]["payload"])
                await asyncio.to_thread(output_stream.write, audio_data)
        except websockets.exceptions.ConnectionClosed:
            print("WebSocket connection closed.")
            break

async def listen_and_send(uri, streamId):
    stream = await asyncio.to_thread(
        pya.open, format=FORMAT, channels=CHANNELS, rate=SEND_SAMPLE_RATE,
        input=True, input_device_index=mic_index, frames_per_buffer=CHUNK_SIZE)
    async with websockets.connect(uri) as websocket:
        print("Connected to WebSocket. Streaming audio...")
        try:
            await asyncio.gather(
                send_audio(stream, websocket, streamId),
                receive_messages(websocket))
        except Exception as e:
            print(f"Error: {e}")
        finally:
            stream.stop_stream()
            stream.close()
            pya.terminate()
            print("Connection closed.")

if __name__ == "__main__":
    superu_client = superu.SuperU('<YOUR_API_KEY>')
    First_message = "Hey there! Ready to explore some fascinating science today?"
    System_prompt = """
        You are a helpful and curious science assistant. Your job is to answer questions clearly, concisely, and in a way that's engaging for someone interested in science.
    """
    superU_call = superu_client.calls.create_call(
        first_message=First_message,
        system_prompt=System_prompt,
        voice_id="90ipbRoKi4CpHXvKVtl0"  # Anika - Customer Care Agent
    )
    asyncio.run(listen_and_send(superU_call['ws_url'], superU_call['streamId']))

# Windows
python superU_voice_assistant.py

# macOS/Linux
python3 superU_voice_assistant.py