Lab046A: validate continuous soft-symbol frame scanning

This commit is contained in:
LittleSam129
2026-09-01 13:18:42 +03:00
parent 1027eef53d
commit e8d85ae4b7
4 changed files with 879 additions and 6 deletions

View File

@@ -32,11 +32,16 @@ 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
@@ -49,6 +54,10 @@ 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
# ============================================================ Детерминированный объект
@@ -421,6 +430,453 @@ def run_object_transport(
)
# ============================================================ 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
@@ -451,6 +907,35 @@ def goodput_report(result: ObjectTransportResult) -> dict:
}
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),
}
# ============================================================ Функциональные тесты
@@ -477,6 +962,12 @@ def run_functional_tests() -> None:
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
@@ -495,6 +986,30 @@ def main() -> int:
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")
@@ -515,6 +1030,32 @@ def main() -> int:
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