346 lines
16 KiB
Python
346 lines
16 KiB
Python
"""Регрессии изменённых вычислений/очередей Lab053, без файлов экспериментов."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from threading import Event, Lock, Thread
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from experiments import lab052_candidate_detection as baseline
|
|
from experiments import lab053_packet_processing as packet
|
|
from experiments import lab053_streaming_receiver as stream
|
|
from experiments import lab053_viterbi as viterbi
|
|
from experiments.lab050_streaming_receiver import streaming_mode
|
|
from experiments.lab053_pipeline import BoundedQueue
|
|
|
|
|
|
def waveform(payload, seed=53001, cfo=10, phase=.73, ebn0=7, bad_crc=False):
|
|
physical = baseline.phy.lab045
|
|
raw_packet = physical.build_packet(payload, physical.MESSAGE_TYPE, physical.SEQUENCE_NUMBER)
|
|
if bad_crc:
|
|
raw_packet = raw_packet[:-1] + bytes([raw_packet[-1] ^ 1])
|
|
coded = physical.conv.encode(physical.lab044.bytes_to_bits(raw_packet))
|
|
permutation = physical.interleaver.build_permutation(len(coded), physical.INTERLEAVER_COLUMNS)
|
|
bits = np.concatenate((physical.lab044.preamble_bits(), physical.encode_phy_header(len(raw_packet)),
|
|
physical.interleaver.interleave(coded, permutation)))
|
|
raw = baseline.phy.build_parameterized_tx_cs8(SimpleNamespace(frame_bits=bits, frame_symbol_count=len(bits)), baseline.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*np.arange(len(values))/8000000)).astype(np.complex64)
|
|
random_generator = np.random.default_rng(seed)
|
|
sigma = np.sqrt(float(np.sum(np.abs(values)**2)) / len(bits) / (2 * 10**(ebn0/10)))
|
|
values += sigma * (random_generator.normal(size=len(values)) + 1j*random_generator.normal(size=len(values)))
|
|
return values.astype(np.complex64)
|
|
|
|
|
|
def receive(values, receiver):
|
|
packets = []
|
|
try:
|
|
for offset in range(0, len(values), receiver.block_samples):
|
|
packets.extend(receiver.push_iq_block(values[offset:offset + receiver.block_samples]))
|
|
packets.extend(receiver.finish())
|
|
return [(item.start_sample, item.end_sample, item.payload) for item in packets]
|
|
finally:
|
|
if hasattr(receiver, "close"):
|
|
receiver.close()
|
|
|
|
|
|
@pytest.mark.parametrize("count", [1, 7, 96, 2152, 8192])
|
|
@pytest.mark.parametrize("kind", ["noise", "zeros", "near_tie", "large"])
|
|
def test_viterbi_bits_match_both_previous_decoders(count, kind):
|
|
physical_count = viterbi.conv.encoded_bit_count(count)
|
|
values = np.random.default_rng(53002 + count).normal(size=physical_count)
|
|
if kind == "zeros":
|
|
values[:] = 0
|
|
elif kind == "near_tie":
|
|
values *= 1e-12
|
|
elif kind == "large":
|
|
values *= 1e100
|
|
assert len(values) == physical_count
|
|
original = viterbi.previous.ORIGINAL_VITERBI(values, count)
|
|
assert np.array_equal(viterbi.previous.fast_viterbi(values, count), original)
|
|
assert np.array_equal(viterbi.viterbi_decode_soft(values, count), original)
|
|
|
|
|
|
@pytest.mark.parametrize("values,count,error", [(np.ones(4), 0, ValueError), (np.ones(4), 1.5, TypeError),
|
|
(np.ones((2, 2)), 1, ValueError), (np.ones(4), 1, ValueError), (np.full(14, np.nan), 1, ValueError)])
|
|
def test_viterbi_invalid_inputs_preserve_validation(values, count, error):
|
|
with pytest.raises(error):
|
|
viterbi.viterbi_decode_soft(values, count)
|
|
|
|
|
|
def test_constant_templates_are_bit_identical_to_lab052():
|
|
old, new = baseline.StreamingReceiver(), packet.PacketProcessor()
|
|
assert np.array_equal(old.raw_reference, new.reference)
|
|
assert np.array_equal(old.matched_taps, new.taps)
|
|
assert np.array_equal(old.preamble_bits, new.bits)
|
|
|
|
|
|
@pytest.mark.parametrize("offset", [-799, -37, 0, 513, 799])
|
|
def test_timing_range_equivalence(offset):
|
|
processor = packet.PacketProcessor()
|
|
random_generator = np.random.default_rng(53003)
|
|
values = (random_generator.normal(size=26000) + 1j*random_generator.normal(size=26000)).astype(np.complex64)
|
|
nominal = 1000
|
|
values[nominal + offset:nominal + offset + 1280*16:16] += processor.expected * 4
|
|
with streaming_mode(baseline.phy.PhyMode(500000)):
|
|
expected = baseline.previous.ORIGINAL_TIMING(values, nominal, processor.expected)
|
|
assert processor.timing(values, nominal, processor.expected) == expected == offset
|
|
|
|
|
|
@pytest.mark.parametrize("cfo,phase", [(-100, -1.4), (0, 0), (10, .73), (100, np.pi)])
|
|
def test_cfo_phase_polarity_phy_crc_equivalence(cfo, phase):
|
|
processor = packet.PacketProcessor()
|
|
payload = bytes(range(256))
|
|
values = np.concatenate((np.zeros(10000, np.complex64), waveform(payload, cfo=cfo, phase=phase), np.zeros(5000, np.complex64)))
|
|
candidate = baseline.offline.CoarseCandidate(10000, .8)
|
|
with streaming_mode(baseline.phy.PhyMode(500000)):
|
|
with baseline.previous.optimized_context(baseline.previous.BlockFrontEnd(values, len(values), baseline.Counter())):
|
|
old = baseline.offline.decode_candidate_from_memmap(values.view(np.float32), len(values), candidate, 1,
|
|
read_iq_window=lambda raw, low, high: raw[2*low:2*high].view(np.complex64))
|
|
new, row = processor.decode(values, candidate)
|
|
assert old.accepted and new.accepted
|
|
assert new.refined_start_sample == old.refined_start_sample
|
|
assert new.next_sample == old.next_sample
|
|
assert new.crc == old.crc == "PASS"
|
|
assert new.frame.reception.packet_bytes == old.frame.reception.packet_bytes
|
|
assert new.frame.timing_offset_samples == old.frame.timing_offset_samples
|
|
assert new.frame.bpsk_polarity == old.frame.bpsk_polarity
|
|
assert abs(new.frame.estimated_cfo_hz - old.frame.estimated_cfo_hz) < .002
|
|
assert abs(new.frame.estimated_phase_rad - old.frame.estimated_phase_rad) < 2e-5
|
|
|
|
|
|
def test_header_failure_does_not_request_body_or_viterbi():
|
|
processor = packet.PacketProcessor()
|
|
values = np.concatenate((np.zeros(10000, np.complex64), processor.reference, np.zeros(100000, np.complex64)))
|
|
outcome, row = processor.decode(values, baseline.offline.CoarseCandidate(10000, .8))
|
|
assert not outcome.accepted
|
|
assert row["calls"].get("body_processing", 0) == 0
|
|
assert row["calls"].get("viterbi", 0) == 0
|
|
|
|
|
|
def test_bad_crc_is_rejected_after_identical_fec_bits():
|
|
processor = packet.PacketProcessor()
|
|
values = np.concatenate((np.zeros(10000, np.complex64), waveform(b"CRC failure", bad_crc=True), np.zeros(5000, np.complex64)))
|
|
candidate = baseline.offline.CoarseCandidate(10000, .8)
|
|
with streaming_mode(baseline.phy.PhyMode(500000)):
|
|
old = baseline.offline.decode_candidate_from_memmap(values.view(np.float32), len(values), candidate, 1,
|
|
read_iq_window=lambda raw, low, high: raw[2*low:2*high].view(np.complex64))
|
|
new, unused = processor.decode(values, candidate)
|
|
assert not old.accepted and not new.accepted
|
|
assert new.crc == old.crc == "FAIL"
|
|
assert new.frame.reception.packet_bytes == old.frame.reception.packet_bytes
|
|
|
|
|
|
def test_existing_object_assembler_accepts_reverse_fragment_order():
|
|
from experiments.lab051_receiver_performance import video
|
|
data = bytes(range(251)) * 5
|
|
frames = video.transport.build_object_frames(data, 42)
|
|
assembler = video.transport.ObjectAssembler()
|
|
for frame in reversed(frames):
|
|
fragment = video.transport.recover_fragment_from_payload(frame.plan.payload)
|
|
assembler.offer(fragment)
|
|
assert assembler.recovered_object() == data
|
|
|
|
|
|
@pytest.mark.parametrize("workers", [1, 2, 4])
|
|
def test_parallel_verdict_and_ring_wrap_match_lab052(workers):
|
|
payload = b"ring and worker order"
|
|
samples = np.zeros(1100000, np.complex64)
|
|
clean = waveform(payload)
|
|
for start in (65536 - 7, 410000, 848001):
|
|
samples[start:start + len(clean)] += clean
|
|
receiver = stream.StreamingReceiver(workers=workers)
|
|
old = receive(samples, baseline.StreamingReceiver())
|
|
new = receive(samples, receiver)
|
|
assert old == new and len(new) == 3
|
|
assert receiver.candidate_queue_peak <= receiver.candidate_capacity
|
|
assert receiver.pending_jobs == 0 and receiver.ring.nbytes == 0
|
|
|
|
|
|
def test_ring_views_preserve_all_samples_without_moving_retained_iq():
|
|
receiver = stream.StreamingReceiver()
|
|
source = (np.arange(1800000) % 127).astype(np.complex64)
|
|
owner = receiver.ring
|
|
def scan(end):
|
|
assert np.shares_memory(receiver.buffer, owner)
|
|
assert np.array_equal(receiver.buffer, source[receiver.base_sample:receiver.total_samples])
|
|
receiver.scan_sample = end
|
|
receiver.base_sample = max(0, end - receiver.margin)
|
|
start = receiver.base_sample % receiver.ring_capacity
|
|
receiver.buffer = owner[start:start + receiver.total_samples - receiver.base_sample]
|
|
return []
|
|
receiver._scan = scan
|
|
receive(source, receiver)
|
|
|
|
|
|
def test_iq_queue_overflow_is_explicit_and_bounded():
|
|
queue = BoundedQueue(2)
|
|
assert queue.offer(np.zeros(65536, np.complex64))
|
|
assert queue.offer(np.zeros(65536, np.complex64))
|
|
assert not queue.offer(np.zeros(65536, np.complex64))
|
|
assert queue.overflows == 1 and queue.peak == queue.qsize() == 2
|
|
|
|
|
|
def test_candidate_queue_overflow_is_explicit():
|
|
release = Event()
|
|
class Probe:
|
|
def decode(self, *args):
|
|
assert release.wait(2)
|
|
return None
|
|
receiver = stream.StreamingReceiver(workers=2, candidate_capacity=2, processor=Probe())
|
|
try:
|
|
receiver.submit(baseline.offline.CoarseCandidate(0, .9))
|
|
receiver.submit(baseline.offline.CoarseCandidate(8, .9))
|
|
with pytest.raises(stream.CandidateQueueOverflow):
|
|
receiver.submit(baseline.offline.CoarseCandidate(0, .9))
|
|
assert receiver.candidate_overflow == 1 and receiver.pending_jobs == 2
|
|
finally:
|
|
release.set()
|
|
receiver.close()
|
|
assert receiver.pending_jobs == 0
|
|
|
|
|
|
def test_context_and_fixed_mode_are_isolated_between_threads():
|
|
processor = packet.PacketProcessor()
|
|
values = np.concatenate((np.zeros(10000, np.complex64), waveform(b"independent"), np.zeros(10000, np.complex64)))
|
|
candidate = baseline.offline.CoarseCandidate(10000, .8)
|
|
functions = (packet.lab043._refine_frame_start, packet.physical.conv.viterbi_decode_soft)
|
|
def decode(unused):
|
|
result = processor.decode(values, candidate)
|
|
assert packet.CURRENT.get() is None
|
|
return result[0].frame.reception.packet_bytes
|
|
with streaming_mode(baseline.phy.PhyMode(20000)), ThreadPoolExecutor(4) as executor:
|
|
results = list(executor.map(decode, range(8)))
|
|
assert all(item == results[0] for item in results)
|
|
assert functions == (packet.lab043._refine_frame_start, packet.physical.conv.viterbi_decode_soft)
|
|
|
|
|
|
def test_worker_exception_waits_for_all_jobs_and_preserves_first_error(monkeypatch):
|
|
entered, release, lock = Event(), Event(), Lock()
|
|
state = {"calls": 0, "active": 0}
|
|
class FailingProcessor:
|
|
def decode(self, samples, candidate, number=1):
|
|
with lock:
|
|
state["calls"] += 1
|
|
number = state["calls"]
|
|
state["active"] += 1
|
|
try:
|
|
if number <= 2:
|
|
raise RuntimeError(f"early_{number}")
|
|
entered.set()
|
|
assert release.wait(2)
|
|
raise RuntimeError("late")
|
|
finally:
|
|
with lock:
|
|
state["active"] -= 1
|
|
receiver = stream.StreamingReceiver(workers=2, processor=FailingProcessor())
|
|
receiver.buffer = np.zeros(100000, np.complex64)
|
|
receiver.total_samples = len(receiver.buffer)
|
|
scores = np.zeros(13000, np.float32)
|
|
scores[[100, 2800, 5500, 8200]] = .9
|
|
def coarse(self, samples, reference):
|
|
self.detection_values = np.zeros(15000, np.complex64)
|
|
return scores
|
|
monkeypatch.setattr(stream.DetectionWindow, "coarse_scores", coarse)
|
|
monkeypatch.setattr(stream.DetectionWindow, "split_scores", lambda *args: np.ones((len(scores), 2)))
|
|
outcome = []
|
|
def run():
|
|
try:
|
|
receiver._scan(receiver.total_samples)
|
|
except RuntimeError as error:
|
|
outcome.append(str(error))
|
|
thread = Thread(target=run)
|
|
thread.start()
|
|
try:
|
|
assert entered.wait(2)
|
|
assert thread.is_alive()
|
|
finally:
|
|
release.set()
|
|
thread.join(3)
|
|
receiver.close()
|
|
assert not thread.is_alive()
|
|
assert outcome == ["early_1"]
|
|
assert state["active"] == receiver.pending_jobs == receiver.candidate_state_count == 0
|
|
assert receiver.closed
|
|
|
|
|
|
def test_alternative_cannot_create_extra_worker_outside_pool(monkeypatch):
|
|
busy, release, alternative = Event(), Event(), Event()
|
|
lock = Lock()
|
|
state = {"active": 0, "peak": 0, "blocked": 0}
|
|
class Probe:
|
|
def decode(self, samples, candidate, number=1):
|
|
with lock:
|
|
state["active"] += 1
|
|
state["peak"] = max(state["peak"], state["active"])
|
|
try:
|
|
if candidate.sample_offset == 110*8:
|
|
alternative.set()
|
|
if candidate.sample_offset in (2800*8, 5500*8):
|
|
with lock:
|
|
state["blocked"] += 1
|
|
if state["blocked"] == 2:
|
|
busy.set()
|
|
assert release.wait(3)
|
|
return SimpleNamespace(accepted=False, reason="FAIL", refined_start_sample=0), {"seconds": .001}
|
|
finally:
|
|
with lock:
|
|
state["active"] -= 1
|
|
receiver = stream.StreamingReceiver(workers=2, processor=Probe())
|
|
receiver.buffer = np.zeros(100000, np.complex64)
|
|
receiver.total_samples = len(receiver.buffer)
|
|
scores = np.zeros(13000, np.float32)
|
|
scores[[100, 110, 2800, 5500]] = [.9, .8, .9, .9]
|
|
def coarse(self, samples, reference):
|
|
self.detection_values = np.zeros(15000, np.complex64)
|
|
return scores
|
|
monkeypatch.setattr(stream.DetectionWindow, "coarse_scores", coarse)
|
|
monkeypatch.setattr(stream.DetectionWindow, "split_scores", lambda *args: np.ones((len(scores), 2)))
|
|
errors = []
|
|
def run():
|
|
try:
|
|
receiver._scan(receiver.total_samples)
|
|
except BaseException as error:
|
|
errors.append(error)
|
|
thread = Thread(target=run)
|
|
thread.start()
|
|
try:
|
|
assert busy.wait(3)
|
|
assert not alternative.wait(.05)
|
|
finally:
|
|
release.set()
|
|
thread.join(3)
|
|
receiver.close()
|
|
assert not thread.is_alive() and not errors
|
|
assert state["peak"] == 2 and state["active"] == 0
|
|
|
|
|
|
@pytest.mark.parametrize("empty", [False, True])
|
|
def test_source_failure_returns_diagnostics_after_cleanup(tmp_path, monkeypatch, empty):
|
|
from experiments import lab053_pipeline as pipeline
|
|
monkeypatch.setattr(pipeline.study, "sequence_fixture", lambda: [SimpleNamespace(jpeg=b"")])
|
|
source = tmp_path / "source.c64"
|
|
if empty:
|
|
source.write_bytes(b"")
|
|
result = pipeline.run_pipeline(source=source, paced=False)
|
|
assert result["errors"][0]["reason"] == ("EMPTY_IQ_SOURCE" if empty else "SOURCE_EXCEPTION")
|
|
assert result["processing_per_rf_second"] is None
|
|
assert result["retained_after_close_bytes"] == result["pending_jobs_after_close"] == 0
|
|
|
|
|
|
def test_paced_overflow_stops_without_concatenating_a_gap(tmp_path, monkeypatch):
|
|
from experiments import lab053_pipeline as pipeline
|
|
monkeypatch.setattr(pipeline.study, "sequence_fixture", lambda: [SimpleNamespace(jpeg=b"")])
|
|
source = tmp_path / "overload.c64"
|
|
source.write_bytes(np.zeros(65536 * 20, np.complex64).tobytes())
|
|
result = pipeline.run_pipeline(source=source, paced=True, iq_capacity=1, processing_delay=.04)
|
|
assert result["iq_overflow"] == result["iq_queue_peak"] == 1
|
|
assert any(row["reason"] == "IQ_QUEUE_OVERFLOW" for row in result["errors"])
|
|
assert result["consumed_iq_seconds"] < result["iq_seconds"]
|
|
assert result["retained_after_close_bytes"] == result["pending_jobs_after_close"] == 0
|