Update app.py

app.py

@@ -1,527 +1,300 @@
+#!/usr/bin/env python3
+#
+#  san_integration_script.py (v5 - Provider Format Fix)
+#  ======================================================
+#  Description:
+#  - Establishes a real-time, two-way audio bridge between a SAN system
+#    and the Millis AI platform.
+#  - Dynamically detects the audio format from the SAN `start` event.
+#  - Forwards inbound audio to Millis AI at 16kHz for processing.
+#  - Receives the AI's audio response at 16kHz.
+#  - Streams the audio back to the SAN system using the exact format
+#    it originally specified.
+#
+#  Changes in this version:
+#  - Fixed the `reverse-media` event payload to match the provider's
+#    expected format (simplified JSON, lowercase 'callid').
+#  - Fixed `ImportError` by changing `starlette.websockets.State` to
+#    `starlette.websockets.WebSocketState`.
+#  - Updated the final connection check to use `WebSocketState.DISCONNECTED`.
+#  -------------------------------------------------------------------
+
 import os
-import sys
 import json
-import signal
-import argparse
-import uvicorn
 import base64
-import wave
 import asyncio
+import logging
 from datetime import datetime
+
+# Third-party libraries
+import numpy as np
+from scipy import signal as scipy_signal
+import websockets
+from websockets.connection import State as WsState
 from fastapi import FastAPI, WebSocket, WebSocketDisconnect
-from fastapi.responses import JSONResponse
-from websockets.exceptions import ConnectionClosed
+import uvicorn
+
+# Import WebSocketState instead of State
+from starlette.websockets import WebSocketState
 
+# ---------- Logging Configuration -----------------------------------------
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger("san-integration-app")
+
+# ---------- FastAPI Application -------------------------------------------
 app = FastAPI()
 
