Files
SDR-Rover/experiments/lab046_bpsk_object_transport.py

1065 lines
45 KiB
Python
Raw 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.
"""Lab046: многопакетная передача большого бинарного объекта поверх Lab045.
Разделены два уровня, они не смешиваются:
- Уровень 1 (внутри одного радиокадра) — Lab045: свёрточный код R=1/2 K=7,
перемежитель, мягкий декодер Витерби, CRC пакета. Исправляет битовые
ошибки внутри принятого кадра.
- Уровень 2 (большой объект) — существующий ``protocol/image_fragments``:
фрагментация, нумерация фрагментов, идентификатор объекта, сборка. Собирает
объект из нескольких независимо переданных кадров.
Новый код здесь минимален: он только связывает эти два готовых уровня и
добавляет fail-closed сборщик объекта, который при недостающем или
повреждённом фрагменте возвращает ``object_complete = False`` без исключений
и без фиктивного объекта. Формат фрагмента не дублируется — используется
существующий 12-байтовый заголовок ``image_fragments`` (``>IHHI``:
object_id, fragment_index, fragment_count, original_size).
Раскладка нагрузки пакета (256 байт, как в доказанной точке Lab045):
[12 байт метаданных фрагмента][244 байта данных фрагмента]
Последний фрагмент короче: его нагрузка добивается нулями до 256 байт перед
Lab045 и обрезается на приёме по длине, вычисляемой из полей заголовка
(object size и fragment count). Erasure-код 12+3 здесь НЕ используется — это
отдельное будущее расширение для потери целых фрагментов.
RF в этой лабе не запускается: RF TX = 0, только программная проверка.
"""
from __future__ import annotations
import argparse
import hashlib
import sys
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Callable
import numpy as np
if __package__ in (None, ""):
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from experiments import lab045_hackrf_bpsk_fec as lab045
from protocol import image_fragments as fragments
# ============================================================ Константы Lab046
PACKET_PAYLOAD_SIZE = lab045.PAYLOAD_SIZE_BYTES # 256, доказанная точка Lab045
FRAGMENT_DATA_SIZE = PACKET_PAYLOAD_SIZE - fragments.FRAGMENT_HEADER_SIZE # 244
DEFAULT_OBJECT_SIZE = 16384 # 16 KiB, ровный размер, не JPEG
DEFAULT_OBJECT_ID = 0x4C303436 # "L046"
DEFAULT_OBJECT_SEED = 46
CONTINUOUS_PREAMBLE_THRESHOLD = 0.80
DEFAULT_MAX_FRAMES = 256
DEFAULT_MAX_SEARCH_ITERATIONS = 1024
# ============================================================ Детерминированный объект
def make_deterministic_object(
size: int = DEFAULT_OBJECT_SIZE,
seed: int = DEFAULT_OBJECT_SEED,
) -> bytes:
"""Собрать детерминированный псевдослучайный бинарный объект (не JPEG)."""
if size <= 0:
raise ValueError("Размер объекта должен быть положительным")
generator = np.random.default_rng(seed)
return generator.integers(0, 256, size=size, dtype=np.uint8).tobytes()
def object_sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
# ============================================================ Уровень объекта -> кадры
@dataclass(frozen=True)
class FramePlan:
"""Один фрагмент объекта, уже упакованный в кадр Lab045."""
fragment_index: int
fragment_count: int
encoded_payload_length: int # реальная длина 12+data до добивки нулями
plan: lab045.TransmissionPlan
def _pad_payload_to_packet(payload: bytes) -> bytes:
if len(payload) > PACKET_PAYLOAD_SIZE:
raise ValueError("Нагрузка фрагмента больше размера пакета Lab045")
return payload + bytes(PACKET_PAYLOAD_SIZE - len(payload))
def build_object_frames(
data: bytes,
object_id: int = DEFAULT_OBJECT_ID,
) -> list[FramePlan]:
"""Разбить объект и упаковать каждый фрагмент в кадр Lab045."""
object_fragments = fragments.split_image_bytes(
data,
object_id,
fragment_data_size=FRAGMENT_DATA_SIZE,
)
frame_plans: list[FramePlan] = []
for fragment in object_fragments:
payload = fragments.encode_image_fragment(fragment)
packet_payload = _pad_payload_to_packet(payload)
plan = lab045.build_transmission_plan(packet_payload)
frame_plans.append(
FramePlan(
fragment_index=fragment.fragment_index,
fragment_count=fragment.fragment_count,
encoded_payload_length=len(payload),
plan=plan,
)
)
return frame_plans
def expected_fragment_data_length(
fragment_index: int,
fragment_count: int,
original_size: int,
) -> int | None:
"""Длина данных фрагмента, вычисленная только из полей заголовка.
Возвращает None при несогласованных полях (fail-closed), не бросая.
"""
if fragment_count <= 0 or original_size <= 0:
return None
if not 0 <= fragment_index < fragment_count:
return None
full = FRAGMENT_DATA_SIZE
# Верхняя граница: объект из fragment_count кусков по full байт не может
# быть больше fragment_count*full. Это отсекает абсурдный original_size
# для любого индекса и не даёт полю управлять размером аллокаций.
if original_size > fragment_count * full:
return None
if (fragment_count - 1) * full >= original_size and fragment_count > 1:
return None
if fragment_index < fragment_count - 1:
return full
last = original_size - (fragment_count - 1) * full
if not 1 <= last <= full:
return None
return last
def recover_fragment_from_payload(packet_payload: bytes) -> fragments.ImageFragment | None:
"""Снять добивку и разобрать фрагмент из 256-байтной нагрузки. Fail-closed.
Использует только принятую нагрузку. Переданный объект неизвестен.
"""
try:
fragment = fragments.decode_image_fragment(packet_payload)
except (fragments.ImageFragmentError, TypeError, ValueError):
return None
real_length = expected_fragment_data_length(
fragment.fragment_index,
fragment.fragment_count,
fragment.original_size,
)
if real_length is None or len(fragment.data) < real_length:
return None
return replace(fragment, data=fragment.data[:real_length])
# ============================================================ Fail-closed сборщик
@dataclass
class AssemblerCounters:
accepted: int = 0
duplicates: int = 0
conflicting: int = 0
rejected: int = 0
class ObjectAssembler:
"""Собрать объект из независимо принятых фрагментов, ничего не бросая.
Любой плохой вход (чужой object_id, несогласованный счётчик/размер,
невозможный индекс, конфликтующий дубликат, malformed) отвергается и не
выходит наружу исключением. Пока не собраны все индексы —
``object_complete()`` равен False, ``recovered_object()`` равен None.
Фиктивный объект не создаётся.
"""
def __init__(self) -> None:
self._object_id: int | None = None
self._fragment_count: int | None = None
self._original_size: int | None = None
self._by_index: dict[int, fragments.ImageFragment] = {}
self.counters = AssemblerCounters()
def offer(self, fragment: fragments.ImageFragment | None) -> str:
"""Предложить фрагмент. Вернуть accepted/duplicate/conflicting/rejected."""
if fragment is None:
self.counters.rejected += 1
return "rejected"
try:
index = int(fragment.fragment_index)
count = int(fragment.fragment_count)
size = int(fragment.original_size)
object_id = int(fragment.image_id)
if count <= 0 or size <= 0 or not 0 <= index < count:
self.counters.rejected += 1
return "rejected"
if len(fragment.data) < 1:
self.counters.rejected += 1
return "rejected"
if self._object_id is None:
self._object_id = object_id
self._fragment_count = count
self._original_size = size
elif (
object_id != self._object_id
or count != self._fragment_count
or size != self._original_size
):
self.counters.rejected += 1
return "rejected"
existing = self._by_index.get(index)
if existing is not None:
if existing.data == fragment.data:
self.counters.duplicates += 1
return "duplicate"
self.counters.conflicting += 1
return "conflicting"
self._by_index[index] = fragment
self.counters.accepted += 1
return "accepted"
except Exception:
self.counters.rejected += 1
return "rejected"
def missing_indexes(self) -> list[int]:
if self._fragment_count is None:
return []
return [i for i in range(self._fragment_count) if i not in self._by_index]
def object_complete(self) -> bool:
return (
self._fragment_count is not None
and len(self._by_index) == self._fragment_count
and not self.missing_indexes()
)
def recovered_object(self) -> bytes | None:
"""Собрать объект, только если он полон. Иначе None. Не бросает."""
if not self.object_complete():
return None
try:
return fragments.reassemble_image(list(self._by_index.values()))
except fragments.ImageFragmentError:
return None
# ============================================================ Канал (программный)
ChannelFunction = Callable[[lab045.TransmissionPlan, int], np.ndarray]
def ideal_channel(plan: lab045.TransmissionPlan, frame_index: int) -> np.ndarray:
del frame_index
return lab045.modulate_bits_to_ideal_soft(plan.post_preamble_bits)
def make_noisy_channel(
noise_sigma: float,
base_seed: int,
) -> ChannelFunction:
"""Детерминированный гауссов канал на уровне soft BPSK-символов.
Гауссов шум реалистичнее жёстких переворотов знака и сохраняет мягкую
информацию, которую использует декодер Витерби. Каждый кадр получает свой
воспроизводимый seed ``base_seed + frame_index``. Уровень sigma выбран
так, чтобы soft Viterbi исправлял все кадры (см. docs/lab046_current_state):
при большем sigma отдельный кадр может не декодироваться, и без erasure-
FEC объект окажется неполным — это отдельный сценарий, а не подгон.
"""
def channel(plan: lab045.TransmissionPlan, frame_index: int) -> np.ndarray:
return lab045.simulate_soft_channel(
plan,
noise_sigma=noise_sigma,
seed=base_seed + frame_index,
)
return channel
# ============================================================ Прогон объекта
@dataclass
class ObjectTransportResult:
object_size: int
object_sha256: str
fragment_count: int
protocol_packet_bytes: int
bpsk_frame_count: int
total_bpsk_symbols: int
total_waveform_samples: int
airtime_seconds: float
crc_pass_count: int
crc_fail_count: int
pre_fec_coded_bit_errors: int
post_fec_protocol_bit_errors: int
valid_fragments: int
duplicates: int
conflicting: int
rejected: int
missing_indexes: list[int]
object_complete: bool
recovered_object_sha256: str | None
object_byte_errors: int | None
sha_match: bool
def run_object_transport(
data: bytes,
object_id: int = DEFAULT_OBJECT_ID,
channel: ChannelFunction = ideal_channel,
delivery_order: list[int] | None = None,
duplicate_indexes: tuple[int, ...] = (),
drop_indexes: tuple[int, ...] = (),
corrupt_indexes: tuple[int, ...] = (),
) -> ObjectTransportResult:
"""Полный программный тракт объект -> кадры -> канал -> приём -> сборка.
Приёмник независим: TX-объект и TX-биты не используются для решений о
приёме, FEC, метаданных или завершении объекта. Диагностическое сравнение
с оригиналом делается только после независимой сборки.
"""
frame_plans = build_object_frames(data, object_id)
frame_by_index = {frame.fragment_index: frame for frame in frame_plans}
order = delivery_order if delivery_order is not None else [f.fragment_index for f in frame_plans]
assembler = ObjectAssembler()
crc_pass = 0
crc_fail = 0
pre_fec_errors = 0
post_fec_errors = 0
def deliver(frame_index: int, corrupt: bool) -> None:
nonlocal crc_pass, crc_fail, pre_fec_errors, post_fec_errors
frame = frame_by_index[frame_index]
soft = channel(frame.plan, frame_index)
if corrupt:
# Детерминированно инвертируем весь coded-поток (после защищённого
# PHY header): CRC гарантированно не пройдёт, пакет отвергается.
soft = soft.copy()
soft[lab045.PHY_HEADER_BIT_COUNT:] = -soft[lab045.PHY_HEADER_BIT_COUNT:]
reception = lab045.analyze_post_preamble_soft(soft)
# Диагностика pre/post-FEC только после независимого CRC-решения.
diagnostic = lab045.diagnostic_pre_and_post_fec(reception, soft, frame.plan)
pre = diagnostic["pre_fec_coded"]["bit_errors"]
post = diagnostic["post_fec_packet"]["bit_errors"]
if pre is not None:
pre_fec_errors += pre
if post is not None:
post_fec_errors += post
if reception.analysis["crc"] == lab045.CRC_PASS and reception.payload_bytes is not None:
crc_pass += 1
fragment = recover_fragment_from_payload(reception.payload_bytes)
assembler.offer(fragment)
else:
crc_fail += 1
for frame_index in order:
deliver(frame_index, corrupt=frame_index in corrupt_indexes)
for frame_index in duplicate_indexes:
deliver(frame_index, corrupt=False)
for frame_index in drop_indexes:
assembler._by_index.pop(frame_index, None)
recovered = assembler.recovered_object()
recovered_sha = object_sha256(recovered) if recovered is not None else None
original_sha = object_sha256(data)
if recovered is not None and len(recovered) == len(data):
byte_errors = int(np.count_nonzero(
np.frombuffer(recovered, dtype=np.uint8) != np.frombuffer(data, dtype=np.uint8)
))
elif recovered is not None:
byte_errors = max(len(recovered), len(data))
else:
byte_errors = None
total_symbols = sum(f.plan.frame_symbol_count for f in frame_plans)
total_samples = sum(f.plan.waveform_sample_count for f in frame_plans)
airtime = sum(f.plan.airtime_seconds for f in frame_plans)
return ObjectTransportResult(
object_size=len(data),
object_sha256=original_sha,
fragment_count=len(frame_plans),
protocol_packet_bytes=len(frame_plans[0].plan.packet),
bpsk_frame_count=len(frame_plans),
total_bpsk_symbols=total_symbols,
total_waveform_samples=total_samples,
airtime_seconds=airtime,
crc_pass_count=crc_pass,
crc_fail_count=crc_fail,
pre_fec_coded_bit_errors=pre_fec_errors,
post_fec_protocol_bit_errors=post_fec_errors,
valid_fragments=assembler.counters.accepted,
duplicates=assembler.counters.duplicates,
conflicting=assembler.counters.conflicting,
rejected=assembler.counters.rejected,
missing_indexes=assembler.missing_indexes(),
object_complete=assembler.object_complete(),
recovered_object_sha256=recovered_sha,
object_byte_errors=byte_errors,
sha_match=(recovered_sha == original_sha and recovered is not None),
)
# ============================================================ Lab046A: один continuous soft stream
def _bits_to_float32_soft(bits: np.ndarray) -> np.ndarray:
"""Преобразовать BPSK-биты в компактный нормализованный soft stream."""
values = np.asarray(bits)
if values.ndim != 1 or not np.all((values == 0) | (values == 1)):
raise ValueError("bits должны быть одномерной двоичной последовательностью")
soft = values.astype(np.float32)
soft *= np.float32(2.0)
soft -= np.float32(1.0)
return soft
def malformed_header_candidate_soft() -> np.ndarray:
"""Преамбула PRBS7 и намеренно неверная контрольная сумма PHY header."""
header = lab045.encode_phy_header(lab045.MIN_PACKET_BYTES).copy()
checksum_bit = lab045.PHY_HEADER_FIELDS_BITS - 1
first_copy = checksum_bit * lab045.HEADER_REPETITION
header[first_copy:first_copy + 2] ^= 1
candidate_bits = np.concatenate((lab045.lab044.preamble_bits(), header))
return _bits_to_float32_soft(candidate_bits)
@dataclass(frozen=True)
class ContinuousSoftStreamPlan:
"""TX-side описание потока; ``frame_starts`` запрещено передавать scanner-у."""
soft_symbols: np.ndarray
frame_plans: tuple[FramePlan, ...]
frame_starts: tuple[int, ...]
leading_background_symbols: int
inter_frame_background_symbols: int
trailing_background_symbols: int
malformed_candidate_symbols: int
noise_sigma: float
@property
def total_frame_symbols(self) -> int:
return sum(frame.plan.frame_symbol_count for frame in self.frame_plans)
@property
def frame_waveform_samples(self) -> int:
return sum(frame.plan.waveform_sample_count for frame in self.frame_plans)
@property
def extra_background_samples(self) -> int:
symbols = (
self.leading_background_symbols
+ self.inter_frame_background_symbols
+ self.trailing_background_symbols
)
return symbols * lab045.SAMPLES_PER_SYMBOL
@property
def equivalent_total_samples(self) -> int:
"""Полные Lab045-формы плюс намеренно добавленные фоновые области."""
auxiliary = self.malformed_candidate_symbols * lab045.SAMPLES_PER_SYMBOL
return self.frame_waveform_samples + self.extra_background_samples + auxiliary
@property
def equivalent_duration_seconds(self) -> float:
return self.equivalent_total_samples / lab045.SAMPLE_RATE_HZ
def build_continuous_soft_stream(
frame_plans: list[FramePlan] | tuple[FramePlan, ...],
*,
leading_background_symbols: int = 0,
inter_frame_background: dict[int, int] | None = None,
trailing_background_symbols: int = 0,
background_sigma: float = 0.12,
noise_sigma: float = 0.0,
seed: int = 46001,
prepend_malformed_candidate: bool = False,
) -> ContinuousSoftStreamPlan:
"""Собрать один float32 stream без передачи RX известных границ кадров.
В stream представлены нормализованные soft BPSK-символы на выходе уже
проверенного Lab045 timing/CFO/phase front-end. Полные sample-level формы
Lab045 не материализуются: их точная длина учитывается свойствами плана.
"""
frames = tuple(frame_plans)
if not frames:
raise ValueError("нужен хотя бы один кадр")
if leading_background_symbols < 0 or trailing_background_symbols < 0:
raise ValueError("длина фоновой области не может быть отрицательной")
if background_sigma < 0.0 or noise_sigma < 0.0:
raise ValueError("sigma не может быть отрицательной")
gaps = dict(inter_frame_background or {})
for frame_index, symbol_count in gaps.items():
if not 0 <= frame_index < len(frames) - 1 or symbol_count < 0:
raise ValueError("некорректная inter-frame область")
random_generator = np.random.default_rng(seed)
chunks: list[np.ndarray] = []
frame_starts: list[int] = []
current_offset = 0
def append_background(symbol_count: int) -> None:
nonlocal current_offset
if symbol_count == 0:
return
background = random_generator.normal(
0.0,
background_sigma,
size=symbol_count,
).astype(np.float32)
chunks.append(background)
current_offset += symbol_count
append_background(leading_background_symbols)
malformed_symbols = 0
if prepend_malformed_candidate:
malformed = malformed_header_candidate_soft()
chunks.append(malformed)
malformed_symbols = int(malformed.size)
current_offset += malformed_symbols
inter_frame_symbols = 0
for frame_index, frame in enumerate(frames):
frame_starts.append(current_offset)
frame_soft = _bits_to_float32_soft(frame.plan.frame_bits)
chunks.append(frame_soft)
current_offset += int(frame_soft.size)
if frame_index < len(frames) - 1:
gap = int(gaps.get(frame_index, 0))
inter_frame_symbols += gap
append_background(gap)
append_background(trailing_background_symbols)
stream = np.concatenate(chunks, dtype=np.float32)
if noise_sigma > 0.0:
noise = random_generator.normal(0.0, noise_sigma, size=stream.size)
stream += noise.astype(np.float32)
return ContinuousSoftStreamPlan(
soft_symbols=stream,
frame_plans=frames,
frame_starts=tuple(frame_starts),
leading_background_symbols=leading_background_symbols,
inter_frame_background_symbols=inter_frame_symbols,
trailing_background_symbols=trailing_background_symbols,
malformed_candidate_symbols=malformed_symbols,
noise_sigma=noise_sigma,
)
def _normalized_soft_preamble_correlation(soft_stream: np.ndarray) -> np.ndarray:
"""Нормированная корреляция PRBS7 через проверенный helper Lab043."""
preamble = _bits_to_float32_soft(lab045.lab044.preamble_bits())
if soft_stream.size < preamble.size:
return np.empty(0, dtype=np.float32)
scores = np.asarray(
lab045.lab044.lab043._normalized_sliding_correlation(soft_stream, preamble),
dtype=np.float32,
)
scores[~np.isfinite(scores)] = 0.0
return np.clip(scores, 0.0, 1.0)
def _preamble_candidate_indexes(scores: np.ndarray, threshold: float) -> np.ndarray:
"""Вернуть все пороговые кандидаты в порядке потока, ничего не удаляя заранее."""
return np.flatnonzero(scores >= threshold)
@dataclass(frozen=True)
class ContinuousFrameReception:
candidate_start: int
frame_end: int
preamble_score: float
packet_bytes: int
fragment_index: int | None
reception: lab045.IndependentReception
@dataclass(frozen=True)
class ContinuousReceiveResult:
stream_symbol_count: int
preamble_candidates: int
valid_phy_headers: int
fec_pass_count: int
crc_pass_count: int
crc_fail_count: int
valid_frame_count: int
rejected_candidates: int
valid_fragments: int
duplicate_fragments: int
conflicting_fragments: int
rejected_fragments: int
missing_indexes: tuple[int, ...]
object_complete: bool
recovered_object: bytes | None
candidate_starts: tuple[int, ...]
cursor_history: tuple[int, ...]
search_iterations: int
termination_reason: str
max_frames_reached: bool
max_search_iterations_reached: bool
correlation_storage_bytes: int
frame_receptions: tuple[ContinuousFrameReception, ...]
def scan_continuous_soft_stream(
soft_stream: np.ndarray,
*,
preamble_threshold: float = CONTINUOUS_PREAMBLE_THRESHOLD,
max_frames: int = DEFAULT_MAX_FRAMES,
max_search_iterations: int = DEFAULT_MAX_SEARCH_ITERATIONS,
) -> ContinuousReceiveResult:
"""Независимо найти и декодировать все кадры в одном soft stream.
Единственный вход с данными — сам stream. TX offsets, число кадров,
fragment order, packets, coded bits и исходный объект функции неизвестны.
"""
try:
stream = np.asarray(soft_stream, dtype=np.float32)
except (TypeError, ValueError):
stream = np.empty((0, 0), dtype=np.float32)
if not 0.0 < preamble_threshold <= 1.0:
raise ValueError("preamble_threshold должен лежать в (0, 1]")
if max_frames < 0 or max_search_iterations < 0:
raise ValueError("safety limits не могут быть отрицательными")
assembler = ObjectAssembler()
scores = np.empty(0, dtype=np.float32)
if stream.ndim == 1 and stream.size and np.all(np.isfinite(stream)):
scores = _normalized_soft_preamble_correlation(stream)
candidate_indexes = _preamble_candidate_indexes(scores, preamble_threshold)
cursor = 0
cursor_history = [cursor]
candidate_starts: list[int] = []
frame_receptions: list[ContinuousFrameReception] = []
preamble_candidates = 0
valid_headers = 0
fec_pass = 0
crc_pass = 0
crc_fail = 0
valid_frames = 0
rejected_candidates = 0
search_iterations = 0
termination_reason = "END_OF_STREAM"
max_frames_reached = False
max_iterations_reached = False
while True:
position = int(np.searchsorted(candidate_indexes, cursor, side="left"))
if position >= candidate_indexes.size:
break
candidate_start = int(candidate_indexes[position])
if search_iterations >= max_search_iterations:
termination_reason = "MAX_SEARCH_ITERATIONS_REACHED"
max_iterations_reached = True
break
if valid_frames >= max_frames:
termination_reason = "MAX_FRAMES_REACHED"
max_frames_reached = True
break
search_iterations += 1
preamble_candidates += 1
candidate_starts.append(candidate_start)
preamble_end = candidate_start + lab045.PREAMBLE_BIT_COUNT
header_end = preamble_end + lab045.PHY_HEADER_BIT_COUNT
header = lab045.decode_phy_header_soft(stream[preamble_end:header_end])
if not header["passed"]:
rejected_candidates += 1
cursor = candidate_start + 1
cursor_history.append(cursor)
continue
valid_headers += 1
packet_bytes = int(header["packet_bytes"])
coded_length = lab045.conv.encoded_bit_count(packet_bytes * 8)
frame_end = header_end + coded_length
if frame_end > stream.size:
rejected_candidates += 1
cursor = candidate_start + 1
cursor_history.append(cursor)
continue
reception = lab045.analyze_post_preamble_soft(stream[preamble_end:frame_end])
if reception.analysis["fec_decode_pass"]:
fec_pass += 1
fragment = None
fragment_index = None
if reception.analysis["crc"] == lab045.CRC_PASS and reception.payload_bytes is not None:
crc_pass += 1
valid_frames += 1
fragment = recover_fragment_from_payload(reception.payload_bytes)
if fragment is not None:
fragment_index = int(fragment.fragment_index)
assembler.offer(fragment)
else:
if reception.analysis["crc"] == lab045.CRC_FAIL:
crc_fail += 1
rejected_candidates += 1
frame_receptions.append(ContinuousFrameReception(
candidate_start=candidate_start,
frame_end=frame_end,
preamble_score=float(scores[candidate_start]),
packet_bytes=packet_bytes,
fragment_index=fragment_index,
reception=reception,
))
cursor = frame_end
cursor_history.append(cursor)
recovered = assembler.recovered_object()
if stream.ndim != 1 or not stream.size or not np.all(np.isfinite(stream)):
termination_reason = "INVALID_STREAM"
return ContinuousReceiveResult(
stream_symbol_count=int(stream.size),
preamble_candidates=preamble_candidates,
valid_phy_headers=valid_headers,
fec_pass_count=fec_pass,
crc_pass_count=crc_pass,
crc_fail_count=crc_fail,
valid_frame_count=valid_frames,
rejected_candidates=rejected_candidates,
valid_fragments=assembler.counters.accepted,
duplicate_fragments=assembler.counters.duplicates,
conflicting_fragments=assembler.counters.conflicting,
rejected_fragments=assembler.counters.rejected,
missing_indexes=tuple(assembler.missing_indexes()),
object_complete=assembler.object_complete(),
recovered_object=recovered,
candidate_starts=tuple(candidate_starts),
cursor_history=tuple(cursor_history),
search_iterations=search_iterations,
termination_reason=termination_reason,
max_frames_reached=max_frames_reached,
max_search_iterations_reached=max_iterations_reached,
correlation_storage_bytes=int(scores.nbytes),
frame_receptions=tuple(frame_receptions),
)
@dataclass(frozen=True)
class ContinuousValidationResult:
stream_plan: ContinuousSoftStreamPlan
receive: ContinuousReceiveResult
object_sha256: str
recovered_object_sha256: str | None
object_byte_errors: int | None
sha_match: bool
pre_fec_coded_bit_errors: int
post_fec_protocol_bit_errors: int
def _continuous_post_receive_diagnostics(
receive: ContinuousReceiveResult,
stream_plan: ContinuousSoftStreamPlan,
) -> tuple[int, int]:
"""Сравнить с TX только после независимых detector/header/FEC/CRC решений."""
frame_by_index = {frame.fragment_index: frame for frame in stream_plan.frame_plans}
pre_fec_errors = 0
post_fec_errors = 0
for record in receive.frame_receptions:
if record.fragment_index is None:
continue
frame = frame_by_index.get(record.fragment_index)
if frame is None:
continue
post_start = record.candidate_start + lab045.PREAMBLE_BIT_COUNT
post_soft = stream_plan.soft_symbols[post_start:record.frame_end]
diagnostic = lab045.diagnostic_pre_and_post_fec(
record.reception,
post_soft,
frame.plan,
)
pre = diagnostic["pre_fec_coded"]["bit_errors"]
post = diagnostic["post_fec_packet"]["bit_errors"]
if pre is not None:
pre_fec_errors += int(pre)
if post is not None:
post_fec_errors += int(post)
return pre_fec_errors, post_fec_errors
def run_continuous_object_transport(
data: bytes,
*,
object_id: int = DEFAULT_OBJECT_ID,
leading_background_symbols: int = 0,
inter_frame_background: dict[int, int] | None = None,
trailing_background_symbols: int = 0,
noise_sigma: float = 0.0,
seed: int = 46001,
prepend_malformed_candidate: bool = False,
max_frames: int = DEFAULT_MAX_FRAMES,
max_search_iterations: int = DEFAULT_MAX_SEARCH_ITERATIONS,
) -> ContinuousValidationResult:
"""Построить TX stream, независимо принять его и лишь затем сравнить с TX."""
frame_plans = build_object_frames(data, object_id)
stream_plan = build_continuous_soft_stream(
frame_plans,
leading_background_symbols=leading_background_symbols,
inter_frame_background=inter_frame_background,
trailing_background_symbols=trailing_background_symbols,
noise_sigma=noise_sigma,
seed=seed,
prepend_malformed_candidate=prepend_malformed_candidate,
)
receive = scan_continuous_soft_stream(
stream_plan.soft_symbols,
max_frames=max_frames,
max_search_iterations=max_search_iterations,
)
pre_fec_errors, post_fec_errors = _continuous_post_receive_diagnostics(
receive,
stream_plan,
)
recovered = receive.recovered_object
recovered_sha = object_sha256(recovered) if recovered is not None else None
original_sha = object_sha256(data)
if recovered is None:
byte_errors = None
elif len(recovered) != len(data):
byte_errors = max(len(recovered), len(data))
else:
byte_errors = int(np.count_nonzero(
np.frombuffer(recovered, dtype=np.uint8)
!= np.frombuffer(data, dtype=np.uint8)
))
return ContinuousValidationResult(
stream_plan=stream_plan,
receive=receive,
object_sha256=original_sha,
recovered_object_sha256=recovered_sha,
object_byte_errors=byte_errors,
sha_match=(recovered is not None and recovered_sha == original_sha),
pre_fec_coded_bit_errors=pre_fec_errors,
post_fec_protocol_bit_errors=post_fec_errors,
)
# ============================================================ Airtime / goodput
def goodput_report(result: ObjectTransportResult) -> dict:
"""Честная baseline-оценка эфирного времени и полезной пропускной способности."""
airtime = result.airtime_seconds
object_bytes = result.object_size
goodput_bytes_per_second = object_bytes / airtime if airtime > 0 else 0.0
goodput_kbit_per_second = goodput_bytes_per_second * 8 / 1000.0
useful_data_bits = object_bytes * 8
total_symbols = result.total_bpsk_symbols
return {
"airtime_seconds": airtime,
"goodput_bytes_per_second": goodput_bytes_per_second,
"goodput_kbit_per_second": goodput_kbit_per_second,
"useful_data_bits": useful_data_bits,
"total_bpsk_symbols": total_symbols,
"useful_bits_per_symbol": useful_data_bits / total_symbols if total_symbols else 0.0,
"overhead": {
"fragment_metadata_bytes_per_packet": fragments.FRAGMENT_HEADER_SIZE,
"protocol_header_plus_crc_bytes_per_packet": lab045.HEADER_SIZE + lab045.CRC_SIZE,
"convolutional_rate": "1/2",
"phy_preamble_bits_per_frame": lab045.PREAMBLE_BIT_COUNT,
"protected_phy_header_bits_per_frame": lab045.PHY_HEADER_BIT_COUNT,
"guard_plus_rrc_symbol_times_per_frame": 2 * lab045.GUARD_SYMBOL_COUNT + lab045.RRC_SPAN_SYMBOLS,
},
}
def continuous_accounting_report(result: ContinuousValidationResult) -> dict:
"""Длины полной Lab045-формы и фактических тестовых областей Lab046A."""
plan = result.stream_plan
baseline_duration = plan.frame_waveform_samples / lab045.SAMPLE_RATE_HZ
actual_duration = plan.equivalent_duration_seconds
useful_bits = len(result.receive.recovered_object or b"") * 8
return {
"total_frame_symbols": plan.total_frame_symbols,
"frame_waveform_samples": plan.frame_plans[0].plan.waveform_sample_count,
"all_frame_waveform_samples": plan.frame_waveform_samples,
"leading_samples": plan.leading_background_symbols * lab045.SAMPLES_PER_SYMBOL,
"inter_frame_gap_samples": plan.inter_frame_background_symbols * lab045.SAMPLES_PER_SYMBOL,
"trailing_samples": plan.trailing_background_symbols * lab045.SAMPLES_PER_SYMBOL,
"total_stream_samples": plan.equivalent_total_samples,
"baseline_duration_seconds": baseline_duration,
"actual_duration_seconds": actual_duration,
"baseline_goodput_kbit_per_second": (
useful_bits / baseline_duration / 1000.0 if baseline_duration else 0.0
),
"actual_goodput_kbit_per_second": (
useful_bits / actual_duration / 1000.0 if actual_duration else 0.0
),
"soft_stream_bytes": int(plan.soft_symbols.nbytes),
"correlation_scores_bytes": result.receive.correlation_storage_bytes,
"cumulative_energy_bytes": int((plan.soft_symbols.size + 1) * 8),
}
# ============================================================ Функциональные тесты
def run_functional_tests() -> None:
"""Быстрые самопроверки без данных эксперимента (вызывается quick_gate)."""
obj = make_deterministic_object(2048, seed=1)
ideal = run_object_transport(obj, channel=ideal_channel)
assert ideal.object_complete and ideal.sha_match and ideal.object_byte_errors == 0
assert ideal.crc_fail_count == 0
frames = build_object_frames(obj)
assert frames[0].plan.frame_symbol_count == 5724
assert all(len(f.plan.packet) == 268 for f in frames)
# Fragment round trip через существующий слой без канала.
fs = fragments.split_image_bytes(obj, DEFAULT_OBJECT_ID, FRAGMENT_DATA_SIZE)
rebuilt = fragments.reassemble_image(fs)
assert rebuilt == obj
# Последний короткий фрагмент восстанавливается по длине из заголовка.
last = fs[-1]
padded = _pad_payload_to_packet(fragments.encode_image_fragment(last))
recovered = recover_fragment_from_payload(padded)
assert recovered is not None and recovered.data == last.data
# Короткая Lab046A-проверка: два кадра находятся в одном stream без offsets.
continuous = run_continuous_object_transport(obj[:300])
assert continuous.receive.valid_frame_count == 2
assert continuous.receive.preamble_candidates == 2
assert continuous.sha_match
# ============================================================ main
def main() -> int:
parser = argparse.ArgumentParser(description="Lab046 multi-packet object transport (software)")
parser.add_argument("--object-size", type=int, default=DEFAULT_OBJECT_SIZE)
parser.add_argument("--seed", type=int, default=DEFAULT_OBJECT_SEED)
parser.add_argument("--sigma", type=float, default=0.40)
arguments = parser.parse_args()
obj = make_deterministic_object(arguments.object_size, arguments.seed)
ideal = run_object_transport(obj, channel=ideal_channel)
noisy = run_object_transport(
obj,
channel=make_noisy_channel(arguments.sigma, base_seed=46000),
)
report = goodput_report(ideal)
background_gaps = {
frame_index: symbol_count
for frame_index, symbol_count in ((7, 11), (23, 19), (51, 7))
if frame_index < ideal.fragment_count - 1
}
continuous_ideal = run_continuous_object_transport(obj)
continuous_background = run_continuous_object_transport(
obj,
leading_background_symbols=37,
inter_frame_background=background_gaps,
trailing_background_symbols=43,
seed=464,
)
continuous_noisy = run_continuous_object_transport(
obj,
noise_sigma=arguments.sigma,
seed=46001,
)
continuous_malformed = run_continuous_object_transport(
obj,
prepend_malformed_candidate=True,
)
continuous_accounting = continuous_accounting_report(continuous_ideal)
background_accounting = continuous_accounting_report(continuous_background)
print("Lab046 software object transport")
print(f"object size = {ideal.object_size} bytes")
print(f"object sha256 = {ideal.object_sha256}")
print(f"fragment data size = {FRAGMENT_DATA_SIZE} bytes")
print(f"fragment count = {ideal.fragment_count}")
print(f"protocol packet = {ideal.protocol_packet_bytes} bytes")
print(f"bpsk frames = {ideal.bpsk_frame_count}")
print(f"total bpsk symbols = {ideal.total_bpsk_symbols}")
print(f"total waveform samples = {ideal.total_waveform_samples}")
print(f"estimated airtime = {report['airtime_seconds']:.2f} s")
print(f"goodput = {report['goodput_bytes_per_second']:.1f} bytes/s = {report['goodput_kbit_per_second']:.2f} kbit/s")
print("--- ideal ---")
print(f"crc pass/fail = {ideal.crc_pass_count}/{ideal.crc_fail_count}")
print(f"object complete = {ideal.object_complete}, sha match = {ideal.sha_match}, byte errors = {ideal.object_byte_errors}")
print(f"--- noisy (gaussian sigma={arguments.sigma}) ---")
print(f"pre-FEC coded errors = {noisy.pre_fec_coded_bit_errors}")
print(f"post-FEC protocol errors = {noisy.post_fec_protocol_bit_errors}")
print(f"crc pass/fail = {noisy.crc_pass_count}/{noisy.crc_fail_count}")
print(f"object complete = {noisy.object_complete}, sha match = {noisy.sha_match}, byte errors = {noisy.object_byte_errors}")
print("--- Lab046A ideal continuous soft stream ---")
print(f"preamble candidates = {continuous_ideal.receive.preamble_candidates}")
print(f"valid headers/frames = {continuous_ideal.receive.valid_phy_headers}/{continuous_ideal.receive.valid_frame_count}")
print(f"fec/crc pass = {continuous_ideal.receive.fec_pass_count}/{continuous_ideal.receive.crc_pass_count}")
print(f"rejected candidates = {continuous_ideal.receive.rejected_candidates}")
print(f"object complete = {continuous_ideal.receive.object_complete}, sha match = {continuous_ideal.sha_match}, byte errors = {continuous_ideal.object_byte_errors}")
print("--- Lab046A background continuous soft stream ---")
print(f"leading/inter-frame/trailing symbols = {continuous_background.stream_plan.leading_background_symbols}/{continuous_background.stream_plan.inter_frame_background_symbols}/{continuous_background.stream_plan.trailing_background_symbols}")
print(f"valid frames = {continuous_background.receive.valid_frame_count}, sha match = {continuous_background.sha_match}")
print(f"actual equivalent samples/duration = {background_accounting['total_stream_samples']}/{background_accounting['actual_duration_seconds']:.6f} s")
print(f"--- Lab046A noisy continuous soft stream (sigma={arguments.sigma}) ---")
print(f"valid frames = {continuous_noisy.receive.valid_frame_count}")
print(f"pre/post-FEC errors = {continuous_noisy.pre_fec_coded_bit_errors}/{continuous_noisy.post_fec_protocol_bit_errors}")
print(f"crc pass/fail = {continuous_noisy.receive.crc_pass_count}/{continuous_noisy.receive.crc_fail_count}")
print(f"object complete = {continuous_noisy.receive.object_complete}, sha match = {continuous_noisy.sha_match}")
print("--- Lab046A malformed candidate ---")
print(f"preamble candidates = {continuous_malformed.receive.preamble_candidates}")
print(f"rejected candidates = {continuous_malformed.receive.rejected_candidates}")
print(f"valid frames = {continuous_malformed.receive.valid_frame_count}, sha match = {continuous_malformed.sha_match}")
print("--- Lab046A accounting ---")
print(f"frame waveform samples = {continuous_accounting['frame_waveform_samples']}")
print(f"all frame waveform samples = {continuous_accounting['all_frame_waveform_samples']}")
print(f"inter-frame gap samples = {continuous_accounting['inter_frame_gap_samples']}")
print(f"airtime = {continuous_accounting['baseline_duration_seconds']:.6f} s")
print(f"useful goodput = {continuous_accounting['baseline_goodput_kbit_per_second']:.6f} kbit/s")
print(f"soft/correlation/cumulative-energy bytes = {continuous_accounting['soft_stream_bytes']}/{continuous_accounting['correlation_scores_bytes']}/{continuous_accounting['cumulative_energy_bytes']}")
print("RF TX = 0 (software-only)")
return 0
if __name__ == "__main__":
raise SystemExit(main())