### Initialize Outbound ESL Server Source: https://github.com/evoluxbr/greenswitch/blob/master/README.rst Configures and starts an OutboundESLServer instance to listen for incoming FreeSWITCH connections. It binds to a specific address and port while defining the application handler class. ```python server = greenswitch.OutboundESLServer(bind_address='0.0.0.0', bind_port=5000, application=MyApplication, max_connections=5) server.listen() ``` -------------------------------- ### Handle Outbound Calls with GreenSWITCH Source: https://github.com/evoluxbr/greenswitch/blob/master/README.rst Demonstrates handling outbound calls using GreenSWITCH with Gevent. It includes setting up an application to receive calls, managing events, answering, playing audio, capturing DTMF digits, and performing background tasks. The example highlights the mix of synchronous and asynchronous operations. ```python import gevent import greenswitch import logging logging.basicConfig(level=logging.DEBUG) class MyApplication(object): def __init__(self, session): self.session = session def run(self): """ Main function that is called when a call comes in. """ try: self.handle_call() except: logging.exception('Exception raised when handling call') self.session.stop() def handle_call(self): # We want to receive events related to this call # They are also needed to know when an application is done running # for example playback self.session.myevents() print("myevents") # Send me all the events related to this call even if the call is already # hangup self.session.linger() print("linger") self.session.answer() print("answer") gevent.sleep(1) print("sleep") # Now block until the end of the file. pass block=False to # return immediately. self.session.playback('ivr/ivr-welcome') print("welcome") # blocks until the caller presses a digit, see response_timeout and take # the audio length in consideration when choosing this number digit = self.session.play_and_get_digits('1', '1', '3', '5000', '#', 'conference/conf-pin.wav', 'invalid.wav', 'test', '\d', '1000', "''", block=True, response_timeout=5) print("User typed: %s" % digit) # Start music on hold in background without blocking code execution # block=False makes the playback function return immediately. self.session.playback('local_stream://default', block=False) print("moh") # Now we can do a long task, for example, processing a payment, # consuming an APIs or even some database query to find our customer :) gevent.sleep(5) print("sleep 5") ``` -------------------------------- ### Connect to FreeSWITCH Inbound ESL Source: https://github.com/evoluxbr/greenswitch/blob/master/README.rst Establishes a connection to the FreeSWITCH Event Socket Layer in inbound mode. It requires the host, port, and password for authentication. The example demonstrates sending an API command ('list_users') and printing the response data. ```python import greenswitch fs = greenswitch.InboundESL(host='127.0.0.1', port=8021, password='ClueCon') fs.connect() r = fs.send('api list_users') print r.data ``` -------------------------------- ### Connect to FreeSWITCH Server with InboundESL Source: https://context7.com/evoluxbr/greenswitch/llms.txt Demonstrates how to establish a connection to a FreeSWITCH server using the InboundESL class. It covers connecting, authenticating, sending API and BGAPI commands, and using a context manager for automatic cleanup. Requires FreeSWITCH host, port (default 8021), and ESL password. ```python import greenswitch import gevent # Basic connection to FreeSWITCH fs = greenswitch.InboundESL( host='127.0.0.1', port=8021, password='ClueCon', timeout=5 # Connection timeout in seconds ) # Connect and authenticate fs.connect() # Send API command and get response response = fs.send('api list_users') print(response.data) # Send bgapi command for background execution response = fs.send('bgapi originate sofia/gateway/mygateway/1234567890 &park') print(response.data) # Using context manager for automatic cleanup with greenswitch.InboundESL(host='127.0.0.1', port=8021, password='ClueCon') as fs: response = fs.send('api status') print(response.data) # Clean shutdown fs.stop() ``` -------------------------------- ### InboundESL - Connect to FreeSWITCH Server Source: https://context7.com/evoluxbr/greenswitch/llms.txt Demonstrates how to establish a connection to a FreeSWITCH server using the InboundESL class, send API commands, and manage the connection lifecycle. ```APIDOC ## InboundESL - Connect to FreeSWITCH Server ### Description Establishes a connection to a FreeSWITCH server's Event Socket interface, allowing for sending commands, subscribing to events, and receiving real-time notifications. ### Method N/A (This is a class instantiation and method calls) ### Endpoint N/A (Connects to a FreeSWITCH ESL port, typically 8021) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python import greenswitch import gevent # Basic connection to FreeSWITCH fs = greenswitch.InboundESL( host='127.0.0.1', port=8021, password='ClueCon', timeout=5 # Connection timeout in seconds ) # Connect and authenticate fs.connect() # Send API command and get response response = fs.send('api list_users') print(response.data) # Send bgapi command for background execution response = fs.send('bgapi originate sofia/gateway/mygateway/1234567890 &park') print(response.data) # Using context manager for automatic cleanup with greenswitch.InboundESL(host='127.0.0.1', port=8021, password='ClueCon') as fs: response = fs.send('api status') print(response.data) # Clean shutdown fs.stop() ``` ### Response #### Success Response (200) Responses vary based on the API command sent. The `send` method returns an `Event` object with a `data` attribute containing the response string. #### Response Example ```json { "data": "+OK list_users: 1 2 3" } ``` ``` -------------------------------- ### Subscribe to FreeSWITCH Events with Event Handlers Source: https://context7.com/evoluxbr/greenswitch/llms.txt Illustrates how to subscribe to specific FreeSWITCH events using the register_handle method. It shows how to define handler functions for different event types (e.g., CHANNEL_ANSWER, CHANNEL_HANGUP, sofia_register_failure) and a wildcard handler for all events. Requires a connection to FreeSWITCH and event subscription commands. ```python import greenswitch import gevent import logging logging.basicConfig(level=logging.DEBUG) def on_channel_answer(event): """Handle call answer events""" caller_id = event.headers.get('Caller-Caller-ID-Number') destination = event.headers.get('Caller-Destination-Number') uuid = event.headers.get('Unique-ID') print(f"Call answered: {caller_id} -> {destination} (UUID: {uuid})") def on_channel_hangup(event): """Handle call hangup events""" cause = event.headers.get('Hangup-Cause') duration = event.headers.get('variable_billsec', '0') print(f"Call ended: {cause}, Duration: {duration}s") def on_sofia_register_failure(event): """Handle failed SIP registration attempts""" message = 'Failed register attempt from {network-ip} to user {to-user} profile {profile-name}' print(message.format(**event.headers)) def on_all_events(event): """Catch-all handler using wildcard""" event_name = event.headers.get('Event-Name') print(f"Event received: {event_name}") # Connect to FreeSWITCH fs = greenswitch.InboundESL(host='192.168.50.4', port=8021, password='ClueCon') fs.connect() # Register event handlers fs.register_handle('CHANNEL_ANSWER', on_channel_answer) fs.register_handle('CHANNEL_HANGUP', on_channel_hangup) fs.register_handle('sofia::register_failure', on_sofia_register_failure) fs.register_handle('*', on_all_events) # Wildcard handler for all events # Subscribe to all events fs.send('EVENTS PLAIN ALL') # Or subscribe to specific events only # fs.send('EVENTS PLAIN CHANNEL_ANSWER CHANNEL_HANGUP') print('Connected to FreeSWITCH!') # Keep the connection alive try: while True: gevent.sleep(1) except KeyboardInterrupt: # Unregister handler and disconnect fs.unregister_handle('CHANNEL_ANSWER', on_channel_answer) fs.stop() print('ESL Disconnected.') ``` -------------------------------- ### Process FreeSWITCH events with ESLEvent Source: https://context7.com/evoluxbr/greenswitch/llms.txt Demonstrates how to register event handlers and extract specific header data from ESLEvent objects, such as call UUIDs, caller information, and channel states. It also shows how to handle specific event types like DTMF and CHANNEL_HANGUP. ```python import greenswitch import gevent def detailed_event_handler(event): event_name = event.headers.get('Event-Name') event_subclass = event.headers.get('Event-Subclass') uuid = event.headers.get('Unique-ID') if event_name == 'DTMF': digit = event.headers.get('DTMF-Digit') print(f"DTMF: {digit}") if event_name == 'CHANNEL_HANGUP': hangup_cause = event.headers.get('Hangup-Cause') print(f"Hangup: {hangup_cause}") fs = greenswitch.InboundESL(host='127.0.0.1', port=8021, password='ClueCon') fs.connect() fs.register_handle('CHANNEL_HANGUP', detailed_event_handler) fs.send('EVENTS PLAIN ALL') while True: gevent.sleep(1) ``` -------------------------------- ### Event Handlers - Subscribe to FreeSWITCH Events Source: https://context7.com/evoluxbr/greenswitch/llms.txt Shows how to register callback functions to handle specific FreeSWITCH events, enabling real-time monitoring and reaction to telephony events. ```APIDOC ## Event Handlers - Subscribe to FreeSWITCH Events ### Description Allows subscribing to specific FreeSWITCH events using the `register_handle` method. When a registered event occurs, the associated handler function is invoked with the event data. ### Method N/A (This is a class instantiation and method calls) ### Endpoint N/A (Connects to a FreeSWITCH ESL port, typically 8021) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python import greenswitch import gevent import logging logging.basicConfig(level=logging.DEBUG) def on_channel_answer(event): """Handle call answer events""" caller_id = event.headers.get('Caller-Caller-ID-Number') destination = event.headers.get('Caller-Destination-Number') uuid = event.headers.get('Unique-ID') print(f"Call answered: {caller_id} -> {destination} (UUID: {uuid})") def on_channel_hangup(event): """Handle call hangup events""" cause = event.headers.get('Hangup-Cause') duration = event.headers.get('variable_billsec', '0') print(f"Call ended: {cause}, Duration: {duration}s") def on_sofia_register_failure(event): """Handle failed SIP registration attempts""" message = 'Failed register attempt from {network-ip} to user {to-user} profile {profile-name}' print(message.format(**event.headers)) def on_all_events(event): """Catch-all handler using wildcard""" event_name = event.headers.get('Event-Name') print(f"Event received: {event_name}") # Connect to FreeSWITCH fs = greenswitch.InboundESL(host='192.168.50.4', port=8021, password='ClueCon') fs.connect() # Register event handlers fs.register_handle('CHANNEL_ANSWER', on_channel_answer) fs.register_handle('CHANNEL_HANGUP', on_channel_hangup) fs.register_handle('sofia::register_failure', on_sofia_register_failure) fs.register_handle('*', on_all_events) # Wildcard handler for all events # Subscribe to all events fs.send('EVENTS PLAIN ALL') # Or subscribe to specific events only # fs.send('EVENTS PLAIN CHANNEL_ANSWER CHANNEL_HANGUP') print('Connected to FreeSWITCH!') # Keep the connection alive try: while True: gevent.sleep(1) except KeyboardInterrupt: # Unregister handler and disconnect fs.unregister_handle('CHANNEL_ANSWER', on_channel_answer) fs.stop() print('ESL Disconnected.') ``` ### Response #### Success Response (200) Event handlers are functions that are called asynchronously when matching events are received from FreeSWITCH. The `event` object passed to the handler contains headers with event details. #### Response Example ``` Event received: CHANNEL_CREATE Call answered: 1001 -> 1002 (UUID: a1b2c3d4-e5f6-7890-1234-567890abcdef) Event received: CHANNEL_ANSWER Call ended: NORMAL_CLEARING, Duration: 60s Event received: CHANNEL_HANGUP ``` ``` -------------------------------- ### Python: Safe Call Operations with while_connected Source: https://context7.com/evoluxbr/greenswitch/llms.txt Demonstrates using the while_connected context manager and decorator in Python to ensure operations are performed only when a call session is active. It automatically handles disconnections by raising OutboundSessionHasGoneAway, preventing errors on dead sessions. Requires the greenswitch library. ```python import gevent import greenswitch from greenswitch.esl import OutboundSessionHasGoneAway class SafeCallApplication: def __init__(self, session): self.session = session def run(self): try: self.session.myevents() self.session.linger() self.session.answer() # Using as context manager - checks connection on enter and exit with self.session.while_connected(): self.session.playback('ivr/ivr-welcome', block=True) self.do_processing() self.session.playback('ivr/ivr-goodbye', block=True) self.session.hangup() except OutboundSessionHasGoneAway: print(f"Caller {self.session.caller_id_number} hung up") finally: self.session.stop() # Using as decorator @property def session_guard(self): return self.session.while_connected() def do_processing(self): # Manual check if caller is still connected self.session.raise_if_disconnected() # Process something gevent.sleep(2) # Check again self.session.raise_if_disconnected() server = greenswitch.OutboundESLServer( bind_address='0.0.0.0', bind_port=5000, application=SafeCallApplication, max_connections=25 ) server.listen() ``` -------------------------------- ### Control Calls with OutboundSession Methods in Python Source: https://context7.com/evoluxbr/greenswitch/llms.txt Demonstrates various call control methods provided by the OutboundSession class in Greenswitch. This includes initiating blocking and non-blocking playback, using text-to-speech, bridging calls, parking calls, and hanging up with specific causes. It highlights how to manage call flows with concurrent operations. ```python import gevent import greenswitch import logging class CallControlApplication: def __init__(self, session): self.session = session def run(self): try: self.session.myevents() self.session.linger() self.session.answer() # Blocking playback - waits for file to finish self.session.playback('ivr/ivr-welcome', block=True) # Non-blocking playback - returns immediately (music on hold) self.session.playback('local_stream://default', block=False) # Do some processing while music plays result = self.process_external_api() # Stop the background audio self.session.uuid_break() # Text-to-speech using FreeSWITCH say application self.session.say( module_name='en', say_type='NUMBER', say_method='pronounced', gender='FEMININE', text='12345', block=True ) # Bridge to another destination self.session.bridge( args='sofia/gateway/provider/18005551234', block=True, response_timeout=60 ) # Park the call (put on hold) self.session.park() # Add delay gevent.sleep(2) # Hangup with specific cause self.session.hangup(cause='NORMAL_CLEARING') except greenswitch.esl.OutboundSessionHasGoneAway: logging.info('Call disconnected') finally: self.session.stop() def process_external_api(self): # Simulate API call while caller hears music gevent.sleep(3) return {'status': 'success'} server = greenswitch.OutboundESLServer( bind_address='0.0.0.0', bind_port=5000, application=CallControlApplication, max_connections=50 ) server.listen() ``` -------------------------------- ### Handle Incoming Calls with OutboundESLServer in Python Source: https://context7.com/evoluxbr/greenswitch/llms.txt Sets up a socket server using Greenswitch's OutboundESLServer to handle incoming calls from FreeSWITCH. This is ideal for IVR applications, allowing fine-grained control over each call. FreeSWITCH needs to be configured to route calls to this socket server. The IVRApplication class handles call logic like answering, playing messages, collecting digits, and hanging up. ```python import gevent import greenswitch import logging logging.basicConfig(level=logging.DEBUG) """ FreeSWITCH Dialplan Configuration (conf/dialplan/default.xml): """ class IVRApplication: def __init__(self, session): self.session = session def run(self): try: self.handle_call() except greenswitch.esl.OutboundSessionHasGoneAway: logging.info('Caller hung up') except Exception: logging.exception('Error handling call') finally: self.session.stop() def handle_call(self): # Subscribe to events for this call self.session.myevents() # Keep receiving events after hangup self.session.linger() # Answer the call self.session.answer() # Play welcome message (blocking - waits for completion) self.session.playback('ivr/ivr-welcome') # Get user input with DTMF digit = self.session.play_and_get_digits( min_digits='1', max_digits='4', max_attempts='3', timeout='5000', terminators='#', prompt_file='ivr/ivr-enter_ext.wav', error_file='ivr/ivr-invalid.wav', variable='user_input', digits_regex=r'\d+', digit_timeout='3000', transfer_on_fail="''", block=True, response_timeout=30 ) print(f"User entered: {digit}") # Access session properties print(f"Call UUID: {self.session.uuid}") print(f"Caller ID: {self.session.caller_id_number}") # End the call self.session.hangup() # Create and start the server server = greenswitch.OutboundESLServer( bind_address='0.0.0.0', bind_port=5000, application=IVRApplication, max_connections=100 ) server.listen() # Blocks and handles incoming calls ``` -------------------------------- ### Terminate FreeSWITCH Session Source: https://github.com/evoluxbr/greenswitch/blob/master/README.rst Demonstrates how to gracefully stop music on hold, hang up the call, and close the session socket. This ensures the FreeSWITCH server releases the connection properly. ```python self.session.uuid_break() print("break") self.session.hangup() print("hangup") self.session.stop() ``` -------------------------------- ### Handle GreenSWITCH connection exceptions Source: https://context7.com/evoluxbr/greenswitch/llms.txt Provides patterns for safely managing inbound connections and command execution using specific GreenSWITCH exceptions like NotConnectedError and OutboundSessionHasGoneAway. ```python import greenswitch from greenswitch.esl import NotConnectedError import logging def safe_inbound_connection(): try: fs = greenswitch.InboundESL(host='127.0.0.1', port=8021, password='ClueCon', timeout=5) fs.connect() return fs except NotConnectedError as e: logging.error(f"Failed to connect: {e}") return None def safe_send(fs, command): try: return fs.send(command) except NotConnectedError: logging.error("Connection lost") return None ``` -------------------------------- ### Python: Interactive DTMF Collection with play_and_get_digits Source: https://context7.com/evoluxbr/greenswitch/llms.txt Illustrates the use of the play_and_get_digits method in Python for interactive voice response (IVR) systems. This method plays prompts, collects DTMF input with configurable options like digit limits, timeouts, regex validation, and retries. It's crucial for gathering user choices or data. Requires the greenswitch library. ```python import gevent import greenswitch class DTMFMenuApplication: def __init__(self, session): self.session = session def run(self): try: self.session.myevents() self.session.linger() self.session.answer() # Simple single digit menu choice = self.session.play_and_get_digits( min_digits='1', max_digits='1', max_attempts='3', timeout='5000', # 5 seconds to start input terminators='#', # # key ends input early prompt_file='ivr/ivr-menu.wav', error_file='ivr/ivr-invalid.wav', variable='menu_choice', digits_regex=r'[1-5]', # Only accept 1-5 digit_timeout='3000', # 3 seconds between digits transfer_on_fail="''", block=True, response_timeout=30 ) if choice == '1': self.session.playback('ivr/ivr-sales.wav') elif choice == '2': self.session.playback('ivr/ivr-support.wav') # Collect multi-digit PIN pin = self.session.play_and_get_digits( min_digits='4', max_digits='6', max_attempts='3', timeout='10000', terminators='#', prompt_file='conference/conf-pin.wav', error_file='conference/conf-bad-pin.wav', variable='user_pin', digits_regex=r'\d{4,6}', # 4-6 digits only digit_timeout='5000', transfer_on_fail="''", block=True, response_timeout=60 ) print(f"PIN entered: {pin}") # Collect phone number phone = self.session.play_and_get_digits( min_digits='10', max_digits='11', max_attempts='2', timeout='15000', terminators='#', prompt_file='ivr/ivr-enter_destination.wav', error_file='ivr/ivr-invalid.wav', variable='destination', digits_regex=r'1?\d{10}', # US phone format digit_timeout='3000', transfer_on_fail="''", block=True, response_timeout=45 ) print(f"Dialing: {phone}") self.session.hangup() except Exception as e: print(f"Error: {e}") finally: self.session.stop() server = greenswitch.OutboundESLServer( bind_address='0.0.0.0', bind_port=5000, application=DTMFMenuApplication, max_connections=20 ) server.listen() ``` -------------------------------- ### Robust IVR Application with Error Handling (Python) Source: https://context7.com/evoluxbr/greenswitch/llms.txt Implements an IVR application that handles outbound calls with comprehensive error management. It catches specific exceptions like `OutboundSessionHasGoneAway` and `gevent.Timeout`, as well as general exceptions, ensuring proper session termination and logging. Dependencies include `gevent` and `greenswitch`. ```Python import gevent import logging # Assuming OutboundSessionHasGoneAway and greenswitch are defined elsewhere # from greenswitch import OutboundSessionHasGoneAway, OutboundESLServer class RobustIVRApplication: def __init__(self, session): self.session = session def run(self): try: self.handle_call() except OutboundSessionHasGoneAway: logging.info(f"Caller {self.session.caller_id_number} disconnected") except gevent.Timeout: logging.warning("Operation timed out") self.session.hangup(cause='RECOVERY_ON_TIMER_EXPIRE') except Exception as e: logging.exception(f"Unexpected error: {e}") try: self.session.hangup(cause='NORMAL_TEMPORARY_FAILURE') except: pass finally: self.session.stop() def handle_call(self): self.session.myevents() self.session.linger() self.session.answer() self.session.raise_if_disconnected() with self.session.while_connected(): self.session.playback('ivr/ivr-welcome', block=True) with gevent.Timeout(30, False): digit = self.session.play_and_get_digits( '1', '1', '3', '5000', '#', 'ivr/ivr-menu.wav', 'ivr/ivr-invalid.wav', 'choice', r'\d', '3000', "''", block=True, response_timeout=25 ) if digit: self.session.playback(f'ivr/ivr-option-{digit}.wav', block=True) self.session.hangup() # Example server setup (assuming greenswitch is imported) # server = greenswitch.OutboundESLServer( # bind_address='0.0.0.0', # bind_port=5000, # application=RobustIVRApplication, # max_connections=50 # ) # server.listen() ``` -------------------------------- ### while_connected - Connection State Guard Source: https://context7.com/evoluxbr/greenswitch/llms.txt Provides a context manager and decorator to safely handle operations that require an active call session, raising an exception if the caller disconnects. ```APIDOC ## while_connected ### Description Ensures that operations are only performed while the call session is active. It prevents errors by checking the connection state before and after blocks of code. ### Method Context Manager / Decorator ### Parameters - **None** ### Request Example with self.session.while_connected(): self.session.playback('ivr/ivr-welcome.wav') ### Response #### Success Response - **None** (Executes block if connected) #### Error Handling - **OutboundSessionHasGoneAway**: Raised if the caller disconnects during the execution of the block. ``` -------------------------------- ### play_and_get_digits - Interactive DTMF Collection Source: https://context7.com/evoluxbr/greenswitch/llms.txt Plays a prompt and collects DTMF input from the caller with support for regex validation, timeouts, and retry logic. ```APIDOC ## play_and_get_digits ### Description Plays an audio prompt to the caller and waits for DTMF input. It validates the input against a regex and handles retries if the input is invalid or times out. ### Parameters - **min_digits** (string) - Required - Minimum number of digits to collect. - **max_digits** (string) - Required - Maximum number of digits to collect. - **max_attempts** (string) - Required - Number of retry attempts allowed. - **timeout** (string) - Required - Time in ms to wait for the first digit. - **prompt_file** (string) - Required - Path to the audio file to play. - **digits_regex** (string) - Optional - Regex pattern to validate the collected digits. ### Request Example self.session.play_and_get_digits(min_digits='4', max_digits='6', prompt_file='pin.wav', digits_regex=r'\d{4,6}') ### Response #### Success Response (200) - **digits** (string) - The collected DTMF input string. #### Response Example "1234" ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.