-# Try to create audio folder with proper error handling
-AUDIO_FOLDER = os.path.join(os.getcwd(), "audio")
-try:
-    if not os.path.exists(AUDIO_FOLDER):
-        os.makedirs(AUDIO_FOLDER, exist_ok=True)
-    # Test if we can write to the folder
-    test_file = os.path.join(AUDIO_FOLDER, "test.txt")
-    with open(test_file, 'w') as f:
-        f.write("test")
-    os.remove(test_file)
-    print(f"[*] Successfully created and verified audio folder at: {AUDIO_FOLDER}")
-except Exception as e:
-    print(f"[!] Warning: Could not create or write to audio folder: {str(e)}")
-    print(f"[!] Audio saving will be disabled")
-    AUDIO_FOLDER = None
-
-def show_exact_websocket_data(message, message_count, client_ip):
-    """Show the EXACT raw WebSocket message as received"""
-    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')
-    
-    print(f"\n" + "="*100)
-    print(f"[*] EXACT WEBSOCKET MESSAGE #{message_count} FROM {client_ip}")
-    print(f"[*] Timestamp: {timestamp}")
-    print(f"[*] Message Object Type: {type(message).__name__}")
-    print(f"[*] Message Keys: {list(message.keys()) if isinstance(message, dict) else 'N/A'}")
-    print(f"[*] Message Type Field: {message.get('type', 'NO TYPE FIELD')}")
-    print("="*100)
-    
-    # Show the COMPLETE raw message object
-    print(f"[*] === COMPLETE RAW MESSAGE OBJECT ===")
-    print(f"EXACT MESSAGE: {message}")
-    print(f"MESSAGE REPR: {repr(message)}")
-    print(f"MESSAGE STR: {str(message)}")
-    
-    # Handle different message content types
-    if message.get("type") == "websocket.receive":
-        if "text" in message:
-            text_data = message["text"]
-            print(f"\n[*] === TEXT DATA (EXACT AS RECEIVED) ===")
-            print(f"Text Data Type: {type(text_data).__name__}")
-            print(f"Text Data Length: {len(text_data)} characters")
-            print(f"EXACT TEXT DATA:")
-            print(f"'{text_data}'")
-            print(f"TEXT DATA REPR:")
-            print(f"{repr(text_data)}")
-            
-            # Show character-by-character breakdown if not too long
-            if len(text_data) <= 1000:
-                print(f"\n[*] CHARACTER-BY-CHARACTER BREAKDOWN:")
-                for i, char in enumerate(text_data):
-                    if i < 100:  # Limit to first 100 chars for readability
-                        print(f"  [{i:3d}]: '{char}' (ord: {ord(char)}, hex: 0x{ord(char):02x})")
-                    elif i == 100:
-                        print(f"  ... (showing first 100 characters only)")
-                        break
-            
-        elif "bytes" in message:
-            binary_data = message["bytes"]
-            print(f"\n[*] === BINARY DATA (EXACT AS RECEIVED) ===")
-            print(f"Binary Data Type: {type(binary_data).__name__}")
-            print(f"Binary Data Length: {len(binary_data)} bytes")
-            print(f"\nEXACT BASE64 REPRESENTATION:")
-            print(f"{base64.b64encode(binary_data).decode('utf-8')}")
-                    
-        else:
-            print(f"\n[*] === OTHER MESSAGE CONTENT ===")
-            for key, value in message.items():
-                if key != "type":
-                    print(f"{key}: {repr(value)}")
-                    
-    else:
-        print(f"\n[*] === NON-RECEIVE MESSAGE ===")
-        print(f"Complete message content: {message}")
-    
-    print("="*100 + "\n")
-
-def save_exact_raw_data(call_id, message_count, message):
-    """Save the exact raw WebSocket message to file"""
-    if AUDIO_FOLDER is None:
-        return False
-    
-    try:
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
-        filename = f"{AUDIO_FOLDER}/raw_websocket_{call_id}_{message_count}_{timestamp}.txt"
-        
-        with open(filename, 'w', encoding='utf-8') as f:
-            f.write(f"WebSocket Message #{message_count}\n")
-            f.write(f"Timestamp: {timestamp}\n")
-            f.write(f"Call ID: {call_id}\n")
-            f.write(f"Message Type: {type(message).__name__}\n")
-            f.write(f"Message Keys: {list(message.keys()) if isinstance(message, dict) else 'N/A'}\n")
-            f.write(f"\n=== EXACT RAW MESSAGE ===\n")
-            f.write(f"{message}\n")
-            f.write(f"\n=== MESSAGE REPR ===\n")
-            f.write(f"{repr(message)}\n")
-            
-            if message.get("type") == "websocket.receive":
-                if "text" in message:
-                    f.write(f"\n=== TEXT DATA ===\n")
-                    f.write(f"Length: {len(message['text'])}\n")
-                    f.write(f"Content: {repr(message['text'])}\n")
-                    f.write(f"Raw Text:\n{message['text']}\n")
-                    
-                elif "bytes" in message:
-                    f.write(f"\n=== BINARY DATA ===\n")
-                    f.write(f"Length: {len(message['bytes'])}\n")
-                    f.write(f"Hex: {message['bytes'].hex()}\n")
-                    f.write(f"Base64: {base64.b64encode(message['bytes']).decode('utf-8')}\n")
-                    f.write(f"Repr: {repr(message['bytes'])}\n")
-        
-        print(f"[*] ✅ Exact raw data saved to: {filename}")
-        return True
-        
-    except Exception as e:
-        print(f"[!] ❌ Failed to save exact raw data: {e}")
-        return False
-
-def save_audio_chunk(call_id, payload, sample_rate=44100, encoding="LINEAR16"):
-    """Save audio chunk to WAV file with proper format handling."""
-    if AUDIO_FOLDER is None:
-        print("[!] Audio saving is disabled due to folder permission issues")
-        return False
-        
-    try:
-        # Decode base64 payload
-        audio_data = base64.b64decode(payload)
-        
-        # Create filename with timestamp
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
-        filename = f"{AUDIO_FOLDER}/call_{call_id}_{timestamp}.wav"
-        
-        # Save as WAV file with correct format
-        with wave.open(filename, 'wb') as wav_file:
-            wav_file.setnchannels(1)  # Mono audio
-            
-            # Handle different encodings
-            if encoding == "LINEAR16":
-                wav_file.setsampwidth(2)  # 16-bit PCM
-            else:
-                wav_file.setsampwidth(2)  # Default to 16-bit
-                
-            wav_file.setframerate(sample_rate)
-            wav_file.writeframes(audio_data)
-        
-        print(f"[*] Saved audio chunk to {filename} (Rate: {sample_rate}Hz, Encoding: {encoding})")
-        return True
-    except Exception as e:
-        print(f"[!] Error saving audio chunk: {str(e)}")
-        return False
-
-def save_binary_audio_chunk(call_id, audio_data, sample_rate=44100, encoding="LINEAR16"):
-    """Save binary audio data directly to WAV file."""
-    if AUDIO_FOLDER is None:
-        print("[!] Audio saving is disabled due to folder permission issues")
-        return False
-        
-    try:
-        # Create filename with timestamp
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
-        filename = f"{AUDIO_FOLDER}/call_{call_id}_{timestamp}.wav"
-        
-        # Save as WAV file with correct format
-        with wave.open(filename, 'wb') as wav_file:
-            wav_file.setnchannels(1)  # Mono audio
-            
-            # Handle different encodings
-            if encoding == "LINEAR16":
-                wav_file.setsampwidth(2)  # 16-bit PCM
-            else:
-                wav_file.setsampwidth(2)  # Default to 16-bit
-                
-            wav_file.setframerate(sample_rate)
-            wav_file.writeframes(audio_data)
-        
-        print(f"[*] Saved binary audio chunk to {filename} (Rate: {sample_rate}Hz, Encoding: {encoding})")
-        return True
-    except Exception as e:
-        print(f"[!] Error saving binary audio chunk: {str(e)}")
-        return False
+# ---------- Environment & Configuration -----------------------------------
+AGENT_ID = os.getenv("MILLIS_AGENT_ID", "-OTBEKt8tHp6GI6AeRJ2")
+PUBLIC_KEY = os.getenv("MILLIS_PUBLIC_KEY", "Dhr5TEtwlpACHNrDmdxQZXDDtM3PgEJi")
+MILLIS_WS_URI = "wss://api-west.millis.ai:8080/millis"
+
+# ---------------------------------------------------------------------------#
+#                          REAL-TIME AUDIO PROCESSOR                         #
+# ---------------------------------------------------------------------------#
+class RealTimeAudioProcessor:
+    """
+    Manages a single live call, bridging audio between the SAN system and Millis AI.
+    """
+    PHONE_RATE = 8000
+    MILLIS_RATE = 16000
+    CHUNK_MS = 20
+    BYTES_PER_SAMPLE = 2
+
+    MILLIS_CHUNK_SIZE = int(MILLIS_RATE * CHUNK_MS / 1000 * BYTES_PER_SAMPLE)
+    PHONE_CHUNK_SIZE = int(PHONE_RATE * CHUNK_MS / 1000 * BYTES_PER_SAMPLE)
+
+    def __init__(self, agent_id: str, public_key: str):
+        self.agent_id = agent_id
+        self.public_key = public_key
+        self.ws: websockets.WebSocketClientProtocol | None = None
+        self.connected = False
+
+        self.inbound = bytearray()
+        self.outbound = bytearray()
+        self.in_lock = asyncio.Lock()
+        self.out_lock = asyncio.Lock()
+
+        self.is_paused = False
+        self.stream_id: str | None = None
+        self.call_id: str | None = None
+        self.media_format: dict = {
+            "encoding": "PCM", "sampleRate": self.PHONE_RATE, "channels": 1
+        }
+        self._packet_counter = 0
+
+    async def connect(self) -> bool:
+        logger.info("🤖 Connecting to Millis AI...")
+        try:
+            self.ws = await websockets.connect(MILLIS_WS_URI, open_timeout=10)
+            await self.ws.send(
+                json.dumps({
+                    "method": "initiate",
+                    "data": {"agent": {"agent_id": self.agent_id}, "public_key": self.public_key},
+                })
+            )
+            msg = await asyncio.wait_for(self.ws.recv(), timeout=10)
+            if json.loads(msg).get("method") != "onready":
+                raise RuntimeError("Millis AI did not send 'onready' confirmation.")
+            self.connected = True
+            logger.info("✅ Successfully connected to Millis AI.")
+            return True
+        except Exception as e:
+            logger.error(f"❌ Millis AI connection failed: {e}")
+            self.connected = False
+            return False
+
+    async def disconnect(self):
+        if self.ws and self.ws.state != WsState.CLOSED:
+            await self.ws.close()
+        self.connected = False
+        self.ws = None
+        logger.info("🔌 Disconnected from Millis AI.")
+
+    @staticmethod
+    def _resample(data: bytes, from_rate: int, to_rate: int) -> bytes:
+        if not data: return b""
+        arr = np.frombuffer(data, dtype=np.int16)
+        if arr.size == 0: return b""
+        new_len = int(arr.size * to_rate / from_rate)
+        resampled = scipy_signal.resample(arr, new_len).astype(np.int16)
+        return resampled.tobytes()
+
+    async def _pump_inbound_to_millis(self):
+        while self.connected:
+            chunk8 = None
+            async with self.in_lock:
+                if len(self.inbound) >= self.PHONE_CHUNK_SIZE:
+                    chunk8 = self.inbound[:self.PHONE_CHUNK_SIZE]
+                    del self.inbound[:self.PHONE_CHUNK_SIZE]
+            if not chunk8:
+                await asyncio.sleep(0.005)
+                continue
+            try:
+                chunk16 = self._resample(chunk8, self.PHONE_RATE, self.MILLIS_RATE)
+                await self.ws.send(chunk16)
+                self._packet_counter += 1
+                if self._packet_counter >= 1_000:
+                    await self.ws.send(json.dumps({"method": "ping"}))
+                    self._packet_counter = 0
+            except Exception as e:
+                logger.error(f"❌ Error in _pump_inbound_to_millis: {e}")
+                self.connected = False
+
+    async def _pump_millis_to_outbound(self):
+        while self.connected and self.ws and self.ws.state == WsState.OPEN:
+            try:
+                msg = await self.ws.recv()
+                if isinstance(msg, bytes):
+                    async with self.out_lock: self.outbound.extend(msg)
+                    continue
+                evt = json.loads(msg)
+                method = evt.get("method")
+                logger.info(f"🤖 JSON from Millis: {evt}")
+                if method == "pause": self.is_paused = True
+                elif method == "unpause": self.is_paused = False
+                elif method in ("clear", "start_answering"):
+                    async with self.out_lock: self.outbound.clear()
+                    self.is_paused = False
+            except websockets.exceptions.ConnectionClosed:
+                logger.warning("🔌 Millis AI closed the connection.")
+                self.connected = False
+            except Exception as e:
+                logger.warning(f"⚠️ Error reading from Millis AI: {e}")
+                self.connected = False
 
