Create app.py

app.py

@@ -0,0 +1,101 @@
+from fastapi import FastAPI, WebSocket
+from twilio.twiml.voice_response import VoiceResponse, Connect, Stream
+from pydub import AudioSegment
+import base64
+import asyncio
+import gradio as gr
+from utils import transcribe_audio, generate_response, text_to_speech
+import os
+
+# FastAPI app
+app = FastAPI()
+
+# Twilio voice webhook
+@app.get("/voice")
+async def handle_call():
+    """Handle incoming Twilio voice calls."""
+    response = VoiceResponse()
+    connect = Connect()
+    connect.stream(url="wss://iajitpanday-vBot-1-7.hf.space/media-stream")
+    response.append(connect)
+    return response
+
+# Twilio media stream WebSocket
+@app.websocket("/media-stream")
+async def media_stream(websocket: WebSocket):
+    """Handle Twilio media streams via WebSocket."""
+    await websocket.accept()
+    while True:
+        try:
+            data = await websocket.receive_json()
+            if data["event"] == "media":
+                # Decode base64 audio
+                audio_data = base64.b64decode(data["media"]["payload"])
+                input_path = "input.wav"
+                with open(input_path, "wb") as f:
+                    f.write(audio_data)
+                
+                # Process audio: STT -> NLP -> TTS
+                text = transcribe_audio(input_path)
+                response_text = generate_response(text)
+                output_path = text_to_speech(response_text)
+                
+                if output_path and os.path.exists(output_path):
+                    # Convert to 8kHz MULAW for Twilio
+                    audio = AudioSegment.from_wav(output_path).set_frame_rate(8000).set_channels(1).set_sample_width(2)
+                    audio.export("output.mulaw", format="raw", codec="pcm_mulaw")
+                    with open("output.mulaw", "rb") as f:
+                        response_audio = base64.b64encode(f.read()).decode("utf-8")
+                    
+                    # Send audio back to Twilio
+                    await websocket.send_json({
+                        "event": "media",
+                        "streamSid": data["streamSid"],
+                        "media": {"payload": response_audio}
+                    })
+                else:
+                    print("TTS failed, skipping response.")
+                
+                # Clean up
+                for path in [input_path, output_path, "output.mulaw"]:
+                    if os.path.exists(path):
+                        os.remove(path)
+            elif data["event"] == "stop":
+                break
+        except Exception as e:
+            print(f"WebSocket Error: {e}")
+            break
+    await websocket.close()
+
+# Gradio interface for testing
+def test_voice_bot(audio):
+    """Test the voice bot pipeline via Gradio UI."""
+    if audio is None:
+        return "No audio provided.", None
+    input_path = "test_input.wav"
+    sf.write(input_path, audio[1], audio[0])  # audio[0] is sample rate, audio[1] is data
+    text = transcribe_audio(input_path)
+    response_text = generate_response(text)
+    output_path = text_to_speech(response_text)
+    os.remove(input_path)
+    if output_path and os.path.exists(output_path):
+        return response_text, output_path
+    return response_text, None
+
+# Gradio UI
+with gr.Blocks() as demo:
+    gr.Markdown("# Voice AI Bot Tester")
+    gr.Markdown("Upload or record audio to test the bot's response.")
+    audio_input = gr.Audio(sources=["microphone", "upload"], type="numpy")
+    text_output = gr.Textbox(label="Bot Response Text")
+    audio_output = gr.Audio(label="Bot Response Audio")
+    submit_btn = gr.Button("Test")
+    submit_btn.click(
+        fn=test_voice_bot,
+        inputs=audio_input,
+        outputs=[text_output, audio_output]
+    )
+
+# Launch Gradio app
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)