Quickstart

"""Talk to your Pearl from the terminal: your microphone in, Pearl's voice through your speakers. Ctrl+C hangs up."""
import asyncio
import json
import os
import queue
import urllib.error
import urllib.request

import av                   # pip install websockets sounddevice av
import sounddevice as sd
import websockets

API_KEY = os.environ["NLPEARL_API_KEY"]    # "<AccountId>:<SecretKey>"
PEARL_ID = os.environ["NLPEARL_PEARL_ID"]
CALL_DATA = {}    # the variables your opening sentence uses, such as {"firstName": "Dana"}

RATE, FRAME = 24000, 480              # the microphone: 20 ms at 24 kHz, coded to opus_24000
PLAY_RATE, PLAY_FRAME = 48000, 960    # Pearl's voice: the Opus decoder returns 48 kHz

encoder = av.CodecContext.create("libopus", "w")
encoder.sample_rate, encoder.layout, encoder.format, encoder.bit_rate = RATE, "mono", "s16", 32000
decoder = av.CodecContext.create("libopus", "r")
decoder.sample_rate, decoder.layout = PLAY_RATE, "mono"


def encode(pcm):
    """20 ms of the microphone, as 16-bit PCM, to one Opus packet."""
    frame = av.AudioFrame(format="s16", layout="mono", samples=FRAME)
    frame.sample_rate = RATE
    frame.planes[0].update(pcm)
    return [bytes(packet) for packet in encoder.encode(frame)]


def decode(packet):
    """One Opus packet from Pearl to 20 ms of 16-bit PCM at 48 kHz."""
    return b"".join(bytes(frame.planes[0])[: frame.samples * 2] for frame in decoder.decode(av.Packet(packet)))


def create_call():
    request = urllib.request.Request(
        f"https://api.nlpearl.ai/v2/Pearl/{PEARL_ID}/Stream/Token",
        method="POST",
        headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
        data=json.dumps({"endUserId": "quickstart", "audioFormat": "opus_24000", "callData": CALL_DATA}).encode(),
    )
    try:
        with urllib.request.urlopen(request) as response:
            return json.load(response)    # callId, url, token, expiresAt, audio
    except urllib.error.HTTPError as error:
        raise SystemExit(f"Create Stream Token failed ({error.code}): {error.read().decode()}")


async def main():
    call = create_call()
    print("call", call["callId"])

    loop = asyncio.get_running_loop()
    microphone = asyncio.Queue()    # the end user's 20 ms frames, to send
    speaker = queue.Queue()         # Pearl's 20 ms frames, to play
    playing = False

    def on_microphone(data, frames, time, status):
        # Called every 20 ms by the sound card: the pacing Pearl expects, silence included.
        loop.call_soon_threadsafe(microphone.put_nowait, bytes(data))

    def on_speaker(out, frames, time, status):
        nonlocal playing
        while speaker.qsize() > 15:                    # more than 300 ms behind: drop the oldest
            speaker.get_nowait()
        if not playing and speaker.qsize() >= 4:       # start with 80 ms buffered
            playing = True
        try:
            out[:] = speaker.get_nowait() if playing else bytes(len(out))
        except queue.Empty:
            out[:] = bytes(len(out))
            playing = False                            # ran dry: buffer again

    async with websockets.connect(f"{call['url']}?token={call['token']}") as ws:
        print(json.loads(await ws.recv()))    # ready: always the first message
        print("Talking. Press Ctrl+C to hang up.")

        async def send_microphone():
            while True:
                for packet in encode(await microphone.get()):
                    await ws.send(packet)

        with sd.RawInputStream(samplerate=RATE, channels=1, dtype="int16", blocksize=FRAME, callback=on_microphone), \
             sd.RawOutputStream(samplerate=PLAY_RATE, channels=1, dtype="int16", blocksize=PLAY_FRAME, callback=on_speaker):
            sender = asyncio.create_task(send_microphone())
            try:
                async for message in ws:
                    if isinstance(message, bytes):
                        speaker.put(decode(message))  # Pearl's voice
                    else:
                        print(json.loads(message))    # transfer, ended
            except asyncio.CancelledError:            # Ctrl+C: hang up, then wait for ended
                await ws.send(json.dumps({"type": "hangup"}))
                async for message in ws:
                    if isinstance(message, str):
                        print(json.loads(message))
            finally:
                sender.cancel()


try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

call 6720f3c1a8b4e5d6f7a8b9c0
{'type': 'ready', 'protocol': 1, 'callId': '6720f3c1a8b4e5d6f7a8b9c0', 'audio': {'encoding': 'opus', 'sampleRate': 24000, 'channels': 1, 'frameMs': 20}}
Talking. Press Ctrl+C to hang up.
{'type': 'ended', 'reason': 'client_hangup', 'durationSec': 26}