Samples for call automation APIs

OutboundCallReminder/CallConfiguration.py

@@ -0,0 +1,15 @@
+from EventHandler.EventAuthHandler import EventAuthHandler
+
+
+class CallConfiguration:
+
+    def __init__(self, connection_string, source_identity, source_phone_number, app_base_url, audio_file_name):
+        self.connection_string: str = str(connection_string)
+        self.source_identity: str = str(source_identity)
+        self.source_phone_number: str = str(source_phone_number)
+        self.app_base_url: str = str(app_base_url)
+        self.audio_file_name: str = str(audio_file_name)
+        eventhandler = EventAuthHandler()
+        self.app_callback_url: str = app_base_url + \
+            "/api/outboundcall/callback?" + eventhandler.get_secret_querystring()
+        self.audio_file_url: str = app_base_url + "/audio/" + audio_file_name

OutboundCallReminder/CommunicationIdentifierKind.py

@@ -0,0 +1,8 @@
+import enum
+
+
+class CommunicationIdentifierKind(enum.Enum):
+
+    USER_IDENTITY = 1
+    PHONE_IDENTITY = 2
+    UNKNOWN_IDENTITY = 3

OutboundCallReminder/ConfigurationManager.py

@@ -0,0 +1,23 @@
+import configparser
+
+
+class ConfigurationManager:
+    __configuration = None
+    __instance = None
+
+    def __init__(self):
+        if(self.__configuration == None):
+            self.__configuration = configparser.ConfigParser()
+            self.__configuration.read('config.ini')
+
+    @staticmethod
+    def get_instance():
+        if(ConfigurationManager.__instance == None):
+            ConfigurationManager.__instance = ConfigurationManager()
+
+        return ConfigurationManager.__instance
+
+    def get_app_settings(self, key):
+        if (key != None):
+            return self.__configuration.get('default', key)
+        return None

OutboundCallReminder/Controller/OutboundCallController.py

@@ -0,0 +1,30 @@
+from aiohttp import web
+from EventHandler.EventAuthHandler import EventAuthHandler
+from EventHandler.EventDispatcher import EventDispatcher
+
+
+class OutboundCallController():
+
+    app = web.Application()
+
+    def __init__(self):
+        self.app.add_routes(
+            [web.post('/api/outboundcall/callback', self.on_incoming_request_async)])
+        self.app.add_routes([web.get("/audio/{file_name}", self.load_file)])
+        web.run_app(self.app, port=9007)
+
+    async def on_incoming_request_async(self, request):
+        param = request.rel_url.query
+        content = await request.content.read()
+
+        eventhandler = EventAuthHandler()
+        if (param.get('secret') and eventhandler.authorize(param['secret'])):
+            eventDispatcher: EventDispatcher = EventDispatcher.get_instance()
+            eventDispatcher.process_notification(str(content.decode('UTF-8')))
+
+        return "OK"
+
+    async def load_file(self, request):
+        file_name = request.match_info.get('file_name', "Anonymous")
+        resp = web.FileResponse(f'audio/{file_name}')
+        return resp

OutboundCallReminder/EventHandler/EventAuthHandler.py

@@ -0,0 +1,22 @@
+from ConfigurationManager import ConfigurationManager
+from Logger import Logger
+
+
+class EventAuthHandler:
+
+    secret_value = None
+
+    def __init__(self):
+        configuration = ConfigurationManager.get_instance()
+        self.secret_value = configuration.get_app_settings("SecretPlaceholder")
+
+        if (self.secret_value == None or self.secret_value == ''):
+            Logger.log_message(Logger.INFORMATION, "SecretPlaceholder is null")
+            self.secret_value = "h3llowW0rld"
+
+    def authorize(self, requestSecretValue):
+        return ((requestSecretValue != None) and (requestSecretValue == self.secret_value))
+
+    def get_secret_querystring(self):
+        secretKey = "secret"
+        return (secretKey + "=" + self.secret_value)

OutboundCallReminder/EventHandler/EventDispatcher.py