+    async def _pump_outbound_to_carrier(self, client_ws: WebSocket):
+        sent_packets = 0
+        while self.connected:
+            if self.is_paused:
+                await asyncio.sleep(0.01)
+                continue
+            chunk16 = None
+            async with self.out_lock:
+                if len(self.outbound) >= self.MILLIS_CHUNK_SIZE:
+                    chunk16 = self.outbound[:self.MILLIS_CHUNK_SIZE]
+                    del self.outbound[:self.MILLIS_CHUNK_SIZE]
+            if not chunk16:
+                await asyncio.sleep(0.005)
+                continue
+            try:
+                target_rate = self.media_format.get("sampleRate", self.PHONE_RATE)
+                chunk_resampled = self._resample(chunk16, self.MILLIS_RATE, target_rate)
+                payload = base64.b64encode(chunk_resampled).decode()
+                sent_packets += 1
+                if sent_packets % 100 == 1:
+                    logger.info(f"⬆️  Sending upstream audio packet #{sent_packets} to SAN...")
+
+                ### --- FIX: Modified the JSON payload to match the provider's simple format --- ###
+                await client_ws.send_json({
+                    "event": "reverse-media",
+                    "callid": self.call_id,  # Changed from "callId" to "callid"
+                    "payload": payload,
+                    # Removed "streamId" and "mediaFormat" fields
+                })
+                ### --- END FIX --- ###
+
+            except Exception as e:
+                logger.error(f"❌ Error in _pump_outbound_to_carrier: {e}")
+                break
+
+    async def start(self, client_ws: WebSocket) -> list[asyncio.Task]:
+        if not await self.connect(): return []
+        tasks = [
+            asyncio.create_task(self._pump_millis_to_outbound()),
+            asyncio.create_task(self._pump_inbound_to_millis()),
+            asyncio.create_task(self._pump_outbound_to_carrier(client_ws)),
+        ]
+        return tasks
+
+async def stop_processor(proc: RealTimeAudioProcessor | None, tasks: list[asyncio.Task]):
+    if not proc: return
+    for t in tasks:
+        if not t.done(): t.cancel()
+    await proc.disconnect()
+
+# ---------------------------------------------------------------------------#
+#                          FASTAPI /media ENDPOINT                           #
+# ---------------------------------------------------------------------------#
 @app.websocket("/media")
