[Python] React js with Pipecat Websocket Implement

Backend guy develop a backend with PY. And send me this script which works fine in windows terminal. I try to convert it in frontend using reactjs. but i get some issue.

First of all the frontend and backend is connected with websocket. But from frontend am sending audio and its not receiving it in backend.

Can someone please help me to convert this in reactjs? i cant share full code because of text limiation

Am using `localhost` in frontend and server url also `http`

This is the Test.py which working fine in terminal

import asyncio
import aiohttp
import websockets
import wave
import time
import os
import numpy as np
import pyaudio
from datetime import datetime
import threading
import queue
import keyboard
import sys


# Configuration
SERVER_URL = ""
TARGET_SAMPLE_RATE = 16000
TARGET_CHANNELS = 1
TARGET_SAMPLE_WIDTH = 2
CHUNK_SIZE = 1024
SILENCE_THRESHOLD = 30
RECEIVE_POLL_TIMEOUT = 0.1
RESPONSE_END_SILENCE = 2.0
MICROPHONE_BOOST = 4.0 # Reduced boost to prevent overflow


class VoiceChatBot:
def __init__(self):
self.session_start = None
self.total_sent_bytes = 0
self.total_received_bytes = 0
self.audio_chunks_sent = 0
self.audio_chunks_received = 0
self.recording = False
self.speaking = False
self.ws = None

# Initialize PyAudio
self.pyaudio = pyaudio.PyAudio()

# Get default input device
try:
default_input = self.pyaudio.get_default_input_device_info()
print(f" Using microphone: {default_input['name']}")
self.input_device_index = default_input['index']
except Exception as e:
print(f"❌ No default input device found: {e}")
self.input_device_index = None
return

# Setup input stream
try:
self.input_stream = self.pyaudio.open(
format=pyaudio.paInt16,
channels=TARGET_CHANNELS,
rate=TARGET_SAMPLE_RATE,
input=True,
input_device_index=self.input_device_index,
frames_per_buffer=CHUNK_SIZE,
start=False
)
print("✅ Microphone stream created with enhanced error handling")
except Exception as e:
print(f"❌ Failed to create microphone stream: {e}")
self.input_stream = None
return

# Setup output stream
try:
self.output_stream = self.pyaudio.open(
format=pyaudio.paInt16,
channels=TARGET_CHANNELS,
rate=24000,
output=True,
frames_per_buffer=CHUNK_SIZE
)
print("✅ Speaker stream created successfully")
except Exception as e:
print(f"❌ Failed to create speaker stream: {e}")
self.output_stream = None

def log_stat(self, event, data=""):
timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
print(f"[{timestamp}] {event}: {data}")

def get_audio_level(self, data):
"""SAFE audio level calculation with NaN protection."""
try:
if not data or len(data) == 0:
return 0.0

audio_data = np.frombuffer(data, dtype=np.int16)

if audio_data.size == 0:
return 0.0

# Calculate mean square with safety checks
mean_sq = np.mean(audio_data.astype(np.float64)**2)

# Check for invalid values
if np.isnan(mean_sq) or np.isinf(mean_sq) or mean_sq < 0:
return 0.0

level = np.sqrt(mean_sq)

# Final safety check
if np.isnan(level) or np.isinf(level):
return 0.0

return float(level)

except Exception as e:
# Silent fallback for audio processing
return 0.0

def boost_audio(self, audio_data):
"""SAFE audio amplification with overflow protection."""
try:
if not audio_data or len(audio_data) == 0:
return audio_data

# Convert to numpy array
audio_array = np.frombuffer(audio_data, dtype=np.int16)

if audio_array.size == 0:
return audio_data

# Apply amplification with float64 precision
boosted = audio_array.astype(np.float64) * MICROPHONE_BOOST

# Strict clipping to prevent overflow
boosted = np.clip(boosted, -32767.0, 32767.0)

# Convert back to int16 with safety
result = boosted.astype(np.int16)

return result.tobytes()

