"""Lab046B: несколько Lab045-кадров в одном непрерывном IQ-потоке. Эксперимент программный. Передатчик формирует настоящие Lab045 waveform, а приёмник получает только один ``complex64`` массив. Известные TX-границы, payload, packet и coded bits не входят в API приёмника. Сравнение с TX выполняется отдельно и только после независимого CRC PASS. RF TX = 0. Аппаратура не используется. """ from __future__ import annotations import argparse import math import sys from dataclasses import dataclass from pathlib import Path if __package__ in (None, ""): sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import numpy as np from experiments import lab045_hackrf_bpsk_fec as lab045 from protocol import bpsk_radio as radio FRAME_COUNT = 5 LEADING_BACKGROUND_SAMPLES = 137 INTER_FRAME_GAPS = (211, 353, 587, 809) TRAILING_BACKGROUND_SAMPLES = 233 IMPAIRED_CFO_HZ = 0.2 IMPAIRED_PHASE_RAD = 0.7 IMPAIRED_AMPLITUDE = 0.65 NOISE_SEED = 46002 NOISE_SIGMA = 0.17 IQ_PREAMBLE_THRESHOLD = 0.65 MAX_FRAMES = 32 MAX_SEARCH_ITERATIONS = 256 IMPAIRMENT_BLOCK_SAMPLES = 262_144 def _amplitude_db(ratio: float) -> float: return 20.0 * math.log10(max(float(ratio), 1e-300)) def _wrap_phase(value: float) -> float: return float((value + math.pi) % (2.0 * math.pi) - math.pi) def deterministic_payload(frame_index: int) -> bytes: """Создать различимый детерминированный payload одного Lab045-кадра.""" if frame_index < 0: raise ValueError("frame_index не может быть отрицательным") random_generator = np.random.Generator(np.random.PCG64(46_000 + frame_index)) payload = random_generator.integers( 0, 256, lab045.PAYLOAD_SIZE_BYTES, dtype=np.uint8, ) payload[:4] = np.frombuffer(frame_index.to_bytes(4, "big"), dtype=np.uint8) return payload.tobytes() def build_frame_plans(frame_count: int = FRAME_COUNT) -> tuple[lab045.TransmissionPlan, ...]: if frame_count <= 0: raise ValueError("нужен хотя бы один кадр") return tuple( lab045.build_transmission_plan(deterministic_payload(frame_index)) for frame_index in range(frame_count) ) def malformed_candidate_waveform() -> np.ndarray: """Сформировать настоящий IQ-prefix PRBS7 с заведомо неверным PHY magic.""" header = lab045.encode_phy_header(lab045.MIN_PACKET_BYTES).copy() header[: lab045.HEADER_REPETITION] ^= 1 bits = np.concatenate((lab045.lab044.preamble_bits(), header)) shaped = lab045.lab044._scale_and_carrier(radio.bpsk_modulate(bits)) raw = lab045.lab044.lab043.to_cs8_bytes(shaped) return lab045.lab044.signed_iq_bytes_to_complex64(raw) @dataclass(frozen=True) class ContinuousIqPlan: """TX-side описание; ``frame_starts`` запрещено передавать scanner-у.""" samples: np.ndarray frame_plans: tuple[lab045.TransmissionPlan, ...] frame_starts: tuple[int, ...] leading_background_samples: int inter_frame_gaps: tuple[int, ...] trailing_background_samples: int cfo_hz: float phase_rad: float amplitude: float noise_sigma: float noise_seed: int malformed_candidate_start: int | None malformed_candidate_sample_count: int @property def sample_count(self) -> int: return int(self.samples.size) @property def memory_bytes(self) -> int: return int(self.samples.nbytes) def _apply_impairments_in_place( samples: np.ndarray, *, cfo_hz: float, phase_rad: float, amplitude: float, noise_sigma: float, noise_seed: int, ) -> None: """Применить канал блоками без постоянной full-size complex128 копии.""" if amplitude <= 0.0 or not np.isfinite(amplitude): raise ValueError("amplitude должна быть положительной и конечной") if noise_sigma < 0.0 or not np.isfinite(noise_sigma): raise ValueError("noise_sigma должна быть неотрицательной и конечной") if not np.isfinite(cfo_hz) or not np.isfinite(phase_rad): raise ValueError("CFO и phase должны быть конечными") phase_step = 2.0 * math.pi * cfo_hz / lab045.SAMPLE_RATE_HZ random_generator = np.random.Generator(np.random.PCG64(noise_seed)) for start in range(0, samples.size, IMPAIRMENT_BLOCK_SAMPLES): end = min(samples.size, start + IMPAIRMENT_BLOCK_SAMPLES) indexes = np.arange(start, end, dtype=np.float64) rotation = np.exp(1j * (phase_rad + phase_step * indexes)).astype(np.complex64) samples[start:end] *= np.complex64(amplitude) * rotation if noise_sigma > 0.0: real_noise = random_generator.standard_normal(end - start, dtype=np.float32) imaginary_noise = random_generator.standard_normal(end - start, dtype=np.float32) samples.real[start:end] += np.float32(noise_sigma) * real_noise samples.imag[start:end] += np.float32(noise_sigma) * imaginary_noise def build_continuous_iq_stream( frame_plans: tuple[lab045.TransmissionPlan, ...] | list[lab045.TransmissionPlan], *, leading_background_samples: int = LEADING_BACKGROUND_SAMPLES, inter_frame_gaps: tuple[int, ...] | list[int] | None = None, trailing_background_samples: int = TRAILING_BACKGROUND_SAMPLES, cfo_hz: float = 0.0, phase_rad: float = 0.0, amplitude: float = 1.0, noise_sigma: float = 0.0, noise_seed: int = NOISE_SEED, prepend_malformed_candidate: bool = False, malformed_to_frame_gap_samples: int = 317, ) -> ContinuousIqPlan: """Собрать один непрерывный ``complex64`` stream из полных Lab045 waveform.""" plans = tuple(frame_plans) if not plans: raise ValueError("нужен хотя бы один кадр") if leading_background_samples < 0 or trailing_background_samples < 0: raise ValueError("фоновые интервалы не могут быть отрицательными") if malformed_to_frame_gap_samples < 0: raise ValueError("интервал после ложного кандидата не может быть отрицательным") if inter_frame_gaps is None: gaps = tuple(0 for _ in range(len(plans) - 1)) else: gaps = tuple(int(value) for value in inter_frame_gaps) if len(gaps) != len(plans) - 1 or any(value < 0 for value in gaps): raise ValueError("нужен один неотрицательный gap между каждой парой кадров") malformed = malformed_candidate_waveform() if prepend_malformed_candidate else None malformed_count = 0 if malformed is None else int(malformed.size) total_samples = ( leading_background_samples + malformed_count + (malformed_to_frame_gap_samples if malformed is not None else 0) + sum(plan.waveform_sample_count for plan in plans) + sum(gaps) + trailing_background_samples ) samples = np.zeros(total_samples, dtype=np.complex64) cursor = leading_background_samples malformed_start = None if malformed is not None: malformed_start = cursor samples[cursor : cursor + malformed_count] = malformed cursor += malformed_count + malformed_to_frame_gap_samples frame_starts: list[int] = [] for frame_index, plan in enumerate(plans): frame_starts.append(cursor) waveform = lab045.lab044.signed_iq_bytes_to_complex64(lab045.build_tx_cs8(plan)) if waveform.dtype != np.complex64 or waveform.size != plan.waveform_sample_count: raise ValueError("Lab045 waveform имеет неожиданную форму") samples[cursor : cursor + waveform.size] = waveform cursor += int(waveform.size) if frame_index < len(gaps): cursor += gaps[frame_index] if cursor + trailing_background_samples != total_samples: raise RuntimeError("внутренняя ошибка учёта длины IQ stream") _apply_impairments_in_place( samples, cfo_hz=cfo_hz, phase_rad=phase_rad, amplitude=amplitude, noise_sigma=noise_sigma, noise_seed=noise_seed, ) return ContinuousIqPlan( samples=samples, frame_plans=plans, frame_starts=tuple(frame_starts), leading_background_samples=leading_background_samples, inter_frame_gaps=gaps, trailing_background_samples=trailing_background_samples, cfo_hz=cfo_hz, phase_rad=phase_rad, amplitude=amplitude, noise_sigma=noise_sigma, noise_seed=noise_seed, malformed_candidate_start=malformed_start, malformed_candidate_sample_count=malformed_count, ) @dataclass(frozen=True) class IqPreambleCandidate: coarse_start_sample: int score: float over_median_db: float over_second_db: float def _decimate_then_downconvert(samples: np.ndarray) -> np.ndarray: """Получить компактный detection stream без full-size baseband копии.""" decimation = lab045.lab044.lab043.DETECTION_DECIMATION decimated = lab045.lab044.lab043.resample_poly( samples, 1, decimation, window=("kaiser", 8), ).astype(np.complex64, copy=False) decimated -= np.mean(decimated, dtype=np.complex128).astype(np.complex64) sample_rate = lab045.SAMPLE_RATE_HZ / decimation indexes = np.arange(decimated.size, dtype=np.float64) rotation = np.exp( -2j * math.pi * lab045.CARRIER_OFFSET_HZ * indexes / sample_rate ).astype(np.complex64) decimated *= rotation return decimated def _local_candidate_indexes(scores: np.ndarray, threshold: float) -> np.ndarray: """Вернуть все локальные максимумы выше порога без global peak/second gate.""" values = np.asarray(scores, dtype=np.float32) if values.ndim != 1 or values.size == 0: return np.empty(0, dtype=np.int64) if not 0.0 < threshold <= 1.0: raise ValueError("threshold должен лежать в (0, 1]") left = np.empty_like(values) right = np.empty_like(values) left[0] = -np.inf left[1:] = values[:-1] right[-1] = -np.inf right[:-1] = values[1:] return np.flatnonzero( (values >= np.float32(threshold)) & (values >= left) & (values > right) ) def _detect_iq_preamble_candidates( samples: np.ndarray, threshold: float, ) -> tuple[tuple[IqPreambleCandidate, ...], int]: reference = lab045.lab044._preamble_reference_samples() rx_decimated = _decimate_then_downconvert(samples) reference_decimated = _decimate_then_downconvert(reference) if rx_decimated.size < reference_decimated.size: return (), int(rx_decimated.nbytes + reference_decimated.nbytes) scores = np.asarray( lab045.lab044.lab043._normalized_sliding_correlation( rx_decimated, reference_decimated, ), dtype=np.float32, ) scores[~np.isfinite(scores)] = 0.0 scores = np.clip(scores, 0.0, 1.0) indexes = _local_candidate_indexes(scores, threshold) median = float(np.median(scores)) candidate_scores = scores[indexes] decimation = lab045.lab044.lab043.DETECTION_DECIMATION candidates: list[IqPreambleCandidate] = [] for position, index in enumerate(indexes): if candidate_scores.size > 1: others = np.delete(candidate_scores, position) second = float(np.max(others)) else: second = 0.0 score = float(scores[index]) candidates.append(IqPreambleCandidate( coarse_start_sample=int(index) * decimation, score=score, over_median_db=( _amplitude_db(score / median) if median > 0.0 else float("inf") ), over_second_db=( _amplitude_db(score / second) if second > 0.0 else float("inf") ), )) storage_bytes = int( rx_decimated.nbytes + reference_decimated.nbytes + scores.nbytes ) return tuple(candidates), storage_bytes @dataclass(frozen=True) class SoftWindow: passed: bool normalized_real: np.ndarray timing_offset_samples: int symbol_sample_phase: int nominal_sample: int carrier: dict reason: str | None def _extract_soft_window( samples: np.ndarray, frame_start: int, symbol_count: int, ) -> SoftWindow: waveform_samples = ( (symbol_count + 2 * lab045.GUARD_SYMBOL_COUNT) * lab045.SAMPLES_PER_SYMBOL + lab045.RRC_SPAN_SYMBOLS * lab045.SAMPLES_PER_SYMBOL ) if frame_start < 0 or frame_start + waveform_samples > samples.size: return SoftWindow(False, np.empty(0), 0, 0, 0, {}, "TRUNCATED_IQ_WINDOW") segment = samples[frame_start : frame_start + waveform_samples] matched, nominal = lab045.lab044.lab043._matched_waveform( segment, 0, waveform_samples, ) preamble_bits = lab045.lab044.preamble_bits() preamble_symbols = radio.bpsk_modulate(preamble_bits) timing = lab045.lab044.lab043._preamble_timing_offset( matched, nominal, preamble_symbols, ) carrier = lab045.lab044.lab043._evaluate_symbols( matched, nominal, timing, preamble_bits, lab045.PREAMBLE_BIT_COUNT, ) if carrier["abs_h"] == 0.0: return SoftWindow(False, np.empty(0), timing, 0, nominal, carrier, "ZERO_GAIN") symbols = matched[ nominal + timing : nominal + timing + symbol_count * lab045.SAMPLES_PER_SYMBOL : lab045.SAMPLES_PER_SYMBOL ].astype(np.complex128) if symbols.size != symbol_count: return SoftWindow(False, np.empty(0), timing, 0, nominal, carrier, "TRUNCATED_SYMBOLS") symbol_time = (np.arange(symbol_count, dtype=np.float64) + 0.5) / lab045.SYMBOL_RATE normalized = ( symbols * np.exp(-2j * math.pi * carrier["cfo_hz"] * symbol_time) / (carrier["abs_h"] * np.exp(1j * carrier["phase_rad"])) ) sample_phase = int((frame_start + nominal + timing) % lab045.SAMPLES_PER_SYMBOL) return SoftWindow( True, normalized.real.astype(np.float64, copy=False), int(timing), sample_phase, int(nominal), carrier, None, ) @dataclass(frozen=True) class IqFrameReception: detection_number: int coarse_start_sample: int refined_start_sample: int frame_end_sample: int preamble_score: float over_median_db: float over_second_db: float timing_offset_samples: int symbol_sample_phase: int nominal_sample: int estimated_cfo_hz: float estimated_phase_rad: float bpsk_polarity: str preamble_errors: int snr_db: float post_preamble_soft: np.ndarray reception: lab045.IndependentReception @dataclass(frozen=True) class CandidateDecode: refined_start_sample: int next_sample: int phy_header_pass: bool fec_pass: bool crc: str accepted: bool reason: str frame: IqFrameReception | None def _decode_iq_candidate( samples: np.ndarray, candidate: IqPreambleCandidate, detection_number: int, ) -> CandidateDecode: reference = lab045.lab044._preamble_reference_samples() try: refined = lab045.lab044.lab043._refine_frame_start( samples, reference, candidate.coarse_start_sample, ) except (ValueError, IndexError): return CandidateDecode( candidate.coarse_start_sample, candidate.coarse_start_sample + 1, False, False, lab045.CRC_NOT_REACHED, False, "REFINEMENT_FAILED", None, ) probe_symbols = lab045.PREAMBLE_BIT_COUNT + lab045.PHY_HEADER_BIT_COUNT probe = _extract_soft_window(samples, refined, probe_symbols) if not probe.passed: return CandidateDecode( refined, candidate.coarse_start_sample + 1, False, False, lab045.CRC_NOT_REACHED, False, probe.reason or "PROBE_FAILED", None, ) header = lab045.decode_phy_header_soft( probe.normalized_real[ lab045.PREAMBLE_BIT_COUNT : lab045.PREAMBLE_BIT_COUNT + lab045.PHY_HEADER_BIT_COUNT ] ) if not header["passed"]: return CandidateDecode( refined, candidate.coarse_start_sample + 1, False, False, lab045.CRC_NOT_REACHED, False, f"PHY_HEADER_{header['reason']}", None, ) packet_bytes = int(header["packet_bytes"]) coded_length = lab045.conv.encoded_bit_count(packet_bytes * 8) frame_symbols = lab045.PREAMBLE_BIT_COUNT + lab045.PHY_HEADER_BIT_COUNT + coded_length full = _extract_soft_window(samples, refined, frame_symbols) if not full.passed: return CandidateDecode( refined, candidate.coarse_start_sample + 1, True, False, lab045.CRC_NOT_REACHED, False, full.reason or "FULL_EXTRACTION_FAILED", None, ) post_preamble = full.normalized_real[lab045.PREAMBLE_BIT_COUNT:] reception = lab045.analyze_post_preamble_soft(post_preamble) crc = str(reception.analysis["crc"]) accepted = crc == lab045.CRC_PASS waveform_samples = ( (frame_symbols + 2 * lab045.GUARD_SYMBOL_COUNT) * lab045.SAMPLES_PER_SYMBOL + lab045.RRC_SPAN_SYMBOLS * lab045.SAMPLES_PER_SYMBOL ) frame_end = refined + waveform_samples preamble_hard = ( full.normalized_real[: lab045.PREAMBLE_BIT_COUNT] >= 0.0 ).astype(np.uint8) preamble_errors = int(np.count_nonzero( preamble_hard != lab045.lab044.preamble_bits() )) carrier = full.carrier frame = IqFrameReception( detection_number=detection_number, coarse_start_sample=candidate.coarse_start_sample, refined_start_sample=refined, frame_end_sample=frame_end, preamble_score=candidate.score, over_median_db=candidate.over_median_db, over_second_db=candidate.over_second_db, timing_offset_samples=full.timing_offset_samples, symbol_sample_phase=full.symbol_sample_phase, nominal_sample=full.nominal_sample, estimated_cfo_hz=float(carrier["cfo_hz"]), estimated_phase_rad=float(carrier["phase_rad"]), bpsk_polarity=str(carrier["polarity"]), preamble_errors=preamble_errors, snr_db=float(carrier["snr_db"]), post_preamble_soft=post_preamble, reception=reception, ) reason = "CRC_PASS" if accepted else f"CRC_{crc}" return CandidateDecode( refined, frame_end if accepted else candidate.coarse_start_sample + 1, True, bool(reception.analysis["fec_decode_pass"]), crc, accepted, reason, frame, ) @dataclass(frozen=True) class ContinuousIqReceiveResult: iq_sample_count: int iq_dtype: str preamble_candidates: int processed_candidates: int rejected_candidates: int phy_header_pass_count: int fec_pass_count: int crc_pass_count: int crc_fail_count: int accepted_frames: tuple[IqFrameReception, ...] rejected_reasons: tuple[str, ...] candidate_starts: tuple[int, ...] cursor_history: tuple[int, ...] search_iterations: int termination_reason: str max_frames_reached: bool max_search_iterations_reached: bool detection_storage_bytes: int def scan_continuous_iq( iq_samples: np.ndarray, *, preamble_threshold: float = IQ_PREAMBLE_THRESHOLD, max_frames: int = MAX_FRAMES, max_search_iterations: int = MAX_SEARCH_ITERATIONS, ) -> ContinuousIqReceiveResult: """Найти несколько кадров только по одному IQ stream, без TX offsets.""" try: samples = np.asarray(iq_samples, dtype=np.complex64) except (TypeError, ValueError): samples = np.empty((0, 0), dtype=np.complex64) 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 не могут быть отрицательными") if samples.ndim != 1 or samples.size == 0 or not np.all(np.isfinite(samples)): return ContinuousIqReceiveResult( int(samples.size), str(samples.dtype), 0, 0, 0, 0, 0, 0, 0, (), ("INVALID_IQ_STREAM",), (), (0,), 0, "INVALID_IQ_STREAM", False, False, 0, ) candidates, storage_bytes = _detect_iq_preamble_candidates( samples, preamble_threshold, ) cursor = 0 cursor_history = [cursor] accepted_frames: list[IqFrameReception] = [] rejected_reasons: list[str] = [] candidate_starts: list[int] = [] phy_header_pass = 0 fec_pass = 0 crc_pass = 0 crc_fail = 0 iterations = 0 termination = "END_OF_STREAM" max_frames_reached = False max_iterations_reached = False for candidate in candidates: if candidate.coarse_start_sample < cursor: continue if len(accepted_frames) >= max_frames: termination = "MAX_FRAMES_REACHED" max_frames_reached = True break if iterations >= max_search_iterations: termination = "MAX_SEARCH_ITERATIONS_REACHED" max_iterations_reached = True break iterations += 1 candidate_starts.append(candidate.coarse_start_sample) decoded = _decode_iq_candidate(samples, candidate, iterations) phy_header_pass += int(decoded.phy_header_pass) fec_pass += int(decoded.fec_pass) crc_pass += int(decoded.crc == lab045.CRC_PASS) crc_fail += int(decoded.crc == lab045.CRC_FAIL) if decoded.accepted and decoded.frame is not None: accepted_frames.append(decoded.frame) else: rejected_reasons.append(decoded.reason) next_cursor = max(cursor + 1, int(decoded.next_sample)) cursor = next_cursor cursor_history.append(cursor) return ContinuousIqReceiveResult( iq_sample_count=int(samples.size), iq_dtype=str(samples.dtype), preamble_candidates=len(candidates), processed_candidates=iterations, rejected_candidates=len(rejected_reasons), phy_header_pass_count=phy_header_pass, fec_pass_count=fec_pass, crc_pass_count=crc_pass, crc_fail_count=crc_fail, accepted_frames=tuple(accepted_frames), rejected_reasons=tuple(rejected_reasons), candidate_starts=tuple(candidate_starts), cursor_history=tuple(cursor_history), search_iterations=iterations, termination_reason=termination, max_frames_reached=max_frames_reached, max_search_iterations_reached=max_iterations_reached, detection_storage_bytes=storage_bytes, ) @dataclass(frozen=True) class FrameDiagnostic: detection_number: int tx_frame_index: int sample_offset: int sample_offset_error: int symbol_sample_phase: int estimated_cfo_hz: float estimated_channel_phase_rad: float preamble_errors: int over_median_db: float over_second_db: float snr_db: float pre_fec_coded_bit_errors: int post_fec_protocol_bit_errors: int crc: str def diagnostics_after_crc( receive: ContinuousIqReceiveResult, stream_plan: ContinuousIqPlan, ) -> tuple[FrameDiagnostic, ...]: """Сопоставить RX с TX только после независимого CRC PASS.""" plan_by_packet = { plan.packet: (frame_index, plan) for frame_index, plan in enumerate(stream_plan.frame_plans) } rows: list[FrameDiagnostic] = [] for frame in receive.accepted_frames: packet = frame.reception.packet_bytes if packet not in plan_by_packet: continue frame_index, plan = plan_by_packet[packet] comparison = lab045.diagnostic_pre_and_post_fec( frame.reception, frame.post_preamble_soft, plan, ) pre_fec_errors = comparison["pre_fec_coded"]["bit_errors"] post_fec_errors = comparison["post_fec_packet"]["bit_errors"] if pre_fec_errors is None or post_fec_errors is None: raise RuntimeError("CRC PASS frame оказался несопоставим с TX diagnostic") phase_reference_sample = ( frame.refined_start_sample + frame.nominal_sample + frame.timing_offset_samples - 0.5 * lab045.SAMPLES_PER_SYMBOL ) estimated_channel_phase = _wrap_phase( frame.estimated_phase_rad - 2.0 * math.pi * frame.estimated_cfo_hz * phase_reference_sample / lab045.SAMPLE_RATE_HZ ) rows.append(FrameDiagnostic( detection_number=frame.detection_number, tx_frame_index=frame_index, sample_offset=frame.refined_start_sample, sample_offset_error=( frame.refined_start_sample - stream_plan.frame_starts[frame_index] ), symbol_sample_phase=frame.symbol_sample_phase, estimated_cfo_hz=frame.estimated_cfo_hz, estimated_channel_phase_rad=estimated_channel_phase, preamble_errors=frame.preamble_errors, over_median_db=frame.over_median_db, over_second_db=frame.over_second_db, snr_db=frame.snr_db, pre_fec_coded_bit_errors=int(pre_fec_errors), post_fec_protocol_bit_errors=int(post_fec_errors), crc=str(frame.reception.analysis["crc"]), )) return tuple(rows) @dataclass(frozen=True) class IqScenarioResult: name: str frame_count: int sample_count: int iq_dtype: str iq_memory_bytes: int frame_starts: tuple[int, ...] inter_frame_gaps: tuple[int, ...] leading_background_samples: int trailing_background_samples: int cfo_hz: float phase_rad: float amplitude: float noise_sigma: float noise_seed: int malformed_candidate_start: int | None receive: ContinuousIqReceiveResult diagnostics: tuple[FrameDiagnostic, ...] @property def pre_fec_coded_bit_errors(self) -> int: return sum(row.pre_fec_coded_bit_errors for row in self.diagnostics) @property def post_fec_protocol_bit_errors(self) -> int: return sum(row.post_fec_protocol_bit_errors for row in self.diagnostics) @property def estimated_cfo_hz(self) -> float | None: if not self.diagnostics: return None return float(np.median([row.estimated_cfo_hz for row in self.diagnostics])) @property def estimated_phase_rad(self) -> float | None: if not self.diagnostics: return None values = np.asarray( [row.estimated_channel_phase_rad for row in self.diagnostics], dtype=np.float64, ) return float(np.angle(np.mean(np.exp(1j * values)))) def run_iq_scenario( name: str, frame_plans: tuple[lab045.TransmissionPlan, ...], **stream_arguments, ) -> IqScenarioResult: stream_plan = build_continuous_iq_stream(frame_plans, **stream_arguments) receive = scan_continuous_iq(stream_plan.samples) diagnostics = diagnostics_after_crc(receive, stream_plan) return IqScenarioResult( name=name, frame_count=len(frame_plans), sample_count=stream_plan.sample_count, iq_dtype=str(stream_plan.samples.dtype), iq_memory_bytes=stream_plan.memory_bytes, frame_starts=stream_plan.frame_starts, inter_frame_gaps=stream_plan.inter_frame_gaps, leading_background_samples=stream_plan.leading_background_samples, trailing_background_samples=stream_plan.trailing_background_samples, cfo_hz=stream_plan.cfo_hz, phase_rad=stream_plan.phase_rad, amplitude=stream_plan.amplitude, noise_sigma=stream_plan.noise_sigma, noise_seed=stream_plan.noise_seed, malformed_candidate_start=stream_plan.malformed_candidate_start, receive=receive, diagnostics=diagnostics, ) @dataclass(frozen=True) class Lab046bValidation: ideal: IqScenarioResult impaired: IqScenarioResult noisy: IqScenarioResult malformed: IqScenarioResult def run_lab046b_validation(frame_count: int = FRAME_COUNT) -> Lab046bValidation: plans = build_frame_plans(frame_count) gaps = INTER_FRAME_GAPS[: frame_count - 1] if len(gaps) != frame_count - 1: gaps = tuple(137 + 2 * index for index in range(frame_count - 1)) common = { "leading_background_samples": LEADING_BACKGROUND_SAMPLES, "inter_frame_gaps": gaps, "trailing_background_samples": TRAILING_BACKGROUND_SAMPLES, } ideal = run_iq_scenario("IDEAL", plans, **common) impaired = run_iq_scenario( "CFO_PHASE_AMPLITUDE", plans, **common, cfo_hz=IMPAIRED_CFO_HZ, phase_rad=IMPAIRED_PHASE_RAD, amplitude=IMPAIRED_AMPLITUDE, ) noisy = run_iq_scenario( "COMPLEX_GAUSSIAN_NOISE", plans, **common, noise_sigma=NOISE_SIGMA, noise_seed=NOISE_SEED, ) malformed = run_iq_scenario( "MALFORMED_CANDIDATE", plans[:1], leading_background_samples=LEADING_BACKGROUND_SAMPLES, inter_frame_gaps=(), trailing_background_samples=TRAILING_BACKGROUND_SAMPLES, prepend_malformed_candidate=True, ) return Lab046bValidation(ideal, impaired, noisy, malformed) def run_functional_tests() -> None: scores = np.asarray([0.0, 0.8, 0.1, 0.9, 0.0], dtype=np.float32) assert _local_candidate_indexes(scores, 0.65).tolist() == [1, 3] assert scan_continuous_iq(np.empty(0, dtype=np.complex64)).termination_reason == "INVALID_IQ_STREAM" def _print_scenario(result: IqScenarioResult) -> None: print(f"{result.name}:") print(f" samples = {result.sample_count}") print(f" dtype = {result.iq_dtype}") print(f" preamble candidates = {result.receive.preamble_candidates}") print(f" processed candidates = {result.receive.processed_candidates}") print(f" accepted frames = {len(result.receive.accepted_frames)}") print(f" rejected candidates = {result.receive.rejected_candidates}") print(f" PHY header PASS = {result.receive.phy_header_pass_count}") print(f" FEC PASS = {result.receive.fec_pass_count}") print(f" CRC PASS/FAIL = {result.receive.crc_pass_count}/{result.receive.crc_fail_count}") print(f" pre/post-FEC bit errors = {result.pre_fec_coded_bit_errors}/{result.post_fec_protocol_bit_errors}") print(f" estimated CFO = {result.estimated_cfo_hz}") print(f" estimated phase = {result.estimated_phase_rad}") for row in result.diagnostics: print( " frame " f"{row.tx_frame_index}: offset={row.sample_offset}, " f"offset_error={row.sample_offset_error}, " f"sample_phase={row.symbol_sample_phase}, " f"CFO={row.estimated_cfo_hz:.9f} Hz, " f"phase={row.estimated_channel_phase_rad:.9f} rad, " f"preamble_errors={row.preamble_errors}, " f"peak/median={row.over_median_db:.3f} dB, " f"peak/second={row.over_second_db:.3f} dB, " f"SNR={row.snr_db:.3f} dB, " f"pre/post-FEC={row.pre_fec_coded_bit_errors}/" f"{row.post_fec_protocol_bit_errors}, CRC={row.crc}" ) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--frame-count", type=int, default=FRAME_COUNT) arguments = parser.parse_args() result = run_lab046b_validation(arguments.frame_count) print("LAB046B SOFTWARE CONTINUOUS IQ VALIDATION") print("RF TX = 0") print("known TX frame boundaries at RX = NO") print( f"main IQ = {result.ideal.sample_count} samples, " f"{result.ideal.iq_dtype}, {result.ideal.iq_memory_bytes} bytes" ) for scenario in (result.ideal, result.impaired, result.noisy, result.malformed): _print_scenario(scenario) return 0 if __name__ == "__main__": raise SystemExit(main())