使用python统计FreeSWITCH呼叫并发

操作系统 ：CentOS 7.6_x64
FreeSWITCH版本 ：1.10.7
Python版本：3.9.12
在FreeSWITCH作为线路接入服务器的场景下（低并发），需要定时采集呼叫的统计数据。
本日整理下使用python脚本统计FreeSWITCH呼叫并发的功能，我将从以下几个方面展开：

• 使用api统计
  
• 基于ESL事故统计
  
• 数据推送
  
• 综合示例
  

一、使用api统计

大致思路如下：
1）通过freeswitch提供的cli工具，定时调用"show calls"命令，获取数据；
2）解析命令行返回的数据，实行统计操作；
api统计可直接使用fs_cli工具，也可以使用ESL方式，这里分开描述。
1、基于fs_cli实现

通过fs_cli工具实行"show calls"命令，效果如下：

命令行输出的文本内容如下：

1. uuid,direction,created,created_epoch,name,state,cid_name,cid_num,ip_addr,dest,presence_id,presence_data,accountcode,callstate,callee_name,callee_num,callee_direction,call_uuid,hostname,sent_callee_name,sent_callee_num,b_uuid,b_direction,b_created,b_created_epoch,b_name,b_state,b_cid_name,b_cid_num,b_ip_addr,b_dest,b_presence_id,b_presence_data,b_accountcode,b_callstate,b_callee_name,b_callee_num,b_callee_direction,b_sent_callee_name,b_sent_callee_num,call_created_epoch
2. f0c93f532af74c73ad55a041d612a4cf,inbound,2026-04-20 19:10:35,1776737435,sofia/internal/[email protected],CS_EXECUTE,1000,1000,192.168.137.1,10087,[email protected],,,RING_WAIT,,,,,host100,,,,,,,,,,,,,,,,,,,,,,
3. 64f34077-45a5-46a7-a144-a1f8f864fe5c,outbound,2026-04-20 19:10:35,1776737435,sofia/external/[email protected]:5060,CS_CONSUME_MEDIA,1000,1000,192.168.137.1,654321,,,,DOWN,Outbound Call,654321,,f0c93f532af74c73ad55a041d612a4cf,host100,,,,,,,,,,,,,,,,,,,,,,
4. 2 total.

复制代码

可以看到，返回的内容是csv格式，可以通过解析csv数据，根据指定特征（比如：呼叫方向、被叫号码等）统计并发数据。
当然，如果无需关注呼叫特征，可直接使用如下命令获取呼叫量：

1. fs_cli -x "show calls count"

复制代码

这里接纳关注呼叫特征的方案（呼叫方向、被叫号码）。
示例代码（cliCount1.py）可从如下渠道获取：
运行效果如下：

 
2、基于ESL实现

该模式的核心还是"show calls"这个api命令，区别在用基于ESL实现。
关于如何编译Python3.9版本的ESL，可参考如下文章：
自定义Python版本ESL库访问FreeSWITCH
示例代码（cliCount2.py）可从如下渠道获取：
运行效果如下：

二、使用ESL事故统计

可基于ESL事故来统计：

1. 计数器累加： CHANNEL_CREATE
2. 计数器减少 ：CHANNEL_HANGUP_COMPLETE 或 CHANNEL_DESTROY

复制代码

关于如何编译Python3.9版本的ESL，可参考如下文章：
自定义Python版本ESL库访问FreeSWITCH
示例代码（eventCount1.py）可从如下渠道获取：
运行效果如下：

三、数据推送

监控数据一般需要推送给监控大屏，这里通过requests库发送HTTP POST请求为例。
数据格式如下（json）：

1. {
2.   "datetime": "2026-04-20 23:07:03",
3.   "inbound": {
4.     "10087": 1
5.   },
6.   "outbound": {
7.     "654321": 1
8.   }
9. }

复制代码

推送示比方下（requestsPost1.py）：

1. #! /usr/bin/env python
2. # -*-coding:utf-8-*-
3. import requests,json
4. url = 'http://127.0.0.1:8080/CallStats'
5. dtmp = {
6.     'datetime': '2026-04-20 23:07:03',
7.     'inbound': {'10087': 1},
8.     'outbound': {'654321': 1}
9. }
10. s = json.dumps(dtmp)
11. r = requests.post(url, data=s)
12. print(r.text)

复制代码

示例服务端（httpServer1.py）可从如下渠道获取：
运行效果如下：

四、综合示例

这里基于ESL使用api实现呼叫数据统计，大致思路如下：
1）主线程定时调用api命令获取并发数据；
2）子线程实行推送操作；
示例代码如下（statsPushTest1.py）：

1. pushThrd = threading.Thread(target=doPush,args=(gPushUrl,))
2. #pushThrd.daemon = True
3. pushThrd.start()
4. con = ESLconnection("127.0.0.1","8021","ClueCon")
5. if con.connected:
6.     while True:
7.         e = con.api(gCountCmd)
9.         cmdRet = e.getBody()
10.         print(f"cmdRet :\n{cmdRet}")
11.         gDStats = count_calls(cmdRet)
12.         #print("按方向统计被叫号码（dest字段）结果：")
13.         #for direction, dest_counts in gDStats.items():
14.         #    print(f"\n{direction.upper()} 方向：")
15.         #    for dest, count in dest_counts.items():
16.         #        print(f"  {dest}: {count} 次")
18.         time.sleep(3)
19. else:   
20.     print("Not connected")

复制代码

完整代码可从如下渠道获取：
运行效果如下：