惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

博客园_首页
IT之家
IT之家
博客园 - Franky
Stack Overflow Blog
Stack Overflow Blog
宝玉的分享
宝玉的分享
Recent Announcements
Recent Announcements
Engineering at Meta
Engineering at Meta
S
SegmentFault 最新的问题
V
Visual Studio Blog
freeCodeCamp Programming Tutorials: Python, JavaScript, Git & More
OSCHINA 社区最新新闻
OSCHINA 社区最新新闻
Last Week in AI
Last Week in AI
H
Help Net Security
V
V2EX
H
Hackread – Cybersecurity News, Data Breaches, AI and More
量子位
博客园 - 叶小钗
J
Java Code Geeks
博客园 - 【当耐特】
月光博客
月光博客
爱范儿
爱范儿
人人都是产品经理
人人都是产品经理
酷 壳 – CoolShell
酷 壳 – CoolShell
小众软件
小众软件

博客园 - Mike_Zhang

从pcap文件提取sip信令文本 FreeSWITCH开启silk编码及转码 FreeSWITCH使用mod_fail2ban模块来提升安全 FreeSWITCH使用RNNoise进行实时通话降噪 使用RNNoise进行音频降噪 python3.14版本的free-threading功能体验 一种FreeSWITCH流量镜像WebSocket音频推流方案 C++使用WinHTTP访问http/https服务 cJSON静态库编译及vs2022使用示例 openssl静态库编译及vs2022使用示例 FreeSWITCH日志功能分析及apr模拟 使用pjsip封装自定义软电话sdk pjsip编译、说明及vs2022使用示例 使用kamailio进行分机注册及互拨 apr库编译及队列使用笔记 debian10环境安装rtpengine FreeSWITCH对接http协议的tts服务 python3解析wav文件获取dtmf值 音频文件降噪及python示例 FreeSWITCH使用soundtouch进行变声
使用python统计FreeSWITCH呼叫并发
Mike_Zhang · 2026-08-20 · via 博客园 - Mike_Zhang

操作系统 :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"命令,效果如下:

image

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

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
f0c93f532af74c73ad55a041d612a4cf,inbound,2026-04-20 19:10:35,1776737435,sofia/internal/1000@192.168.137.100,CS_EXECUTE,1000,1000,192.168.137.1,10087,1000@192.168.137.100,,,RING_WAIT,,,,,host100,,,,,,,,,,,,,,,,,,,,,,
64f34077-45a5-46a7-a144-a1f8f864fe5c,outbound,2026-04-20 19:10:35,1776737435,sofia/external/654321@192.168.137.71:5060,CS_CONSUME_MEDIA,1000,1000,192.168.137.1,654321,,,,DOWN,Outbound Call,654321,,f0c93f532af74c73ad55a041d612a4cf,host100,,,,,,,,,,,,,,,,,,,,,,
2 total.

可以看到,返回的内容是csv格式,可以通过解析csv数据,根据指定特征(比如:呼叫方向、被叫号码等)统计并发数据。

当然,如果无需关注呼叫特征,可直接使用如下命令获取呼叫量:

fs_cli -x "show calls count"

image

这里采用关注呼叫特征的方案(呼叫方向、被叫号码)。

示例代码(cliCount1.py)可从如下渠道获取:

 关注微信公众号(聊聊博文,文末可扫码)后回复 20260421 获取。

运行效果如下:

image

2、基于ESL实现

该模式的核心还是"show calls"这个api命令,区别在用基于ESL实现。

关于如何编译Python3.9版本的ESL,可参考如下文章:

自定义Python版本ESL库访问FreeSWITCH

示例代码(cliCount2.py)可从如下渠道获取:

关注微信公众号(聊聊博文,文末可扫码)后回复 20260421 获取。

运行效果如下:

image

二、使用ESL事件统计

可基于ESL事件来统计:

计数器累加: CHANNEL_CREATE
计数器减少 :CHANNEL_HANGUP_COMPLETE 或 CHANNEL_DESTROY

关于如何编译Python3.9版本的ESL,可参考如下文章:

自定义Python版本ESL库访问FreeSWITCH

示例代码(eventCount1.py)可从如下渠道获取:

关注微信公众号(聊聊博文,文末可扫码)后回复 20260421 获取。

运行效果如下:

image

三、数据推送

监控数据一般需要推送给监控大屏,这里通过requests库发送HTTP POST请求为例。

数据格式如下(json):

{
  "datetime": "2026-04-20 23:07:03",
  "inbound": {
    "10087": 1
  },
  "outbound": {
    "654321": 1
  }
}

推送示例如下(requestsPost1.py):

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

示例服务端(httpServer1.py)可从如下渠道获取:

关注微信公众号(聊聊博文,文末可扫码)后回复 20260421 获取。

运行效果如下:

image

四、综合示例

这里基于ESL使用api实现呼叫数据统计,大致思路如下:

1)主线程定时调用api命令获取并发数据;

2)子线程执行推送操作;

示例代码如下(statsPushTest1.py):

pushThrd = threading.Thread(target=doPush,args=(gPushUrl,))
#pushThrd.daemon = True 
pushThrd.start()
con = ESLconnection("127.0.0.1","8021","ClueCon")
if con.connected:
    while True:
        e = con.api(gCountCmd)

        cmdRet = e.getBody()
        print(f"cmdRet :\n{cmdRet}")
        gDStats = count_calls(cmdRet)
        #print("按方向统计被叫号码(dest字段)结果:")
        #for direction, dest_counts in gDStats.items():
        #    print(f"\n{direction.upper()} 方向:")
        #    for dest, count in dest_counts.items():
        #        print(f"  {dest}: {count} 次")

        time.sleep(3)
else:   
    print("Not connected")

完整代码可从如下渠道获取:

关注微信公众号(聊聊博文,文末可扫码)后回复 20260421 获取。

运行效果如下:

image