#!/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 SUPERVISOR_TOKEN = os.environ.get('SUPERVISOR_TOKEN') def register_service(): """Register service with Home Assistant via Supervisor.""" if not SUPERVISOR_TOKEN: _LOGGER.warning("No SUPERVISOR_TOKEN, skipping service registration") return try: headers = { 'Authorization': f'Bearer {SUPERVISOR_TOKEN}', 'Content-Type': 'application/json', } # Call Home Assistant service registration endpoint response = requests.post( 'http://supervisor/services', headers=headers, json={ 'domain': 'sip_notifier', 'service': 'send_notification', 'addon': '088d3b92_sip-notifier' }, timeout=10 ) if response.status_code in [200, 201]: _LOGGER.info("✅ Service registered: sip_notifier.send_notification") else: _LOGGER.warning(f"Service registration returned: {response.status_code}") except Exception as e: _LOGGER.error(f"Failed to register service: {e}") 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)) } # Register service with Home Assistant _LOGGER.info("Registering service with Home Assistant...") time.sleep(2) # Wait for supervisor to be ready register_service() _LOGGER.info("SIP Voice Notifier ready") _LOGGER.info("Service: sip_notifier.send_notification") _LOGGER.info("Starting Flask service on port 8099") app.run(host='0.0.0.0', port=8099, debug=False)