290 lines
14 KiB
Python
290 lines
14 KiB
Python
"""VIDEO-002: воспроизводимый файловый прогон H.264 через фрагменты и потери."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
from fractions import Fraction
|
||
import hashlib
|
||
import json
|
||
import math
|
||
from pathlib import Path
|
||
import random
|
||
|
||
import av
|
||
|
||
from protocol.video_stream_transport import (
|
||
BoundedFrameQueue, FrameReassembler, FrameType, VideoFrame,
|
||
encode_fragment,
|
||
)
|
||
from tools.video002_radio_budget import estimate, matrix, margin_fits
|
||
from tools.video_compare import new_h264_encoder, source_frames
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
SOURCE = ROOT / "data/raw/video001/udacity_project_video.mp4"
|
||
RESULT_160 = ROOT / "data/processed/video001/comparison_traffic_160k.json"
|
||
RESULT_320 = ROOT / "data/processed/video001/comparison_traffic_320k.json"
|
||
DEFAULT_OUTPUT = ROOT / "data/processed/video002/results.json"
|
||
SEED = 20260916
|
||
SCENARIOS = tuple([("random", value) for value in (0, 1, 2, 5, 10, 20)]
|
||
+ [("burst", value) for value in (2, 5, 10)]
|
||
+ [("keyframe", 1), ("delta", 1)])
|
||
MODES = (("ROBUST", 320, 240, 5, 160_000),
|
||
("NORMAL", 640, 360, 10, 320_000),
|
||
("HIGH_QUALITY", 854, 480, 10, 320_000))
|
||
BUDGET_ONLY_MODES = (("640x480_10FPS", 640, 480, 10, 320_000),
|
||
("1280x720_5FPS", 1280, 720, 5, 320_000))
|
||
LINKS = (("CURRENT_BPSK_256_GAP5", "BPSK", 500_000, 256, 5),
|
||
("CALCULATED_QPSK_1024_GAP0", "QPSK", 500_000, 1024, 0),
|
||
("SOFTWARE_2MBPS", None, 2_000_000, 1024, 0))
|
||
|
||
|
||
def _result_case(results: dict, window: str, width: int, height: int,
|
||
fps: int) -> dict:
|
||
matches = [case for case in results["cases"]
|
||
if case["window"] == window and case["resolution"] == [width, height]
|
||
and case["fps"] == fps and case["codec"] == "H264_LIBX264"]
|
||
if len(matches) != 1:
|
||
raise ValueError("Не найдена единственная исходная серия VIDEO-001")
|
||
return matches[0]
|
||
|
||
|
||
def load_real_stream(source: Path, window: dict, width: int, height: int,
|
||
fps: int, maxrate_bps: int, expected_sizes: list[int]) -> list[VideoFrame]:
|
||
"""Повторить кодирование VIDEO-001 и сверить каждый размер с исходным JSON."""
|
||
frames = source_frames(source, width, height, fps,
|
||
window["start_seconds"], window["duration_seconds"])
|
||
encoder, _ = new_h264_encoder(width, height, fps, maxrate_bps)
|
||
encoded: list[VideoFrame] = []
|
||
for index, pixels in enumerate(frames):
|
||
frame = av.VideoFrame.from_ndarray(pixels, format="rgb24").reformat(format="yuv420p")
|
||
frame.pts = index
|
||
packets = encoder.encode(frame)
|
||
if len(packets) != 1:
|
||
raise ValueError("Кодировщик не дал один пакет на кадр")
|
||
packet = packets[0]
|
||
encoded.append(VideoFrame(0x56303032, index,
|
||
FrameType.KEYFRAME if packet.is_keyframe
|
||
else FrameType.DELTA_FRAME,
|
||
round(index * 1000 / fps), bytes(packet)))
|
||
if encoder.encode(None):
|
||
raise ValueError("Неожиданные отложенные пакеты кодировщика")
|
||
actual_sizes = [len(frame.data) for frame in encoded]
|
||
if actual_sizes != expected_sizes:
|
||
raise ValueError("Поток не совпал с VIDEO-001 по размерам кадров")
|
||
return encoded
|
||
|
||
|
||
def _drop(kind: str, level: int, packet_index: int, frame_id: int,
|
||
fragment_index: int, fps: int, random_generator: random.Random,
|
||
burst_start: int | None) -> bool:
|
||
if kind == "random":
|
||
return random_generator.random() < level / 100
|
||
if kind == "burst":
|
||
return burst_start is not None and burst_start <= packet_index < burst_start + level
|
||
target_id = fps if kind == "keyframe" else fps + 1
|
||
return frame_id == target_id and fragment_index == 0
|
||
|
||
|
||
def simulate(frames: list[VideoFrame], fps: int, link: tuple,
|
||
scenario: tuple[str, int], seed: int = SEED,
|
||
max_age_ms: int = 300,
|
||
airtime_override_ms: float | None = None) -> dict:
|
||
"""Дискретные моменты захвата и завершения пакетов, затем настоящий H.264 decode."""
|
||
link_name, modulation, speed, payload, gap = link
|
||
kind, level = scenario
|
||
if airtime_override_ms is not None:
|
||
if not math.isfinite(airtime_override_ms) or airtime_override_ms <= 0:
|
||
raise ValueError("эфирное время должно быть положительным")
|
||
airtime_ms = airtime_override_ms
|
||
elif modulation:
|
||
airtime_ms = estimate(modulation, speed, payload, gap).airtime_ms
|
||
else:
|
||
# Контрольная программная линия: 2 Мбит/с полезной PHY-нагрузки без RF.
|
||
airtime_ms = payload * 8000 / speed
|
||
queue = BoundedFrameQueue(payload, max_frames=2, max_age_ms=max_age_ms)
|
||
reassembler = FrameReassembler(frames[0].stream_id, payload,
|
||
max_pending_frames=2, max_age_ms=max_age_ms)
|
||
decoder = av.CodecContext.create("h264", "r")
|
||
decoder.open()
|
||
random_generator = random.Random(seed)
|
||
next_arrival = 0
|
||
next_ready = 0.0
|
||
packet_index = 0
|
||
burst_start: int | None = None
|
||
lost_packets = 0
|
||
lost_frame_ids: set[int] = set()
|
||
displayed: list[tuple[int, float, str]] = []
|
||
decoder_errors = 0
|
||
synchronized = False
|
||
last_displayed_id: int | None = None
|
||
until_ms = frames[-1].capture_ms + max_age_ms
|
||
while next_arrival < len(frames) or queue.depth:
|
||
arrival_ms = (frames[next_arrival].capture_ms
|
||
if next_arrival < len(frames) else math.inf)
|
||
if arrival_ms <= next_ready or not queue.depth:
|
||
if next_arrival >= len(frames):
|
||
break
|
||
queue.enqueue(frames[next_arrival], arrival_ms)
|
||
next_ready = max(next_ready, float(arrival_ms))
|
||
next_arrival += 1
|
||
continue
|
||
start_ms = next_ready
|
||
end_ms = start_ms + airtime_ms
|
||
fragment = queue.next_fragment(math.floor(start_ms), end_ms)
|
||
if fragment is None:
|
||
continue
|
||
wire = encode_fragment(fragment, payload)
|
||
if kind == "burst" and burst_start is None:
|
||
burst_start = packet_index
|
||
packet_lost = _drop(kind, level, packet_index, fragment.frame_id,
|
||
fragment.fragment_index, fps, random_generator,
|
||
burst_start)
|
||
queue.mark_sent()
|
||
packet_index += 1
|
||
next_ready = end_ms + gap
|
||
if packet_lost:
|
||
lost_packets += 1
|
||
lost_frame_ids.add(fragment.frame_id)
|
||
continue
|
||
complete = reassembler.offer(wire, math.ceil(end_ms))
|
||
if complete is None:
|
||
continue
|
||
if complete.frame_type == FrameType.KEYFRAME:
|
||
decoder = av.CodecContext.create("h264", "r")
|
||
decoder.open()
|
||
synchronized = True
|
||
elif last_displayed_id is None or complete.frame_id != last_displayed_id + 1:
|
||
synchronized = False
|
||
if not synchronized:
|
||
continue
|
||
encoded_packet = av.Packet(complete.data)
|
||
encoded_packet.pts = encoded_packet.dts = complete.frame_id
|
||
encoded_packet.time_base = Fraction(1, fps)
|
||
try:
|
||
decoded = decoder.decode(encoded_packet)
|
||
except av.error.FFmpegError:
|
||
decoder_errors += 1
|
||
synchronized = False
|
||
continue
|
||
if len(decoded) != 1 or int(decoded[0].pts) != complete.frame_id:
|
||
decoder_errors += 1
|
||
synchronized = False
|
||
continue
|
||
digest = hashlib.sha256(decoded[0].to_ndarray(format="rgb24").tobytes()).hexdigest()
|
||
if last_displayed_id is not None and complete.frame_id <= last_displayed_id:
|
||
raise AssertionError("Повторный показ старого кадра")
|
||
displayed.append((complete.frame_id, end_ms - frames[complete.frame_id].capture_ms,
|
||
digest))
|
||
last_displayed_id = complete.frame_id
|
||
reassembler.expire(math.ceil(until_ms + 1))
|
||
displayed_ids = {item[0] for item in displayed}
|
||
if len(displayed_ids) != len(displayed):
|
||
raise AssertionError("Повторный номер показанного кадра")
|
||
lost_ids = sorted(lost_frame_ids)
|
||
recovery_ms = None
|
||
recovered_keyframe = None
|
||
if lost_ids:
|
||
first_lost = lost_ids[0]
|
||
next_keys = [frame_id for frame_id, age, _ in displayed
|
||
if frame_id > first_lost
|
||
and frames[frame_id].frame_type == FrameType.KEYFRAME]
|
||
if next_keys:
|
||
recovered_keyframe = next_keys[0]
|
||
shown = next(item for item in displayed if item[0] == recovered_keyframe)
|
||
recovery_ms = shown[1] + frames[recovered_keyframe].capture_ms - frames[first_lost].capture_ms
|
||
ages = [item[1] for item in displayed]
|
||
p95_age = (sorted(ages)[math.ceil(0.95 * len(ages)) - 1]
|
||
if ages else None)
|
||
return {
|
||
"link": link_name, "scenario": kind, "level": level, "seed": seed,
|
||
"input_frames": len(frames), "packet_attempts": packet_index,
|
||
"lost_packets": lost_packets, "affected_frames": len(lost_frame_ids),
|
||
"displayed_frames": len(displayed),
|
||
"displayed_fps": len(displayed) / (len(frames) / fps),
|
||
"frame_loss_fraction": 1 - len(displayed) / len(frames),
|
||
"displayed_frame_ids": [item[0] for item in displayed],
|
||
"keyframe_sha256_by_id": {
|
||
str(frame_id): digest for frame_id, _, digest in displayed
|
||
if frames[frame_id].frame_type == FrameType.KEYFRAME},
|
||
"unique_decoded_images": len({item[2] for item in displayed}),
|
||
"no_old_frame_replay": len(displayed_ids) == len(displayed),
|
||
"mean_end_to_end_age_ms": sum(ages) / len(ages) if ages else None,
|
||
"p95_end_to_end_age_ms": p95_age,
|
||
"max_end_to_end_age_ms": max(ages) if ages else None,
|
||
"recovery_ms_after_first_lost_fragment": recovery_ms,
|
||
"recovered_keyframe_id": recovered_keyframe,
|
||
"max_queue_depth": queue.max_depth,
|
||
"max_oldest_frame_age_ms": queue.max_oldest_age_ms,
|
||
"dropped_stale_frames": queue.stale_frame_drops,
|
||
"dropped_capacity_frames": queue.capacity_frame_drops,
|
||
"incomplete_expired_frames": reassembler.expired_frames,
|
||
"decoder_errors": decoder_errors,
|
||
}
|
||
|
||
|
||
def run(source: Path = SOURCE) -> dict:
|
||
source_digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
||
results_160 = json.loads(RESULT_160.read_text(encoding="utf-8"))
|
||
results_320 = json.loads(RESULT_320.read_text(encoding="utf-8"))
|
||
if source_digest != results_160["source_sha256"] or source_digest != results_320["source_sha256"]:
|
||
raise ValueError("SHA-256 ролика не совпадает с VIDEO-001")
|
||
cases = []
|
||
for mode, width, height, fps, maxrate in MODES:
|
||
source_results = results_160 if maxrate == 160_000 else results_320
|
||
for window in source_results["windows"]:
|
||
original = _result_case(source_results, window["name"], width, height, fps)
|
||
frames = load_real_stream(source, window, width, height, fps, maxrate,
|
||
original["frame_sizes_bytes"])
|
||
window_cases = []
|
||
for link in LINKS:
|
||
for scenario in SCENARIOS:
|
||
measured = simulate(frames, fps, link, scenario)
|
||
measured.update({"mode": mode, "window": window["name"],
|
||
"resolution": [width, height], "fps": fps,
|
||
"original_average_bitrate_bps": original["average_bitrate_bps"],
|
||
"original_peak_bitrate_bps": original["peak_one_second_bitrate_bps"]})
|
||
window_cases.append(measured)
|
||
reference = next(item for item in window_cases
|
||
if item["link"] == "SOFTWARE_2MBPS"
|
||
and item["scenario"] == "random" and item["level"] == 0)
|
||
for item in window_cases:
|
||
for frame_id, digest in item["keyframe_sha256_by_id"].items():
|
||
if digest != reference["keyframe_sha256_by_id"][frame_id]:
|
||
raise AssertionError("Опорный кадр после потери отличается от чистого декодирования")
|
||
cases.extend(window_cases)
|
||
peaks = {mode: max(_result_case(results_160 if rate == 160_000 else results_320,
|
||
window["name"], width, height, fps)["peak_one_second_bitrate_bps"]
|
||
for window in results_320["windows"])
|
||
for mode, width, height, fps, rate in MODES + BUDGET_ONLY_MODES}
|
||
budgets = []
|
||
for row in matrix():
|
||
item = row.as_dict()
|
||
item["fits_peak_with_20pct_margin"] = {
|
||
name: margin_fits(row.useful_video_bps, peak, 0.2)
|
||
for name, peak in peaks.items()}
|
||
item["fits_peak_with_50pct_margin"] = {
|
||
name: margin_fits(row.useful_video_bps, peak, 0.5)
|
||
for name, peak in peaks.items()}
|
||
budgets.append(item)
|
||
return {"source_sha256": source_digest, "seed": SEED,
|
||
"max_age_ms": 300, "sender_queue_frames": 2,
|
||
"receiver_pending_frames": 2, "peak_bitrate_bps": peaks,
|
||
"radio_estimates": budgets, "simulations": cases,
|
||
"rf_tx": 0}
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--source", type=Path, default=SOURCE)
|
||
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
|
||
args = parser.parse_args()
|
||
result = run(args.source)
|
||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(f"VIDEO-002: {len(result['simulations'])} прогонов; {args.output}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|