-async def media_endpoint(websocket: WebSocket):
-    client_ip = websocket.client.host if websocket.client else "unknown"
-    print(f"[*] WebSocket connection attempt from {client_ip}")
-    
-    # Connection state tracking
-    connection_start_time = datetime.now()
-    message_count = 0
-    current_media_format = None
-    last_message_time = None
-    
+async def media_socket(ws: WebSocket):
+    await ws.accept()
+    logger.info("🔗 SAN system WebSocket accepted.")
+
+    processor: RealTimeAudioProcessor | None = None
+    tasks: list[asyncio.Task] = []
+    active_call_id: str | None = None
+
     try:
-        await websocket.accept()
-        print(f"[*] WebSocket connection accepted from {client_ip} at {connection_start_time}")
-        
         while True:
-            try:
-                # Receive any type of message (text or binary)
-                message = await asyncio.wait_for(websocket.receive(), timeout=60.0)
-                message_count += 1
-                last_message_time = datetime.now()
-                
-                # Show EXACT raw WebSocket data as received
-                show_exact_websocket_data(message, message_count, client_ip)
-                
-                # Save exact raw data to file
-                call_id = f"session_{connection_start_time.strftime('%H%M%S')}"
-                save_exact_raw_data(call_id, message_count, message)
-                
-                # Process the message for application logic
-                if message["type"] == "websocket.receive":
-                    if "text" in message:
-                        text_data = message["text"]
-                        
-                        try:
-                            json_data = json.loads(text_data)
-                            
-                            # Handle media format setup
-                            if "mediaFormat" in json_data:
-                                print(f"[*] 📡 Media format detected: {json_data['mediaFormat']}")
-                                current_media_format = json_data["mediaFormat"]
-                                print(f"[*] Media format updated globally")
-                                continue
-                            
-                            # Handle media events with payload
-                            if json_data.get("event") == "media" and "payload" in json_data:
-                                call_id = json_data.get("callId", "unknown")
-                                
-                                if current_media_format:
-                                    sample_rate = current_media_format.get("sampleRate", 44100)
-                                    encoding = current_media_format.get("encoding", "LINEAR16")
-                                else:
-                                    sample_rate = json_data.get("sampleRate", 44100)
-                                    encoding = json_data.get("encoding", "LINEAR16")
-                                
-                                print(f"[*] 🎵 Processing audio chunk: Rate={sample_rate}Hz, Encoding={encoding}, CallID={call_id}")
-                                
-                                if save_audio_chunk(call_id, json_data["payload"], sample_rate, encoding):
-                                    print(f"[*] ✅ Audio chunk saved successfully for callId: {call_id}")
-                                continue
-                            
-                            # Handle base64 payload messages
-                            if "payload" in json_data and json_data.get("encoding") == "base64":
-                                call_id = json_data.get("callId", json_data.get("streamId", f"session_{connection_start_time.strftime('%H%M%S')}"))
-                                print(f"[*] 🎵 Processing base64 payload for callId: {call_id}")
-                                
-                                sample_rate = 44100
-                                encoding = "LINEAR16"
-                                if current_media_format:
-                                    sample_rate = current_media_format.get("sampleRate", 44100)
-                                    encoding = current_media_format.get("encoding", "LINEAR16")
-                                
-                                if save_audio_chunk(call_id, json_data["payload"], sample_rate, encoding):
-                                    print(f"[*] ✅ Base64 audio chunk saved successfully")
-                                continue
-                            
-                            # Handle events that need responses
-                            if json_data.get("event") in ["connected", "start", "ready"]:
-                                print(f"[*] 📞 Setup event received: {json_data['event']}")
-                                
-                                response = {
-                                    "event": "media-ready",
-                                    "callId": json_data.get("callId", ""),
-                                    "streamId": json_data.get("streamId", ""),
-                                    "status": "ready",
-                                    "timestamp": datetime.now().isoformat()
-                                }
-                                await websocket.send_json(response)
-                                print(f"[*] ✅ Acknowledgment sent for {json_data['event']} event")
-                                continue
-                            
-                            # Handle answer events
-                            required_fields = {
-                                "timestamp": str, "streamId": str, "callerId": str,
-                                "channelId": str, "event": str, "callDirection": str,
-                                "did": str, "callId": str, "cid": str, "extraParams": str
-                            }
-                            
-                            if (all(field in json_data and isinstance(json_data[field], field_type) 
-                                   for field, field_type in required_fields.items()) and 
-                                json_data["event"] == "answer"):
-                                
-                                print(f"[*] 📞 Answer event detected, preparing response...")
-                                
-                                response_format = {
-                                    "encoding": "LINEAR16",
-                                    "sampleRate": 44100,
-                                    "channels": 1
-                                }
-                                
-                                if current_media_format:
-                                    response_format.update(current_media_format)
-                                
-                                response = {
-                                    "event": "reverse-media",
-                                    "callId": json_data["callId"],
-                                    "streamId": json_data["streamId"],
-                                    "chunk": 1,
-                                    "chunk_durn_ms": 20,
-                                    "mediaFormat": response_format,
-                                    "payload": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAP//AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA//8AAAAAAAAAAAAAAAAAAAAA//8AAAAAAAD//////////////////wAA/////wAAAAD/////AAD//wAAAAABAP//AAAAAAEAAAAAAAEAAQABAAAAAQACAAEAAAABAAMAAAD//wEAAQABAAEAAAABAAAAAgD+/wAA//8BAPr/7f/z/+r/7P/w/+H/1//a/+L/5f/j/+r/8/////b/5v/j//3/AgDt/9r/6f8DAAAA7//9/xYADQDs/+//BgAJAAUAAgAJAA4ADwARABgAIAAlADIAMwAxAEQAUQBZAFIASgA/ADgAOwBFAEUARwBRAFMAOwAvAEIATQBPAEMAQgBJAEQAOgA8AEEAPQA9ADUAHwAgACYALQAaABIAFQATAAQA9f8FAPP/7v/w/+v/8f/X/83/x//L/83/yf/Q/7n/v/+9/8P/wP++/8j/yf/B/67/uv/X/9T/0v/I/9n/3v/K/9D/0v/0//P/8f/O/9z/8f/4//r/7v8LAAQAAwANAAwADgAaABkACwAZACMAMAA3ACMAHAAnACkAKwAoADoAPQA5AD0ANQAuACMARgBEADAANgBAAD8AKwApADIAOwAnAAsAEAASABEAGwAWAA4A/f/6//3/AQD6/+7/9v/+//X/4f/k/+r/5//s//X/7P/a/9j/4P/v/+v/2v/k/+X/5P/i/9T/2v/r//3/+P/7/wAA+f8IAAIABgAKACIAFAD9/xIAGwAsACMANgArAC4AJQAYACEAMwBTAEMAIgApADQAMAAWAC0ARgBFACAAGAAnACUAHwAPACAAIwAVAAkABQDy//T/FQAVAO7/5//1//D/3f/d//j//P/a/8L/wv+9/8X/2f/h/87/s//A/7v/u//C/73/vv+8/7//vP+//8T/y//I/8D/vf/K/9N/1v/P/9b/2//T/9f/2v/f/+r/4v/X/9f/7f/6//D/7//q//P/8P/1//7/AgALAA0ACgAHAAkAEAAQAA4ADgAZABMABgAJABYAGAAGAAwAEwAXABAABwAHAAoAEAAVABIACAABAAIABgAGABAAEAAWAA0A/P/4/wQACAADAAcABwAAAPP/8f/+/woABgALAAAA6//q//3/BAD1/wEABAD3//b/AwAKAPz/+P8JABoAEAAOAB0AGAAOAAcADAAaACYALQAoABgAHAAoACgAJwAiABgALQAwAB8AEQAkADgAMQAbAAcAIgAmACEABQAVACcAGwAIAPT/BgAXABwA/P/x/+//9f/x//n////0/+r/3f/b/+D/6v/u/+n/3P/U/8b/yP/T/93/4f/g/9d/vv+7/8j/2//c/9b/z//C/7j/wP/W/9//5v/d/83/u//G/9N/3f/Z/97/3//T/9H/2//1/+X/3P/j//r/9v/k/+r////+/+n/8P8AAAQA9f/1//v/BAD///v/AwD7/+z/9f8DAAAA+v/3//H/6v/t//P/9//y//f/9P/k/9j/4f/2/+7/6f/f/+L/6f/w/+f/3v/m/+P/5f/g/+f/7f/1/+X/4v/s//z/+//x//r/BgAIAPz/AAACAAoAHgAhABEADAAdACEAGwAdACkAMwAjABQAFwAmADQAOgBBADgAMgAwADQAOwBAAEQARABAADkAOwBDAEYAQABDAEIAPQAzADIANQA5ADMAMAAsACQAIAAhABoAFAAXAB0AEgAIAAkADQANAAcACQAOABAA+f/p/+f/+/8EAAAA/f/0//D/4f/m//T/9P/j/9z/3//Z/9b/3//u/w=="
-                                }
-                                
-                                try:
-                                    await websocket.send_json(response)
-                                    print(f"[*] ✅ Response sent successfully for answer event from callId: {json_data['callId']}")
-                                except Exception as send_error:
-                                    print(f"[!] ❌ Failed to send response: {send_error}")
-                            
-                            else:
-                                print(f"[*] 📨 Other JSON message: Event={json_data.get('event', 'unknown')}")
-                                
-                        except json.JSONDecodeError:
-                            print(f"[*] Non-JSON text data received")
-                    
-                    elif "bytes" in message:
-                        binary_data = message["bytes"]
-                        
-                        if current_media_format:
-                            sample_rate = current_media_format.get("sampleRate", 44100)
-                            encoding = current_media_format.get("encoding", "LINEAR16")
-                            channels = current_media_format.get("channels", 1)
-                            
-                            duration_ms = (len(binary_data) * 1000) / (sample_rate * channels * 2)
-                            print(f"[*] Audio info: Rate={sample_rate}Hz, Encoding={encoding}, Duration={duration_ms:.1f}ms")
-                            
-                            call_id = f"call_{connection_start_time.strftime('%H%M%S')}"
-                            
-                            if save_binary_audio_chunk(call_id, binary_data, sample_rate, encoding):
-                                print(f"[*] ✅ Binary audio chunk saved successfully")
-                        else:
-                            print(f"[!] ⚠️ No media format, using defaults...")
-                            call_id = f"call_{connection_start_time.strftime('%H%M%S')}"
-                            if save_binary_audio_chunk(call_id, binary_data, 44100, "LINEAR16"):
-                                print(f"[*] ✅ Binary audio chunk saved with defaults")
-                
-                elif message["type"] == "websocket.disconnect":
-                    print(f"[*] 🔌 Disconnect message received")
-                    break
-                    
-                else:
-                    print(f"[!] ⚠️ Unknown message type: {message['type']}")
-                
-            except asyncio.TimeoutError:
-                print(f"[!] ⏰ No message received from {client_ip} for 60 seconds")
-                try:
-                    await websocket.ping()
-                    print(f"[*] ✅ Ping sent successfully")
-                except Exception as ping_error:
-                    print(f"[!] ❌ Ping failed: {ping_error}")
-                    break
+            raw = await ws.receive_text()
+            msg = json.loads(raw)
+            event = msg.get("event")
+
+            if event == "start":
+                new_call_id = msg.get("callId")
+                stream_id = msg.get("streamId")
+
+                if processor and new_call_id != active_call_id:
+                    logger.info(f"🔄 New call detected ({active_call_id} -> {new_call_id}). Stopping old processor.")
+                    await stop_processor(processor, tasks)
+                    processor, tasks = None, []
+
+                if processor is None:
+                    logger.info(f"📞 Starting processor for call: {new_call_id}")
+                    processor = RealTimeAudioProcessor(AGENT_ID, PUBLIC_KEY)
+                    processor.stream_id = stream_id
+                    processor.call_id = new_call_id
                     
