如何使用Python检测RTMP直播流中的音频削波（Clipping）问题

整体实现流程 #

• 第一步：拉流与音频解码
  首先拉取RTMP直播流，分离出音频轨道后，将OGG/MP3/AAC编码的压缩音频解码为原始PCM采样数据，同时记录音频的采样位深、采样率参数，用于后续削波阈值的确定：比如16位有符号整型PCM的阈值为±32767，32位浮点型PCM的阈值为±1.0。
• 第二步：削波特征检测
  削波的核心判定规则为连续多个采样点触达量化区间的最大/最小阈值，单采样点触达大概率是偶发噪声，通常设置连续3~5个采样点顶到阈值即可判定为出现削波。你也可以根据需求统计固定时长内的削波发生次数、削波时长占比，用来评估失真的严重程度。
• 第三步：（可选）结果输出
  可以将检测结果同步输出为日志、告警，或者结合时间戳标记削波发生的具体流时间点。

适用Python库 #

• ffmpeg-python：FFmpeg的Python绑定库，支持直接拉取RTMP流，内置所有主流音频编码的解码能力，无需单独处理OGG/MP3/AAC的兼容逻辑，可直接输出原始PCM数据，是流处理场景的首选工具。
• numpy：高性能数值计算库，将PCM数据转换为numpy数组后，仅需几行代码即可完成阈值匹配、连续采样点统计的逻辑，运行效率足够满足实时直播流的检测需求。
• 可选补充库：如果你的场景不需要严格的实时处理，也可以用pydub做音频解码，它的API封装更简洁，底层同样依赖FFmpeg；如果需要后续做更复杂的音频质量分析，可以搭配librosa使用。

极简示例代码 #

import ffmpeg
import numpy as np

# 配置RTMP流地址与音频参数
rtmp_url = "你的RTMP流地址"
sample_rate = 44100
channels = 2
sample_format = "s16le" # 16位有符号整型PCM
threshold = 32767 # 对应s16le的最大幅值

# 启动ffmpeg拉流解码，输出PCM流
process = (
    ffmpeg
    .input(rtmp_url, listen=0, rtmp_buffer=1000)
    .output('pipe:', format=sample_format, ac=channels, ar=sample_rate)
    .run_async(pipe_stdout=True, pipe_stderr=True)
)

# 实时检测逻辑
while True:
    # 每次读取1秒的PCM数据
    in_bytes = process.stdout.read(sample_rate * channels * 2) # 16位每个采样占2字节
    if not in_bytes:
        break
    # 转成numpy数组
    audio_array = np.frombuffer(in_bytes, dtype=np.int16)
    # 检测连续达到阈值的采样点
    is_clipping = np.abs(audio_array) >= threshold
    # 找连续3个及以上的削波点
    clipping_count = np.sum(np.convolve(is_clipping, np.ones(3, dtype=int), mode='valid') >= 3)
    if clipping_count > 0:
        print(f"检测到削波，当前秒内削波段数量：{clipping_count}")

示例代码仅做逻辑参考，实际使用时可根据流的实际音频参数调整采样率、通道数、位深对应的阈值，也可根据需求调整连续削波的判定长度。

内容的提问来源于stack exchange，提问作者Eelco