Files
SDR-Rover/experiments/lab050_video_stream.py
2026-09-08 14:15:36 +03:00

471 lines
27 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Lab050: JPEG-профили, свежесть видеопотока и блочный RX без оборудования."""
from __future__ import annotations
import argparse
from collections import Counter
from dataclasses import asdict, dataclass
import hashlib
from io import BytesIO
import json
import math
from pathlib import Path
import statistics
import sys
import time
if __package__ in (None, ""):
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import numpy as np
from PIL import Image, ImageChops
from experiments import lab046_bpsk_object_transport as transport
from experiments import lab049_bpsk_symbol_rate as phy
from experiments.lab050_streaming_receiver import StreamingReceiver
from protocol import packet as packet_protocol
from protocol.link_packet import Direction, LinkPacket, TrafficClass, HEADER_SIZE
from protocol.video_age_policy import AgePolicyPacket
from protocol.video_frame_scheduler import FramePolicy, VideoFrameGroup, schedule_video_frames
ROOT = Path(__file__).resolve().parents[1]
OUTPUT = ROOT / "data/processed/lab050"
SYMBOL_RATES = (200_000, 400_000, 500_000, 800_000, 1_000_000)
INPUT_RATES = (3, 5, 10)
AGE_LIMITS = (0.200, 0.333, 0.500, 1.000)
FRAME_INTERVALS = 5766
FRAGMENT_BYTES = transport.FRAGMENT_DATA_SIZE
@dataclass(frozen=True)
class JpegProfile:
name: str
width: int
height: int
color_mode: str
quality: int
PROFILES = (
JpegProfile("A", 320, 240, "RGB", 35),
JpegProfile("B", 160, 120, "RGB", 35),
JpegProfile("C", 128, 96, "RGB", 25),
JpegProfile("D", 160, 120, "L", 30),
)
@dataclass(frozen=True)
class EncodedImage:
image_id: int
jpeg: bytes
encode_seconds: float
decode_seconds: float
psnr_db: float
def fragment_count(jpeg_bytes: int) -> int:
if type(jpeg_bytes) is not int or jpeg_bytes <= 0:
raise ValueError("Размер JPEG должен быть положительным целым")
return (jpeg_bytes + FRAGMENT_BYTES - 1) // FRAGMENT_BYTES
def maximum_jpeg_bytes(symbol_rate: int, fps: int) -> int:
"""Точный ступенчатый предел с добивкой последнего физического кадра."""
phy.PhyMode(symbol_rate)
if type(fps) is not int or fps <= 0:
raise ValueError("Частота изображений должна быть положительным целым")
return (symbol_rate // (fps * FRAME_INTERVALS)) * FRAGMENT_BYTES
def encode_sequence(source: Image.Image, profile: JpegProfile, count: int = 30) -> list[EncodedImage]:
"""Детерминированные сдвиги фотографии; это не видео реальной камеры."""
if count < 1:
raise ValueError("Нужна непустая последовательность")
sequence = []
for index in range(count):
# Первый кадр совпадает с подготовкой Lab047. Остальные — циклический
# сдвиг исходного фото до уменьшения; амплитуда задана, не подобрана по байтам.
shifted = source if index == 0 else ImageChops.offset(source, (index * 7) % 61, (index * 3) % 31)
reference = shifted.convert(profile.color_mode).resize(
(profile.width, profile.height), Image.Resampling.LANCZOS)
started = time.perf_counter()
stream = BytesIO()
reference.save(stream, format="JPEG", quality=profile.quality, subsampling=2,
optimize=False, progressive=False)
jpeg = stream.getvalue()
encode_seconds = time.perf_counter() - started
started = time.perf_counter()
with Image.open(BytesIO(jpeg)) as image:
image.load()
decoded = np.asarray(image).copy()
decode_seconds = time.perf_counter() - started
difference = np.asarray(reference, dtype=np.float64) - decoded.astype(np.float64)
mean_square = float(np.mean(difference * difference))
psnr = 10 * math.log10(255**2 / mean_square) if mean_square else 999.0
sequence.append(EncodedImage(index + 1, jpeg, encode_seconds, decode_seconds, psnr))
return sequence
def make_frame_groups(sequence: list[EncodedImage], input_fps: int) -> list[VideoFrameGroup]:
"""Использовать существующий планировщик; LinkPacket здесь только модель времени.
В эфире остаются packet/fragment Lab046, без LinkPacket и без стирающего FEC.
Все элементы планировщика имеют 256 байт, его эквивалентная скорость ниже
выбирается так, чтобы каждый занимал ровно 5766/Rs секунд.
"""
if input_fps <= 0:
raise ValueError("Частота камеры должна быть положительной")
frames = []
order = 0
for index, image in enumerate(sequence):
capture_us = round(index / input_fps * 1_000_000)
ready_us = capture_us + math.ceil(image.encode_seconds * 1_000_000)
fragments = transport.fragments.split_image_bytes(image.jpeg, image.image_id, FRAGMENT_BYTES)
packets = []
for fragment in fragments:
payload = transport.fragments.encode_image_fragment(fragment)
payload += bytes(256 - len(payload))
model_packet = LinkPacket(TrafficClass.VIDEO, Direction.ROVER_TO_GROUND,
50, order, capture_us, 0, payload)
packets.append(AgePolicyPacket(model_packet, order, ready_us, image.image_id))
order += 1
frames.append(VideoFrameGroup(image.image_id, capture_us, tuple(packets), ready_us))
return frames
def simulate_sequence(sequence: list[EncodedImage], input_fps: int, symbol_rate: int,
max_video_age: float, *, rx_seconds_per_packet: float = 0.0,
receiver_lookahead_seconds: float = 0.0) -> dict:
"""Очередь и идеальная доставка пакетов, не измерение эфира/камеры/экрана.
Измеренная стоимость RX может быть подставлена как отдельная оценка:
последовательный обработчик пакетов, устаревший объект не допускается
к обработке. Эта оценка не заменяет живой IQ-прогон.
"""
phy.PhyMode(symbol_rate)
if not math.isfinite(rx_seconds_per_packet) or rx_seconds_per_packet < 0:
raise ValueError("Нужна конечная неотрицательная стоимость RX")
if not math.isfinite(receiver_lookahead_seconds) or receiver_lookahead_seconds < 0:
raise ValueError("Нужна конечная неотрицательная задержка буфера RX")
frames = make_frame_groups(sequence, input_fps)
image_lookup = {image.image_id: image for image in sequence}
equivalent_bitrate = (256 + HEADER_SIZE) * 8 * symbol_rate / FRAME_INTERVALS
def minimum_post_rf_latency(frame: VideoFrameGroup) -> float:
# Идеальный конвейер RF -> последовательный RX без предыдущего backlog.
# Если уже эта нижняя оценка не укладывается в возраст, TX бесполезен.
packet_time = FRAME_INTERVALS / symbol_rate
return (receiver_lookahead_seconds + rx_seconds_per_packet
+ (len(frame.packets)-1) * max(0.0, rx_seconds_per_packet-packet_time)
+ image_lookup[frame.composite_frame_id].decode_seconds)
schedule = schedule_video_frames(
frames, (), FramePolicy.LATEST_ONLY, equivalent_bitrate,
max_video_age_seconds=max_video_age,
completion_latency_seconds=minimum_post_rf_latency,
)
arrivals: dict[int, list] = {}
for sent in schedule.transmitted:
arrivals.setdefault(sent.item.composite_frame_id, []).append(sent)
frame_lookup = {frame.composite_frame_id: frame for frame in frames}
horizon = len(sequence) / input_fps
rx_cursor = 0.0
displayed = []
traces = []
received = 0
discarded_rx = 0
for frame_id, sent in arrivals.items():
frame = frame_lookup[frame_id]
image = image_lookup[frame_id]
first_tx, last_tx = sent[0].start_seconds, sent[-1].end_seconds
# Идеальный packet-level путь действительно проверяет сериализованный
# CRC и собирает JPEG; FEC/IQ проверяется отдельным интеграционным опытом.
assembler = transport.ObjectAssembler()
for item in sent:
wire = packet_protocol.build_packet(item.item.packet.payload,
phy.lab045.MESSAGE_TYPE, phy.lab045.SEQUENCE_NUMBER)
parsed = packet_protocol.parse_packet(wire)
assembler.offer(transport.recover_fragment_from_payload(parsed.payload))
if assembler.recovered_object() != image.jpeg:
raise RuntimeError("Идеальная сборка изменила JPEG")
first_rx = sent[0].end_seconds
predicted_end = rx_cursor
for item in sent:
predicted_end = max(predicted_end, item.end_seconds + receiver_lookahead_seconds) + rx_seconds_per_packet
predicted_end += image.decode_seconds
dropped = predicted_end - frame.generation_time_seconds > max_video_age + 1e-12
if dropped:
discarded_rx += 1
object_complete = decode_complete = display_time = None
else:
received += 1
rx_cursor = predicted_end
decode_complete = predicted_end
object_complete = decode_complete - image.decode_seconds
display_time = decode_complete # Только модель мгновенного экрана.
displayed.append((frame_id, display_time, display_time - frame.generation_time_seconds))
traces.append({
"image_id": frame_id, "capture_timestamp": frame.generation_time_seconds,
"encode_complete_timestamp": frame.available_time_seconds,
"queue_enter_timestamp": frame.available_time_seconds,
"first_tx_timestamp": first_tx, "last_tx_timestamp": last_tx,
"first_rx_timestamp": first_rx, "object_complete_timestamp": object_complete,
"decode_complete_timestamp": decode_complete, "display_timestamp": display_time,
"queue_age_seconds": first_tx - frame.available_time_seconds,
"rf_age_seconds": last_tx - first_tx,
"end_to_end_image_age_seconds": None if display_time is None else display_time - frame.generation_time_seconds,
"dropped_at_rx_admission": dropped,
})
dropped_before = {drop.frame.composite_frame_id: drop for drop in schedule.dropped_frames}
for frame in frames:
if frame.composite_frame_id not in dropped_before:
continue
drop = dropped_before[frame.composite_frame_id]
traces.append({"image_id": frame.composite_frame_id,
"capture_timestamp": frame.generation_time_seconds,
"encode_complete_timestamp": frame.available_time_seconds,
"queue_enter_timestamp": frame.available_time_seconds,
**{key: None for key in ("first_tx_timestamp", "last_tx_timestamp", "first_rx_timestamp",
"object_complete_timestamp", "decode_complete_timestamp", "display_timestamp",
"queue_age_seconds", "rf_age_seconds", "end_to_end_image_age_seconds")},
"drop_reason": drop.reason, "drop_timestamp": drop.drop_time_seconds})
active_received = sum(row["object_complete_timestamp"] is not None
and row["object_complete_timestamp"] <= horizon + 1e-12 for row in traces)
active_displayed = [item for item in displayed if item[1] <= horizon + 1e-12]
ages = [item[2] for item in displayed]
drops = Counter(drop.reason for drop in schedule.dropped_frames)
return {
"scope": "packet-level ideal RF model, measured JPEG CPU durations; display/camera not measured",
"input_fps": input_fps, "symbol_rate": symbol_rate, "max_video_age_seconds": max_video_age,
"input_frames": len(sequence), "transmitted_frames": len(arrivals),
"dropped_before_tx": len(schedule.dropped_frames), "drop_reasons": dict(drops),
"rf_delivered_objects": len(arrivals), "received_frames": received,
"dropped_at_rx_admission": discarded_rx,
"displayed_frames": len(displayed), "maximum_waiting_frames": schedule.maximum_waiting_frames,
"observation_seconds": horizon, "active_received_frames": active_received,
"active_displayed_frames": len(active_displayed),
"raw_received_fps": active_received / horizon,
"raw_display_fps": len(active_displayed) / horizon,
"fresh_display_fps": len(active_displayed) / horizon,
"steady_fresh_display_fps": ((len(displayed)-1)/(displayed[-1][1]-displayed[0][1])
if len(displayed) > 1 else 0.0),
"drain_displayed_frames": len(displayed) - len(active_displayed),
"image_age_min_seconds": min(ages) if ages else None,
"image_age_median_seconds": statistics.median(ages) if ages else None,
"image_age_max_seconds": max(ages) if ages else None,
"rx_cost_seconds_per_packet_modeled": rx_seconds_per_packet,
"receiver_lookahead_seconds_modeled": receiver_lookahead_seconds,
"elrs_control_in_video_queue": False,
"frames": sorted(traces, key=lambda row: row["image_id"]),
}
def benchmark_receiver(sequence: list[EncodedImage], symbol_rate: int, *, noisy: bool = True) -> dict:
"""Подать несколько JPEG частями IQ без TX-метаданных в RX, замерить CPU.
TX хранит лишь очередной физический кадр, RX — несколько блоков.
Проверка SHA производится после независимой выдачи и сборки объекта.
"""
mode = phy.PhyMode(symbol_rate)
receiver = StreamingReceiver(symbol_rate)
random_generator = np.random.default_rng(50_050)
completed = []
current_assembler = None
current_id = None
last_completed_id = -1
object_first_sample = None
assembly_decode_seconds = 0.0
retained_object_bytes_peak = 0
packets_before_finish = 0
emitted_jpegs_before_finish = 0
def consume(packets):
nonlocal current_assembler, current_id, last_completed_id, object_first_sample
nonlocal assembly_decode_seconds, retained_object_bytes_peak
started = time.perf_counter()
for packet in packets:
fragment = transport.recover_fragment_from_payload(packet.payload)
if fragment is None or fragment.original_size > 65536 or fragment.fragment_count > 269:
continue
# Для этой конечной сессии ID возрастают, без wrap/restart. Строгий
# transport assembler защищает согласованность индексов/дубликатов.
if fragment.image_id <= last_completed_id or (current_id is not None and fragment.image_id < current_id):
continue
if current_id != fragment.image_id:
current_id = fragment.image_id
current_assembler = transport.ObjectAssembler()
object_first_sample = packet.start_sample
current_assembler.offer(fragment)
retained_object_bytes_peak = max(retained_object_bytes_peak, fragment.original_size)
recovered = current_assembler.recovered_object()
if recovered is not None:
with Image.open(BytesIO(recovered)) as image:
image.load()
dimensions = image.size
completed.append({"image_id": current_id, "sha256": hashlib.sha256(recovered).hexdigest(),
"bytes": len(recovered), "dimensions": dimensions,
"first_sample": object_first_sample, "last_sample": packet.end_sample,
"emitted_after_sample": packet.emitted_after_sample})
last_completed_id = current_id
current_id = current_assembler = None
assembly_decode_seconds += time.perf_counter() - started
def push_array(values):
for offset in range(0, values.size, receiver.block_samples):
consume(receiver.push_iq_block(values[offset:offset + receiver.block_samples]))
# Ненулевые несогласованные с символом интервалы, неизвестные RX.
push_array((random_generator.normal(0, 0.2, 4321) + 1j * random_generator.normal(0, 0.2, 4321)).astype(np.complex64))
transmitted_packets = 0
air_samples = 0
for image in sequence:
for frame in transport.build_object_frames(image.jpeg, image.image_id):
cs8 = phy.build_parameterized_tx_cs8(frame.plan, mode)
components = np.frombuffer(cs8, dtype=np.int8).astype(np.float32)
values = components[0::2] + 1j * components[1::2]
air_samples += values.size
if noisy:
indexes = np.arange(values.size, dtype=np.float64)
values = values * np.exp(1j * (0.73 + 2*np.pi*10*indexes/mode.sample_rate_hz)) * (-1 if transmitted_packets % 2 else 1)
energy = float(np.sum(abs(values)**2)) / frame.plan.frame_symbol_count
sigma = math.sqrt(energy / (2 * 10**(7/10)))
values += sigma * (random_generator.normal(size=values.size) + 1j * random_generator.normal(size=values.size))
push_array(np.asarray(values, np.complex64))
transmitted_packets += 1
gap = 211 + (transmitted_packets * 137) % 1000
push_array((random_generator.normal(0, 0.2, gap) + 1j * random_generator.normal(0, 0.2, gap)).astype(np.complex64))
# Хвост — ограниченный поток фона, не вся запись. Позволяет проверить
# выдачу последнего JPEG именно из push, до finish.
left = receiver.lookahead + receiver.core_samples
while left:
count = min(left, receiver.block_samples)
push_array((random_generator.normal(0, 0.2, count) + 1j * random_generator.normal(0, 0.2, count)).astype(np.complex64))
left -= count
packets_before_finish = receiver.packet_count
emitted_jpegs_before_finish = len(completed)
consume(receiver.finish())
expected = {image.image_id: hashlib.sha256(image.jpeg).hexdigest() for image in sequence}
exact = {item["image_id"]: item["sha256"] for item in completed} == expected
processing = receiver.processing_seconds + assembly_decode_seconds
capture_seconds = receiver.total_samples / mode.sample_rate_hz
return {
"symbol_rate": symbol_rate, "samples_per_symbol": mode.samples_per_symbol,
"detection_decimation": receiver.decimation, "input_images": len(sequence),
"tx_rf_packets": transmitted_packets, "accepted_rf_packets": receiver.packet_count,
"completed_images": len(completed), "all_sha_match": exact,
"packets_before_finish": packets_before_finish, "images_before_finish": emitted_jpegs_before_finish,
"capture_seconds": capture_seconds, "payload_airtime_seconds": air_samples / mode.sample_rate_hz,
"processing_seconds": processing, "processing_per_rf_second": processing / capture_seconds,
"processing_per_payload_airtime_second": processing / (air_samples / mode.sample_rate_hz),
"coarse_seconds": receiver.coarse_seconds, "strict_seconds": receiver.strict_seconds,
"assembly_jpeg_decode_seconds": assembly_decode_seconds,
"processing_seconds_per_rf_packet": processing / transmitted_packets,
"peak_iq_buffer_samples": receiver.peak_buffer_samples,
"iq_buffer_limit_samples": receiver.buffer_limit_samples,
"working_arrays_estimate_bytes": receiver.memory_estimate_bytes + 2 * 65536,
"memory_is_process_rss_measurement": False,
"peak_object_payload_bytes": retained_object_bytes_peak,
"candidate_count": receiver.candidate_count, "peak_candidates_in_block": receiver.peak_candidate_count,
"rejected_reasons": dict(receiver.rejected_reasons),
"buffer_release_delay_upper_seconds": (receiver.lookahead + 2*receiver.core_samples) / mode.sample_rate_hz,
"noisy": noisy, "coded_bit_ebn0_db": 7 if noisy else None, "cfo_hz": 10 if noisy else 0,
"candidate_threshold": phy.lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD,
"completed": completed, "rf_tx": 0,
}
def run_functional_tests() -> None:
assert maximum_jpeg_bytes(500_000, 5) == 4148
assert maximum_jpeg_bytes(1_000_000, 10) == 4148
assert fragment_count(244) == 1 and fragment_count(245) == 2
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--reuse-benchmarks", action="store_true", help="Переиспользовать текущие локальные замеры IQ")
args = parser.parse_args()
run_functional_tests()
OUTPUT.mkdir(parents=True, exist_ok=True)
source_path = ROOT / "data/raw/lab011_scene.jpg"
with Image.open(source_path) as source:
source.load()
sequences = {profile.name: encode_sequence(source, profile) for profile in PROFILES}
control = sequences["A"][0].jpeg
if hashlib.sha256(control).hexdigest() != phy.EXPECTED_JPEG_SHA256:
raise RuntimeError("Профиль A не воспроизвёл контрольный JPEG Lab047")
summaries = []
for profile in PROFILES:
sequence = sequences[profile.name]
(OUTPUT / f"profile_{profile.name}.jpg").write_bytes(sequence[0].jpeg)
summaries.append({**asdict(profile), "first_jpeg_bytes": len(sequence[0].jpeg),
"min_jpeg_bytes": min(len(item.jpeg) for item in sequence),
"max_jpeg_bytes": max(len(item.jpeg) for item in sequence),
"first_psnr_db": sequence[0].psnr_db,
"median_encode_seconds": statistics.median(item.encode_seconds for item in sequence),
"median_decode_seconds": statistics.median(item.decode_seconds for item in sequence),
"sha256_first": hashlib.sha256(sequence[0].jpeg).hexdigest(),
"airtime_capacity": {str(rate): rate/(fragment_count(len(sequence[0].jpeg))*FRAME_INTERVALS) for rate in SYMBOL_RATES}})
print(json.dumps(summaries, ensure_ascii=False), flush=True)
benchmark_path = OUTPUT / "receiver_benchmarks.json"
if args.reuse_benchmarks:
benchmarks = json.loads(benchmark_path.read_text(encoding="utf-8"))
expected_hashes = {image.image_id: hashlib.sha256(image.jpeg).hexdigest() for image in sequences["C"][:3]}
if [row["symbol_rate"] for row in benchmarks] != list(SYMBOL_RATES):
raise ValueError("Сохранены замеры других скоростей")
for row in benchmarks:
if (row["input_images"] != 3 or row["coded_bit_ebn0_db"] != 7
or row["candidate_threshold"] != phy.lab047.FIXED_HARDWARE_CANDIDATE_THRESHOLD
or any(expected_hashes.get(item["image_id"]) != item["sha256"] for item in row["completed"])):
raise ValueError("Сохранённые замеры относятся к другому сценарию")
else:
benchmarks = []
for rate in SYMBOL_RATES:
print(f"IQ benchmark {rate}: 3 JPEG profile C", flush=True)
result = benchmark_receiver(sequences["C"][:3], rate)
benchmarks.append(result)
print(json.dumps({key: result[key] for key in ("symbol_rate", "completed_images", "all_sha_match", "processing_per_rf_second", "working_arrays_estimate_bytes")}), flush=True)
benchmark_path.write_text(json.dumps(benchmarks, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
matrix = []
receiver_estimates = []
for profile in PROFILES:
for fps in INPUT_RATES:
for rate in SYMBOL_RATES:
for age in AGE_LIMITS:
row = simulate_sequence(sequences[profile.name], fps, rate, age)
matrix.append({"profile": profile.name, **row})
for measured in benchmarks:
if not measured["all_sha_match"]:
continue
for age in AGE_LIMITS:
receiver_estimates.append({"profile": "C", **simulate_sequence(sequences["C"], 5,
measured["symbol_rate"], age, rx_seconds_per_packet=measured["processing_seconds_per_rf_packet"],
receiver_lookahead_seconds=measured["buffer_release_delay_upper_seconds"])})
result = {
"stage": "Lab050 software video stream", "lab049_commit": "cdf5396e807b83f82a264632dc46f71428fcd616",
"source_sha256": hashlib.sha256(source_path.read_bytes()).hexdigest(),
"sequence": "30 deterministic cyclic shifts of one project photo, not camera video",
"profiles": summaries,
"maximum_jpeg_bytes": {str(rate): {str(fps): maximum_jpeg_bytes(rate, fps) for fps in INPUT_RATES} for rate in SYMBOL_RATES},
"receiver_benchmarks": benchmarks, "packet_model_matrix": matrix,
"measured_rx_cost_model": receiver_estimates,
"conditional_accelerated_models": [
{"profile": "C", "is_current_receiver_performance": False,
"required_speedup_vs_measured_cost": next(row["processing_seconds_per_rf_packet"]
for row in benchmarks if row["symbol_rate"] == 500000) / cost,
**simulate_sequence(sequences["C"], fps, 500000, .2, rx_seconds_per_packet=cost,
receiver_lookahead_seconds=next(row["buffer_release_delay_upper_seconds"]
for row in benchmarks if row["symbol_rate"] == 500000))}
for fps, cost in ((5, .016), (10, .012))
],
"modeled_unknown_latencies": {"capture": "unknown, modeled zero", "display": "unknown, modeled zero",
"encode": "JPEG encode CPU measured; source preparation outside timing",
"decode": "JPEG decode CPU measured", "rf": "calculated waveform airtime"},
"elrs_control_path": "SEPARATE", "rf_tx": 0,
}
(OUTPUT / "result.json").write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
selected = next(row for row in matrix if row["profile"] == "C" and row["input_fps"] == 5 and row["symbol_rate"] == 500000 and row["max_video_age_seconds"] == .2)
print(json.dumps({key: value for key, value in selected.items() if key != "frames"}, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()