-            except ConnectionClosed:
-                print(f"[!] 🔌 Connection closed by {client_ip}")
-                break
-                
-            except WebSocketDisconnect:
-                print(f"[!] 🔌 WebSocket disconnected by {client_ip}")
-                break
-                
-            except Exception as message_error:
-                print(f"[!] ❌ Error processing message from {client_ip}: {message_error}")
-                print(f"[!] Error type: {type(message_error).__name__}")
-                
-                if "receive" in str(message_error).lower() or "closed" in str(message_error).lower():
-                    print(f"[!] 🔌 Connection appears to be closed. Exiting receive loop.")
-                    break
-                
+                    if "mediaFormat" in msg:
+                        processor.media_format = msg["mediaFormat"]
+                        logger.info(f"👂 Captured media format from SAN: {processor.media_format}")
+                    else:
+                        logger.warning("⚠️ No mediaFormat in 'start' event. Using default.")
+
+                    tasks = await processor.start(ws)
+                    if not tasks:
+                        await ws.close(code=1011, reason="Could not connect to AI backend.")
+                        return
+                    active_call_id = new_call_id
+                continue
+
+            elif event == "media" and processor:
+                payload_b64 = msg.get("payload")
+                if payload_b64:
+                    pcm = base64.b64decode(payload_b64)
+                    async with processor.in_lock: processor.inbound.extend(pcm)
+                continue
+
+            elif event in ("hangup", "stop", "disconnect"):
+                logger.info(f"📞 Call {active_call_id} ended via '{event}' event.")
+                await stop_processor(processor, tasks)
+                processor, tasks, active_call_id = None, [], None
+                continue
+
+            elif event in ("connected", "answer", "ringing"):
+                logger.debug(f"ℹ️  Informational event received: {event}")
                 continue
