wss://app.openhome.com/websocket/voice-stream/OPENHOME_API_KEY/PERSONALITY_ID
wss://app.openhome.com/websocket/voice-stream/xyzsdsadasannfma/4727
OPENHOME_API_KEY
Checkout API Docs to get OPENHOME_API_KEYAGENT_ID
Checkout Get Agents section to get your account’s agent ids.- 4727: Represents the agent ID.
- Set the agent ID to
0to 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
WebSocket Message Structure
Client to Server Messages
- User to Server Text Messages
{ "data": "MESSAGE_CONTEXT", "type": "transcribed" } - User to Server Audio Messages
{ "data": "BASE64_ENCODED_AUDIO_MESSAGES", "type": "audio" }
Server to Client Messages
-
Server to User Text Messages
live: true indicates live transcription or response logs. final: true provides finalized transcription and response messages.
{ "data": { "content": "MESSAGE CONTENT", "live": true, "role": "assistant" }, "type": "message" } -
Server to User Audio Messages
{ "data": "BASE64_ENCODED_AUDIO_MESSAGES", "type": "audio" }
EXAMPLES
Find More Examples on GitHub<!DOCTYPE html>
<html>
<head>
<title>Audio Stream</title>
<style>
.container {
max-width: 800px;
margin: 0 auto;
padding: 20px;
}
.status {
padding: 10px;
margin-bottom: 10px;
background-color: #f0f0f0;
}
.active { background-color: #90EE90; }
.error { background-color: #ffcccb; }
#retryButton {
padding: 10px;
margin: 10px 0;
display: none;
}
</style>
</head>
<body>
<div class="container">
<div id="connectionStatus" class="status">Connection Status: Disconnected</div>
<div id="micStatus" class="status">Microphone Status: Off</div>
<div class="call-controls">
<button id="startCall" class="call-button">Start Call</button>
<button id="endCall" class="call-button">End Call</button>
</div>
<button id="retryButton" onclick="retryMicrophoneAccess()">Retry Microphone Access</button>
<audio id="audioPlayer" controls></audio>
</div>
<script>
const api_key = "0603b422xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxd0e87"
const wsUrl = 'wss://app.openhome.com/websocket/voice-stream/' + api_key + '/0';
const BUFFER_SIZE = 4096;
const FRAMES_PER_BUFFER = 3200;
const CHANNELS = 1;
const RATE = 16000;
let ws;
let audioContext;
let mediaStream;
let reconnectAttempts = 0;
let microphoneSource;
let audioBuffer = new Float32Array();
// Audio playback setup
let audioElement = document.getElementById('audioPlayer');
let mediaSource = null;
let sourceBuffer = null;
let audioQueue = [];
let isSourceOpen = false;
async function initAudio() {
try {
if (!navigator.mediaDevices || !navigator.mediaDevices.getUserMedia) {
throw new Error('Your browser does not support audio recording');
}
audioContext = new (window.AudioContext || window.webkitAudioContext)({
sampleRate: RATE
});
mediaStream = await navigator.mediaDevices.getUserMedia({
audio: {
channelCount: CHANNELS,
sampleRate: RATE,
echoCancellation: true,
noiseSuppression: true,
autoGainControl: true
}
});
if (!mediaStream || !mediaStream.active) {
throw new Error('Failed to get media stream');
}
const tracks = mediaStream.getAudioTracks();
if (!tracks || tracks.length === 0) {
throw new Error('No audio tracks available');
}
microphoneSource = audioContext.createMediaStreamSource(mediaStream);
const processor = audioContext.createScriptProcessor(BUFFER_SIZE, CHANNELS, CHANNELS);
microphoneSource.connect(processor);
processor.connect(audioContext.destination);
processor.onaudioprocess = (e) => {
const inputData = e.inputBuffer.getChannelData(0);
const newBuffer = new Float32Array(audioBuffer.length + inputData.length);
newBuffer.set(audioBuffer);
newBuffer.set(inputData, audioBuffer.length);
audioBuffer = newBuffer;
while (audioBuffer.length >= FRAMES_PER_BUFFER) {
const samplesTo = audioBuffer.slice(0, FRAMES_PER_BUFFER);
const pcmData = convertFloatTo16BitPCM(samplesTo);
if (ws && ws.readyState === WebSocket.OPEN) {
const base64Audio = btoa(String.fromCharCode(...new Uint8Array(pcmData.buffer)));
ws.send(JSON.stringify({
type: 'audio',
data: base64Audio
}));
}
audioBuffer = audioBuffer.slice(FRAMES_PER_BUFFER);
}
};
document.getElementById('micStatus').textContent = 'Microphone Status: Active';
document.getElementById('micStatus').classList.add('active');
document.getElementById('micStatus').classList.remove('error');
document.getElementById('retryButton').style.display = 'none';
if (audioContext.state === 'suspended') {
await audioContext.resume();
}
} catch (error) {
console.error('Error initializing audio:', error);
document.getElementById('micStatus').textContent = 'Microphone Status: Error - ' + error.message;
document.getElementById('micStatus').classList.add('error');
document.getElementById('micStatus').classList.remove('active');
document.getElementById('retryButton').style.display = 'block';
throw error;
}
}
function convertFloatTo16BitPCM(float32Array) {
const int16Array = new Int16Array(float32Array.length);
for (let i = 0; i < float32Array.length; i++) {
const s = Math.max(-1, Math.min(1, float32Array[i]));
int16Array[i] = s < 0 ? s * 0x8000 : s * 0x7FFF;
}
return int16Array;
}
async function retryMicrophoneAccess() {
try {
if (mediaStream) {
mediaStream.getTracks().forEach(track => track.stop());
}
if (audioContext) {
await audioContext.close();
}
audioBuffer = new Float32Array();
await initAudio();
} catch (error) {
console.error('Error retrying microphone access:', error);
document.getElementById('micStatus').textContent = 'Microphone Status: Error - ' + error.message;
document.getElementById('micStatus').classList.add('error');
}
}
function createMediaSource() {
return new Promise((resolve, reject) => {
try {
if (!isCallActive) {
reject(new Error('Call is not active'));
return;
}
if (mediaSource) {
if (mediaSource.readyState !== 'closed') {
try {
mediaSource.endOfStream();
} catch (e) {
console.error('Error ending previous media source:', e);
}
}
mediaSource = null;
sourceBuffer = null;
}
mediaSource = new MediaSource();
audioElement.src = URL.createObjectURL(mediaSource);
// Set up automatic play when data is available
audioElement.addEventListener('canplay', () => {
audioElement.play().catch(e => console.error('Error playing audio:', e));
});
// Handle audio ending
audioElement.addEventListener('ended', () => {
if (audioQueue.length > 0) {
audioElement.play().catch(e => console.error('Error playing audio:', e));
}
});
mediaSource.addEventListener('sourceopen', () => {
try {
console.log('MediaSource opened');
isSourceOpen = true;
if (!sourceBuffer && mediaSource.readyState === 'open') {
sourceBuffer = mediaSource.addSourceBuffer('audio/mpeg');
sourceBuffer.mode = 'sequence';
sourceBuffer.addEventListener('updateend', () => {
if (audioQueue.length > 0 && !sourceBuffer.updating) {
processAudioQueue();
}
});
}
resolve();
} catch (e) {
console.error('Error in sourceopen:', e);
reject(e);
}
});
mediaSource.addEventListener('sourceended', () => {
console.log('MediaSource ended');
isSourceOpen = false;
});
mediaSource.addEventListener('sourceclose', () => {
console.log('MediaSource closed');
isSourceOpen = false;
if (isCallActive) {
setTimeout(() => createMediaSource().catch(console.error), 1000);
}
});
} catch (e) {
console.error('Error creating MediaSource:', e);
reject(e);
}
});
}
function processAudioQueue() {
if (!sourceBuffer || !isSourceOpen) return;
while (audioQueue.length > 0 && !sourceBuffer.updating) {
try {
const data = audioQueue.shift();
sourceBuffer.appendBuffer(data);
return;
} catch (e) {
console.error('Error appending buffer:', e);
if (e.name === 'QuotaExceededError') {
if (sourceBuffer.buffered.length > 0) {
const start = sourceBuffer.buffered.start(0);
const end = sourceBuffer.buffered.end(0);
if (end - start > 10) {
sourceBuffer.remove(start, end - 10);
}
}
audioQueue.unshift(data);
return;
}
createMediaSource().catch(console.error);
return;
}
}
}
async function initWebSocket() {
try {
ws = new WebSocket(wsUrl);
ws.onopen = async () => {
console.log('WebSocket connected');
document.getElementById('connectionStatus').textContent = 'Connection Status: Connected';
document.getElementById('connectionStatus').classList.add('active');
document.getElementById('connectionStatus').classList.remove('error');
reconnectAttempts = 0;
await createMediaSource();
};
ws.onmessage = (event) => {
try {
const message = JSON.parse(event.data);
if (message.type === 'audio' && message.data) {
ws.send(JSON.stringify({
type: "ack",
data: "audio-received"
}));
const audioData = base64ToUint8Array(message.data);
audioQueue.push(audioData);
if (sourceBuffer && !sourceBuffer.updating) {
processAudioQueue();
}
}
} catch (error) {
console.error('Error processing message:', error);
}
};
ws.onclose = () => {
console.log('WebSocket closed');
document.getElementById('connectionStatus').textContent = 'Connection Status: Disconnected';
document.getElementById('connectionStatus').classList.remove('active');
document.getElementById('connectionStatus').classList.add('error');
const timeout = Math.min(1000 * Math.pow(2, reconnectAttempts), 30000);
reconnectAttempts++;
setTimeout(() => {
initWebSocket();
}, timeout);
};
ws.onerror = (error) => {
console.error('WebSocket error:', error);
document.getElementById('connectionStatus').textContent = 'Connection Status: Error';
document.getElementById('connectionStatus').classList.remove('active');
document.getElementById('connectionStatus').classList.add('error');
};
} catch (error) {
console.error('Error initializing WebSocket:', error);
document.getElementById('connectionStatus').textContent = 'Connection Status: Error - ' + error.message;
document.getElementById('connectionStatus').classList.add('error');
}
}
function base64ToUint8Array(base64) {
const binaryString = atob(base64);
const bytes = new Uint8Array(binaryString.length);
for (let i = 0; i < binaryString.length; i++) {
bytes[i] = binaryString.charCodeAt(i);
}
return bytes;
}
let isCallActive = false;
const startCallButton = document.getElementById('startCall');
const endCallButton = document.getElementById('endCall');
// Add event listeners for the buttons
startCallButton.addEventListener('click', startCall);
endCallButton.addEventListener('click', endCall);
async function startCall() {
if (isCallActive) return;
try {
startCallButton.disabled = true;
// Reset all states before starting new call
await resetStates();
await init();
isCallActive = true;
startCallButton.style.display = 'none';
endCallButton.style.display = 'block';
audioElement.play().catch(e => console.error('Error playing audio:', e));
} catch (error) {
console.error('Error starting call:', error);
startCallButton.disabled = false;
}
}
async function endCall() {
if (!isCallActive) return;
try {
// Stop the microphone
if (mediaStream) {
mediaStream.getTracks().forEach(track => track.stop());
mediaStream = null;
}
// Close audio context
if (audioContext) {
await audioContext.close();
audioContext = null;
}
// Close WebSocket connection
if (ws) {
if (ws.readyState === WebSocket.OPEN) {
ws.close();
}
ws = null;
}
await resetStates();
// Update UI
startCallButton.style.display = 'block';
startCallButton.disabled = false;
endCallButton.style.display = 'none';
document.getElementById('connectionStatus').textContent = 'Connection Status: Disconnected';
document.getElementById('connectionStatus').classList.remove('active');
document.getElementById('micStatus').textContent = 'Microphone Status: Off';
document.getElementById('micStatus').classList.remove('active');
} catch (error) {
console.error('Error ending call:', error);
}
}
async function resetStates() {
// Reset MediaSource
if (mediaSource) {
if (mediaSource.readyState !== 'closed') {
try {
mediaSource.endOfStream();
} catch (e) {
console.error('Error ending media stream:', e);
}
}
mediaSource = null;
}
// Clear source buffer
if (sourceBuffer) {
try {
if (!sourceBuffer.updating) {
sourceBuffer.abort();
}
} catch (e) {
console.error('Error aborting source buffer:', e);
}
sourceBuffer = null;
}
// Reset audio element
try {
audioElement.pause();
audioElement.src = '';
audioElement.load(); // Important: properly reset the audio element
} catch (e) {
console.error('Error resetting audio element:', e);
}
// Reset other states
audioQueue = [];
isSourceOpen = false;
audioBuffer = new Float32Array();
isCallActive = false;
reconnectAttempts = 0;
// Return a promise that resolves after a short delay
return new Promise(resolve => setTimeout(resolve, 100));
}
// Modify the init function to not auto-start
async function init() {
try {
await initWebSocket();
await initAudio();
setInterval(() => {
if (ws && ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify({ type: 'ping' }));
}
}, 30000);
document.addEventListener('click', async () => {
if (audioContext && audioContext.state === 'suspended') {
await audioContext.resume();
}
});
} catch (error) {
console.error('Initialization error:', error);
throw error; // Propagate the error to handle it in startCall
}
}
//init().catch(console.error);
</script>
</body>
</html>
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())
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())