This commit is contained in:
awe
2026-03-25 18:54:59 +03:00
parent fa4870c56c
commit 64e66933e4
14 changed files with 350 additions and 42 deletions

View File

@ -31,6 +31,7 @@ class CliTests(unittest.TestCase):
self.assertEqual(proc.returncode, 0)
self.assertIn("usage:", proc.stdout)
self.assertIn("--parser_16_bit_x2", proc.stdout)
self.assertIn("--parser_complex_ascii", proc.stdout)
def test_backend_mpl_reports_removal(self):
proc = _run("-m", "rfg_adc_plotter.main", "/dev/null", "--backend", "mpl")

View File

@ -56,6 +56,18 @@ class ProcessingTests(unittest.TestCase):
self.assertEqual(calibrated["I"].shape, (32,))
self.assertTrue(np.all(np.diff(calibrated["F"]) >= 0.0))
def test_calibrate_freqs_keeps_complex_payload(self):
sweep = {
"F": np.linspace(3.3, 14.3, 32),
"I": np.exp(1j * np.linspace(0.0, np.pi, 32)).astype(np.complex64),
}
calibrated = calibrate_freqs(sweep)
self.assertEqual(calibrated["F"].shape, (32,))
self.assertEqual(calibrated["I"].shape, (32,))
self.assertTrue(np.iscomplexobj(calibrated["I"]))
self.assertTrue(np.all(np.isfinite(calibrated["I"])))
def test_normalizers_and_envelopes_return_finite_ranges(self):
calib = (np.sin(np.linspace(0.0, 4.0 * np.pi, 64)) * 5.0).astype(np.float32)
raw = calib * 0.75
@ -105,6 +117,15 @@ class ProcessingTests(unittest.TestCase):
self.assertAlmostEqual(float(normalized[1]), 2.0, places=5)
self.assertAlmostEqual(float(normalized[2]), -3.0, places=5)
def test_normalize_by_envelope_supports_complex_input(self):
raw = np.asarray([1.0 + 1.0j, 2.0 - 2.0j], dtype=np.complex64)
envelope = np.asarray([1.0, 2.0], dtype=np.float32)
normalized = normalize_by_envelope(raw, envelope)
self.assertTrue(np.iscomplexobj(normalized))
self.assertTrue(np.all(np.isfinite(normalized)))
self.assertTrue(np.allclose(normalized, np.asarray([1.0 + 1.0j, 1.0 - 1.0j], dtype=np.complex64)))
def test_load_calib_envelope_rejects_empty_payload(self):
with tempfile.TemporaryDirectory() as tmp_dir:
path = os.path.join(tmp_dir, "empty.npy")
@ -247,6 +268,33 @@ class ProcessingTests(unittest.TestCase):
self.assertTrue(np.allclose(spectrum[zero_mask], 0.0))
self.assertTrue(np.any(np.abs(spectrum[pos_idx]) > 0.0))
def test_complex_symmetric_ifft_spectrum_uses_conjugate_mirror(self):
sweep = np.exp(1j * np.linspace(0.0, np.pi, 128)).astype(np.complex64)
freqs = np.linspace(4.0, 10.0, 128, dtype=np.float64)
spectrum = build_symmetric_ifft_spectrum(sweep, freqs, fft_len=FFT_LEN)
self.assertIsNotNone(spectrum)
freq_axis = np.linspace(-10.0, 10.0, FFT_LEN, dtype=np.float64)
neg_idx_all = np.flatnonzero(freq_axis <= (-4.0))
pos_idx_all = np.flatnonzero(freq_axis >= 4.0)
band_len = int(min(neg_idx_all.size, pos_idx_all.size))
neg_idx = neg_idx_all[:band_len]
pos_idx = pos_idx_all[-band_len:]
self.assertTrue(np.iscomplexobj(spectrum))
self.assertTrue(np.allclose(spectrum[neg_idx], np.conj(spectrum[pos_idx][::-1])))
def test_compute_fft_helpers_accept_complex_input(self):
sweep = np.exp(1j * np.linspace(0.0, 2.0 * np.pi, 128)).astype(np.complex64)
freqs = np.linspace(3.3, 14.3, 128, dtype=np.float64)
mag = compute_fft_mag_row(sweep, freqs, 513, mode="positive_only")
row = compute_fft_row(sweep, freqs, 513, mode="positive_only")
self.assertEqual(mag.shape, (513,))
self.assertEqual(row.shape, (513,))
self.assertTrue(np.any(np.isfinite(mag)))
self.assertTrue(np.any(np.isfinite(row)))
def test_symmetric_distance_axis_uses_windowed_frequency_bounds(self):
freqs = np.linspace(4.0, 10.0, 128, dtype=np.float64)
axis = compute_distance_axis(freqs, 513, mode="symmetric")

View File

@ -3,6 +3,7 @@ from __future__ import annotations
import numpy as np
import unittest
from rfg_adc_plotter.processing.fft import compute_fft_mag_row
from rfg_adc_plotter.state.ring_buffer import RingBuffer
@ -72,6 +73,19 @@ class RingBufferTests(unittest.TestCase):
self.assertEqual(ring.last_fft_db.shape, (ring.fft_bins,))
self.assertIsNotNone(ring.distance_axis)
def test_ring_buffer_rebuilds_fft_from_complex_input(self):
ring = RingBuffer(max_sweeps=2)
freqs = np.linspace(3.3, 14.3, 64, dtype=np.float64)
complex_input = np.exp(1j * np.linspace(0.0, 2.0 * np.pi, 64)).astype(np.complex64)
display_sweep = np.abs(complex_input).astype(np.float32)
ring.push(display_sweep, freqs, fft_input=complex_input)
ring.set_fft_mode("direct")
expected = compute_fft_mag_row(complex_input, freqs, ring.fft_bins, mode="direct")
self.assertTrue(np.allclose(ring.get_last_fft_linear(), expected))
self.assertTrue(np.allclose(ring.get_display_raw()[: display_sweep.size, -1], display_sweep))
def test_ring_buffer_reset_clears_cached_history(self):
ring = RingBuffer(max_sweeps=2)
ring.push(np.linspace(0.0, 1.0, 64, dtype=np.float32), np.linspace(4.0, 10.0, 64))

View File

@ -5,6 +5,7 @@ import unittest
from rfg_adc_plotter.io.sweep_parser_core import (
AsciiSweepParser,
ComplexAsciiSweepParser,
LegacyBinaryParser,
LogScale16BitX2BinaryParser,
LogScaleBinaryParser32,
@ -96,6 +97,20 @@ class SweepParserCoreTests(unittest.TestCase):
self.assertEqual(events[1].x, 1)
self.assertEqual(events[1].y, -2.0)
def test_complex_ascii_parser_detects_new_sweep_on_step_reset(self):
parser = ComplexAsciiSweepParser()
events = parser.feed(b"0 3 4\n1 5 12\n0 8 15\n")
self.assertIsInstance(events[0], PointEvent)
self.assertEqual(events[0].x, 0)
self.assertEqual(events[0].y, 5.0)
self.assertEqual(events[0].aux, (3.0, 4.0))
self.assertIsInstance(events[1], PointEvent)
self.assertEqual(events[1].y, 13.0)
self.assertIsInstance(events[2], StartEvent)
self.assertIsInstance(events[3], PointEvent)
self.assertEqual(events[3].aux, (8.0, 15.0))
def test_logscale_32_parser_keeps_channel_and_aux_values(self):
parser = LogScaleBinaryParser32()
stream = _pack_log_start(5) + _pack_log_point(7, 1500, 700, ch=5)