External Meeting Control & Command Patterns

1
import base64
2
import numpy as np
3
4
def encode_pcm16_to_base64(pcm_bytes: bytes) -> str:
5
    """Encode raw PCM16 LE bytes to base64 for the sendaudio command."""
6
    return base64.b64encode(pcm_bytes).decode("utf-8")
7
8
9
def float32_to_pcm16_bytes(float_audio: np.ndarray) -> bytes:
10
    """Convert float32 audio (-1.0 to 1.0) to PCM16 LE bytes."""
11
    clipped = np.clip(float_audio, -1.0, 1.0)
12
    return (clipped * 32767).astype(np.int16).tobytes()
13
14
15
def resample_to_48k(pcm_bytes: bytes, source_rate: int) -> bytes:
16
    """Resample PCM16 LE audio from any sample rate to 48kHz."""
17
    if source_rate == 48000:
18
        return pcm_bytes
19
    samples = np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32)
20
    num_output = int(len(samples) * 48000 / source_rate)
21
    t_in = np.linspace(0, 1, len(samples), endpoint=False)
22
    t_out = np.linspace(0, 1, num_output, endpoint=False)
23
    resampled = np.interp(t_out, t_in, samples)
24
    return np.clip(resampled, -32768, 32767).astype(np.int16).tobytes()
25
26
27
def build_sendaudio_command(bot_id: str, pcm_bytes: bytes, sample_rate: int = 48000) -> dict:
28
    """Build a complete sendaudio command ready to send over the WebSocket."""
29
    return {
30
        "command": "sendaudio",
31
        "bot_id": bot_id,
32
        "audiochunk": encode_pcm16_to_base64(pcm_bytes),
33
        "sample_rate": sample_rate,
34
        "encoding": "pcm16",
35
        "channels": 1,
36
        "endianness": "little",
37
    }

1
import wave
2
3
def send_wav_file(ws, bot_id: str, wav_path: str):
4
    """Read a WAV file and send it as a sendaudio command."""
5
    with wave.open(wav_path, "rb") as wf:
6
        assert wf.getsampwidth() == 2, "WAV must be 16-bit"
7
        assert wf.getnchannels() == 1, "WAV must be mono"
8
        pcm_bytes = wf.readframes(wf.getnframes())
9
        source_rate = wf.getframerate()
10
11
    pcm_48k = resample_to_48k(pcm_bytes, source_rate)
12
    command = build_sendaudio_command(bot_id, pcm_48k)
13
    ws.send(json.dumps(command))

1
CHUNK_SAMPLES = 48000  # 1 second at 48kHz
2
CHUNK_BYTES = CHUNK_SAMPLES * 2  # 2 bytes per sample
3
4
for i in range(0, len(pcm_bytes), CHUNK_BYTES):
5
    chunk = pcm_bytes[i : i + CHUNK_BYTES]
6
    command = build_sendaudio_command(bot_id, chunk)
7
    await ws.send(json.dumps(command))
8
    await asyncio.sleep(0.8)  # pace slightly below real-time to avoid queue gaps

1
def build_chat_command(bot_id: str, text: str) -> dict:
2
    return {
3
        "command": "sendmsg",
4
        "bot_id": bot_id,
5
        "message": text,
6
        "msg": text,
7
    }
8
9
await ws.send(json.dumps(build_chat_command("bot_abc123", "Meeting notes are ready!")))

1
# Stream tokens as they arrive
2
for token in llm_stream:
3
    accumulated_text += token
4
    await ws.send(json.dumps({
5
        "command": "sendchat",
6
        "bot_id": bot_id,
7
        "role": "assistant",
8
        "text": accumulated_text,
9
        "is_final": False,
10
    }))
11
12
# Send the final complete message
13
await ws.send(json.dumps({
14
    "command": "sendchat",
15
    "bot_id": bot_id,
16
    "role": "assistant",
17
    "text": accumulated_text,
18
    "is_final": True,
19
}))

1
async def handle_barge_in(ws, bot_id: str):
2
    """Stop the bot's audio when a user starts speaking."""
3
    await ws.send(json.dumps({
4
        "command": "interrupt",
5
        "bot_id": bot_id,
6
        "action": "clear_audio_queue",
7
    }))

