๐ WebRTC Real-time Voice (Beta)
Enhanced real-time voice interaction with streaming transcription
๐ก WebSocket endpoints: /ws/webrtc/{client_id} |
๐งช Test page: WebRTC Demo |
โก API Status: Test Endpoint
""")
# Text input section
with gr.Row():
text_input = gr.Textbox(
label="๐ฌ Type your message or see voice transcription",
placeholder="Hi! I'm [Your Name]. Book a 30-minute meeting tomorrow at 2 PM...",
lines=2,
scale=4
)
send_btn = gr.Button("Send", variant="primary", scale=1)
with gr.Column(scale=1):
# Quick action buttons
gr.Markdown("### ๐ Quick Actions")
quick_meet = gr.Button(
"๐ฅ Google Meet (30m)",
variant="secondary"
)
quick_availability = gr.Button(
"๐
Check Availability",
variant="secondary"
)
quick_cancel = gr.Button(
"โ Cancel Meeting",
variant="secondary"
)
# Version info
version_btn = gr.Button(
"โน๏ธ Version Info",
variant="secondary"
)
version_display = gr.Textbox(
label="Version Information",
interactive=False,
visible=False
)
# Voice settings
gr.Markdown("### ๐ญ Voice Settings")
voice_enabled = gr.Checkbox(
label="Enable voice responses",
value=True
)
voice_selection = gr.Dropdown(
choices=[
"v2/en_speaker_0",
"v2/en_speaker_1",
"v2/en_speaker_2",
"v2/en_speaker_6",
"v2/en_speaker_9"
],
value="v2/en_speaker_6",
label="AI Voice"
)
# Event handlers
def handle_text_submit(message, history, session):
if message.strip():
# Use asyncio to handle the async function
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
app.process_message(message, history, session)
)
return result
finally:
loop.close()
return history, message
def handle_audio_submit(audio, history, session):
print(f"๐ค AUDIO DEBUG: Received audio input: {type(audio)}")
print(f"๐ค AUDIO DEBUG: Audio data: {audio}")
if audio is not None:
print(f"๐ค AUDIO DEBUG: Processing audio...")
# Convert audio data and process
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
# Debug audio format
if isinstance(audio, tuple) and len(audio) >= 2:
sample_rate, audio_array = audio
print(f"๐ค AUDIO DEBUG: Sample rate: {sample_rate}")
print(f"๐ค AUDIO DEBUG: Audio array type: {type(audio_array)}")
print(f"๐ค AUDIO DEBUG: Audio array shape: {audio_array.shape if hasattr(audio_array, 'shape') else 'No shape'}")
# Use the audio handler's process method instead
transcription = app.audio_handler.process_audio_input(audio)
print(f"๐ค AUDIO DEBUG: Transcription result: {transcription}")
if transcription and transcription != "No audio received":
# Process the transcription as a message
result = loop.run_until_complete(
app.process_message(transcription, history, session)
)
# Return updated history, transcription in text box, and no audio output for now
return result[0], transcription, None
else:
print(f"๐ค AUDIO DEBUG: No valid transcription received")
return history, "No audio transcription available", None
else:
print(f"๐ค AUDIO DEBUG: Invalid audio format")
return history, "Invalid audio format", None
except Exception as e:
print(f"๐ค AUDIO ERROR: {str(e)}")
import traceback
traceback.print_exc()
return history, f"Audio processing error: {str(e)}", None
finally:
loop.close()
else:
print(f"๐ค AUDIO DEBUG: No audio received")
return history, "No audio received", None
def handle_quick_action(action_text, history, session):
"""Handle quick action button clicks."""
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
app.process_message(action_text, history, session)
)
return result[0], "" # Return updated history and clear text input
finally:
loop.close()
# Wire up the event handlers
send_btn.click(
fn=handle_text_submit,
inputs=[text_input, chatbot, session_id],
outputs=[chatbot, text_input]
)
text_input.submit(
fn=handle_text_submit,
inputs=[text_input, chatbot, session_id],
outputs=[chatbot, text_input]
)
audio_input.change(
fn=handle_audio_submit,
inputs=[audio_input, chatbot, session_id],
outputs=[chatbot, text_input, audio_output]
)
# Quick action handlers
quick_meet.click(
fn=lambda hist, sess: handle_quick_action(
"Book a 30-minute Google Meet with Peter for next available time",
hist, sess
),
inputs=[chatbot, session_id],
outputs=[chatbot, text_input]
)
quick_availability.click(
fn=lambda hist, sess: handle_quick_action(
"What is Peter's availability this week?",
hist, sess
),
inputs=[chatbot, session_id],
outputs=[chatbot, text_input]
)
quick_cancel.click(
fn=lambda hist, sess: handle_quick_action(
"Cancel my upcoming meeting with Peter",
hist, sess
),
inputs=[chatbot, session_id],
outputs=[chatbot, text_input]
)
# Version info handler
def show_version():
info = get_version_info()
version_text = f"Version: {info['version']}\nBuild: {info['build_date']}\nDescription: {info['description']}\nStatus: {info['status']}"
return version_text, gr.update(visible=True)
version_btn.click(
fn=show_version,
outputs=[version_display, version_display]
)
return demo
# Global app instance
app = ChatCalVoiceApp()
# Create and launch the interface
if __name__ == "__main__":
import uvicorn
try:
# Create WebRTC-enabled FastAPI app as main app
webrtc_app = create_fastapi_app()
# Create Gradio interface (for future integration)
demo = app.create_interface()
# WebRTC-first approach: Launch FastAPI with WebSocket endpoints
print("๐ ChatCal WebRTC-First Deployment v0.4.3")
print("๐ก WebSocket endpoint: /ws/webrtc/{client_id}")
print("๐งช WebRTC demo page: /webrtc/demo")
print("โก API status: /webrtc/test")
print("โ ๏ธ Gradio interface development - WebRTC priority")
# Launch WebRTC FastAPI app directly
uvicorn.run(webrtc_app, host="0.0.0.0", port=7860)
except Exception as e:
print(f"โ WebRTC integration error: {e}")
print("๐ Falling back to Gradio-only deployment")
import traceback
traceback.print_exc()
# Create stable Gradio interface fallback
demo = app.create_interface()
print("๐ ChatCal Voice-Enabled Assistant v0.4.3")
print("๐ฑ Traditional voice input available via Gradio Audio component")
print("โ๏ธ WebRTC real-time streaming: Debugging in progress")
# Launch configuration for HF Spaces (stable fallback)
demo.launch(
server_name="0.0.0.0",
server_port=7860,
share=False, # HF handles sharing
show_error=True
)