-                
+            
+            logger.warning(f"⚠️ Received unhandled event: {event}")
+
     except WebSocketDisconnect:
-        print(f"[!] 🔌 WebSocket disconnected from {client_ip}")
-    except ConnectionClosed:
-        print(f"[!] 🔌 Connection closed from {client_ip}")
+        logger.info("🚪 SAN system disconnected the WebSocket.")
     except Exception as e:
-        print(f"[!] ❌ WebSocket error from {client_ip}: {str(e)}")
-        print(f"[!] Error type: {type(e).__name__}")
-        import traceback
-        traceback.print_exc()
+        logger.error(f"❌ Unhandled error in media_socket: {e}", exc_info=True)
     finally:
-        connection_end_time = datetime.now()
-        connection_duration = (connection_end_time - connection_start_time).total_seconds()
-        
-        print(f"\n[*] === CONNECTION SUMMARY FOR {client_ip} ===")
-        print(f"[*] Connection started: {connection_start_time}")
-        print(f"[*] Connection ended: {connection_end_time}")
-        print(f"[*] Total duration: {connection_duration:.2f} seconds")
-        print(f"[*] Total messages received: {message_count}")
-        if message_count > 0:
-            print(f"[*] Average messages per second: {message_count/connection_duration:.2f}")
-            print(f"[*] Last message received: {last_message_time}")
-        print(f"[*] =======================================\n")
-
-@app.websocket("/test")
-async def test_websocket(websocket: WebSocket):
-    await websocket.accept()
-    client_ip = websocket.client.host if websocket.client else "unknown"
-    print(f"[*] Test WebSocket connected from {client_ip}")
-    
-    try:
-        counter = 0
-        while True:
-            counter += 1
-            test_message = {
-                "event": "test",
-                "message": f"Test message #{counter}",
-                "timestamp": datetime.now().isoformat(),
-                "client_ip": client_ip
-            }
-            await websocket.send_text(json.dumps(test_message))
-            print(f"[*] Sent test message #{counter} to {client_ip}")
-            await asyncio.sleep(5)
-    except WebSocketDisconnect:
-        print(f"[*] Test WebSocket disconnected from {client_ip}")
-    except Exception as e:
-        print(f"[!] Test WebSocket error from {client_ip}: {e}")
-
-@app.get("/health")
-async def health_check_get():
-    return JSONResponse(content={"message": "GET OK", "timestamp": datetime.now().isoformat()}, status_code=200)
-
-@app.post("/health")
-async def health_check_post():
-    return JSONResponse(content={"message": "POST OK", "timestamp": datetime.now().isoformat()}, status_code=200)
-
-@app.get("/status")
-async def connection_status():
-    audio_file_count = 0
-    log_file_count = 0
-    raw_file_count = 0
-    
-    if AUDIO_FOLDER and os.path.exists(AUDIO_FOLDER):
-        try:
-            all_files = os.listdir(AUDIO_FOLDER)
-            audio_file_count = len([f for f in all_files if f.endswith('.wav')])
-            log_file_count = len([f for f in all_files if f.endswith('.log') or f.endswith('.hex') or f.endswith('.bin')])
-            raw_file_count = len([f for f in all_files if f.startswith('raw_websocket_')])
-        except:
-            audio_file_count = -1
-            log_file_count = -1
-            raw_file_count = -1
-    
-    return JSONResponse(content={
-        "status": "running",
-        "websocket_endpoint": "/media",
-        "test_endpoint": "/test",
-        "audio_folder": AUDIO_FOLDER,
-        "audio_folder_exists": os.path.exists(AUDIO_FOLDER) if AUDIO_FOLDER else False,
-        "audio_files_count": audio_file_count,
-        "log_files_count": log_file_count,
-        "raw_websocket_files_count": raw_file_count,
-        "exact_raw_logging": True,
-        "timestamp": datetime.now().isoformat(),
-        "server_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
-        "endpoints": {
-            "websocket_media": "/media",
-            "websocket_test": "/test",
-            "health_get": "/health",
-            "health_post": "/health",
-            "status": "/status"
-        }
-    })
+        await stop_processor(processor, tasks)
+        if ws.client_state != WebSocketState.DISCONNECTED:
+            await ws.close()
+        logger.info("✅ Cleanup complete for this WebSocket connection.")
 
 @app.get("/")
