81 lines
4.0 KiB
Python
81 lines
4.0 KiB
Python
"""Проверить сохранённые измерения VIDEO-001 без повторного кодирования."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import math
|
||
from pathlib import Path
|
||
|
||
from video_compare import FRAGMENT_BYTES, MATRIX, peak_second, radio_screen
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--source", type=Path, required=True)
|
||
parser.add_argument("results", nargs=3, type=Path)
|
||
arguments = parser.parse_args()
|
||
source_sha = hashlib.sha256(arguments.source.read_bytes()).hexdigest()
|
||
expected_modes = {(width, height, fps)
|
||
for width, height, rates in MATRIX for fps in rates}
|
||
all_cases = 0
|
||
for path in arguments.results:
|
||
document = json.loads(path.read_text(encoding="utf-8"))
|
||
assert document["status"] == "MEASURED_SOFTWARE_ONLY"
|
||
assert document["source_sha256"] == source_sha
|
||
assert document["rf_tx"] == 0
|
||
cases = document["cases"]
|
||
assert len(cases) == 2 * len(expected_modes) * 2
|
||
modes = {(case["resolution"][0], case["resolution"][1], case["fps"])
|
||
for case in cases}
|
||
assert modes == expected_modes
|
||
assert len({(tuple(case["resolution"]), case["fps"], case["window"],
|
||
case["codec"]) for case in cases}) == len(cases)
|
||
durations = {window["name"]: window["duration_seconds"]
|
||
for window in document["windows"]}
|
||
assert len(durations) == 2
|
||
for case in cases:
|
||
frames = case["frame_sizes_bytes"]
|
||
fps = case["fps"]
|
||
assert len(frames) == round(durations[case["window"]] * fps)
|
||
assert all(size > 0 for size in frames)
|
||
assert case["decoded_frame_count"] == len(frames)
|
||
assert case["source_unique_frames"] == len(frames)
|
||
assert case["unique_decoded_frames"] == len(frames)
|
||
assert case["identical_decoded_frame_count"] == 0
|
||
assert case["actual_encoded_fps"] > fps
|
||
file_latency = case["end_to_end_file_pipeline_latency_s"]
|
||
source_time = case["source_read_resize_time_per_frame_s"]
|
||
assert 0 < source_time["mean"] <= file_latency["mean"]
|
||
assert source_time["max"] <= file_latency["max"]
|
||
assert file_latency["mean"] <= file_latency["max"]
|
||
assert case["peak_process_rss_bytes"] > 0
|
||
assert math.isclose(case["average_bitrate_bps"],
|
||
8 * sum(frames) / durations[case["window"]])
|
||
assert case["peak_one_second_bitrate_bps"] == peak_second(frames, fps)
|
||
assert case["average_frame_size_bytes"] == sum(frames) / len(frames)
|
||
assert case["max_frame_size_bytes"] == max(frames)
|
||
expected_radio = radio_screen(frames, fps)
|
||
assert case["radio_screen"] == expected_radio
|
||
assert expected_radio["fits_current_radio_budget"] == \
|
||
"NO_ON_OBSERVED_BURST_SCREEN"
|
||
if case["codec"] == "JPEG_INDEPENDENT":
|
||
assert case["loss_recovery"]["first_exact_frame"] == 1
|
||
assert case["reference_frame_size_bytes"] == max(frames)
|
||
else:
|
||
assert case["codec"] == "H264_LIBX264"
|
||
assert case["loss_recovery"]["lost_reference"]["first_exact_frame"] == fps
|
||
assert case["loss_recovery"]["lost_delta"]["first_exact_frame"] == fps
|
||
assert case["reference_frame_size_bytes"] in case["keyframe_sizes_bytes"]
|
||
assert case["reference_frame_size_bytes"] == max(case["keyframe_sizes_bytes"])
|
||
assert expected_radio["max_packets_per_frame"] == max(
|
||
math.ceil(size / FRAGMENT_BYTES) for size in frames)
|
||
all_cases += len(cases)
|
||
print(f"{path}: {len(cases)} режима/участка, согласовано")
|
||
print(f"Всего проверено: {all_cases} прохода; RF TX = 0")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|