1
import base64
2
3
def build_image_command(bot_id: str, image_path: str) -> dict:
4
    with open(image_path, "rb") as f:
5
        img_b64 = base64.b64encode(f.read()).decode("utf-8")
6
    return {
7
        "command": "sendimg",
8
        "bot_id": bot_id,
9
        "img": img_b64,
10
    }
11
12
await ws.send(json.dumps(build_image_command("bot_abc123", "avatar.png")))

1
import asyncio
2
import base64
3
import json
4
import numpy as np
5
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
6
7
app = FastAPI()
8
9
control_sockets: dict[str, WebSocket] = {}
10
11
12
def decode_audio_frame(data: bytes):
13
    if len(data) < 5 or data[0] != 0x01:
14
        return None
15
    sid_len = int.from_bytes(data[1:3], "little")
16
    speaker_id = data[3 : 3 + sid_len].decode("utf-8")
17
    off = 3 + sid_len
18
    sname_len = int.from_bytes(data[off : off + 2], "little")
19
    off += 2
20
    speaker_name = data[off : off + sname_len].decode("utf-8")
21
    off += sname_len
22
    return speaker_id, speaker_name, data[off:]
23
24
25
async def send_command(bot_id: str, command: dict):
26
    ws = control_sockets.get(bot_id)
27
    if ws:
28
        await ws.send_text(json.dumps(command))
29
30
31
@app.websocket("/{bot_id}/audio")
32
async def audio_endpoint(websocket: WebSocket, bot_id: str):
33
    """Receives live meeting audio from the Meetstream bot."""
34
    await websocket.accept()
35
36
    try:
37
        while True:
38
            raw = await websocket.receive()
39
40
            if "text" in raw and raw["text"]:
41
                handshake = json.loads(raw["text"])
42
                print(f"[{bot_id}] Audio handshake: {handshake}")
43
                continue
44
45
            if "bytes" in raw and raw["bytes"]:
46
                result = decode_audio_frame(raw["bytes"])
47
                if result is None:
48
                    continue
49
                speaker_id, speaker_name, pcm_bytes = result
50
                duration_ms = (len(pcm_bytes) / 2 / 48000) * 1000
51
                print(f"[{bot_id}] [{speaker_name}] {duration_ms:.0f}ms audio")
52
53
    except WebSocketDisconnect:
54
        print(f"[{bot_id}] Audio disconnected")
55
56
57
@app.websocket("/{bot_id}/control")
58
async def control_endpoint(websocket: WebSocket, bot_id: str):
59
    """Receives commands from and sends commands to the Meetstream bot."""
60
    await websocket.accept()
61
    control_sockets[bot_id] = websocket
62
63
    try:
64
        while True:
65
            text = await websocket.receive_text()
66
            data = json.loads(text)
67
68
            if data.get("type") == "ready":
69
                print(f"[{bot_id}] Control ready")
70
71
                # Send a welcome chat message
72
                await send_command(bot_id, {
73
                    "command": "sendmsg",
74
                    "bot_id": bot_id,
75
                    "message": "Bot is now live!",
76
                    "msg": "Bot is now live!",
77
                })
78
79
    except WebSocketDisconnect:
80
        print(f"[{bot_id}] Control disconnected")
81
    finally:
82
        control_sockets.pop(bot_id, None)

1
# Audio channel: receive meeting audio
2
speaker_id, speaker_name, pcm_bytes = decode_audio_frame(binary_message)
3
4
# Your processing pipeline
5
transcript = your_stt_service.transcribe(pcm_bytes)
6
response = your_llm.generate(transcript)
7
tts_audio = your_tts_service.synthesize(response)
8
9
# Control channel: send response back into the meeting
10
await send_command(bot_id, build_sendaudio_command(bot_id, tts_audio))
11
await send_command(bot_id, {
12
    "command": "sendchat",
13
    "bot_id": bot_id,
14
    "role": "assistant",
15
    "text": response,
16
    "is_final": True,
17
})

1
# When you detect the user started speaking while bot audio is playing:
2
await send_command(bot_id, {
3
    "command": "interrupt",
4
    "bot_id": bot_id,
5
    "action": "clear_audio_queue",
6
})
7
8
# Then process the new user input and generate a fresh response