except Exception as e:
print(f"⚠️ Boost error (using original): {e}")
return audio_data

async def record_audio(self):
"""Enhanced recording with bulletproof error handling."""
print("\n **SAFE MODE FOR REALTEK MICROPHONES**")
print(" Press and HOLD SPACE to talk")
print(" Press ESC to exit")

if not self.input_stream:
print("❌ Microphone not available!")
return None

audio_data = bytearray()
consecutive_errors = 0

while True:
try:
if keyboard.is_pressed('space') and not self.recording:
if not self.speaking:
self.recording = True
audio_data = bytearray()
consecutive_errors = 0

try:
if not self.input_stream.is_active():
self.input_stream.start_stream()
print(" Recording... (release SPACE to send)")
except Exception as e:
print(f"❌ Failed to start recording: {e}")
self.recording = False
continue
else:
print("⏳ Bot is speaking, please wait...")

elif not keyboard.is_pressed('space') and self.recording:
self.recording = False
try:
if self.input_stream.is_active():
self.input_stream.stop_stream()
except:
pass

if len(audio_data) > 500:
print(f" Sending audio ({len(audio_data)} bytes)")
return bytes(audio_data)
else:
print("⚠️ Audio too short, try speaking longer")

elif keyboard.is_pressed('esc'):
print(" Exiting...")
return None

if self.recording and not self.speaking:
try:
if self.input_stream.is_active():
# Read with exception handling
data = self.input_stream.read(CHUNK_SIZE, exception_on_overflow=False)

if data and len(data) > 0:
# Boost audio safely
boosted_data = self.boost_audio(data)

# Calculate level safely
audio_level = self.get_audio_level(boosted_data)

# Add to buffer
audio_data.extend(boosted_data)

