RedFS - FreeSWITCH Event Socket
FreeSWITCH Event Socket Protocol Client Library by Python3:Gevent

Why RedFS ?

It's origin fork from greenswitch, with additional implementation:

• Minimalize dependency (gevent only)
• Large scale adaptation
• Bug fix
• Dualstack IPv4/IPv6
• Python3 syntax improvement

Usage

Installation

pip3 install redfs

FreeSWITCH configuration

event socket module

<configuration name="event_socket.conf" description="Socket Client">
  <settings>
    <param name="nat-map" value="false"/>
    <param name="listen-ip" value="127.0.0.1"/>
    <param name="listen-port" value="8021"/>
    <param name="password" value="your-esl-password"/>
  </settings>
</configuration>

dialplan

<include>
  <context name="default">
    <extension name="daemon-ex">
      <condition regex="all">
        <regex field="destination_number" expression="."/>
        <action application="sched_hangup" data="+60 ALLOTTED_TIMEOUT"/>
        <action application="park"/>
        <anti-action application="hangup" data="REQUESTED_CHAN_UNAVAIL"/>
     </condition>
    </extension>
  </context>
</include>

a simple python application for auto answer.

import gevent
import redfs
import traceback


ESL_HOST = '127.0.0.1'
ESL_PORT = 8021
ESL_PASSWORD = 'your-esl-password'
ESL_TIMEOUT = 10

def realtime(event):
    event_name = event.headers.get('Event-Name')
    uuid = event.headers.get('Unique-ID')
    print(f'Event {event_name} with uuid {uuid}')

    conn = redfs.InboundESL(host=ESL_HOST, port=ESL_PORT, password=ESL_PASSWORD, timeout=ESL_TIMEOUT)
    conn.connect()

    if event_name == 'CHANNEL_PARK':
        cmd = f'api uuid_answer {uuid}'
        res = conn.send(cmd)
        print(cmd, res.data)

def printlog(event):
    event_name = event.headers.get('Event-Name')
    uuid = event.headers.get('Unique-ID')
    print(f'Event {event_name} with uuid {uuid}')

cnx = redfs.InboundESL(host=ESL_HOST, port=ESL_PORT, password=ESL_PASSWORD, timeout=ESL_TIMEOUT)
cnx.connect()

cnx.register_handle(['CHANNEL_PARK', 'CHANNEL_ANSWER'], realtime)
cnx.register_handle('*', printlog)
cnx.send('EVENTS PLAIN ALL')
print('connected')
while True:
    try:
        gevent.sleep(1)
    except KeyboardInterrupt:
        cnx.stop()
        break
    except Exception as e:
        print(e, traceback.format_exc())

Enjoy!

License

MIT