External Meeting Control & Command Patterns

import base64
import numpy as np
def encode_pcm16_to_base64(pcm_bytes: bytes) -> str:
    """Encode raw PCM16 LE bytes to base64 for the sendaudio command."""
    return base64.b64encode(pcm_bytes).decode("utf-8")
def float32_to_pcm16_bytes(float_audio: np.ndarray) -> bytes:
    """Convert float32 audio (-1.0 to 1.0) to PCM16 LE bytes."""
    clipped = np.clip(float_audio, -1.0, 1.0)
    return (clipped * 32767).astype(np.int16).tobytes()
def resample_to_48k(pcm_bytes: bytes, source_rate: int) -> bytes:
    """Resample PCM16 LE audio from any sample rate to 48kHz."""
    if source_rate == 48000:
        return pcm_bytes
    samples = np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32)
    num_output = int(len(samples) * 48000 / source_rate)
    t_in = np.linspace(0, 1, len(samples), endpoint=False)
    t_out = np.linspace(0, 1, num_output, endpoint=False)
    resampled = np.interp(t_out, t_in, samples)
    return np.clip(resampled, -32768, 32767).astype(np.int16).tobytes()
def build_sendaudio_command(bot_id: str, pcm_bytes: bytes, sample_rate: int = 48000) -> dict:
    """Build a complete sendaudio command ready to send over the WebSocket."""
    return {
        "command": "sendaudio",
        "bot_id": bot_id,
        "audiochunk": encode_pcm16_to_base64(pcm_bytes),
        "sample_rate": sample_rate,
        "encoding": "pcm16",
        "channels": 1,
        "endianness": "little",
    }

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

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

import struct
def build_binary_audio_frame(pcm_bytes: bytes, sample_rate: int = 48000) -> bytes:
    return bytes([0x02]) + struct.pack("<IB", sample_rate, 1) + pcm_bytes
await ws.send(build_binary_audio_frame(pcm_chunk, 24000))

def build_chat_command(bot_id: str, text: str) -> dict:
    return {
        "command": "sendmsg",
        "bot_id": bot_id,
        "message": text,
        "msg": text,
    }
await ws.send(json.dumps(build_chat_command("bot_abc123", "Meeting notes are ready!")))

async def handle_barge_in(ws, bot_id: str):
    """Stop the bot's audio when a user starts speaking."""
    await ws.send(json.dumps({
        "command": "interrupt",
        "bot_id": bot_id,
        "action": "clear_audio_queue",
    }))
async def start_next_response(ws, bot_id: str, turn_id: str):
    """Call before sending the first sendaudio of the next response."""
    await ws.send(json.dumps({
        "command": "resume_audio",
        "bot_id": bot_id,
        "resume_id": turn_id,
    }))

import base64
def build_image_command(bot_id: str, image_path: str) -> dict:
    with open(image_path, "rb") as f:
        img_b64 = base64.b64encode(f.read()).decode("utf-8")
    return {
        "command": "sendimg",
        "bot_id": bot_id,
        "img": img_b64,
    }
await ws.send(json.dumps(build_image_command("bot_abc123", "avatar.png")))

import asyncio
import base64
import json
import numpy as np
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
control_sockets: dict[str, WebSocket] = {}
def decode_audio_frame(data: bytes):
    """Return (speaker_id, speaker_name, sample_rate, pcm_bytes), or None."""
    if len(data) < 5:
        return None
    if data[0] == 0x01:            # Google Meet and Microsoft Teams: 48 kHz mono
        sample_rate, off = 48000, 1
    elif data[0] == 0x04:          # Zoom: sample rate and channels in the header
        sample_rate, off = int.from_bytes(data[1:5], "little"), 6
    else:
        return None
    sid_len = int.from_bytes(data[off : off + 2], "little")
    off += 2
    speaker_id = data[off : off + sid_len].decode("utf-8")
    off += sid_len
    sname_len = int.from_bytes(data[off : off + 2], "little")
    off += 2
    speaker_name = data[off : off + sname_len].decode("utf-8")
    off += sname_len
    return speaker_id, speaker_name, sample_rate, data[off:]
async def send_command(bot_id: str, command: dict):
    ws = control_sockets.get(bot_id)
    if ws:
        await ws.send_text(json.dumps(command))
@app.websocket("/{bot_id}/audio")
async def audio_endpoint(websocket: WebSocket, bot_id: str):
    """Receives live meeting audio from the MeetStream bot."""
    await websocket.accept()
    try:
        while True:
            raw = await websocket.receive()
            if "text" in raw and raw["text"]:
                handshake = json.loads(raw["text"])
                print(f"[{bot_id}] Audio handshake: {handshake}")
                continue
            if "bytes" in raw and raw["bytes"]:
                result = decode_audio_frame(raw["bytes"])
                if result is None:
                    continue
                speaker_id, speaker_name, sample_rate, pcm_bytes = result
                duration_ms = (len(pcm_bytes) / 2 / sample_rate) * 1000
                print(f"[{bot_id}] [{speaker_name}] {duration_ms:.0f}ms audio")
    except WebSocketDisconnect:
        print(f"[{bot_id}] Audio disconnected")
@app.websocket("/{bot_id}/control")
async def control_endpoint(websocket: WebSocket, bot_id: str):
    """Receives commands from and sends commands to the MeetStream bot."""
    await websocket.accept()
    control_sockets[bot_id] = websocket
    try:
        while True:
            text = await websocket.receive_text()
            data = json.loads(text)
            if data.get("type") == "ready":
                print(f"[{bot_id}] Control ready")
                # Send a welcome chat message
                await send_command(bot_id, {
                    "command": "sendmsg",
                    "bot_id": bot_id,
                    "message": "Bot is now live!",
                    "msg": "Bot is now live!",
                })
    except WebSocketDisconnect:
        print(f"[{bot_id}] Control disconnected")
    finally:
        control_sockets.pop(bot_id, None)

# Audio channel: receive meeting audio
speaker_id, speaker_name, sample_rate, pcm_bytes = decode_audio_frame(binary_message)
# Your processing pipeline
transcript = your_stt_service.transcribe(pcm_bytes)
response = your_llm.generate(transcript)
tts_audio = your_tts_service.synthesize(response)  # 48 kHz PCM16 mono
# Control channel: send the response back into the meeting
await send_command(bot_id, {"command": "resume_audio", "bot_id": bot_id, "resume_id": "turn-1"})
await send_command(bot_id, build_sendaudio_command(bot_id, tts_audio))
await send_command(bot_id, {
    "command": "sendmsg",
    "bot_id": bot_id,
    "message": response,
    "msg": response,
})

# When you detect the user started speaking while bot audio is playing:
await send_command(bot_id, {
    "command": "interrupt",
    "bot_id": bot_id,
    "action": "clear_audio_queue",
})
# Then process the new user input, send resume_audio, and play the fresh response