63 lines
2.6 KiB
Python
63 lines
2.6 KiB
Python
"""Проверка временной модели VIDEO-003 на настоящих малых H.264 кадрах."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
|
|
import av
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from protocol.video_stream_transport import FrameType, VideoFrame
|
|
from tools.video002_simulation import simulate
|
|
from tools.video_compare import new_h264_encoder
|
|
from tools.video003_design import estimate
|
|
|
|
|
|
def _frames() -> list[VideoFrame]:
|
|
encoder, _ = new_h264_encoder(64, 48, 5, 160_000)
|
|
output = []
|
|
for index in range(10):
|
|
pixels = np.zeros((48, 64, 3), dtype=np.uint8)
|
|
pixels[:, :, 1] = index * 17
|
|
pixels[8:20, 3 * index:3 * index + 12, 0] = 240
|
|
raw = av.VideoFrame.from_ndarray(pixels, format="rgb24").reformat(format="yuv420p")
|
|
raw.pts = index
|
|
packets = encoder.encode(raw)
|
|
assert len(packets) == 1
|
|
packet = packets[0]
|
|
output.append(VideoFrame(33, index,
|
|
FrameType.KEYFRAME if packet.is_keyframe
|
|
else FrameType.DELTA_FRAME,
|
|
200 * index, bytes(packet)))
|
|
assert not encoder.encode(None)
|
|
return output
|
|
|
|
|
|
def test_explicit_fec_airtime_controls_video_age_and_p95() -> None:
|
|
frames = _frames()
|
|
row = estimate("QPSK", 500_000, "2/3", 1012, 0)
|
|
link = ("QPSK_500_R23", "QPSK", 500_000, 1012, 0)
|
|
calculated = simulate(frames, 5, link, ("random", 0),
|
|
airtime_override_ms=row.packet_airtime_ms)
|
|
slower = simulate(frames, 5, link, ("random", 0),
|
|
airtime_override_ms=row.packet_airtime_ms * 2)
|
|
assert calculated["displayed_frames"] == 10
|
|
assert slower["displayed_frames"] == 10
|
|
assert calculated["p95_end_to_end_age_ms"] <= slower["p95_end_to_end_age_ms"]
|
|
assert calculated["p95_end_to_end_age_ms"] <= calculated["max_end_to_end_age_ms"]
|
|
assert calculated["no_old_frame_replay"] and slower["no_old_frame_replay"]
|
|
with pytest.raises(ValueError):
|
|
simulate(frames, 5, link, ("random", 0), airtime_override_ms=math.nan)
|
|
|
|
|
|
def test_design_headroom_uses_actual_video001_peaks() -> None:
|
|
from tools.video003_design import measured_video_profiles
|
|
|
|
profiles = measured_video_profiles()
|
|
assert profiles["NORMAL"]["measured_peak_1s_bps"] == pytest.approx(375_776, abs=200)
|
|
assert profiles["HIGH_QUALITY"]["measured_peak_1s_bps"] == pytest.approx(455_600, abs=200)
|
|
for profile in profiles.values():
|
|
assert profile["required_20pct_bps"] == pytest.approx(
|
|
profile["measured_peak_1s_bps"] * 1.2)
|