#!/usr/bin/python3 -O

"""
Self-hosted AMD EAGI client: local five-second no-audio policy.
Python 3.4+ syntax. Release: 2026-09-21.

Zero FD3 bytes after five seconds of the monitoring loop -> MACHINE locally.
The clock begins after WebSocket setup and dialplan pre-playback, not at answer.
Silent PCM is audio. EOF/hangup and network failures keep separate policies.
No new thread or unsolicited receive is needed for the zero-audio decision.
"""

import os
import fcntl
import time
import json
import select
from websocket import create_connection, ABNF, WebSocketTimeoutException
from asterisk.agi import AGI

# =============================================================================
# CONFIGURATION CONSTANTS
# =============================================================================

# Audio Processing
AUDIO_FD = 3                    # Asterisk audio file descriptor (EAGI)
AUDIO_READ_SIZE = 9500         # Audio chunk read size (bytes)

# WebSocket Configuration
# Self-hosted AMD service (usad2/gru champion). Point each dialer here; override via env.
WS_ENDPOINT = os.environ.get("AMD_WS_URL", "ws://144.76.101.53:2700")
# AUTH = source-IP allowlist (NO api key). Add THIS dialer's public IP to the allowlist
# on the service dashboard (Access panel) before it will be allowed to connect.
SAMPLE_RATE = 8000             # Audio sample rate for AMD service

