Files
SDR-Rover/experiments/lab049_bpsk_symbol_rate.py
2026-09-08 13:26:06 +03:00

1173 lines
45 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.
"""Lab049: программное сравнение символьных скоростей BPSK.
Форматы PRBS7, PHY, свёрточного FEC, CRC и фрагментов остаются теми же,
что в Lab045--Lab047. Меняются только символьная скорость и точное целое
число выборок на символ при неизменной частоте дискретизации 8 Мвыб/с.
Основной проход использует полный IQ-поток без переданных границ кадров,
битов, нагрузки или ожидаемого числа кадров. Сравнение с переданными данными
выполняется только после независимого решения PHY/FEC/CRC и сборки JPEG.
"""
from __future__ import annotations
import argparse
from dataclasses import dataclass, replace
from functools import lru_cache
import hashlib
import json
import math
from pathlib import Path
import struct
import sys
import time
if __package__ in (None, ""):
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import numpy as np
from scipy.signal import oaconvolve
from experiments import lab043_pluto_to_rtlsdr as shaping
from experiments import lab045_hackrf_bpsk_fec as lab045
from experiments import lab046_bpsk_object_transport as lab046
from experiments import lab047_jpeg_object_transport as lab047
from protocol import bpsk_radio as radio
from protocol.control_messages import (
ControlStateMessage,
EmergencyStop,
MessageContext,
encode_message,
)
from protocol.link_packet import (
HEADER_SIZE,
Direction,
LinkPacket,
TrafficClass,
encode_link_packet,
)
from protocol.priority_scheduler import (
QueuedPacket,
SchedulerMode,
schedule_packets,
)
SAMPLE_RATE_HZ = 8_000_000
MANDATORY_SYMBOL_RATES = (20_000, 50_000, 100_000)
OPTIONAL_SYMBOL_RATE = 200_000
TESTED_SYMBOL_RATES = (*MANDATORY_SYMBOL_RATES, OPTIONAL_SYMBOL_RATE)
RRC_ROLLOFF = 0.35
NOISY_CODED_BIT_EBN0_DB = 7.0
NOISE_BASE_SEED = 49_000
JPEG_PATH = Path("data/processed/lab047/tx_image.jpg")
EXPECTED_JPEG_BYTES = 8_724
EXPECTED_JPEG_SHA256 = (
"e913dab9ac32795250880a4be67fbbbf89a4f96536842476ab591d1acaafc02a"
)
OUTPUT_DIRECTORY = Path("data/processed/lab049")
RESULT_NAME = "result.json"
CONTROL_PERIOD_SECONDS = 0.050
TELEMETRY_PERIOD_SECONDS = 0.100
SCHEDULER_SOURCE_SECONDS = 2.0
EMERGENCY_TIME_SECONDS = 0.777
VIDEO_BACKLOG_PACKETS = 256
TELEMETRY_APPLICATION_BYTES = 64
@dataclass(frozen=True)
class PhyMode:
"""Одна точная программная конфигурация частоты символов."""
symbol_rate: int
sample_rate_hz: int = SAMPLE_RATE_HZ
def __post_init__(self) -> None:
if any(isinstance(value, bool) or not isinstance(value, (int, np.integer))
for value in (self.symbol_rate, self.sample_rate_hz)):
raise ValueError("Частоты должны быть целыми, без скрытого округления")
if self.symbol_rate <= 0:
raise ValueError("Символьная скорость должна быть положительной")
if self.sample_rate_hz <= 0:
raise ValueError("Частота дискретизации должна быть положительной")
if self.sample_rate_hz % self.symbol_rate != 0:
raise ValueError("Число выборок на символ должно быть целым")
@property
def samples_per_symbol(self) -> int:
return self.sample_rate_hz // self.symbol_rate
@dataclass(frozen=True)
class IqFrameReception:
"""Независимый результат одного выровненного физического кадра."""
phy_header_pass: bool
fec_pass: bool
crc: str
payload_bytes: bytes | None
post_preamble_soft: np.ndarray
timing_offset_samples: int
reception: lab045.IndependentReception
@dataclass
class _RoundtripAccumulator:
assembler: lab046.ObjectAssembler
processing_seconds: float = 0.0
phy_header_pass: int = 0
phy_header_fail: int = 0
fec_pass: int = 0
fec_fail: int = 0
crc_pass: int = 0
crc_fail: int = 0
crc_not_reached: int = 0
pre_fec_errors: int = 0
post_fec_errors: int = 0
noise_sigmas: list[float] | None = None
def phy_mode(symbol_rate: int) -> PhyMode:
"""Вернуть режим только при точном целом отношении Fs/Rs."""
return PhyMode(symbol_rate)
def frame_waveform_sample_count(
plan: lab045.TransmissionPlan,
mode: PhyMode,
) -> int:
"""Точная длина прежней формы при заданном числе выборок на символ."""
symbol_intervals = (
plan.frame_symbol_count
+ 2 * lab045.GUARD_SYMBOL_COUNT
+ lab045.RRC_SPAN_SYMBOLS
)
return symbol_intervals * mode.samples_per_symbol
def build_parameterized_tx_complex(
plan: lab045.TransmissionPlan,
mode: PhyMode,
) -> np.ndarray:
"""Сформировать прежнюю BPSK/RRC-форму с параметрическим Rs."""
symbols = radio.bpsk_modulate(plan.frame_bits)
shaped = shaping.shape_frame_like_lab042(symbols, mode.samples_per_symbol)
indexes = np.arange(shaped.size, dtype=np.float64)
carriered = shaped * np.exp(
2j
* np.pi
* lab045.CARRIER_OFFSET_HZ
* indexes
/ mode.sample_rate_hz
)
rms = float(np.sqrt(np.mean(np.abs(carriered) ** 2)))
if not np.isfinite(rms) or rms <= 0.0:
raise ValueError("Нулевая или нечисловая мощность формы")
result = carriered * (lab045.lab044.TARGET_RMS_CODE / rms)
expected = frame_waveform_sample_count(plan, mode)
if result.size != expected:
raise RuntimeError("Параметрическая форма имеет неверную длину")
return result
def build_parameterized_tx_cs8(
plan: lab045.TransmissionPlan,
mode: PhyMode,
) -> bytes:
"""Квантовать параметрическую форму тем же преобразователем CS8."""
return lab045.lab044.lab043.to_cs8_bytes(
build_parameterized_tx_complex(plan, mode)
)
@lru_cache(maxsize=len(TESTED_SYMBOL_RATES))
def _matched_filter_taps(samples_per_symbol: int) -> np.ndarray:
taps = radio.root_raised_cosine_taps(
RRC_ROLLOFF,
samples_per_symbol,
lab045.RRC_SPAN_SYMBOLS,
).astype(np.float32)
taps.flags.writeable = False
return taps
def _preamble_timing_offset(
matched: np.ndarray,
nominal: int,
samples_per_symbol: int,
) -> int:
"""Найти фазу дискретизации только по известной PRBS7-преамбуле."""
expected = radio.bpsk_modulate(lab045.lab044.preamble_bits())
offsets = np.arange(-2 * samples_per_symbol, 2 * samples_per_symbol + 1)
scores = np.zeros(offsets.size, dtype=np.float64)
first_sample = nominal + int(offsets[0])
for phase in range(samples_per_symbol):
score_indexes = np.arange(phase, offsets.size, samples_per_symbol)
sequence_length = expected.size + score_indexes.size - 1
start = first_sample + phase
stop = start + sequence_length * samples_per_symbol
sequence = matched[start:stop:samples_per_symbol]
if start < 0 or sequence.size != sequence_length:
raise RuntimeError("Недостаточно выборок для поиска синхронизации")
windows = np.lib.stride_tricks.sliding_window_view(
sequence,
expected.size,
)
correlation = windows @ np.conj(expected)
energy = np.sum(np.abs(windows) ** 2, axis=1)
scores[score_indexes] = np.abs(correlation) / np.sqrt(
expected.size * energy + 1e-30
)
return int(offsets[int(np.argmax(scores))])
def receive_iq_frame(
iq_samples: np.ndarray,
mode: PhyMode,
) -> IqFrameReception:
"""Принять один IQ-кадр без переданных битов, пакета или payload."""
samples = np.asarray(iq_samples, dtype=np.complex64)
if samples.ndim != 1 or samples.size == 0 or not np.all(np.isfinite(samples)):
raise ValueError("Нужен конечный одномерный IQ-кадр")
indexes = np.arange(samples.size, dtype=np.float64)
centered = samples - np.complex64(np.mean(samples, dtype=np.complex128))
baseband = (
centered
* np.exp(
-2j
* np.pi
* lab045.CARRIER_OFFSET_HZ
* indexes
/ mode.sample_rate_hz
).astype(np.complex64)
)
taps = _matched_filter_taps(mode.samples_per_symbol)
matched = oaconvolve(baseband, taps, mode="full")
filter_delay = (len(taps) - 1) // 2
nominal = lab045.GUARD_SYMBOL_COUNT * mode.samples_per_symbol + 2 * filter_delay
timing = _preamble_timing_offset(
matched,
nominal,
mode.samples_per_symbol,
)
probe_count = lab045.PREAMBLE_BIT_COUNT + lab045.PHY_HEADER_BIT_COUNT
probe = matched[
nominal + timing :
nominal + timing + probe_count * mode.samples_per_symbol :
mode.samples_per_symbol
].astype(np.complex128)
if probe.size != probe_count:
raise RuntimeError("Физический кадр оборван до PHY-заголовка")
expected_preamble = radio.bpsk_modulate(lab045.lab044.preamble_bits())
gain = complex(np.mean(
probe[: lab045.PREAMBLE_BIT_COUNT] * np.conj(expected_preamble)
))
if abs(gain) <= 1e-30:
raise RuntimeError("Не удалось оценить комплексный коэффициент канала")
normalized_probe = (probe / gain).real
header = lab045.decode_phy_header_soft(
normalized_probe[
lab045.PREAMBLE_BIT_COUNT :
lab045.PREAMBLE_BIT_COUNT + lab045.PHY_HEADER_BIT_COUNT
]
)
if not header["passed"]:
reception = lab045.analyze_post_preamble_soft(
normalized_probe[lab045.PREAMBLE_BIT_COUNT:]
)
return IqFrameReception(
False,
False,
str(reception.analysis["crc"]),
None,
normalized_probe[lab045.PREAMBLE_BIT_COUNT:],
timing,
reception,
)
packet_bytes = int(header["packet_bytes"])
coded_bits = lab045.conv.encoded_bit_count(packet_bytes * 8)
total_symbols = (
lab045.PREAMBLE_BIT_COUNT
+ lab045.PHY_HEADER_BIT_COUNT
+ coded_bits
)
symbols = matched[
nominal + timing :
nominal + timing + total_symbols * mode.samples_per_symbol :
mode.samples_per_symbol
].astype(np.complex128)
if symbols.size != total_symbols:
raise RuntimeError("Физический кадр оборван до конца кодированной части")
normalized = (symbols / gain).real
post_preamble = normalized[lab045.PREAMBLE_BIT_COUNT:]
reception = lab045.analyze_post_preamble_soft(post_preamble)
return IqFrameReception(
True,
bool(reception.analysis["fec_decode_pass"]),
str(reception.analysis["crc"]),
reception.payload_bytes,
post_preamble,
timing,
reception,
)
def add_coded_bit_ebn0_noise(
iq_samples: np.ndarray,
transmitted_symbol_count: int,
ebn0_db: float,
seed: int,
) -> tuple[np.ndarray, float]:
"""Добавить комплексный AWGN при одинаковом coded-bit Eb/N0.
Для BPSK один переданный символ несёт один кодированный бит. Энергия бита
измеряется непосредственно по всей квантованной форме, поэтому при смене
samples/symbol шум физически масштабируется как sqrt(Eb), а не подбирается
отдельно для режима.
"""
samples = np.asarray(iq_samples, dtype=np.complex64)
if transmitted_symbol_count <= 0:
raise ValueError("Число переданных символов должно быть положительным")
if not np.isfinite(ebn0_db):
raise ValueError("Eb/N0 должен быть конечным")
energy_per_bit = float(
np.sum(np.abs(samples) ** 2, dtype=np.float64)
/ transmitted_symbol_count
)
linear = 10.0 ** (ebn0_db / 10.0)
component_sigma = math.sqrt(energy_per_bit / (2.0 * linear))
generator = np.random.default_rng(seed)
noisy = samples.copy()
noisy.real += np.float32(component_sigma) * generator.standard_normal(
samples.size,
dtype=np.float32,
)
noisy.imag += np.float32(component_sigma) * generator.standard_normal(
samples.size,
dtype=np.float32,
)
return noisy, component_sigma
def _offer_reception(
accumulator: _RoundtripAccumulator,
frame: lab046.FramePlan,
received: IqFrameReception,
) -> None:
accumulator.phy_header_pass += int(received.phy_header_pass)
accumulator.phy_header_fail += int(not received.phy_header_pass)
accumulator.fec_pass += int(received.fec_pass)
accumulator.fec_fail += int(received.phy_header_pass and not received.fec_pass)
accumulator.crc_pass += int(received.crc == lab045.CRC_PASS)
accumulator.crc_fail += int(received.crc == lab045.CRC_FAIL)
accumulator.crc_not_reached += int(received.crc == lab045.CRC_NOT_REACHED)
diagnostic = lab045.diagnostic_pre_and_post_fec(
received.reception,
received.post_preamble_soft,
frame.plan,
)
pre_fec = diagnostic["pre_fec_coded"]["bit_errors"]
post_fec = diagnostic["post_fec_packet"]["bit_errors"]
if pre_fec is not None:
accumulator.pre_fec_errors += int(pre_fec)
if post_fec is not None:
accumulator.post_fec_errors += int(post_fec)
if received.crc == lab045.CRC_PASS and received.payload_bytes is not None:
accumulator.assembler.offer(
lab046.recover_fragment_from_payload(received.payload_bytes)
)
def _roundtrip_summary(
accumulator: _RoundtripAccumulator,
jpeg: bytes,
frame_count: int,
capture_seconds: float,
) -> dict:
recovered = accumulator.assembler.recovered_object()
recovered_sha = (
hashlib.sha256(recovered).hexdigest() if recovered is not None else None
)
byte_errors: int | None = None
bit_errors: int | None = None
if recovered is not None and len(recovered) == len(jpeg):
difference = np.bitwise_xor(
np.frombuffer(recovered, dtype=np.uint8),
np.frombuffer(jpeg, dtype=np.uint8),
)
byte_errors = int(np.count_nonzero(difference))
bit_errors = int(np.unpackbits(difference).sum())
sigmas = accumulator.noise_sigmas or []
return {
"jpeg_frame_count": frame_count,
"phy_header_pass": accumulator.phy_header_pass,
"phy_header_fail": accumulator.phy_header_fail,
"fec_pass": accumulator.fec_pass,
"fec_fail": accumulator.fec_fail,
"crc_pass": accumulator.crc_pass,
"crc_fail": accumulator.crc_fail,
"crc_not_reached": accumulator.crc_not_reached,
"valid_frames": accumulator.assembler.counters.accepted,
"missing_fragments": accumulator.assembler.missing_indexes(),
"object_complete": accumulator.assembler.object_complete(),
"pre_fec_coded_bit_errors": accumulator.pre_fec_errors,
"post_fec_protocol_bit_errors": accumulator.post_fec_errors,
"recovered_sha256": recovered_sha,
"sha_match": recovered_sha == hashlib.sha256(jpeg).hexdigest(),
"byte_errors": byte_errors,
"bit_errors": bit_errors,
"receiver_processing_seconds": accumulator.processing_seconds,
"capture_equivalent_seconds": capture_seconds,
"processing_to_capture_ratio": (
accumulator.processing_seconds / capture_seconds
),
"real_time_capable": accumulator.processing_seconds < capture_seconds,
"noise_component_sigma_min": min(sigmas) if sigmas else 0.0,
"noise_component_sigma_median": (
float(np.median(sigmas)) if sigmas else 0.0
),
"noise_component_sigma_max": max(sigmas) if sigmas else 0.0,
}
def run_mode_roundtrips(jpeg: bytes, mode: PhyMode) -> tuple[dict, dict]:
"""Выполнить идеальный и единый шумный IQ-тракт для всего JPEG."""
from experiments.lab049_independent_iq import run_roundtrips
return run_roundtrips(jpeg, mode)
def run_frame_aligned_roundtrips(jpeg: bytes, mode: PhyMode) -> tuple[dict, dict]:
"""Историческая локальная проверка; не измеряет независимый поиск потока."""
frames = tuple(lab046.build_object_frames(jpeg, lab047.OBJECT_ID))
ideal = _RoundtripAccumulator(lab046.ObjectAssembler())
noisy = _RoundtripAccumulator(lab046.ObjectAssembler(), noise_sigmas=[])
capture_samples = 0
for frame_number, frame in enumerate(frames):
tx_cs8 = build_parameterized_tx_cs8(frame.plan, mode)
iq = lab045.lab044.signed_iq_bytes_to_complex64(tx_cs8)
capture_samples += iq.size
started = time.perf_counter()
ideal_received = receive_iq_frame(iq, mode)
ideal.processing_seconds += time.perf_counter() - started
_offer_reception(ideal, frame, ideal_received)
noisy_iq, sigma = add_coded_bit_ebn0_noise(
iq,
frame.plan.frame_symbol_count,
NOISY_CODED_BIT_EBN0_DB,
NOISE_BASE_SEED + frame_number,
)
assert noisy.noise_sigmas is not None
noisy.noise_sigmas.append(sigma)
started = time.perf_counter()
noisy_received = receive_iq_frame(noisy_iq, mode)
noisy.processing_seconds += time.perf_counter() - started
_offer_reception(noisy, frame, noisy_received)
capture_seconds = capture_samples / mode.sample_rate_hz
return (
_roundtrip_summary(ideal, jpeg, len(frames), capture_seconds),
_roundtrip_summary(noisy, jpeg, len(frames), capture_seconds),
)
def single_frame_roundtrip(symbol_rate: int) -> dict:
"""Быстрая сквозная проверка одного кадра для тестов."""
payload = bytes((index * 49 + 7) & 0xFF for index in range(256))
plan = lab045.build_transmission_plan(payload)
mode = phy_mode(symbol_rate)
iq = lab045.lab044.signed_iq_bytes_to_complex64(
build_parameterized_tx_cs8(plan, mode)
)
received = receive_iq_frame(iq, mode)
return {
"phy_header_pass": received.phy_header_pass,
"fec_pass": received.fec_pass,
"crc": received.crc,
"payload_match": received.payload_bytes == payload,
}
def _padded_bytes(prefix: bytes, size: int, sequence: int) -> bytes:
head = prefix + struct.pack("!I", sequence)
tail = bytes((sequence * 37 + index * 17) & 0xFF for index in range(size))
return (head + tail)[:size]
def _control_payload(sequence: int) -> bytes:
context = MessageContext(49, 1, 1, sequence)
return encode_message(ControlStateMessage(context, 1.0, 0.0, False, True))
def _emergency_payload(sequence: int) -> bytes:
context = MessageContext(49, 1, 1, sequence)
return encode_message(EmergencyStop(context, 49))
def _link_packet(
traffic_class: TrafficClass,
sequence: int,
generation_seconds: float,
payload: bytes,
) -> LinkPacket:
directions = {
TrafficClass.EMERGENCY: Direction.GROUND_TO_ROVER,
TrafficClass.CONTROL: Direction.GROUND_TO_ROVER,
TrafficClass.TELEMETRY: Direction.ROVER_TO_GROUND,
TrafficClass.VIDEO: Direction.ROVER_TO_GROUND,
}
deadlines = {
TrafficClass.EMERGENCY: 50,
TrafficClass.CONTROL: 100,
TrafficClass.TELEMETRY: 500,
TrafficClass.VIDEO: 0,
}
return LinkPacket(
traffic_class=traffic_class,
direction=directions[traffic_class],
stream_id=int(traffic_class),
sequence_number=sequence,
generation_time_us=int(round(generation_seconds * 1_000_000)),
deadline_ms=deadlines[traffic_class],
payload=payload,
)
def _schedule_fixed_physical_frames(
queued_packets: tuple[QueuedPacket, ...],
frame_duration_seconds: float,
):
"""Применить штатный планировщик к одинаковым физическим кадрам.
Исходные ``LinkPacket`` содержат неизменённые прикладные сообщения. Для
расчёта только занятости линии создаются внутренние служебные представители
той же очереди с нагрузкой 256 байт: поэтому штатный планировщик отводит
каждому элементу ровно длительность кадра Lab045. Перед возвратом результата
внутренние представители заменяются исходными пакетами и их точными байтами.
"""
if frame_duration_seconds <= 0.0:
raise ValueError("Длительность физического кадра должна быть положительной")
actual_by_order = {item.arrival_order: item for item in queued_packets}
if len(actual_by_order) != len(queued_packets):
raise ValueError("arrival_order должен быть уникальным")
scheduling_packets = tuple(
QueuedPacket(
packet=LinkPacket(
traffic_class=item.packet.traffic_class,
direction=item.packet.direction,
stream_id=item.packet.stream_id,
sequence_number=item.packet.sequence_number,
generation_time_us=item.packet.generation_time_us,
deadline_ms=item.packet.deadline_ms,
payload=bytes(lab046.PACKET_PAYLOAD_SIZE),
),
arrival_order=item.arrival_order,
)
for item in queued_packets
)
occupancy_schedule = schedule_packets(
scheduling_packets,
SchedulerMode.LATEST_STATE,
_scheduler_channel_bitrate(frame_duration_seconds),
)
transmitted = tuple(
replace(
item,
queued=actual_by_order[item.queued.arrival_order],
blocked_by=(
None
if item.blocked_by is None
else actual_by_order[item.blocked_by.arrival_order]
),
wire_packet=encode_link_packet(
actual_by_order[item.queued.arrival_order].packet
),
)
for item in occupancy_schedule.transmitted
)
replacements = tuple(
replace(
item,
removed=actual_by_order[item.removed.arrival_order],
replacement=actual_by_order[item.replacement.arrival_order],
)
for item in occupancy_schedule.replacements
)
return replace(
occupancy_schedule,
transmitted=transmitted,
replacements=replacements,
)
def _latency_statistics(schedule, traffic_class: TrafficClass, *, completion=False) -> dict:
values = np.asarray(
[
(item.end_seconds if completion else item.start_seconds)
- item.queued.generation_time_seconds
for item in schedule.transmitted
if item.queued.packet.traffic_class is traffic_class
],
dtype=np.float64,
)
if values.size == 0:
return {"count": 0, "min_ms": None, "median_ms": None, "max_ms": None}
return {
"count": int(values.size),
"min_ms": float(np.min(values) * 1000.0),
"median_ms": float(np.median(values) * 1000.0),
"max_ms": float(np.max(values) * 1000.0),
}
def _deadline_misses(schedule, traffic_class: TrafficClass) -> int:
"""Посчитать переданные после собственного ненулевого deadline пакеты."""
return sum(
item.queued.packet.traffic_class is traffic_class
and item.queued.packet.deadline_ms > 0
and (
item.end_seconds - item.queued.generation_time_seconds
) * 1000.0 > item.queued.packet.deadline_ms + 1e-9
for item in schedule.transmitted
)
def _scheduler_channel_bitrate(frame_duration_seconds: float) -> float:
scheduler_video_wire_bytes = HEADER_SIZE + lab046.PACKET_PAYLOAD_SIZE
return scheduler_video_wire_bytes * 8.0 / frame_duration_seconds
def channel_budget(
frame_duration_seconds: float,
jpeg_bytes: int,
jpeg_frame_count: int,
) -> dict:
"""Бюджет общего LinkPacket-канала, привязанный к длительности VIDEO."""
if jpeg_frame_count <= 0:
raise ValueError("Число кадров JPEG должно быть положительным")
physical_wire_bytes = HEADER_SIZE + lab046.PACKET_PAYLOAD_SIZE
channel_bitrate = _scheduler_channel_bitrate(frame_duration_seconds)
frame_capacity_per_second = 1.0 / frame_duration_seconds
control_frames_per_second = 1.0 / CONTROL_PERIOD_SECONDS
telemetry_frames_per_second = 1.0 / TELEMETRY_PERIOD_SECONDS
control_airtime_fraction = control_frames_per_second * frame_duration_seconds
telemetry_airtime_fraction = (
telemetry_frames_per_second * frame_duration_seconds
)
control_bps = physical_wire_bytes * 8.0 / CONTROL_PERIOD_SECONDS
telemetry_bps = physical_wire_bytes * 8.0 / TELEMETRY_PERIOD_SECONDS
remaining_link_bps = max(0.0, channel_bitrate - control_bps - telemetry_bps)
available_video_frames_per_second = max(
0.0,
frame_capacity_per_second
- control_frames_per_second
- telemetry_frames_per_second,
)
video_efficiency = jpeg_bytes / (jpeg_frame_count * physical_wire_bytes)
jpeg_bits_per_physical_frame = jpeg_bytes * 8.0 / jpeg_frame_count
available_video_bps = (
available_video_frames_per_second * jpeg_bits_per_physical_frame
)
return {
"scheduler_link_bitrate_bps": channel_bitrate,
"scheduler_video_wire_bytes": physical_wire_bytes,
"physical_payload_bytes_per_packet": lab046.PACKET_PAYLOAD_SIZE,
"fixed_physical_frame_per_packet": True,
"physical_frame_capacity_per_second": frame_capacity_per_second,
"control_application_bytes": len(_control_payload(0)),
"control_application_bps": (
len(_control_payload(0)) * 8.0 / CONTROL_PERIOD_SECONDS
),
"control_reserved_frames_per_second": control_frames_per_second,
"control_reserved_airtime_fraction": control_airtime_fraction,
"control_reserved_link_bps": control_bps,
"telemetry_application_bytes": TELEMETRY_APPLICATION_BYTES,
"telemetry_application_bps": (
TELEMETRY_APPLICATION_BYTES * 8.0 / TELEMETRY_PERIOD_SECONDS
),
"telemetry_load_frames_per_second": telemetry_frames_per_second,
"telemetry_airtime_fraction": telemetry_airtime_fraction,
"telemetry_load_link_bps": telemetry_bps,
"service_requested_airtime_fraction": (
control_airtime_fraction + telemetry_airtime_fraction
),
"remaining_link_bps": remaining_link_bps,
"video_available_frames_per_second": available_video_frames_per_second,
"video_payload_efficiency": video_efficiency,
"video_available_bitrate_bps": available_video_bps,
"theoretical_jpeg_fps": available_video_bps / (jpeg_bytes * 8.0),
"emergency_events_per_second_in_scenario": 1.0 / SCHEDULER_SOURCE_SECONDS,
"emergency_airtime_fraction_in_scenario": (
frame_duration_seconds / SCHEDULER_SOURCE_SECONDS
),
"video_available_bitrate_with_emergency_bps": max(
0.0, available_video_frames_per_second - 1.0 / SCHEDULER_SOURCE_SECONDS
) * jpeg_bits_per_physical_frame,
}
def priority_probe(frame_duration_seconds: float) -> dict:
"""Проверить непрерываемость кадра и выбор класса на его границе."""
arrival = frame_duration_seconds / 4.0
packets = (
_link_packet(
TrafficClass.VIDEO,
0,
0.0,
bytes(lab046.PACKET_PAYLOAD_SIZE),
),
_link_packet(
TrafficClass.VIDEO,
1,
0.0,
bytes(lab046.PACKET_PAYLOAD_SIZE),
),
_link_packet(
TrafficClass.TELEMETRY,
0,
arrival,
_padded_bytes(b"TELEMETRY", TELEMETRY_APPLICATION_BYTES, 0),
),
_link_packet(
TrafficClass.CONTROL,
0,
arrival,
_control_payload(0),
),
_link_packet(
TrafficClass.EMERGENCY,
0,
arrival,
_emergency_payload(0),
),
)
queued = tuple(
QueuedPacket(packet=packet, arrival_order=index)
for index, packet in enumerate(packets)
)
schedule = _schedule_fixed_physical_frames(queued, frame_duration_seconds)
order = [item.queued.packet.traffic_class.name for item in schedule.transmitted]
expected = ["VIDEO", "EMERGENCY", "CONTROL", "TELEMETRY", "VIDEO"]
first = schedule.transmitted[0]
return {
"transmission_order": order,
"expected_order": expected,
"priority_pass": order == expected,
"first_video_not_preempted": (
first.queued.packet.traffic_class is TrafficClass.VIDEO
and math.isclose(
first.end_seconds - first.start_seconds,
frame_duration_seconds,
rel_tol=0.0,
abs_tol=1e-12,
)
),
"control_max_video_blocking_ms": frame_duration_seconds * 1000.0,
"emergency_max_video_blocking_ms": frame_duration_seconds * 1000.0,
"emergency_deadline_ms": 50.0,
"emergency_deadline_met_in_worst_phase": frame_duration_seconds <= 0.050,
"legacy_deadline_field_scope": "blocking_only_not_delivery",
"emergency_worst_delivery_ms": 2.0 * frame_duration_seconds * 1000.0,
"emergency_delivery_deadline_met_in_worst_phase": (
2.0 * frame_duration_seconds <= 0.050
),
}
def loaded_scheduler_scenario(frame_duration_seconds: float) -> dict:
"""Постоянно заполненная VIDEO-очередь и периодический служебный трафик."""
packets: list[LinkPacket] = []
for sequence in range(VIDEO_BACKLOG_PACKETS):
packets.append(
_link_packet(
TrafficClass.VIDEO,
sequence,
0.0,
_padded_bytes(b"VIDEO", lab046.PACKET_PAYLOAD_SIZE, sequence),
)
)
control_times = np.arange(
CONTROL_PERIOD_SECONDS / 2.0,
SCHEDULER_SOURCE_SECONDS,
CONTROL_PERIOD_SECONDS,
)
for sequence, generated in enumerate(control_times):
packets.append(
_link_packet(
TrafficClass.CONTROL,
sequence,
float(generated),
_control_payload(sequence),
)
)
telemetry_times = np.arange(
TELEMETRY_PERIOD_SECONDS / 2.0,
SCHEDULER_SOURCE_SECONDS,
TELEMETRY_PERIOD_SECONDS,
)
for sequence, generated in enumerate(telemetry_times):
packets.append(
_link_packet(
TrafficClass.TELEMETRY,
sequence,
float(generated),
_padded_bytes(
b"TELEMETRY",
TELEMETRY_APPLICATION_BYTES,
sequence,
),
)
)
packets.append(
_link_packet(
TrafficClass.EMERGENCY,
0,
EMERGENCY_TIME_SECONDS,
_emergency_payload(0),
)
)
packets.sort(
key=lambda packet: (
packet.generation_time_us,
int(packet.traffic_class),
packet.sequence_number,
)
)
queued = tuple(
QueuedPacket(packet=packet, arrival_order=index)
for index, packet in enumerate(packets)
)
schedule = _schedule_fixed_physical_frames(queued, frame_duration_seconds)
active_schedule = replace(schedule, transmitted=tuple(
item for item in schedule.transmitted
if item.end_seconds <= SCHEDULER_SOURCE_SECONDS
))
created_by_class = {
traffic.name: sum(packet.traffic_class is traffic for packet in packets)
for traffic in TrafficClass
}
transmitted_by_class = {
traffic.name: sum(
item.queued.packet.traffic_class is traffic
for item in schedule.transmitted
)
for traffic in TrafficClass
}
replacements_by_class = {
traffic.name: sum(
item.removed.packet.traffic_class is traffic
for item in schedule.replacements
)
for traffic in TrafficClass
}
return {
"scheduler": SchedulerMode.LATEST_STATE.value,
"source_duration_seconds": SCHEDULER_SOURCE_SECONDS,
"control_period_seconds": CONTROL_PERIOD_SECONDS,
"telemetry_period_seconds": TELEMETRY_PERIOD_SECONDS,
"emergency_time_seconds": EMERGENCY_TIME_SECONDS,
"video_backlog_packets": VIDEO_BACKLOG_PACKETS,
"created_packets": created_by_class,
"transmitted_packets": transmitted_by_class,
"replaced_waiting_packets": replacements_by_class,
"latency": {
traffic.name: _latency_statistics(schedule, traffic)
for traffic in TrafficClass
},
"latency_scope": "queue_wait_until_drain_end",
"delivery_latency_until_drain_end": {
traffic.name: _latency_statistics(schedule, traffic, completion=True)
for traffic in TrafficClass
},
"active_window_delivery_latency": {
traffic.name: _latency_statistics(active_schedule, traffic, completion=True)
for traffic in TrafficClass
},
"active_window_delivered_packets": {
traffic.name: sum(item.queued.packet.traffic_class is traffic
for item in active_schedule.transmitted)
for traffic in TrafficClass
},
"drain_end_seconds": max(item.end_seconds for item in schedule.transmitted),
"deadline_metric": "end_minus_generation",
"deadline_misses": {
traffic.name: _deadline_misses(schedule, traffic)
for traffic in TrafficClass
},
"scheduler_drops_expired_packets": False,
"application_payloads_padded": False,
"fixed_physical_frame_duration_model": True,
"video_queue_saturated": True,
}
def processing_to_capture_ratio(
processing_seconds: float,
capture_seconds: float,
) -> float:
if processing_seconds < 0.0:
raise ValueError("Время обработки не может быть отрицательным")
if capture_seconds <= 0.0:
raise ValueError("Длительность записи должна быть положительной")
return processing_seconds / capture_seconds
def _roundtrip_pass(result: dict, expected_frames: int) -> bool:
return bool(
result["valid_frames"] == expected_frames
and result["crc_pass"] == expected_frames
and result["crc_fail"] == 0
and result.get("termination_reason", "END_OF_STREAM") == "END_OF_STREAM"
and result.get("jpeg_decode", True)
and result["object_complete"]
and result["sha_match"]
and result["byte_errors"] == 0
and result["bit_errors"] == 0
)
def classify_mode(
exact_roundtrips: bool,
priority_pass: bool,
real_time_capable: bool,
control_target_met: bool,
emergency_deadline_met: bool,
video_capacity_positive: bool,
) -> str:
"""Классифицировать режим по всем критериям Lab049."""
if (
exact_roundtrips
and priority_pass
and real_time_capable
and control_target_met
and emergency_deadline_met
and video_capacity_positive
):
return "PASS"
if exact_roundtrips and priority_pass:
return "PARTIAL"
return "FAIL"
def run_mode(jpeg: bytes, symbol_rate: int) -> dict:
print(f"Lab049: independent IQ {symbol_rate} symbol/s", file=sys.stderr, flush=True)
mode = phy_mode(symbol_rate)
frames = tuple(lab046.build_object_frames(jpeg, lab047.OBJECT_ID))
frame_samples = frame_waveform_sample_count(frames[0].plan, mode)
frame_duration = frame_samples / mode.sample_rate_hz
jpeg_airtime = frame_duration * len(frames)
ideal, noisy = run_mode_roundtrips(jpeg, mode)
probe = priority_probe(frame_duration)
scheduler = loaded_scheduler_scenario(frame_duration)
budget = channel_budget(frame_duration, len(jpeg), len(frames))
goodput_bps = len(jpeg) * 8.0 / jpeg_airtime
baseline_airtime = (
frame_samples
// mode.samples_per_symbol
* len(frames)
/ MANDATORY_SYMBOL_RATES[0]
)
baseline_goodput_bps = len(jpeg) * 8.0 / baseline_airtime
exact_roundtrips = _roundtrip_pass(ideal, len(frames)) and _roundtrip_pass(
noisy,
len(frames),
)
real_time = bool(noisy["real_time_capable"])
control_target = frame_duration < 0.100
video_capacity = budget["video_available_bitrate_bps"] > 0.0
emergency_deadline = bool(probe["emergency_delivery_deadline_met_in_worst_phase"])
status = classify_mode(
exact_roundtrips,
bool(probe["priority_pass"]),
real_time,
control_target,
emergency_deadline,
video_capacity,
)
return {
"symbol_rate": mode.symbol_rate,
"samples_per_symbol": mode.samples_per_symbol,
"sample_rate_hz": mode.sample_rate_hz,
"frame_symbol_intervals": frame_samples // mode.samples_per_symbol,
"frame_samples": frame_samples,
"frame_duration_seconds": frame_duration,
"jpeg_rf_frames": len(frames),
"jpeg_airtime_seconds": jpeg_airtime,
"useful_goodput_bytes_per_second": len(jpeg) / jpeg_airtime,
"useful_goodput_kbit_per_second": goodput_bps / 1000.0,
"goodput_gain_over_20ksps": goodput_bps / baseline_goodput_bps,
"ideal": ideal,
"noisy": {
"coded_bit_ebn0_db": NOISY_CODED_BIT_EBN0_DB,
**noisy,
},
"priority_probe": probe,
"loaded_scheduler": scheduler,
"channel_budget": budget,
"control_under_100ms": control_target,
"control_under_50ms": frame_duration < 0.050,
"emergency_50ms_deadline_met": emergency_deadline,
"video_capacity_positive": video_capacity,
"mode_result": status,
}
def choose_recommended_mode(mode_results: list[dict]) -> tuple[int | None, str]:
"""Выбрать один диагностический PHY-режим с точным независимым приёмом."""
by_rate = {item["symbol_rate"]: item for item in mode_results}
preferred = by_rate.get(100_000)
if (
preferred is not None
and _roundtrip_pass(preferred["ideal"], preferred["jpeg_rf_frames"])
and _roundtrip_pass(preferred["noisy"], preferred["jpeg_rf_frames"])
and preferred["noisy"]["real_time_capable"]
and preferred["control_under_100ms"]
):
return (
100_000,
"Диагностический кандидат отдельного аппаратного PHY-опыта: точный "
"шумный приём, менее 100 мс блокировки CONTROL и real-time. Для "
"объединённого канала не готов: нарушает 50 мс EMERGENCY и при "
"полных кадрах не оставляет полосы VIDEO. 200 ксимв/с исправляет "
"блокировку, но доставка EMERGENCY всё ещё превышает 50 мс.",
)
passing = [item for item in mode_results if item["mode_result"] == "PASS"]
if not passing:
diagnostic = [item for item in mode_results
if _roundtrip_pass(item["ideal"], item["jpeg_rf_frames"])
and _roundtrip_pass(item["noisy"], item["jpeg_rf_frames"])]
if diagnostic:
selected = min(diagnostic, key=lambda item: item["noisy"]["processing_to_capture_ratio"])
return int(selected["symbol_rate"]), (
"Только диагностический PHY-кандидат: среди точных независимых "
"шумных проходов имеет наименьшее отношение обработки к IQ. "
"Ни один режим не выполнил все требования; кандидат также не "
"пригоден для непрерывного объединённого канала. RF не разрешён."
)
return None, "Ни один режим не выполнил все программные критерии"
selected = min(passing, key=lambda item: item["symbol_rate"])
return int(selected["symbol_rate"]), "Выбран наименьший режим со статусом PASS"
def run_lab049(jpeg_path: Path = JPEG_PATH) -> dict:
"""Выполнить все программные режимы Lab049 без аппаратных операций."""
path = Path(jpeg_path)
jpeg = path.read_bytes()
digest = hashlib.sha256(jpeg).hexdigest()
if len(jpeg) != EXPECTED_JPEG_BYTES or digest != EXPECTED_JPEG_SHA256:
raise RuntimeError("JPEG Lab047 не совпал с зафиксированным объектом")
modes = [run_mode(jpeg, rate) for rate in TESTED_SYMBOL_RATES]
recommended, reason = choose_recommended_mode(modes)
return {
"stage": "LAB049_SOFTWARE_SYMBOL_RATE",
"jpeg": {
"path": str(path),
"bytes": len(jpeg),
"sha256": digest,
"width": 320,
"height": 240,
"mode": "RGB",
},
"fixed": {
"sample_rate_hz": SAMPLE_RATE_HZ,
"modulation": "BPSK",
"convolutional_fec": "R=1/2 K=7 generators=171/133 octal",
"crc_changed": False,
"packet_format_changed": False,
"fragment_format_changed": False,
"prbs7_changed": False,
"phy_header_changed": False,
"priority_order": "EMERGENCY > CONTROL > TELEMETRY > VIDEO",
"noisy_coded_bit_ebn0_db": NOISY_CODED_BIT_EBN0_DB,
"noise_tuned_per_mode": False,
},
"receiver_input": {
"frame_aligned_iq": False,
"tx_frame_offsets": False,
"tx_bits": False,
"tx_payload": False,
"expected_frame_count": False,
},
"modes": modes,
"maximum_stable_software_symbol_rate": max((
item["symbol_rate"]
for item in modes
if _roundtrip_pass(item["ideal"], item["jpeg_rf_frames"])
and _roundtrip_pass(item["noisy"], item["jpeg_rf_frames"])
), default=None),
"recommended_rf_symbol_rate": recommended,
"recommendation_reason": reason,
"recommended_for_integrated_operation": False,
"all_mode_requirements_met": any(
item["mode_result"] == "PASS" for item in modes
),
"rf_tx": 0,
"hackrf_started": False,
}
def run_functional_tests() -> None:
assert tuple(phy_mode(rate).samples_per_symbol for rate in TESTED_SYMBOL_RATES) == (
400,
160,
80,
40,
)
assert SAMPLE_RATE_HZ == lab047.SAMPLE_RATE_HZ
assert NOISY_CODED_BIT_EBN0_DB == 7.0
assert [traffic.name for traffic in TrafficClass] == [
"EMERGENCY",
"CONTROL",
"TELEMETRY",
"VIDEO",
]
def main() -> int:
parser = argparse.ArgumentParser(
description="Lab049: программное сравнение скоростей BPSK"
)
parser.add_argument("--jpeg", type=Path, default=JPEG_PATH)
parser.add_argument("--output-directory", type=Path, default=OUTPUT_DIRECTORY)
arguments = parser.parse_args()
result = run_lab049(arguments.jpeg)
arguments.output_directory.mkdir(parents=True, exist_ok=True)
output_path = arguments.output_directory / RESULT_NAME
output_path.write_text(
json.dumps(result, ensure_ascii=False, indent=2),
encoding="utf-8",
)
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())