114 lines
5.2 KiB
Python
114 lines
5.2 KiB
Python
"""VIDEO-001: лёгкий повтор сохранённого IQ через штатный приёмник Lab053.
|
||
|
||
Не обращается к SDR. Внутрь RX передаются только последовательные блоки IQ.
|
||
Ожидаемые SHA читаются лишь после строгого PHY/FEC/CRC и сборки JPEG.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
|
||
import numpy as np
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
||
|
||
SAMPLE_RATE = 8_000_000
|
||
BLOCK_SAMPLES = 65_536
|
||
|
||
|
||
def replay(source: Path) -> dict:
|
||
from experiments.lab051_receiver_performance import ImageSink
|
||
from experiments.lab053_streaming_receiver import StreamingReceiver
|
||
|
||
receiver = StreamingReceiver(workers=1, candidate_capacity=4)
|
||
startup = dict(receiver.processor.startup)
|
||
sink = ImageSink()
|
||
images: list[dict] = []
|
||
bytes_read = 0
|
||
start = time.perf_counter()
|
||
|
||
def consume(packets) -> None:
|
||
for packet in packets:
|
||
item = sink.offer(packet, time.perf_counter() - start)
|
||
if item is not None:
|
||
images.append({"image_id": item["image_id"],
|
||
"sha256": hashlib.sha256(item["jpeg"]).hexdigest(),
|
||
"dimensions": item["dimensions"]})
|
||
|
||
try:
|
||
with source.open("rb") as stream:
|
||
while raw := stream.read(BLOCK_SAMPLES * 8):
|
||
if len(raw) % 8:
|
||
raise ValueError("IQ содержит неполный complex64 отсчёт")
|
||
bytes_read += len(raw)
|
||
consume(receiver.push_iq_block(np.frombuffer(raw, dtype=np.complex64)))
|
||
consume(receiver.finish())
|
||
elapsed = time.perf_counter() - start
|
||
packet_count = receiver.packet_count
|
||
candidate_count = receiver.candidate_count
|
||
finally:
|
||
receiver.close()
|
||
iq_seconds = bytes_read / 8 / SAMPLE_RATE
|
||
return {"startup": startup, "iq_bytes": bytes_read, "iq_seconds": iq_seconds,
|
||
"replay_wall_seconds": elapsed,
|
||
"replay_wall_over_iq_seconds": elapsed / iq_seconds,
|
||
"iq_seconds_over_replay_wall": iq_seconds / elapsed,
|
||
"total_with_jit_warmup_seconds": elapsed + startup["warmup_seconds"],
|
||
"accepted_packets": packet_count, "strict_candidates": candidate_count,
|
||
"completed_images": len(images), "images": images}
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--source", required=True, type=Path)
|
||
parser.add_argument("--output", required=True, type=Path)
|
||
parser.add_argument("--numba-packages", required=True, type=Path)
|
||
parser.add_argument("--fixture-json", required=True, type=Path)
|
||
arguments = parser.parse_args()
|
||
if not arguments.source.is_file() or not arguments.fixture_json.is_file():
|
||
parser.error("IQ или зафиксированная ведомость JPEG отсутствует")
|
||
if not arguments.numba_packages.is_dir():
|
||
parser.error("Локальные пакеты Numba отсутствуют")
|
||
if arguments.source.stat().st_size % 8:
|
||
parser.error("IQ содержит неполный complex64 отсчёт")
|
||
sys.path.insert(0, str(arguments.numba_packages))
|
||
fixture = json.loads(arguments.fixture_json.read_text(encoding="utf-8"))
|
||
expected = {item["image_id"]: item["sha256"] for item in fixture["images"]}
|
||
with tempfile.TemporaryDirectory(prefix="video001_numba_") as cache:
|
||
os.environ["NUMBA_CACHE_DIR"] = cache
|
||
first = replay(arguments.source)
|
||
repeat = replay(arguments.source)
|
||
for result in (first, repeat):
|
||
result["exact_images_after_verdict"] = sum(
|
||
expected.get(item["image_id"]) == item["sha256"] for item in result["images"]
|
||
)
|
||
result["all_expected_images_exact"] = (
|
||
result["completed_images"] == len(expected)
|
||
and result["exact_images_after_verdict"] == len(expected)
|
||
)
|
||
result.pop("images")
|
||
document = {"status": "MEASURED_SAVED_IQ_NO_SDR", "source": str(arguments.source),
|
||
"source_bytes": arguments.source.stat().st_size,
|
||
"comparison": "Первый и повторный проходы в одном процессе; пустой JIT-кеш в начале",
|
||
"path_scope": "Время включает чтение файла, RX, сборку и JPEG decode; без графиков, послеблочной истории и сравнения SHA внутри измерения",
|
||
"first": first, "repeat": repeat, "rf_tx": 0}
|
||
arguments.output.parent.mkdir(parents=True, exist_ok=True)
|
||
arguments.output.write_text(json.dumps(document, ensure_ascii=False, indent=2) + "\n",
|
||
encoding="utf-8")
|
||
print(json.dumps({label: {name: value[name] for name in ("startup", "replay_wall_seconds",
|
||
"iq_seconds_over_replay_wall", "accepted_packets", "completed_images",
|
||
"all_expected_images_exact")} for label, value in (("first", first), ("repeat", repeat))},
|
||
ensure_ascii=False), flush=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|