# Time-Based Chunking Configuration (Core Feature)
SEND_TIMES = [0.5, 1.0, 1.5, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
NO_AUDIO_TIMEOUT = 5.0         # Zero bytes READ from FD3, not acoustic silence
MAX_WAIT_TIME = 10             # Loop deadline; ordinary socket waits can exceed it
SELECT_TIMEOUT = 0.05          # Select() timeout for timing precision (seconds)
FALLBACK_CHUNK_SIZE = 8000     # Fallback size-based threshold after time-based sends
FALLBACK_SEND_INTERVAL = 1.0   # Flush even a small buffer after the timed schedule
FINAL_REPLY_TIMEOUT = 3.0     # Bounded EOF finalization, only after prior audio

# AMD Behavior
MACHINE_DELAY =  0           # Delay after machine detection (original behavior)

# Error Handling
NO_DATA_SLEEP = 0.1            # Fallback sleep when select() fails
CONNECTION_TIMEOUT = 10        # WebSocket connection timeout


# =============================================================================
# UTILITY FUNCTIONS
# =============================================================================

def log_verbose(agi, message, vid=None):
    """
    Consistent logging with AMD prefix and VID for easy tracking
    
    Args:
        agi: AGI instance
        message: Log message string
        vid: Optional VID for call tracking
    """
    if vid:
        agi.verbose("AMD[{}]: {}".format(vid, message))
    else:
        agi.verbose("AMD: {}".format(message))


def set_amd_variables(agi, status, cause, stats=None, vid=None):
    """
    Set AMD result variables in Asterisk channel
    
    Args:
        agi: AGI instance
        status: AMDSTATUS value (HUMAN, MACHINE, HANGUP, NOTSURE)
        cause: AMDCAUSE value (descriptive reason)
        stats: Optional AMDSTATS value (additional data)
        vid: Optional VID for call tracking
    """
    agi.set_variable('AMDSTATUS', status)
    agi.set_variable('AMDCAUSE', cause)
    if stats:
        agi.set_variable('AMDSTATS', stats)
    
    log_verbose(agi, "Variables set - Status: {}, Cause: {}".format(status, cause), vid)


# =============================================================================
# WEBSOCKET MANAGEMENT
# =============================================================================

def create_websocket_connection(agi, caller_name, vid=None):
    """
    Create and configure WebSocket connection to AMD service
    
    Args:
        agi: AGI instance
        caller_name: Caller ID name from AGI environment
        vid: Optional VID for call tracking
        
    Returns:
        WebSocket connection object or None if failed
    """
    try:
        # Establish connection with timeout
        ws = create_connection(WS_ENDPOINT, timeout=CONNECTION_TIMEOUT)
        
        # Send initial configuration using proper JSON formatting
        config = json.dumps({
            "config": {
                "sample_rate": SAMPLE_RATE,
                "VID": caller_name or "Unknown",
            }
        })
        
        ws.send(config)
        log_verbose(agi, "WebSocket connected to {}".format(WS_ENDPOINT), vid)
        log_verbose(agi, "Config sent: sample_rate={}, VID={}".format(SAMPLE_RATE, caller_name), vid)
        
        return ws
        
    except Exception as err:
        log_verbose(agi, "WebSocket connection failed: {}".format(err), vid)
        log_verbose(agi, "AMD service unavailable - defaulting to HUMAN for safety", vid)
        # When AMD service is down, route to human agents
        set_amd_variables(agi, "HUMAN", "CONNECTION_ERROR", vid=vid)
        return None


def cleanup_websocket(ws):
    """
    Properly close WebSocket connection with end-of-file marker
    
    Args:
        ws: WebSocket connection to close
    """
    try:
        if ws:
            ws.settimeout(0.1)
            # Send EOF marker to AMD service
            ws.send(json.dumps({"eof": 1}))
            ws.close(timeout=0.1)
    except Exception:
        # Ignore cleanup errors - connection might already be closed
        pass


# =============================================================================
# AUDIO PROCESSING
# =============================================================================

def setup_audio_stream():
    """
    Configure audio file descriptor for non-blocking I/O operations
    Required for select() optimization to work properly
    """
    flags = fcntl.fcntl(AUDIO_FD, fcntl.F_GETFL)
    fcntl.fcntl(AUDIO_FD, fcntl.F_SETFL, flags | os.O_NONBLOCK)


def finalize_stalled_audio(agi, ws, wait_seconds, vid=None):
    """Ask for a verdict after prior audio stalls, without waiting indefinitely.

    Unlike zero-audio fallback, this actually requests/reads a server result.
    Empty interim replies and ping/pong are not terminal. Restore the socket's
    timeout even on failure. Do not invent a MACHINE on network failure.
    """
    previous_timeout = ws.gettimeout()
    deadline = time.monotonic() + max(0.0, wait_seconds)
    try:
        if wait_seconds <= 0:
            raise WebSocketTimeoutException("No finalization budget remains")
        ws.settimeout(wait_seconds)
        ws.send(json.dumps({"eof": 1}))
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise WebSocketTimeoutException("EOF verdict deadline reached")
            ws.settimeout(remaining)
            opcode, response = ws.recv_data(control_frame=True)
            if opcode == ABNF.OPCODE_CLOSE:
                raise IOError("AMD connection closed before EOF verdict")
            if opcode in (ABNF.OPCODE_PING, ABNF.OPCODE_PONG):
                continue
            if opcode != ABNF.OPCODE_TEXT:
                raise ValueError("Unexpected binary AMD verdict")
            if isinstance(response, bytes):
                response = response.decode('utf-8')
            if not response:
                continue
            if 'HUMAN' in response:
                set_amd_variables(agi, "HUMAN", "HUMAN", response, vid)
                return
            if 'AMD' in response or 'MACHINE' in response:
                set_amd_variables(agi, "MACHINE", response, vid=vid)
                time.sleep(MACHINE_DELAY)
                return
            set_amd_variables(agi, "NOTSURE", "EOF_INCONCLUSIVE", vid=vid)
            return
    except WebSocketTimeoutException:
        set_amd_variables(agi, "NOTSURE", "EOF_TIMEOUT", vid=vid)
    except Exception as err:
        log_verbose(agi, "EOF finalization error: {}".format(err), vid)
        set_amd_variables(agi, "NOTSURE", "EOF_ERROR", vid=vid)
    finally:
        ws.settimeout(previous_timeout)


def process_audio_chunk(agi, ws, audio_buffer, vid=None):
    """
    Send audio chunk to AMD service and process response
    
    Args:
        agi: AGI instance
        ws: WebSocket connection
        audio_buffer: Audio data to send
        vid: Optional VID for call tracking
        
    Returns:
        True if detection is complete (HUMAN/MACHINE detected)
        False if should continue processing
    """
    try:
        # Send binary audio data to AMD service
        ws.send_binary(audio_buffer)
        log_verbose(agi, "Sent audio chunk: {} bytes".format(len(audio_buffer)), vid)
        
        # Receive and process AMD response
        response = ws.recv()
        log_verbose(agi, "AMD response: {}".format(response), vid)
        
        # Parse detection results
        if 'HUMAN' in response:
            log_verbose(agi, "*** HUMAN DETECTED ***", vid)
            set_amd_variables(agi, "HUMAN", "HUMAN", response, vid)
            return True
            
        elif 'AMD' in response or 'MACHINE' in response:
            log_verbose(agi, "*** MACHINE DETECTED ***", vid)
            set_amd_variables(agi, "MACHINE", response, vid=vid)
            # Original behavior: wait for machine to finish speaking
            log_verbose(agi, "Waiting {}s for machine to complete".format(MACHINE_DELAY), vid)
            time.sleep(MACHINE_DELAY)
            return True
            
        # Continue processing for inconclusive responses
        return False
        
    except Exception as err:
        log_verbose(agi, "Audio processing error: {}".format(err), vid)
        log_verbose(agi, "Network/processing error - defaulting to HUMAN", vid)
        set_amd_variables(agi, "HUMAN", "PROCESSING_ERROR", vid=vid)
        return True


def process_audio_stream(agi, ws, channel, vid=None):
    """
    Main audio processing loop with select() optimization and time-based chunking
    
    Args:
        agi: AGI instance
        ws: WebSocket connection
        channel: Asterisk channel name for logging
        vid: Optional VID for call tracking
    """
    start_time = time.monotonic()
    total_data_received = 0    # Bytes actually read from FD3, including unsent data
    total_data_sent = 0
    audio_buffer = b''
    send_index = 0
    no_audio_streak = 0
    last_send_time = 0.0
    
    log_verbose(agi, "Starting select()-optimized audio processing", vid)
    log_verbose(agi, "Channel: {}".format(channel), vid)
    log_verbose(agi, "Send schedule: {} seconds".format(SEND_TIMES), vid)
    log_verbose(agi, "Process optimizations: select() I/O, time-based chunking", vid)
    
    while True:
        current_time = time.monotonic()
        elapsed_time = current_time - start_time
        
        try:
            # Calculate smart timeout for select()
            if send_index < len(SEND_TIMES):
                time_until_next_send = SEND_TIMES[send_index] - elapsed_time
                select_timeout = max(0, min(time_until_next_send, SELECT_TIMEOUT))
            else:
                select_timeout = SELECT_TIMEOUT
            
            # Use select() for efficient I/O
            ready, _, _ = select.select([AUDIO_FD], [], [], select_timeout)
            
            if ready:
                try:
                    audio_chunk = os.read(AUDIO_FD, AUDIO_READ_SIZE)
                    if audio_chunk:
                        audio_buffer += audio_chunk
                        total_data_received += len(audio_chunk)
                        no_audio_streak = 0
                    else:
                        log_verbose(agi, "End of audio stream detected", vid)
                        log_verbose(agi, "Total data processed: {} bytes".format(total_data_received), vid)
                        log_verbose(agi, "Channel [{}] HANGUP".format(channel), vid)
                        set_amd_variables(agi, "NOAUDIO", "NOAUDIO", vid=vid)
                        return
                except OSError as err:
                    if err.errno == 11:
                        log_verbose(agi, "Unexpected EAGAIN after select() - continuing", vid)
                        pass
                    else:
                        raise
            
            # Read ready audio FIRST: an unsent byte still counts as audio.
            # Decide locally; never block waiting for the server on this path.
            elapsed_time = time.monotonic() - start_time
            if total_data_received == 0 and elapsed_time >= NO_AUDIO_TIMEOUT:
                set_amd_variables(agi, "MACHINE", "NO_AUDIO_TIMEOUT", vid=vid)
                return

            # Existing ten-second fallback for calls that did deliver audio.
            if elapsed_time > MAX_WAIT_TIME:
                log_verbose(agi, "Global timeout reached after {}s".format(MAX_WAIT_TIME), vid)
                log_verbose(agi, "Buffer: {} bytes, Total: {} bytes".format(len(audio_buffer), total_data_received), vid)
                log_verbose(agi, "Channel [{}] TIMEOUT".format(channel), vid)
                
                set_amd_variables(agi, "HANGUP", "AUDIO_TIMEOUT", vid=vid)
                return
            
            # Time-based sending logic
            if send_index < len(SEND_TIMES) and elapsed_time >= SEND_TIMES[send_index]:
                send_time = SEND_TIMES[send_index]
                
                if len(audio_buffer) > 0:
                    total_data_sent += len(audio_buffer)
                    log_verbose(agi, "Time-based send #{}: threshold={}s, buffer={} bytes, total={} bytes, elapsed={:.3f}s".format(
                        send_index + 1, send_time, len(audio_buffer), total_data_received, elapsed_time), vid)
                    
                    if process_audio_chunk(agi, ws, audio_buffer, vid):
                        return
                    audio_buffer = b''
                    last_send_time = elapsed_time
                else:
                    no_audio_streak += 1
                    log_verbose(agi, "Time-based send #{}: threshold={}s, NO AUDIO DATA, elapsed={:.3f}s".format(
                        send_index + 1, send_time, elapsed_time), vid)
                    if no_audio_streak >= 2 and total_data_sent > 0:
                        wait_seconds = min(FINAL_REPLY_TIMEOUT,
                                           max(0.0, MAX_WAIT_TIME - elapsed_time))
                        finalize_stalled_audio(agi, ws, wait_seconds, vid)
                        return
                send_index += 1
            
            # Fallback after scheduled sends
            elif (send_index >= len(SEND_TIMES) and audio_buffer and
                  (len(audio_buffer) >= FALLBACK_CHUNK_SIZE or
                   elapsed_time - last_send_time >= FALLBACK_SEND_INTERVAL)):
                total_data_sent += len(audio_buffer)
                log_verbose(agi, "Fallback size-based send: buffer={} bytes, total={} bytes, elapsed={:.3f}s".format(
                    len(audio_buffer), total_data_received, elapsed_time), vid)
                
                if process_audio_chunk(agi, ws, audio_buffer, vid):
                    return
                audio_buffer = b''
                last_send_time = elapsed_time
                
        except Exception as err:
            log_verbose(agi, "Unexpected error in audio processing: {}".format(err), vid)
            log_verbose(agi, "Critical error - defaulting to HUMAN for safety", vid)
            set_amd_variables(agi, "HUMAN", "PROCESSING_ERROR", vid=vid)
            return


# =============================================================================
# MAIN ENTRY POINT
# =============================================================================

def start_agi():
    """
    Main AGI entry point - coordinates all AMD processing
    """
    devnull = open('/dev/null', 'w')
    agi = AGI(stderr=devnull)
    ws = None
    
    try:
        # Extract call information from AGI environment
        caller_id = agi.env.get('agi_callerid', 'Unknown')
        caller_name = agi.env.get('agi_calleridname', 'Unknown')
        extension = agi.env.get('agi_extension', 'Unknown')
        channel = agi.env.get('agi_channel', 'Unknown')
        
        # Use caller_name as VID for tracking
        vid = caller_name
        
        # Initialization logging with VID
        log_verbose(agi, "=" * 50, vid)
        log_verbose(agi, "AMD DETECTION STARTED", vid)
        log_verbose(agi, "Caller ID: {} ({})".format(caller_id, caller_name), vid)
        log_verbose(agi, "Extension: {}".format(extension), vid)
        log_verbose(agi, "Channel: {}".format(channel), vid)
        log_verbose(agi, "Optimizations: select() I/O, time-based chunking", vid)
        log_verbose(agi, "=" * 50, vid)
        
        # Create WebSocket connection
        ws = create_websocket_connection(agi, caller_name, vid)
        if not ws:
            # Connection failed, variables already set by create_websocket_connection()
            return
        
        # Setup audio processing
        setup_audio_stream()
        
        # Main processing with VID tracking
        process_audio_stream(agi, ws, channel, vid)
        
        # Completion logging
        log_verbose(agi, "=" * 50, vid)
        log_verbose(agi, "AMD DETECTION COMPLETE", vid)
        log_verbose(agi, "=" * 50, vid)
        
    except Exception as err:
        # Fatal error handling
        log_verbose(agi, "FATAL ERROR in AMD processing: {}".format(err), vid if 'vid' in locals() else None)
        log_verbose(agi, "Exception occurred - defaulting to HUMAN for call safety", vid if 'vid' in locals() else None)
        set_amd_variables(agi, "HUMAN", "FATAL_ERROR", vid=vid if 'vid' in locals() else None)
        
    finally:
        # Always ensure WebSocket is properly closed
        cleanup_websocket(ws)


# =============================================================================
# SCRIPT EXECUTION
# =============================================================================

if __name__ == "__main__":
    start_agi()