> ## Documentation Index
> Fetch the complete documentation index at: https://docs.openhome.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Websocket Streaming
> How to use our websocket in your application to call your agent
Use the desired agent id under your account in the WebSocket URL.
```wss theme={"system"}
wss://app.openhome.com/websocket/voice-stream/OPENHOME_API_KEY/PERSONALITY_ID
```
Here is an example:
```wss theme={"system"}
wss://app.openhome.com/websocket/voice-stream/xyzsdsadasannfma/4727
```
### OPENHOME\_API\_KEY
Checkout API Docs to get OPENHOME\_API\_KEY
### AGENT\_ID
Checkout Get Agents section to get your account's agent ids.
* **4727**: Represents the agent ID.
* Set the agent ID to `0` to skip this part. This will start the call with the default agent, OpenHome.
***
## WebSocket Flow
### Audio Data Format
Audio data sent to the WebSocket must adhere to the following specifications:
* Format: **16-bit PCM**
* Sample Rate: **16000 Hz**
* Encoding: **Base64**
Ensure that the audio is converted to this format before sending it to the WebSocket.
***
### WebSocket Message Structure
#### Client to Server Messages
1. **User to Server Text Messages**
```json theme={"system"}
{
"data": "MESSAGE_CONTEXT",
"type": "transcribed"
}
```
2. **User to Server Audio Messages**
```json theme={"system"}
{
"data": "BASE64_ENCODED_AUDIO_MESSAGES",
"type": "audio"
}
```
### Server to Client Messages
1. Server to User Text Messages
```json theme={"system"}
{
"data": {
"content": "MESSAGE CONTENT",
"live": true,
"role": "assistant"
},
"type": "message"
}
```
**live: true** indicates live transcription or response logs.
**final: true** provides finalized transcription and response messages.
2. Server to User Audio Messages
```json theme={"system"}
{
"data": "BASE64_ENCODED_AUDIO_MESSAGES",
"type": "audio"
}
```
## EXAMPLES
Find More Examples on GitHub
```HTML HTML theme={"system"}
Audio Stream
Connection Status: Disconnected
Microphone Status: Off
```
```py Python theme={"system"}
import websockets
import asyncio
import base64
import json
import subprocess
import pyaudio
import os
import numpy as np
OPENHOME_API_KEY = "abcxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxyz"
class VoiceStreamer:
def __init__(self, speaker_type="mpv"):
self.api_key = OPENHOME_API_KEY
self.server_url = f"https://app.openhome.com/websocket/voice-stream/{self.api_key}/0"
self.frames_per_buffer = 3200
self.format = pyaudio.paInt16
self.channels = 1
self.rate = 16000
self.speaker_type = speaker_type
self.websocket = None
self.should_send_audio = True
# Noise reduction parameters
self.alpha = 0.95 # Smoothing factor
self.prev_noise_power = None
# INITIALIZE PYAUDIO
self.py_audio_obj = pyaudio.PyAudio()
self.stream = self._create_stream()
# INITIALIZE PLACEHOLDER FOR MPV
self.mpv_process = None
self.speaker = None
# self.playback_complete = asyncio.Event()
self.is_speaking = False
def _create_stream(self):
"""Create the microphone audio stream with better error handling."""
try:
# Get default input device info
device_info = self.py_audio_obj.get_default_input_device_info()
# Adjust parameters based on device capabilities
supported_rate = min(int(device_info['defaultSampleRate']), self.rate)
stream = self.py_audio_obj.open(
format=self.format,
channels=self.channels,
rate=supported_rate,
input=True,
frames_per_buffer=self.frames_per_buffer,
stream_callback=None,
start=False # Don't start immediately
)
# Test the stream
stream.start_stream()
test_data = stream.read(self.frames_per_buffer)
if not test_data:
raise IOError("No audio data received")
return stream
except Exception as e:
print(f"Error creating audio stream: {e}")
# Try fallback parameters
return self.py_audio_obj.open(
format=self.format,
channels=1,
rate=16000,
input=True,
frames_per_buffer=1024
)
def process_audio(self, audio_data):
audio_array = np.frombuffer(audio_data, dtype=np.int16)
audio_float = audio_array.astype(np.float32) / 32768.0
# RMS energy calculation
rms = np.sqrt(np.mean(audio_float**2))
# Initialize state variables if they don't exist
if not hasattr(self, 'energy_history'):
self.energy_history = []
self.baseline_energy = None
self.speaking_threshold = 0.03
self.last_voice_activity = False
self.voice_holdoff_counter = 0
# Update energy history
self.energy_history.append(rms)
if len(self.energy_history) > 30:
self.energy_history.pop(0)
# Calculate dynamic threshold with different bases for speaking/not speaking
if self.is_speaking:
# Much higher threshold when bot is speaking
dynamic_threshold = np.mean(self.energy_history) * 1.8
else:
# Normal threshold when bot is not speaking
dynamic_threshold = np.mean(self.energy_history) * 1.2
# Significant audio detection with hysteresis
if not self.last_voice_activity:
# Higher threshold to start detection
significant_audio = rms > dynamic_threshold * 1.2
else:
# Lower threshold to continue detection
significant_audio = rms > dynamic_threshold * 0.8
# Voice activity state machine with holdoff
if significant_audio:
self.last_voice_activity = True
self.voice_holdoff_counter = 0
else:
self.voice_holdoff_counter += 1
if self.voice_holdoff_counter > 10: # Adjust holdoff period as needed
self.last_voice_activity = False
# If bot is speaking, require much stronger user audio
if self.is_speaking and rms < dynamic_threshold * 2.0:
return b'\x00' * len(audio_data)
# If no significant audio, return silence
if not self.last_voice_activity:
return b'\x00' * len(audio_data)
# Noise reduction
if self.prev_noise_power is None:
self.prev_noise_power = rms**2
noise_power = self.alpha * self.prev_noise_power + (1 - self.alpha) * rms**2
self.prev_noise_power = noise_power
gain = np.maximum(1 - (noise_power / (rms**2 + 1e-10)), 0.1)
# Additional gain reduction when bot is speaking
if self.is_speaking:
gain *= 0.3 # Stronger gain reduction when bot is speaking
processed_audio = audio_float * gain
return (processed_audio * 32768).astype(np.int16).tobytes()
def pause_mic(self):
"""Pause the microphone stream to prevent capturing speaker's audio."""
self.should_send_audio = False
if self.stream.is_active():
self.stream.stop_stream()
print("[+] Microphone stream paused")
def resume_mic(self):
"""Resume the microphone stream after speaker finishes."""
if not self.stream.is_active():
self.stream.start_stream()
self.should_send_audio = True
print("[+] Microphone stream resumed")
async def handle_mpv(self):
"""Handle MPV cleanup after audio playback ends."""
if self.mpv_process and self.mpv_process.stdin:
try:
self.mpv_process.stdin.close()
except BrokenPipeError:
print("[-] Broken pipe error while writing to MPV")
await self.mpv_process.communicate()
self.mpv_process = None
# self.playback_complete.set()
self.is_speaking = False
print("[+] MPV playback completed")
message = {"type": "text", "data": "bot-speak-end"}
await self.websocket.send(json.dumps(message))
async def send_data(self):
buffer_size = 5 # Number of frames to buffer
audio_buffer = []
while True:
try:
if not self.should_send_audio:
await asyncio.sleep(0.01)
continue
audio_bytes = self.stream.read(self.frames_per_buffer)
processed_audio = self.process_audio(audio_bytes)
# Buffer the processed audio
audio_buffer.append(processed_audio)
if len(audio_buffer) < buffer_size:
continue
# Check if any frame in buffer has significant audio
has_audio = any(np.frombuffer(frame, dtype=np.int16).any() for frame in audio_buffer)
if has_audio:
# Send all buffered frames
if self.is_speaking:
print("Interrupting")
self.mpv_process.stdin.write(b"m\n")
message = {"type": "text", "data": "interrupt-event"}
await self.websocket.send(json.dumps(message))
self.mpv_process.stdin.write(b"q\n")
await self.mpv_process.stdin.drain()
self.is_speaking = False
print("[+] STOPPED MPV...")
for frame in audio_buffer:
encoded_bytes = base64.b64encode(frame).decode("utf-8")
json_data = json.dumps({"type": "audio", "data": encoded_bytes})
await self.websocket.send(json_data)
else:
print("NO AUDIO")
# Clear buffer
audio_buffer = []
except websockets.exceptions.ConnectionClosedError:
print("[!] Connection is closed")
break
except Exception as e:
print("[!] Error in send_data:", e)
self.stream = self._create_stream()
await asyncio.sleep(0.01)
async def receive_data(self):
"""Receive data from server and handle it based on speaker type."""
while True:
try:
server_response = await self.websocket.recv()
data = json.loads(server_response)
if data["type"] == "text":
await self.handle_text_message(data)
elif data["type"] == "audio":
await self.handle_audio_message(data)
elif data["type"] == "message":
await self.handle_chat_message(data["data"])
except websockets.exceptions.ConnectionClosedError:
print("[!] Connection is closed")
break
except Exception as e:
print("[!] Error in receive_data:", e)
async def handle_chat_message(self, data):
"""Process 'chat' type messages from server."""
if data.get("final",False):
print("%s: FINAL: %s"%(data.get("role").upper(), data.get("content","")))
else:
print("%s: LIVE: %s..."%(data.get("role").upper(), data.get("content","")))
async def handle_text_message(self, data):
"""Process 'text' type messages from server based on speaker type."""
if data["data"] == "audio-init":
print("[+] BOT SPEAKING EVENT IS SET...")
# self.pause_mic()
# self.playback_complete.clear()
self.is_speaking = True
self.mpv_process = await asyncio.create_subprocess_exec(
"mpv", "--no-cache", "--no-terminal", "--", "fd://0",
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
)
message = {"type": "text", "data": "bot-speaking"}
await self.websocket.send(json.dumps(message))
print("[+] SENT BOT SPEAKING EVENT...")
elif data["data"] == "interrupt":
print("[+] INTERRUPTION RECEIVED...")
if self.speaker_type == "mpv" and self.mpv_process and self.mpv_process.stdin:
self.mpv_process.stdin.write(b"q\n")
await self.mpv_process.stdin.drain()
self.is_speaking = False
print("[+] STOPPED MPV...")
elif data["data"] == "audio-end":
print("[+] AUDIO END RECEIVED...")
if self.speaker_type == "mpv":
asyncio.get_event_loop().create_task(self.handle_mpv())
# await self.playback_complete.wait()
await asyncio.sleep(0.5)
# self.resume_mic()
async def handle_audio_message(self, data):
"""Process 'audio' type messages from server."""
message = {"type": "ack", "data": "audio-received"}
await self.websocket.send(json.dumps(message))
audio_bytes = base64.b64decode(data["data"])
if self.speaker_type == "mpv" and self.mpv_process and self.mpv_process.stdin and self.is_speaking:
self.mpv_process.stdin.write(audio_bytes)
await self.mpv_process.stdin.drain()
async def run(self):
"""Establish a connection to the server and start sending/receiving data."""
try:
async with websockets.connect(self.server_url) as websocket:
self.websocket = websocket
await asyncio.gather(self.send_data(), self.receive_data())
except Exception as e:
print(e)
def __del__(self):
"""Cleanup resources."""
if self.stream:
self.stream.stop_stream()
self.stream.close()
if self.py_audio_obj:
self.py_audio_obj.terminate()
if __name__ == "__main__":
voice_streamer = VoiceStreamer(speaker_type=os.getenv("SPEAKER_TYPE"))
asyncio.run(voice_streamer.run())
```
```python RaspberryPi theme={"system"}
import websockets
import asyncio
import base64
import json
import subprocess
import pyaudio
import os
from dotenv import load_dotenv
from time import time
load_dotenv(dotenv_path=".env")
class VoiceStreamer:
def __init__(self):
self.api_key = os.getenv('API_KEY')
self.default_agent = os.getenv('DEFAULT_AGENT')
self.server_url = f"{os.getenv('STREAM_SERVER_URL_WS')}/websocket/voice-stream/{self.api_key}/{self.default_agent}?devkit=true"
self.frames_per_buffer = 1024
self.format = pyaudio.paInt16
self.channels = 1
self.rate = 16000
self.websocket = None
self.last_live_transcription = time()
self.py_audio_obj = pyaudio.PyAudio()
self.stream = None
self.stream = self._create_stream()
self.mpv_process = None
self.bot_speaking = True
def _create_stream(self):
if self.stream is not None:
self.stream.stop_stream()
self.stream.close()
return self.py_audio_obj.open(
format=self.format,
channels=self.channels,
rate=self.rate,
input=True,
frames_per_buffer=self.frames_per_buffer,
start=True
)
def pause_mic(self):
if self.stream.is_active():
self.stream.stop_stream()
def resume_mic(self):
if not self.stream.is_active():
self.stream.start_stream()
async def handle_mpv(self):
if self.mpv_process and self.mpv_process.stdin:
try:
self.mpv_process.stdin.close()
await self.mpv_process.wait()
except BrokenPipeError:
pass
self.mpv_process = None
self.bot_speaking = False
message = {"type": "text", "data": "bot-speak-end"}
await self.websocket.send(json.dumps(message))
async def send_data(self):
while True:
try:
audio_bytes = self.stream.read(self.frames_per_buffer, exception_on_overflow=False)
if audio_bytes and not self.bot_speaking:
encoded_bytes = base64.b64encode(audio_bytes).decode("utf-8")
await self.websocket.send(json.dumps({"type": "audio", "data": encoded_bytes}))
except Exception:
self.stream = self._create_stream()
await asyncio.sleep(0.01)
async def receive_data(self):
while True:
try:
server_response = await self.websocket.recv()
data = json.loads(server_response)
if data["type"] == "text":
await self.handle_text_message(data)
elif data["type"] == "audio":
await self.handle_audio_message(data)
elif data["type"] == "message" and data["data"].get("final",False):
print(data["data"], flush=True)
except websockets.exceptions.ConnectionClosedError:
break
except Exception:
continue
async def handle_text_message(self, data):
if data["data"] == "audio-init":
self.mpv_process = await asyncio.create_subprocess_exec(
"mpv", "--no-cache", "--no-terminal", "--", "fd://0",
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
)
await self.websocket.send(json.dumps({"type": "text", "data": "bot-speaking"}))
self.bot_speaking = True
elif data["data"] == "interrupt" and self.mpv_process:
self.mpv_process.stdin.write(b"q\n")
await self.mpv_process.stdin.drain()
elif data["data"] == "audio-end":
await self.handle_mpv()
async def handle_audio_message(self, data):
while not self.bot_speaking:
await asyncio.sleep(0.2)
if self.mpv_process and self.mpv_process.stdin:
self.mpv_process.stdin.write(base64.b64decode(data["data"]))
await self.mpv_process.stdin.drain()
async def run(self):
try:
async with websockets.connect(self.server_url) as websocket:
self.websocket = websocket
await asyncio.gather(self.send_data(), self.receive_data())
except Exception:
pass
if __name__ == "__main__":
voice_streamer = VoiceStreamer()
asyncio.run(voice_streamer.run())
```