环境：CentOS 7.9_x64
FreeSWITCH版本： 1.10.11（docker 23.0.6部署）
Python版本：3.9.12
之前遇到过一个需求：
在不影响生产系统业务（FreeSWITCH服务）的情况下，实时推送WebSocket格式语音流给解析服务器，以完成后续服务（比如实时辅助、实时质检等）。
今天整理下该需求的可行性验证笔记并提供示例代码，如需商业使用请参考文章自行实现。
本文使用的FreeSWITCH是基于docker部署，如有使用docker部署FreeSWITCH的需求，可参考如下文章：
为了演示方便，本文使用基于主机的流量镜像方案，生产环境可使用交换机的镜像口来替换。
此外，关于编码方面，这里做下简化：
1）流量镜像rtp编码是g711a（pcma）；
2）ws推流的目标编码是 L16（16k）；
整体结构如下：
整体结构如下：
说明：
1）192.168.137.100机器部署freeswitch，并配置流量镜像推流；
2）192.168.137.101机器部署流量镜像推流程序、ws模拟服务程序；
这里使用iptables的tee模块实现流量镜像功能，为了演示方便，这里使用udp流量全推的方案。
1）推流策略配置
流量镜像配置示例：
[root@host100 pbx]# cat /root/tee1.sh #! /bin/bash iptables -t mangle -F iptables -t mangle -I PREROUTING 1 -p udp -j TEE --gateway 192.168.137.101 iptables -t mangle -I POSTROUTING 1 -p udp -j TEE --gateway 192.168.137.101 iptables -t mangle -L -v [root@host100 pbx]#
2）推流效果验证
在100机器上执行如下命令：
nc -u 192.168.137.108 33330
输入 123456 ，然后执行 ctrl + d 键进行发送。
在101机器上进行抓包：
tcpdump -i enp0s3 udp port 33330 -vvvvvvvvvvvvvvvv
使用tcpdump验证效果如下：
使用wireshark验证效果如下：
为了方便方案验证，这里使用python模拟个ws服务器，该服务器会将收到的音频数据存储为本地pcm文件以便验证。
这里用的python3.9环境，环境部署可参考如下文章：
配合验证的ws服务器代码如下（wsServer2.py）：
import os,json,time from websockets.sync.server import serve gPcmDir="pcmFiles" def handler(ws): callid,tag = "",0 fout,fName = None,"" clientId = id(ws) for msg in ws: #print("%s,len msg : %d "% (fName,len(msg))) #msg2 = "I got your message: {}".format(msg) #websocket.send(msg) if type(msg) == str: #if msg.find("initConfig") > -1: print(msg) dtmp = json.loads(msg) var1 = dtmp.get("call_id") if var1 : call_id = var1 var2 = dtmp.get("tag") if None != var2 : tag = var2 print(call_id,tag) #fName = "%s-%d.pcm" % (call_id,role) fName = "%s/%s.pcm" % (gPcmDir,tag) if None == fout : fout = open(fName,"wb") if "end" == dtmp.get("type"): fout.close() if type(msg) == bytes: fout.write(msg) return None def main(): if not os.path.isdir(gPcmDir):os.makedirs(gPcmDir) with serve(handler, "0.0.0.0", 8765) as server: server.serve_forever() if __name__ == "__main__": main()
启动脚本如下（start.sh）
#! /bin/bash #pydir=$PWD pydir=/root/python39 export CFLAGS="-I$pydir/include" export LDFLAGS="-L$pydir/lib" export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$pydir/lib $pydir/bin/python3.9 wsServer2.py
运行效果如下：
这里列举下核心功能，并提供python示例代码。
关键点如下：
1）抓包功能通过libpcap实现；
2）解析sdp获取rtp端口信息；
3）根据rtp协议从payload中解析并提取音频数据；
4）使用ffmpeg将提取的alaw格式音频转换为wav格式，以便验证音频的正确性；
示例代码如下（pcap2alaw.py）：
完整代码可从如下渠道获取：
关注微信公众号（聊聊博文，文末可扫码）后回复 20250816 获取。
alaw2pcm示例代码：
import os,struct,socket import ctypes as ct import audioop rawFile = b"g711a_export1.raw" sampwidth = 4 i,step = 0,160//sampwidth fout = open("pcm_export1.raw","wb") with open(rawFile,"rb") as fin: while True: data = fin.read(step) if not data : break data2 = audioop.alaw2lin(data, sampwidth) fout.write(data2)
提取的音频导入效果如下：
关于pcm音频的播放可参考如下文章：
pcm转ws示例代码如下（pcm2ws.py）：
完整代码可从如下渠道获取：
关注微信公众号（聊聊博文，文末可扫码）后回复 20250816 获取。
运行效果如下：
这里描述下独立程序的推流方案，即：所有功能都在一个程序里面实现，比如单独的python脚本。
模块设计如下：
大致设计思路如下：
1）启动独立线程ThrdLiveCap，使用libpcap实时抓取流量镜像过来的数据，并将数据推送到指定队列；
2）启动独立线程ThrdPktParse，解析sip协议，提取uuid、rtp端口等关键信息，并将解析的数据存储到全局缓存中；
3）启动独立线程ThrdRtp2Ws，监测并启动独立线程，进行ws推流；
这里提供下示例代码。
完整代码可从如下渠道获取：
关注微信公众号（聊聊博文，文末可扫码）后回复 20250816 获取。
1）编写拨号方案
内容如下：
<extension name="playTest"> <condition field="destination_number" expression="^10087$"> <action application="answer"/> <action application="playback" data="local_stream://moh"/> </condition> </extension>
2）注册1000分机，拨打10087号码，在101机器上启动rtp2ws脚本、wsServer脚本，可成功接收rtp文件。
运行效果如下：
文件导入效果如下（s16le 16K）：
可正常播放。
对应的pcm文件可从文末提供的渠道获取。
本文相关资源及运行环境，可从如下渠道获取：
关注微信公众号（聊聊博文，文末可扫码）后回复 20250816 获取。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。