Vonage Voice and SMS API

VonageAPIWrapper.py

#!/usr/bin/env python3
import time
import os
import sys
import signal
from pprint import pprint
from box import Box
import datetime
import uuid
import jwt
import requests
import json
from zoneinfo import ZoneInfo

class VonageWrapper:
	def __init__( self , options={} ):
		self.options = Box( options )
		pprint( self.options )
		signal.signal( signal.SIGINT , self.hangup_call ) # Ctrl+C
		signal.signal( signal.SIGTSTP , self.hangup_call ) # Ctrl+Z
		self.ANSWERED = False
		self.COMPLETED = False
		self.CALL = None
		self.STATUS = None
		self.CALL_COMPLETED_TIME = None
		self.CALL_ANSWERED_TIME = None
		self.CALL_IN_PROGRESS_DURATION = None
		self.CALL_TOTAL_DURATION = None
		self.TimeZone = ZoneInfo( "America/New_York" )
		self.generate_jwt()
		self.HEADERS = {
			"User-Agent": "vonage-python/3.16.0 python/3.12.3",
			"Accept": "application/json",
			"Authorization": f"Bearer {self.JWT}"
		}

	def read_private_key( self ):
		with open( self.options.private_key , 'rb' ) as key_file:
			return key_file.read()

	def generate_jwt( self ):
		private_key = self.read_private_key()
		now = int( time.time() )
		payload = {
			"application_id": self.options.application_id ,
			"iat": now ,
			"jti": str( uuid.uuid4() ) ,
			"exp": now + ( 15 * 60 )
		}
		headers = { 'alg': 'RS256', 'typ': 'JWT' }
		self.JWT = jwt.encode( payload , private_key , algorithm='RS256' , headers=headers )
		return

	def convert_time_to_string( self , epoch_time ):
		dt = datetime.datetime.fromtimestamp( epoch_time , self.TimeZone )
		milliseconds = round( dt.microsecond / 1000 )
		milliseconds = str( milliseconds ).zfill( 3 )
		dt_string = dt.strftime( "%d%b%Y === %H:%M:%S" ).upper()
		return f"{dt_string}.{milliseconds}"

	def hangup_call( self , signum , frame ):
		print( "Ending call early..." )
		try:
			if self.CALL is not None:
				if "uuid" in self.CALL:
					url = f"https://api.nexmo.com/v1/calls/{self.CALL['uuid']}"
					payload = {
						"action": "hangup"
					}
					response = requests.put( url , headers=self.HEADERS , json=payload )
					response.raise_for_status()
		except Exception as e:
			print( f"Error hanging up call: {e}" )
		self.ANSWERED = False
		self.COMPLETED = False
		self.CALL = None
		self.STATUS = None
		self.CALL_COMPLETED_TIME = None
		self.CALL_ANSWERED_TIME = None
		self.CALL_START_TIME = None
		self.CALL_IN_PROGRESS_DURATION = None
		self.CALL_TOTAL_DURATION = None
		sys.exit( 1 )

	def send_sms( self , _to , _message ):
		url = "https://api.nexmo.com/v1/messages"
		payload = {
			"channel": "sms",
			"message_type": "text",
			"to": _to ,
			"from": self.options.sms_number ,
			"text": _message ,
		}
		response = requests.post( url , headers=self.HEADERS , json=payload )
		response.raise_for_status()
		return response.json()

	def call_get_info( self ):
		url = f"https://api.nexmo.com/v1/calls/{self.CALL['uuid']}"
		response = requests.get( url , headers=self.HEADERS )
		response.raise_for_status()
		return response.json()

	def call_wait_on_finished( self ):
		self.CALL_START_TIME = time.time()
		for i in range( 500 ):
			time.sleep( 1 )
			c = self.call_get_info()
			self.STATUS = c[ "status" ]
			if self.STATUS == "started":
				print( "Starting" )
			if self.STATUS == "ringing":
				print( "Ringing ..." )
			if self.STATUS == "answered":
				print( "In Progress ..." )
				self.ANSWERED = True
				self.CALL_ANSWERED_TIME = time.time()
			if self.STATUS == "completed":
				self.FINISHED = True
				self.CALL_COMPLETED_TIME = time.time()
				self.CALL_TOTAL_DURATION = ( self.CALL_COMPLETED_TIME - self.CALL_START_TIME )
				self.CALL_IN_PROGRESS_DURATION = ( self.CALL_COMPLETED_TIME - self.CALL_ANSWERED_TIME )
				print( "Ended" )
				pprint( c )
				print( f"Call Start Time:" , self.convert_time_to_string( self.CALL_START_TIME ) )
				print( f"Call Answered Time:" , self.convert_time_to_string( self.CALL_ANSWERED_TIME ) )
				print( f"Call Completed Time:" , self.convert_time_to_string( self.CALL_COMPLETED_TIME ) )
				print( f"Call In Progress Duration: {self.CALL_IN_PROGRESS_DURATION} seconds" )
				print( f"Call Total Duration: {self.CALL_TOTAL_DURATION} seconds" )
				return c
		return False

	def call_say_message( self , _to , _message , loop=1 ):
		url = "https://api.nexmo.com/v1/calls"
		payload = {
			"to": [ { "type": "phone" , "number": _to } ] ,
			"from": { "type": "phone" , "number": self.options.voice_number } ,
			"ncco": [ { "action": "talk" , "text": _message , "loop": loop } ]
		}
		response = requests.post( url , headers=self.HEADERS , json=payload )
		response.raise_for_status()
		self.CALL = response.json()
		pprint( self.CALL )
		self.call_wait_on_finished()
		return

	def call_play_mp3( self , _to , answer_url ):
		url = "https://api.nexmo.com/v1/calls"
		payload = {
			"to": [ { "type": "phone", "number": _to } ] ,
			"from": { "type": "phone", "number": self.options.voice_number } ,
			"answer_url": [ answer_url ]
		}
		response = requests.post( url , headers=self.HEADERS , json=payload )
		response.raise_for_status()
		self.CALL = response.json()
		pprint( self.CALL )
		self.call_wait_on_finished()
		return

if __name__ == "__main__":
	v = VonageWrapper({
		"sms_number": "1937asdf" ,
		"voice_number": "1937asdf" ,
		"application_id": "asdf" ,
		"private_key": "/Users/morpheous/Downloads/private(1).key" ,
	})
	_to = "1asdf"
	# v.call_say_message( _to , "testing 123" )
	v.call_play_mp3( _to , "https://files.example.org/ub.json" )