如何用Python轻量实现网络上下行速率、jitter检测及通话可用性判断

核心实现思路 #

要兼顾检测准确性和低网络负担，需要将不同指标的检测频率拆分：

• 时延/丢包类的ping检测保持30秒一次的高频，这类检测单次仅发送3个ICMP包，几乎不占用带宽
• 上下行速率类检测降低频率到10~30分钟一次，且仅用小体积测试资源完成测速，避免挤占通话带宽

完整实现代码 #

依赖安装 #

仅需额外安装requests用于轻量HTTP测速：

pip install requests

完整代码 #

import platform
import subprocess
import re
import logging
import requests
import time

# 配置项
PING_CHECK_HOST = "223.5.5.5"  # 可替换为常用的通话服务节点地址
SPEED_TEST_INTERVAL_COUNT = 30  # 每30次ping检测（即15分钟）执行一次速率检测
DOWNLOAD_TEST_URL = "替换为你的小体积测试文件地址（建议500KB以内）"
UPLOAD_TEST_URL = "替换为你的上传测试接口地址"
UPLOAD_TEST_DATA = b"x" * 1024 * 512  # 上传测试用512KB随机数据
REQUEST_TIMEOUT = 10

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

def check_ping(host):
    """
    返回ping结果 (min, avg, max, mdev)，检测失败返回None
    """
    param = '-n' if platform.system().lower() == 'windows' else '-c'
    command = ['ping', param, '3', host]
    ping = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, error = ping.communicate()
    matcher = re.compile("(\d+.\d+)/(\d+.\d+)/(\d+.\d+)/(\d+.\d+)")
    ping_list = r"Minimum = (\d+)ms, Maximum = (\d+)ms, Average = (\d+)ms"
    try:
        if not error:
            if platform.system().lower() == 'windows':
                response = re.findall(ping_list, out.decode())[0]
                # 补齐Windows下没有的mdev字段为0
                return (response[0], response[2], response[1], "0")
            else:
                response = matcher.search(out.decode()).group().split('/')
                return tuple(response)
    except Exception as e:
        logging.error(f"ping检测失败: {e}")
        return None

def check_download_speed():
    """
    返回下载速率（单位Mbps），检测失败返回None
    """
    try:
        start_time = time.time()
        response = requests.get(DOWNLOAD_TEST_URL, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()
        duration = time.time() - start_time
        file_size = len(response.content)
        # 字节转Mbps: (字节*8)/1024/1024 / 秒
        speed_mbps = (file_size * 8) / (1024 * 1024) / duration
        return round(speed_mbps, 2)
    except Exception as e:
        logging.error(f"下载速率检测失败: {e}")
        return None

def check_upload_speed():
    """
    返回上传速率（单位Mbps），检测失败返回None
    """
    try:
        start_time = time.time()
        response = requests.post(UPLOAD_TEST_URL, data=UPLOAD_TEST_DATA, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()
        duration = time.time() - start_time
        data_size = len(UPLOAD_TEST_DATA)
        speed_mbps = (data_size * 8) / (1024 * 1024) / duration
        return round(speed_mbps, 2)
    except Exception as e:
        logging.error(f"上传速率检测失败: {e}")
        return None

if __name__ == "__main__":
    count = 0
    while True:
        # 每次都执行ping检测
        ping_res = check_ping(PING_CHECK_HOST)
        if ping_res:
            logging.info(f"ping检测结果: 最小{ping_res[0]}ms, 平均{ping_res[1]}ms, 最大{ping_res[2]}ms, 抖动{ping_res[3]}ms")
        count += 1
        # 达到间隔次数时执行速率检测
        if count >= SPEED_TEST_INTERVAL_COUNT:
            logging.info("开始执行速率检测...")
            down_speed = check_download_speed()
            up_speed = check_upload_speed()
            if down_speed:
                logging.info(f"下载速率: {down_speed} Mbps")
            if up_speed:
                logging.info(f"上传速率: {up_speed} Mbps")
            count = 0
        time.sleep(30)

低负担优化说明 #

• 速率检测仅按配置的低频执行，默认15分钟一次，单次测速总流量不超过1MB，不会造成网络卡顿
• 测试资源体积可根据需求调整，带宽紧张的场景可将测试文件缩小到100KB以内，进一步降低占用
• 可在代码中增加通话状态判断逻辑，检测到正在进行互联网通话时自动跳过速率检测，完全避免带宽抢占
• 所有请求都增加了超时控制，异常场景下不会长时间占用网络连接

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