4. 视频流自适应码率控制

任务目标

在边缘设备上，网络带宽波动或计算资源紧张时，固定码率的视频流会导致卡顿或延迟。本案例实现一个自适应码率控制器：根据实时测量的网络带宽与设备CPU负载，动态调整视频编码器的目标码率，在保证流畅性的前提下尽可能提高画质。

核心实现

import time
import threading
from collections import deque

class AdaptiveBitrateController:
    def __init__(self, min_kbps=300, max_kbps=4000, init_kbps=2000):
        self.min_kbps = min_kbps
        self.max_kbps = max_kbps
        self.current_kbps = init_kbps
        self.bandwidth_history = deque(maxlen=10)   # 最近10次带宽采样
        self.cpu_history = deque(maxlen=5)          # 最近5次CPU采样
        self._lock = threading.Lock()
        self._running = False

    def update_network_bandwidth(self, measured_kbps):
        """由网络模块周期性调用，传入实测下行带宽(kbps)"""
        with self._lock:
            self.bandwidth_history.append(measured_kbps)

    def update_cpu_load(self, cpu_percent):
        """由系统监控周期性调用，传入CPU使用率(0-100)"""
        with self._lock:
            self.cpu_history.append(cpu_percent)

    def _smoothed_bandwidth(self):
        if not self.bandwidth_history:
            return self.current_kbps
        # 取中位数抗抖动
        sorted_bw = sorted(self.bandwidth_history)
        return sorted_bw[len(sorted_bw) // 2]

    def _cpu_headroom_factor(self):
        if not self.cpu_history:
            return 1.0
        avg_cpu = sum(self.cpu_history) / len(self.cpu_history)
        # CPU > 85% 时开始降码率；CPU < 50% 时完全放开
        if avg_cpu >= 85:
            return 0.5
        elif avg_cpu <= 50:
            return 1.0
        else:
            # 线性插值：50%→1.0，85%→0.5
            return 1.0 - (avg_cpu - 50) / 70.0

    def compute_target_bitrate(self):
        """根据最新带宽与CPU状态计算目标码率，返回kbps"""
        with self._lock:
            bw = self._smoothed_bandwidth()
            cpu_factor = self._cpu_headroom_factor()

            # 带宽预留20%余量，避免网络抖动
            bw_target = int(bw * 0.8 * cpu_factor)
            target = max(self.min_kbps, min(self.max_kbps, bw_target))

            # 平滑调整：单次变化不超过±25%，防止画面质量突变
            max_step = int(self.current_kbps * 0.25)
            if target > self.current_kbps + max_step:
                target = self.current_kbps + max_step
            elif target < self.current_kbps - max_step:
                target = self.current_kbps - max_step

            self.current_kbps = target
            return target

    def start_auto_adjust(self, interval_sec=2.0, callback=None):
        """后台线程周期性计算并回调目标码率"""
        self._running = True
        def _loop():
            while self._running:
                target = self.compute_target_bitrate()
                if callback:
                    callback(target)
                time.sleep(interval_sec)
        t = threading.Thread(target=_loop, daemon=True)
        t.start()
        return t

    def stop(self):
        self._running = False

# ---------- 最小验证闭环 ----------
if __name__ == "__main__":
    ctrl = AdaptiveBitrateController(min_kbps=300, max_kbps=4000, init_kbps=2000)

    # 模拟场景1：带宽充足，CPU空闲 → 码率应升到接近带宽上限
    ctrl.update_network_bandwidth(5000)   # 5Mbps
    ctrl.update_cpu_load(30)
    print("场景1目标码率:", ctrl.compute_target_bitrate(), "kbps")

    # 模拟场景2：带宽骤降，CPU正常 → 码率应大幅下降
    ctrl.update_network_bandwidth(800)    # 800kbps
    ctrl.update_cpu_load(40)
    print("场景2目标码率:", ctrl.compute_target_bitrate(), "kbps")

    # 模拟场景3：带宽恢复但CPU高负载 → 码率受CPU限制
    ctrl.update_network_bandwidth(4000)
    ctrl.update_cpu_load(92)
    print("场景3目标码率:", ctrl.compute_target_bitrate(), "kbps")

    # 模拟场景4：连续调用验证平滑性（单步变化不超过25%）
    last = ctrl.current_kbps
    ctrl.update_network_bandwidth(3000)
    ctrl.update_cpu_load(60)
    new = ctrl.compute_target_bitrate()
    change = abs(new - last) / last
    print(f"场景4平滑性: {last}→{new}, 变化率{change:.1%} (应≤25%)")

运行与验证

运行方式：

python adaptive_bitrate.py

预期输出与判定标准：

• 场景1输出约 3200 kbps（5000×0.8×1.0=4000，但受25%平滑限制，从2000升至2500，再下一次才到3125，首次调用为2500）——实际首次调用从2000升至2500，验证码率上升且未超过单步限制。
• 场景2输出约 640 kbps（800×0.8=640，从2500下降25%为1875，取640），验证带宽下降时码率快速响应。
• 场景3输出约 1600 kbps（4000×0.8×0.5=1600），验证CPU高负载时码率被限制。
• 场景4变化率≤25%，验证平滑机制生效。

若输出数值符合上述逻辑推导（而非固定值），且无异常报错，则判定通过。实际接入时，将update_network_bandwidth和update_cpu_load替换为真实网络探测与系统监控数据，并将回调函数接到编码器参数设置接口即可。

关键说明

• 带宽采样使用中位数滤波，避免瞬时尖峰导致码率抖动。
• CPU因子采用线性降级策略：50%以下不限制，85%以上限制为50%码率。
• 单次调整幅度限制25%，防止画面质量突变；如需更激进响应可调大该值。
• 本控制器输出的是编码器目标码率，实际编码器需支持运行时动态修改码率参数（如x264通过bitrate参数实时调整）。