-async def root():
-    return JSONResponse(content={
-        "message": "FastAPI WebSocket Server with EXACT Raw Data Display",
-        "status": "✅ Connected and ready to show EXACT websocket data as received",
-        "features": [
-            "Shows EXACT raw WebSocket message as received",
-            "Complete message object display",
-            "Character-by-character breakdown for text",
-            "Byte-by-byte breakdown for binary",
-            "Multiple representations (raw, repr, hex, base64)",
-            "Saves exact raw data to files"
-        ],
-        "endpoints": {
-            "websocket_media": "/media",
-            "websocket_test": "/test",
-            "health_check": "/health",
-            "status": "/status"
-        },
-        "timestamp": datetime.now().isoformat()
-    })
-
-def signal_handler(sig, frame):
-    print("\n[*] 🛑 Shutting down server...")
-    print("[*] Cleaning up resources...")
-    sys.exit(0)
+async def health():
+    return {"status": "ok", "timestamp": datetime.now().isoformat()}
 
 if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description='FastAPI WebSocket Server - EXACT Raw Data Display')
-    parser.add_argument('--port', type=int, default=8000, help='Port to run the server on')
-    parser.add_argument('--host', type=str, default="0.0.0.0", help='Host to bind the server to')
-    parser.add_argument('--debug', action='store_true', help='Enable debug mode')
-    args = parser.parse_args()
-
-    signal.signal(signal.SIGINT, signal_handler)
-
-    print(f"[*] ===== FastAPI WebSocket Server - EXACT Raw Data Display =====")
-    print(f"[*] 🚀 Server running at ws://{args.host}:{args.port}/media")
-    print(f"[*] 📊 Status check at http://{args.host}:{args.port}/status")
-    print(f"[*] 📁 Audio folder: {AUDIO_FOLDER}")
-    print(f"[*] 🔍 EXACT Raw Data Logging: ENABLED")
-    print(f"[*] 📝 Raw WebSocket files: raw_websocket_*.txt")
-    print(f"[*] 📄 Shows complete message objects, character/byte breakdowns")
-    print(f"[*] 🕐 Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
-    print(f"[*] =============================================================")
-    
-    log_level = "debug" if args.debug else "info"
-    uvicorn.run("app:app", host=args.host, port=args.port, log_level=log_level, reload=args.debug)
+    print("🚀 Starting SAN to Millis AI Integration Server (v5 - Provider Format Fix)...")
+    uvicorn.run(app, host="0.0.0.0", port=7860)