"""Свойства локального обнаружителя Lab052 без оборудования и внешних данных.""" from __future__ import annotations import numpy as np import pytest from types import SimpleNamespace from experiments import lab052_candidate_detection as lab from experiments.lab050_streaming_receiver import streaming_mode def waveform(payload: bytes, *, seed: int = 52001, ebn0_db: float = 7, phase: float = .73, cfo_hz: float = 10, scale: float = 1) -> np.ndarray: random_generator = np.random.default_rng(seed) # Генератор Lab045 ограничен payload=256, но приёмный PHY допускает # packet 12..1024. Собираем короткий кадр штатными кодерами для границ. physical = lab.phy.lab045 packet = physical.build_packet(payload, physical.MESSAGE_TYPE, physical.SEQUENCE_NUMBER) coded = physical.conv.encode(physical.lab044.bytes_to_bits(packet)) permutation = physical.interleaver.build_permutation(len(coded), physical.INTERLEAVER_COLUMNS) bits = np.concatenate((physical.lab044.preamble_bits(), physical.encode_phy_header(len(packet)), physical.interleaver.interleave(coded, permutation))) plan = SimpleNamespace(frame_bits=bits, frame_symbol_count=len(bits)) raw = lab.phy.build_parameterized_tx_cs8(plan, lab.phy.PhyMode(500000)) components = np.frombuffer(raw, np.int8).astype(np.float32) values = components[0::2] + 1j * components[1::2] values *= np.exp(1j * (phase + 2 * np.pi * cfo_hz * np.arange(len(values)) / 8000000)).astype(np.complex64) energy = float(np.sum(abs(values) ** 2)) / plan.frame_symbol_count sigma = np.sqrt(energy / (2 * 10 ** (ebn0_db / 10))) values += sigma * (random_generator.normal(size=len(values)) + 1j * random_generator.normal(size=len(values))) return (values * scale).astype(np.complex64) def decode(samples, receiver): packets = [] for start in range(0, len(samples), receiver.block_samples): packets.extend(receiver.push_iq_block(samples[start:start + receiver.block_samples])) assert receiver.candidate_state_count == 0 if hasattr(receiver, "candidate_state_count") else True packets.extend(receiver.finish()) return packets def test_cluster_of_false_peaks_has_one_strongest_representative(): scores = np.zeros(1600, np.float32) scores[[600, 620, 654, 700, 730]] = [.05, .07, .12, .06, .04] assert lab.cluster_peak_indexes(scores, 256, .03).tolist() == [654] def test_two_minimum_frames_do_not_compete_even_at_opposite_coarse_errors(): distance = lab.MIN_FRAME_SAMPLES - 2 * lab.COARSE_ERROR_SAMPLES scores = np.zeros(8000, np.float32) indexes = [1000, 1000 + distance // 8] scores[indexes] = [.3, .9] assert lab.cluster_peak_indexes(scores, lab.CLUSTER_WIDTH_SAMPLES // 8, .03).tolist() == indexes def test_equal_peaks_use_reproducible_local_tie(): scores = np.zeros(1000, np.float32) scores[[100, 120, 600]] = .2 assert lab.cluster_peak_indexes(scores, 100, .03).tolist() == [120, 600] def test_equal_peak_chain_does_not_extend_cluster_width(): scores = np.zeros(500, np.float32) scores[[100, 180, 260]] = .2 assert lab.cluster_peak_indexes(scores, 100, .03).tolist() == [180, 260] def test_single_loud_half_cannot_pass_split_check(): random_generator = np.random.default_rng(52002) reference = random_generator.choice([-1., 1.], 2560).astype(np.complex64) samples = (random_generator.normal(size=2560) + 1j * random_generator.normal(size=2560)).astype(np.complex64) samples[:1280] = reference[:1280] * 100 assert lab.part_correlations(samples, reference, 1)[0] > lab.part_threshold(1) assert np.any(lab.part_correlations(samples, reference, 2) < lab.part_threshold(2)) @pytest.mark.parametrize("phase,scale", [(0, .001), (.73, 1), (np.pi, 100), (-1.6, .1)]) def test_true_preamble_preserves_unknown_gain_phase_and_polarity(phase, scale): random_generator = np.random.default_rng(52003) reference = random_generator.choice([-1., 1.], 2560).astype(np.complex64) samples = reference * np.exp(1j * phase) * scale assert np.all(lab.part_correlations(samples, reference, 2) >= lab.part_threshold(2)) def test_noise_is_rejected_before_strict_timing(monkeypatch): random_generator = np.random.default_rng(52004) samples = (random_generator.normal(size=750000) + 1j * random_generator.normal(size=750000)).astype(np.complex64) def forbidden(*args, **kwargs): pytest.fail("Чистый шум этого seed не должен достигнуть строгого тракта") monkeypatch.setattr(lab.offline, "decode_candidate_from_memmap", forbidden) receiver = lab.StreamingReceiver() assert decode(samples, receiver) == [] assert receiver.detection_statistics["cheap_stage_rejected"] > 0 assert receiver.front_statistics["unique_soft_windows"] == 0 @pytest.mark.parametrize("ebn0_db,scale,cfo", [(7, 1, 10), (2, .02, -100), (1, .2, 100)]) def test_weak_true_packet_matches_lab051(ebn0_db, scale, cfo): payload = bytes(range(128)) values = waveform(payload, seed=52005, ebn0_db=ebn0_db, scale=scale, cfo_hz=cfo) samples = np.concatenate((np.zeros(63217, np.complex64), values, np.zeros(350000, np.complex64))) baseline = decode(samples, lab.previous.StreamingReceiver()) optimized = decode(samples, lab.StreamingReceiver()) assert [packet.payload for packet in optimized] == [packet.payload for packet in baseline] assert [packet.payload for packet in baseline] == ([payload] if ebn0_db >= 2 else []) @pytest.mark.parametrize("block_samples,prefix", [(32749, 32749 - 3), (65536, 65536 - 8), (65536, 65536 + 7)]) def test_minimum_adjacent_frames_at_block_boundary_are_both_decoded(block_samples, prefix): # Минимальный packet: пустой payload, заголовок+CRC=12 байт. first = waveform(b"", seed=52006, phase=.4) second = waveform(b"x", seed=52007, phase=-2) assert len(first) == lab.MIN_FRAME_SAMPLES samples = np.concatenate((np.zeros(prefix, np.complex64), first, second, np.zeros(350000, np.complex64))) receiver = lab.StreamingReceiver(block_samples=block_samples) assert [packet.payload for packet in decode(samples, receiver)] == [b"", b"x"] def test_changing_background_with_weak_packet(): random_generator = np.random.default_rng(52008) samples = (random_generator.normal(size=700000) + 1j * random_generator.normal(size=700000)).astype(np.complex64) samples[:180000] *= .001 samples[180000:300000] *= 20 samples[300000:] *= .03 payload = b"varying local background" values = waveform(payload, seed=52009, ebn0_db=2, scale=.02) samples[370013:370013 + len(values)] = values assert [packet.payload for packet in decode(samples, lab.StreamingReceiver())] == [payload] @pytest.mark.parametrize("true_start", [30000, 65536 - 8, 65536 + 7]) def test_false_strong_preamble_does_not_hide_weak_valid_packet(true_start): decoy_start = true_start - 20000 samples = np.zeros(700000, np.complex64) decoy = waveform(b"", seed=52022, ebn0_db=100)[:lab.PREAMBLE_SAMPLES + 512] payload = b"actual valid short packet" true = waveform(payload, seed=52021, ebn0_db=7, scale=.2) samples[decoy_start:decoy_start + len(decoy)] += decoy samples[true_start:true_start + len(true)] += true expected = decode(samples, lab.previous.StreamingReceiver()) actual = decode(samples, lab.StreamingReceiver()) assert [packet.payload for packet in expected] == [payload] assert [(packet.start_sample, packet.payload) for packet in actual] == [ (packet.start_sample, packet.payload) for packet in expected] def test_constant_templates_are_exact_and_restored_on_exception(): receiver = lab.StreamingReceiver() owner = lab.phy.lab045.lab044 originals = (owner._preamble_reference_samples, owner.preamble_bits, lab.previous.lab043._matched_filter_taps) with streaming_mode(lab.phy.PhyMode(500000)): np.testing.assert_array_equal(receiver.raw_reference, originals[0]()) np.testing.assert_array_equal(receiver.preamble_bits, originals[1]()) np.testing.assert_array_equal(receiver.matched_taps, originals[2]()) with pytest.raises(RuntimeError): with receiver.constant_templates(): raise RuntimeError("restore templates") assert originals == (owner._preamble_reference_samples, owner.preamble_bits, lab.previous.lab043._matched_filter_taps) @pytest.mark.parametrize("count", [1, 2, 3]) def test_shared_split_scores_equal_individual_vdot(count): random_generator = np.random.default_rng(52024) values = (random_generator.normal(size=9000) + 1j * random_generator.normal(size=9000)).astype(np.complex64) values[4000:] *= .003 reference = random_generator.choice([-1., 1.], size=2560).astype(np.complex64) front = object.__new__(lab.CandidateFrontEnd) front.detection_values = values scores = front.split_scores(reference, count, len(values) - len(reference) + 1) for index in (0, 1001, 3999, len(scores) - 1): np.testing.assert_allclose(scores[index], lab.part_correlations(values[index:index+len(reference)], reference, count), atol=2e-6, rtol=2e-5) def test_bounded_candidate_state_and_finish(): random_generator = np.random.default_rng(52010) receiver = lab.StreamingReceiver() samples = (random_generator.normal(size=65536) + 1j * random_generator.normal(size=65536)).astype(np.complex64) for index in range(120): receiver.push_iq_block(samples * (.01 if index % 2 else 10)) assert receiver.candidate_state_count == 0 assert receiver.buffer.size <= receiver.buffer_limit_samples assert receiver.decimator.values.size <= (receiver.buffer_limit_samples + 7) // 8 receiver.finish() assert receiver.buffer.size == receiver.decimator.values.size == receiver.candidate_state_count == 0 assert receiver.peak_cluster_count <= 20 assert receiver.observer is None def test_20k_does_not_enter_new_filter_and_mode_is_restored(): old = lab.previous.lab045.SYMBOL_RATE with pytest.raises(ValueError): lab.StreamingReceiver(20000) lab.run_functional_tests() assert lab.previous.lab045.SYMBOL_RATE == old with streaming_mode(lab.phy.PhyMode(20000)): assert lab.previous.lab045.SAMPLES_PER_SYMBOL == 400 assert lab.previous.lab045.SYMBOL_RATE == old @pytest.mark.parametrize("rule", [dict(parts=4), dict(parts=-1), dict(cluster_width_samples=26720), dict(cluster_width_samples=3)]) def test_invalid_rule_rejected(rule): with pytest.raises(ValueError): lab.DetectorRule(**rule)