WebRTC、会话状态与延迟预算

New-Item -ItemType Directory -Force l9-realtime
cd l9-realtime

mkdir -p l9-realtime
cd l9-realtime

BUFFER_MS = 40
ARRIVALS = [(0, 0), (2, 10), (1, 30)]
STAGES = {"端点等待": 100, "网络": 40, "生成": 180}

def playback(arrivals, buffer_ms):
    if buffer_ms < 0:
        raise ValueError("缓冲时间不能为负数")
    if len({seq for seq, _ in arrivals}) != len(arrivals):
        raise ValueError("音频包序号重复")
    if any(seq < 0 or stamp < 0 for seq, stamp in arrivals):
        raise ValueError("序号和到达时间不能为负数")
    # 固定每包20ms，按序号安排播放时刻；超过时刻的包计为迟到。
    ordered = sorted(arrivals)
    if [seq for seq, _ in ordered] != list(range(len(ordered))):
        raise ValueError("缺少音频包")
    result = []
    for seq, arrival_ms in ordered:
        scheduled = buffer_ms + seq * 20
        result.append((seq, scheduled, "播放" if arrival_ms <= scheduled else "迟到"))
    return result


class Session:
    NAMES = {"new": "连接", "listening": "收音", "generating": "生成",
             "playing": "播放", "closed": "关闭"}

    def __init__(self):
        self.state = "new"
        self.trace = ["连接"]
        self.current = None
        self.counter = 0
        self.queue = []

    def event(self, action):
        # 控制事件合法迁移；取消与断线都会清理尚未播放的媒体包。
        if action == "disconnect" and self.state != "closed":
            self.queue.clear()
            self.current = None
            self.state = "closed"
        elif action == "connect" and self.state == "new":
            self.state = "listening"
        elif action == "commit" and self.state == "listening":
            self.counter += 1
            self.current = "r" + str(self.counter)
            self.state = "generating"
        elif action == "cancel" and self.state in ("generating", "playing"):
            self.queue.clear()
            self.current = None
            self.state = "listening"
        else:
            raise ValueError("当前状态不接受事件：" + action)
        self.trace.append(self.NAMES[self.state])
        return self.current

    def audio(self, request_id, packet):
        # 旧响应尾包不进入新会话；合法包让生成状态进入播放状态。
        if request_id != self.current or self.state not in ("generating", "playing"):
            return False
        self.queue.append(packet)
        if self.state == "generating":
            self.state = "playing"
            self.trace.append("播放")
        return True

if __name__ == "__main__":
    packets = playback(ARRIVALS, BUFFER_MS)
    session = Session()
    session.event("connect")
    request = session.event("commit")
    for packet in ARRIVALS:
        session.audio(request, packet)
    print("会话：" + " → ".join(session.trace))
    print("包序号：" + ",".join(str(seq) for seq, _, _ in packets))
    print("迟到包：" + str(sum(state == "迟到" for _, _, state in packets)))
    print("首音预算：" + str(sum(STAGES.values()) + BUFFER_MS) + "ms")
    session.event("cancel")
    print("取消后：" + session.NAMES[session.state] + "，待播包=" + str(len(session.queue)))
    print("旧响应尾包：" + str(session.audio(request, (3, 90))))
    session.event("disconnect")
    print("断线后：" + session.NAMES[session.state])

import unittest
import session_lab as lab


class MechanismTests(unittest.TestCase):
    # 正常、边界和失败输入都由这里提供，直接运行即可。

    def test_normal_reorder(self):
        self.assertEqual([x[0] for x in lab.playback(lab.ARRIVALS, 40)], [0, 1, 2])
        session = lab.Session()
        session.event("connect")
        r1 = session.event("commit")
        self.assertTrue(session.audio(r1, (0, 0)))
        session.event("cancel")
        r2 = session.event("commit")
        self.assertFalse(session.audio(r1, (1, 20)))
        self.assertTrue(session.audio(r2, (0, 0)))
        self.assertEqual(session.trace, ["连接", "收音", "生成", "播放", "收音", "生成", "播放"])

    def test_boundary_deadline(self):
        self.assertEqual(lab.playback([(0, 40)], 40)[0][2], "播放")
        self.assertEqual(lab.playback([], 0), [])
        session = lab.Session()
        session.event("connect")
        r1 = session.event("commit")
        session.audio(r1, (0, 0))
        session.event("disconnect")
        self.assertEqual(session.queue, [])
        self.assertFalse(session.audio(r1, (1, 20)))

    def test_failure_late_and_missing(self):
        self.assertEqual(lab.playback([(0, 41)], 40)[0][2], "迟到")
        with self.assertRaises(ValueError):
            lab.playback([(1, 0)], 40)
        session = lab.Session()
        with self.assertRaises(ValueError):
            session.event("commit")
        session.event("disconnect")
        with self.assertRaises(ValueError):
            session.event("connect")


if __name__ == "__main__":
    unittest.main()

# 在刚才创建的练习目录检查文件，再运行程序。
ls
python -X utf8 session_lab.py
python -X utf8 -m unittest -v test_session_lab.py

会话：连接 → 收音 → 生成 → 播放
包序号：0,1,2
迟到包：0
首音预算：360ms
取消后：收音，待播包=0
旧响应尾包：False
断线后：关闭

python -X utf8 session_lab.py