@@ -0,0 +1,101 @@
+from Logger import Logger
+from threading import Lock
+import threading
+import json
+from azure.core.messaging import CloudEvent
+from azure.communication.callingserver import CallingServerEventType, \
+    CallConnectionStateChangedEvent, ToneReceivedEvent, \
+    PlayAudioResultEvent, AddParticipantResultEvent
+
+
+class EventDispatcher:
+    __instance = None
+    notification_callbacks: dict = None
+    subscription_lock = None
+
+    def __init__(self):
+        self.notification_callbacks = dict()
+        self.subscription_lock = Lock()
+
+    @staticmethod
+    def get_instance():
+        if EventDispatcher.__instance is None:
+            EventDispatcher.__instance = EventDispatcher()
+
+        return EventDispatcher.__instance
+
+    def subscribe(self, event_type: str, event_key: str, notification_callback):
+        self.subscription_lock.acquire
+        event_id: str = self.build_event_key(event_type, event_key)
+        self.notification_callbacks[event_id] = notification_callback
+        self.subscription_lock.release
+
+    def unsubscribe(self, event_type: str, event_key: str):
+        self.subscription_lock.acquire
+        event_id: str = self.build_event_key(event_type, event_key)
+        del self.notification_callbacks[event_id]
+        self.subscription_lock.release
+
+    def build_event_key(self, event_type: str, event_key: str):
+        return event_type + "-" + event_key
+
+    def process_notification(self, request: str):
+        call_event = self.extract_event(request)
+        if call_event is not None:
+            self.subscription_lock.acquire
+            notification_callback = self.notification_callbacks.get(
+                self.get_event_key(call_event))
+            if (notification_callback != None):
+                threading.Thread(target=notification_callback,
+                                 args=(call_event,)).start()
+
+    def get_event_key(self, call_event_base):
+        if type(call_event_base) == CallConnectionStateChangedEvent:
+            call_leg_id = call_event_base.call_connection_id
+            key = self.build_event_key(
+                CallingServerEventType.CALL_CONNECTION_STATE_CHANGED_EVENT, call_leg_id)
+            return key
+        elif type(call_event_base) == ToneReceivedEvent:
+            call_leg_id = call_event_base.call_connection_id
+            key = self.build_event_key(
+                CallingServerEventType.TONE_RECEIVED_EVENT, call_leg_id)
+            return key
+        elif type(call_event_base) == PlayAudioResultEvent:
+            operation_context = call_event_base.operation_context
+            key = self.build_event_key(
+                CallingServerEventType.PLAY_AUDIO_RESULT_EVENT, operation_context)
+            return key
+        elif type(call_event_base) == AddParticipantResultEvent:
+            operation_context = call_event_base.operation_context
+            key = self.build_event_key(
+                CallingServerEventType.ADD_PARTICIPANT_RESULT_EVENT, operation_context)
+            return key
+        return None
+
+    def extract_event(self, content: str):
+        try:
+            event = CloudEvent.from_dict(json.loads(content)[0])
+            if event.type == CallingServerEventType.CALL_CONNECTION_STATE_CHANGED_EVENT:
+                call_connection_state_changed_event = CallConnectionStateChangedEvent.deserialize(
+                    event.data)
+                return call_connection_state_changed_event
+
+            if event.type == CallingServerEventType.PLAY_AUDIO_RESULT_EVENT:
+                play_audio_result_event = PlayAudioResultEvent.deserialize(
+                    event.data)
+                return play_audio_result_event
+
+            if event.type == CallingServerEventType.ADD_PARTICIPANT_RESULT_EVENT:
+                add_participant_result_event = AddParticipantResultEvent.deserialize(
+                    event.data)
+                return add_participant_result_event
+
+            if event.type == CallingServerEventType.TONE_RECEIVED_EVENT:
+                tone_received_event = ToneReceivedEvent.deserialize(event.data)
+                return tone_received_event
+
+        except Exception as ex:
+            Logger.log_message(
+                Logger.ERROR, "Failed to parse request content Exception: " + str(ex))
+
+        return None

OutboundCallReminder/Logger.py

@@ -0,0 +1,12 @@
+import enum
+
+
+class Logger(enum.Enum):
+
+    INFORMATION = 1
+    ERROR = 2
+
+    @staticmethod
+    def log_message(message_type, message):
+        log_message = message_type.name + " : " + message
+        print(log_message)

OutboundCallReminder/Ngrok/NgrokConnector.py

@@ -0,0 +1,13 @@
+import requests
+import json
+
+
+class NgrokConnector:
+
+    def __init__(self):
+        self.ngrok_tunnel_url = "http://127.0.0.1:4040/api/tunnels"
+
+    def get_all_tunnels(self):
+        tunnel_url = requests.get(self.ngrok_tunnel_url).text
+        tunnel_list = json.loads(tunnel_url)
+        return tunnel_list

OutboundCallReminder/Ngrok/NgrokService.py

@@ -0,0 +1,82 @@
+from time import sleep
+from Ngrok.NgrokConnector import NgrokConnector
+import os
+import subprocess
+import psutil
+import signal
+from pathlib import Path
+
+
+class NgrokService:
+    # The NGROK process
+    __ngrokProcess = None  # Process
+    # NgrokConnector connector;
+    __connector = None  # NgrokConnector
+
+    def __init__(self, ngrokPath, authToken):
+        self.__connector = NgrokConnector()
+        self.ensure_ngrok_not_running()
+        self.create_ngrok_process(ngrokPath, authToken)
+
+    # <summary>
+    # Ensures that NGROK is not running.
+    # </summary>
+
+    def ensure_ngrok_not_running(self):
+        process_name = "ngrok.exe"
+
+        for proc in psutil.process_iter():
+            try:
+                # Check if process name contains the given name string.
+                if process_name.lower() in proc.name().lower():
+                    raise(
+                        "Looks like NGROK is still running. Please kill it before running the provider again.")
+
+            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
+                pass
+
+    # <summary>
+    # Kill ngrok.exe process
+    # </summary>
+
+    def dispose(self):
+        out, err = self.__ngrokProcess.communicate()
+        for line in out.splitlines():
+            if 'ngrok.exe' in line:
+                pid = int(line.split(None, 1)[0])
+                os.kill(pid, signal.SIGKILL)
+
+    # <summary>
+    # Creates the NGROK process.
+    # </summary>
+
+    def create_ngrok_process(self, ngrokPath, authToken):
+        auth_token_args = ""
+        if (authToken and len(authToken)):
+            auth_token_args = " --authtoken " + authToken
+
+        executable = str(Path(ngrokPath, "ngrok.exe"))
+        os.chmod(executable, 0o777)
+        self.__ngrokProcess = subprocess.Popen(
+            [executable, "http", "http://localhost:9007/", "-host-header", "localhost:9007"])
+
+    # <summary>
+    # Get Ngrok URL
+    # </summary>
+
+    def get_ngrok_url(self):
+        try:
+            totalAttempts = 4
+            while(totalAttempts > 0):
+                # Wait for fetching the ngrok url as ngrok process might not be started yet.
+                sleep(2)
+                tunnels = self.__connector.get_all_tunnels()
+                if (tunnels and len(tunnels['tunnels'])):
+                    # Do the parsing of the get
+                    ngrok_url = tunnels['tunnels'][0]['public_url']
+                    return ngrok_url
+
+                totalAttempts = totalAttempts - 1
+
+        except Exception as ex:
+            raise Exception("Failed to retrieve ngrok url --> " + str(ex))