ChTheo/RFG_stm32_ADC_receiver_GUI
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

giga fix

Browse Source
This commit is contained in:
awe
2026-04-10 16:20:48 +03:00
parent 9aac162320
commit 934ca33d58
4 changed files with 531 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
tests/test_processing.py
View File

@ -9,9 +9,15 @@ from rfg_adc_plotter.constants import FFT_LEN, SWEEP_FREQ_MAX_GHZ, SWEEP_FREQ_MI
from rfg_adc_plotter.gui.pyqtgraph_backend import (
apply_working_range,
apply_working_range_to_aux_curves,
build_main_window_layout,
coalesce_packets_for_ui,
compute_background_subtracted_bscan_levels,
decimate_curve_for_display,
resolve_axis_bounds,
resolve_heavy_refresh_stride,
resolve_initial_window_size,
sanitize_curve_data_for_display,
sanitize_image_for_display,
resolve_visible_fft_curves,
resolve_visible_aux_curves,
)
@ -253,6 +259,72 @@ class ProcessingTests(unittest.TestCase):
self.assertEqual(len(kept), 1)
self.assertEqual(int(kept[0][1]["sweep"]), 1)
def test_coalesce_packets_for_ui_switches_to_latest_only_on_large_backlog(self):
packets = [
(np.asarray([float(idx)], dtype=np.float32), {"sweep": idx}, None)
for idx in range(40)
]
kept, skipped = coalesce_packets_for_ui(packets, max_packets=8, backlog_packets=40)
self.assertEqual(skipped, 39)
self.assertEqual(len(kept), 1)
self.assertEqual(int(kept[0][1]["sweep"]), 39)
def test_resolve_heavy_refresh_stride_increases_with_backlog(self):
self.assertEqual(resolve_heavy_refresh_stride(0, max_packets=8), 1)
self.assertEqual(resolve_heavy_refresh_stride(20, max_packets=8), 2)
self.assertEqual(resolve_heavy_refresh_stride(40, max_packets=8), 4)
def test_sanitize_curve_data_for_display_rejects_fully_nonfinite_series(self):
xs, ys = sanitize_curve_data_for_display(
np.asarray([np.nan, np.nan], dtype=np.float64),
np.asarray([np.nan, np.nan], dtype=np.float32),
)
self.assertEqual(xs.shape, (0,))
self.assertEqual(ys.shape, (0,))
def test_sanitize_image_for_display_rejects_fully_nonfinite_frame(self):
data = sanitize_image_for_display(np.full((4, 4), np.nan, dtype=np.float32))
self.assertIsNone(data)
def test_resolve_axis_bounds_rejects_nonfinite_ranges(self):
bounds = resolve_axis_bounds(np.asarray([np.nan, np.inf], dtype=np.float64))
self.assertIsNone(bounds)
def test_resolve_initial_window_size_stays_within_small_screen(self):
width, height = resolve_initial_window_size(800, 480)
self.assertLessEqual(width, 800)
self.assertLessEqual(height, 480)
self.assertGreaterEqual(width, 640)
self.assertGreaterEqual(height, 420)
def test_build_main_window_layout_uses_splitter_and_scroll_area(self):
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
try:
from PyQt5 import QtCore, QtWidgets
except Exception as exc: # pragma: no cover - environment-dependent
self.skipTest(f"Qt unavailable: {exc}")
app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([])
main_window = QtWidgets.QWidget()
try:
_layout, splitter, _plot_layout, settings_widget, settings_layout, settings_scroll = build_main_window_layout(
QtCore,
QtWidgets,
main_window,
)
self.assertIsInstance(splitter, QtWidgets.QSplitter)
self.assertIsInstance(settings_scroll, QtWidgets.QScrollArea)
self.assertIs(settings_scroll.widget(), settings_widget)
self.assertIsInstance(settings_layout, QtWidgets.QVBoxLayout)
finally:
main_window.close()
def test_background_subtracted_bscan_levels_ignore_zero_floor(self):
disp_fft_lin = np.zeros((4, 8), dtype=np.float32)
disp_fft_lin[1, 2:6] = np.asarray([0.05, 0.1, 0.5, 2.0], dtype=np.float32)

41
tests/test_ring_buffer.py
View File

@ -2,6 +2,8 @@ from __future__ import annotations
import numpy as np
import unittest
import warnings
from unittest.mock import patch
from rfg_adc_plotter.processing.fft import compute_fft_mag_row
from rfg_adc_plotter.state.ring_buffer import RingBuffer
@ -116,6 +118,45 @@ class RingBufferTests(unittest.TestCase):
self.assertEqual(ring.width, 0)
self.assertEqual(ring.head, 0)
def test_ring_buffer_push_ignores_all_nan_fft_without_runtime_warning(self):
ring = RingBuffer(max_sweeps=2)
freqs = np.linspace(3.3, 14.3, 64, dtype=np.float64)
ring.push(np.linspace(0.0, 1.0, 64, dtype=np.float32), freqs)
fft_before = ring.last_fft_db.copy()
y_min_before = ring.y_min_fft
y_max_before = ring.y_max_fft
with warnings.catch_warnings():
warnings.simplefilter("error", RuntimeWarning)
with patch(
"rfg_adc_plotter.state.ring_buffer.compute_fft_mag_row",
return_value=np.full((ring.fft_bins,), np.nan, dtype=np.float32),
):
ring.push(np.linspace(1.0, 2.0, 64, dtype=np.float32), freqs)
self.assertFalse(ring.last_push_fft_valid)
self.assertTrue(np.allclose(ring.last_fft_db, fft_before))
self.assertEqual(ring.y_min_fft, y_min_before)
self.assertEqual(ring.y_max_fft, y_max_before)
def test_ring_buffer_set_fft_mode_ignores_all_nan_rebuild_without_runtime_warning(self):
ring = RingBuffer(max_sweeps=2)
freqs = np.linspace(3.3, 14.3, 64, dtype=np.float64)
ring.push(np.linspace(0.0, 1.0, 64, dtype=np.float32), freqs)
fft_before = ring.last_fft_db.copy()
with warnings.catch_warnings():
warnings.simplefilter("error", RuntimeWarning)
with patch(
"rfg_adc_plotter.state.ring_buffer.compute_fft_mag_row",
return_value=np.full((ring.fft_bins,), np.nan, dtype=np.float32),
):
ring.set_fft_mode("direct")
self.assertFalse(ring.last_push_fft_valid)
self.assertTrue(np.allclose(ring.last_fft_db, fft_before))
self.assertEqual(ring.fft_mode, "direct")
if __name__ == "__main__":
unittest.main()