ha-sip-notifier/sip-notifier/sip_service.py

#!/usr/bin/env python3
"""SIP Voice Notifier Add-on Service using pyVoIP."""
import json
import logging
import os
import tempfile
import time
from urllib.parse import urlparse

import requests
from flask import Flask, request, jsonify
from pydub import AudioSegment
from pyVoIP.VoIP import VoIPPhone, InvalidStateError, CallState

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
_LOGGER = logging.getLogger(__name__)

app = Flask(__name__)

# Global config
CONFIG = {}
DEFAULT_SAMPLE_RATE = 8000


def download_and_convert_audio(url: str) -> str:
    """Download and convert audio to WAV."""
    parsed_url = urlparse(url)
    extension = os.path.splitext(parsed_url.path)[1] or ".mp3"
    
    with tempfile.NamedTemporaryFile(delete=False, suffix=extension) as tmp:
        download_path = tmp.name
    
    response = requests.get(url, stream=True, timeout=30)
    response.raise_for_status()
    
    with open(download_path, 'wb') as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    
    _LOGGER.debug(f"Downloaded: {download_path}")
    
    try:
        audio = AudioSegment.from_file(download_path)
        audio = audio.set_frame_rate(DEFAULT_SAMPLE_RATE)
        audio = audio.set_channels(1)
        audio = audio.set_sample_width(2)
        
        with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
            wav_path = tmp.name
        
        audio.export(wav_path, format="wav")
        _LOGGER.debug(f"Converted: {wav_path}")
        
        os.remove(download_path)
        return wav_path
        
    except Exception as e:
        if os.path.exists(download_path):
            os.remove(download_path)
        raise Exception(f"Audio conversion failed: {e}")


def generate_tts_audio(message: str) -> str:
    """Generate TTS audio."""
    try:
        from gtts import gTTS
        
        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp:
            mp3_path = tmp.name
        
        tts = gTTS(text=message, lang='en')
        tts.save(mp3_path)
        
        audio = AudioSegment.from_mp3(mp3_path)
        audio = audio.set_frame_rate(DEFAULT_SAMPLE_RATE)
        audio = audio.set_channels(1)
        audio = audio.set_sample_width(2)
        
        with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
            wav_path = tmp.name
        
        audio.export(wav_path, format="wav")
        os.remove(mp3_path)
        
        return wav_path
        
    except ImportError:
        raise Exception("gTTS not available")


def place_sip_call(destination: str, audio_file: str, duration: int):
    """Place a SIP call using pyVoIP."""
    sip_user = CONFIG['sip_user']
    sip_server = CONFIG['sip_server']
    sip_password = CONFIG['sip_password']
    
    _LOGGER.info(f"Connecting to SIP server: {sip_server}")
    
    try:
        # Create VoIP phone
        phone = VoIPPhone(
            sip_server,
            5060,
            sip_user,
            sip_password,
            callCallback=None,
            myIP=None,
            sipPort=5060,
            rtpPortLow=10000,
            rtpPortHigh=20000
        )
        
        phone.start()
        time.sleep(2)  # Wait for registration
        
        _LOGGER.info(f"Calling: {destination}")
        
        # Make call
        call = phone.call(destination)
        
        # Wait for call to be answered
        timeout = 10
        elapsed = 0
        while call.state != CallState.ANSWERED and elapsed < timeout:
            time.sleep(0.5)
            elapsed += 0.5
        
        if call.state != CallState.ANSWERED:
            _LOGGER.warning("Call not answered within timeout")
            call.hangup()
            phone.stop()
            return
        
        _LOGGER.info("Call answered! Playing audio...")
        
        # Transmit audio file
        call.write_audio(audio_file)
        
        # Keep call active
        _LOGGER.info(f"Call active for {duration}s")
        time.sleep(duration)
        
        # Hangup
        _LOGGER.info("Hanging up...")
        call.hangup()
        
        # Cleanup
        phone.stop()
        
    except Exception as e:
        _LOGGER.error(f"Call error: {e}", exc_info=True)
        raise


@app.route('/health', methods=['GET'])
def health():
    """Health check endpoint."""
    return jsonify({'status': 'ok'}), 200


@app.route('/send_notification', methods=['POST'])
def handle_send_notification():
    """Handle send_notification service call."""
    try:
        data = request.json
        destination = data.get('destination')
        audio_url = data.get('audio_url')
        message = data.get('message')
        duration = data.get('duration', CONFIG.get('default_duration', 30))
        
        if not destination:
            return jsonify({'error': 'destination required'}), 400
        
        if not audio_url and not message:
            return jsonify({'error': 'audio_url or message required'}), 400
        
        temp_files = []
        
        try:
            # Prepare audio
            if audio_url:
                _LOGGER.info(f"Downloading: {audio_url}")
                audio_file = download_and_convert_audio(audio_url)
                temp_files.append(audio_file)
            elif message:
                _LOGGER.info(f"Generating TTS: {message}")
                audio_file = generate_tts_audio(message)
                temp_files.append(audio_file)
            
            # Place call
            _LOGGER.info(f"Calling: {destination}")
            place_sip_call(destination, audio_file, duration)
            
            return jsonify({'status': 'success'}), 200
            
        finally:
            # Cleanup
            for f in temp_files:
                try:
                    if os.path.exists(f):
                        os.remove(f)
                except Exception as e:
                    _LOGGER.warning(f"Cleanup failed: {e}")
                    
    except Exception as e:
        _LOGGER.error(f"Request error: {e}", exc_info=True)
        return jsonify({'error': str(e)}), 500


if __name__ == '__main__':
    # Load config from Home Assistant add-on options
    options_file = '/data/options.json'
    if os.path.exists(options_file):
        with open(options_file, 'r') as f:
            CONFIG = json.load(f)
            _LOGGER.info("Config loaded from options.json")
    else:
        _LOGGER.warning("No options.json found, using environment variables")
        CONFIG = {
            'sip_server': os.getenv('SIP_SERVER', ''),
            'sip_user': os.getenv('SIP_USER', ''),
            'sip_password': os.getenv('SIP_PASSWORD', ''),
            'default_duration': int(os.getenv('DEFAULT_DURATION', 30))
        }
    
    _LOGGER.info("SIP Voice Notifier ready - service will be auto-registered by Supervisor")
    _LOGGER.info("Starting service on port 8099")
    app.run(host='0.0.0.0', port=8099, debug=False)