# Visual feedback (every ~0.5 seconds)
if len(audio_data) % (CHUNK_SIZE * 8) == 0:
try:
level_int = int(audio_level) # Safe conversion
level_bars = "█" * min(10, max(0, level_int // 200))
print(f"\r Level: {level_int:4d} {level_bars:10} ({len(audio_data)} bytes)", end="", flush=True)
except:
print(f"\r Recording: {len(audio_data)} bytes", end="", flush=True)

consecutive_errors = 0 # Reset error counter
else:
print("⚠️ Empty audio data received")

except Exception as e:
consecutive_errors += 1
if consecutive_errors < 5: # Allow some errors
continue
else:
print(f"\n❌ Too many recording errors: {e}")
self.recording = False
break

await asyncio.sleep(0.01)

except Exception as e:
print(f"\n❌ Record loop error: {e}")
break

return None

async def play_response(self):
"""Enhanced response playback."""
last_receive_time = time.time()
response_started = False

try:
while True:
try:
msg = await asyncio.wait_for(self.ws.recv(), timeout=RECEIVE_POLL_TIMEOUT)

if isinstance(msg, bytes) and len(msg) > 0:
current_time = time.time()

if not response_started:
response_started = True
self.speaking = True
print("\n Bot is speaking...")

if self.output_stream:
try:
self.output_stream.write(msg)
except Exception as e:
# Continue even if playback has issues
pass

self.total_received_bytes += len(msg)
self.audio_chunks_received += 1
last_receive_time = current_time

except asyncio.TimeoutError:
current_time = time.time()
if response_started and current_time - last_receive_time > RESPONSE_END_SILENCE:
print("\n✅ Bot finished speaking")
response_started = False
self.speaking = False
last_receive_time = current_time

except websockets.ConnectionClosed:
break

except Exception as e:
self.log_stat("PLAYBACK_ERROR", str(e))

async def send_audio_chunk(self, audio_data):
"""Send audio with proper chunking."""
if not audio_data or not self.ws:
return

chunk_size = int(TARGET_SAMPLE_RATE * TARGET_SAMPLE_WIDTH * 0.25)
pos = 0

print(f" Sending audio ({len(audio_data)} bytes)...")

while pos < len(audio_data):
end = min(pos + chunk_size, len(audio_data))
chunk = audio_data[pos:end]

if len(chunk) > 0:
await self.ws.send(chunk)
self.total_sent_bytes += len(chunk)
self.audio_chunks_sent += 1
await asyncio.sleep(0.01)

pos = end

# Send silence to trigger processing
silence_data = np.zeros(int(TARGET_SAMPLE_RATE * 0.5), dtype=np.int16)
await self.ws.send(silence_data.tobytes())
print("✅ Audio sent successfully")

async def chat_loop(self, ws_url):
"""Main chat loop with enhanced error handling."""
self.log_stat("SESSION_START", f"WebSocket URL: {ws_url}")

async with websockets.connect(ws_url) as ws:
self.ws = ws
self.log_stat("WEBSOCKET", "Connected successfully")

response_task = asyncio.create_task(self.play_response())

print("\n" + "="*60)
print("️ SAFE MODE ARABIC VOICE CHAT - READY!")
print("="*60)
print("ENHANCED ERROR HANDLING:")
print("• Safe audio processing (no NaN errors)")
print("• Overflow protection")
print("• Robust microphone handling")
print("="*60)
print("CONTROLS:")
print("• Press and HOLD SPACE to talk")
print("• Release SPACE to send")
print("• Press ESC to exit")
print("="*60)

try:
conversation_count = 0
while True:
audio_data = await self.record_audio()

if audio_data is None:
break

if len(audio_data) == 0:
continue

conversation_count += 1
self.log_stat("CONVERSATION", f"Turn {conversation_count}")

await self.send_audio_chunk(audio_data)
print("⏳ Waiting for response...")

except KeyboardInterrupt:
print("\n Chat ended by user")
except Exception as e:
self.log_stat("CHAT_ERROR", str(e))
finally:
response_task.cancel()
try:
await response_task
except asyncio.CancelledError:
pass

self.print_session_stats()

def print_session_stats(self):
print("\n" + "="*60)
print(" SESSION STATISTICS")
print("="*60)
print(f"Audio sent: {self.audio_chunks_sent} chunks, {self.total_sent_bytes} bytes")
print(f"Audio received: {self.audio_chunks_received} chunks, {self.total_received_bytes} bytes")
print("="*60)

def cleanup(self):
try:
if self.input_stream:
if self.input_stream.is_active():
self.input_stream.stop_stream()
self.input_stream.close()
if self.output_stream:
if self.output_stream.is_active():
self.output_stream.stop_stream()
self.output_stream.close()
self.pyaudio.terminate()
print(" Audio resources cleaned up")
except Exception as e:
print(f"⚠️ Cleanup error: {e}")


async def create_session():
async with aiohttp.ClientSession() as session:
async with session.post(f"{SERVER_URL}/api/v1/ws/voice/session/create") as response:
if response.status == 200:
data = await response.json()
return data.get("websocket_url")
else:
raise Exception(f"Failed to create session: {response.status}")


async def main():
bot = None
try:
print(" Starting Arabic Voice Chat Bot - SAFE MODE")
print(f"Server: {SERVER_URL}")
print("-" * 60)

print("Creating session...")
ws_url = await create_session()

if ws_url:
print(f"✅ Session created successfully")

bot = VoiceChatBot()

if bot.input_stream is None:
print("❌ Cannot start without working microphone")
return

await bot.chat_loop(ws_url)

else:
print("❌ Failed to create session")

except Exception as e:
print(f"❌ Error: {e}")
import traceback
traceback.print_exc()
finally:
if bot:
bot.cleanup()


if __name__ == "__main__":
print("️ SAFE MODE - Enhanced error handling for audio processing")
print(" Fixes NaN errors and overflow issues")
print()

required_packages = ['pyaudio', 'websockets', 'aiohttp', 'numpy', 'keyboard']
missing = []

for package in required_packages:
try:
__import__(package)
except ImportError:
missing.append(package)

if missing:
print(f"❌ Install: pip install {' '.join(missing)}")
exit(1)

asyncio.run(main())


How can i solve it?

Thank you

Continue reading...