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

fix ui

Browse Source
This commit is contained in:
awe
2026-04-27 18:28:56 +03:00
parent c40df97085
commit 75bc502fe1
4 changed files with 74 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
rfg_adc_plotter/gui/pyqtgraph_backend.py
View File

@ -50,9 +50,9 @@ RAW_PLOT_MAX_POINTS = 4096
RAW_WATERFALL_MAX_POINTS = 2048
UI_MAX_PACKETS_PER_TICK = 8
DEBUG_FRAME_LOG_EVERY = 10
UI_BACKLOG_TAIL_THRESHOLD_MULTIPLIER = 2
UI_BACKLOG_LATEST_ONLY_THRESHOLD_MULTIPLIER = 4
UI_HEAVY_REFRESH_BACKLOG_MULTIPLIER = 2
UI_BACKLOG_TAIL_THRESHOLD_MULTIPLIER = 1
UI_BACKLOG_LATEST_ONLY_THRESHOLD_MULTIPLIER = 2
UI_HEAVY_REFRESH_BACKLOG_MULTIPLIER = 1
UI_HEAVY_REFRESH_MAX_STRIDE = 4
UI_DATA_WAIT_NOTE_AFTER_S = 3.0
FFT_LOW_CUT_SLIDER_SCALE = 10
@ -594,6 +594,28 @@ def coalesce_packets_for_ui(
return packet_list[-limit:], skipped
def is_short_sweep(sweep_size: int, expected_sweep_width: int) -> bool:
"""Return True when current sweep is much shorter than the learned baseline."""
sweep_width = max(0, int(sweep_size))
expected_width = max(0, int(expected_sweep_width))
if sweep_width <= 0 or expected_width <= 0:
return False
return sweep_width < max(8, expected_width // 2)
def update_expected_sweep_width(expected_sweep_width: int, sweep_size: int) -> int:
"""Update dynamic expected sweep width while ignoring tiny/short outliers."""
sweep_width = max(0, int(sweep_size))
expected_width = max(0, int(expected_sweep_width))
if sweep_width < 8:
return expected_width
if expected_width <= 0:
return sweep_width
if is_short_sweep(sweep_width, expected_width):
return expected_width
return int(round((0.9 * float(expected_width)) + (0.1 * float(sweep_width))))
def resolve_visible_fft_curves(
fft_complex: Optional[np.ndarray],
fft_mag: Optional[np.ndarray],
@ -1081,6 +1103,8 @@ def run_pyqtgraph(args) -> None:
last_queue_backlog = 0
last_backlog_skipped = 0
last_heavy_refresh_stride = 1
expected_sweep_width = 0
base_freqs_cache: Dict[int, np.ndarray] = {}
last_packet_processed_at: Optional[float] = None
fixed_ylim: Optional[Tuple[float, float]] = None
if args.ylim:
@ -2087,7 +2111,7 @@ def run_pyqtgraph(args) -> None:
def drain_queue() -> int:
nonlocal processed_frames, ui_frames_skipped, last_queue_backlog, last_backlog_skipped, last_heavy_refresh_stride
nonlocal last_packet_processed_at
nonlocal expected_sweep_width, base_freqs_cache, last_packet_processed_at
pending_packets: List[SweepPacket] = []
while True:
try:
@ -2114,13 +2138,20 @@ def run_pyqtgraph(args) -> None:
f"ui backlog:{drained} keep:{len(pending_packets)} skipped:{skipped_packets}",
)
for sweep, info, aux_curves in pending_packets:
if runtime.ring.width > 0 and sweep.size < max(8, runtime.ring.width // 2):
sweep_width = int(getattr(sweep, "size", 0))
short_sweep = is_short_sweep(sweep_width, expected_sweep_width)
if short_sweep:
set_status_note(f"короткий свип: {int(sweep.size)} точек", ttl_s=2.0)
log_debug_event(
"short_sweep",
f"ui short sweep width:{int(sweep.size)} expected:{int(runtime.ring.width)}",
f"ui short sweep width:{int(sweep.size)} expected:{int(expected_sweep_width)}",
)
base_freqs = np.linspace(SWEEP_FREQ_MIN_GHZ, SWEEP_FREQ_MAX_GHZ, sweep.size, dtype=np.float64)
expected_sweep_width = update_expected_sweep_width(expected_sweep_width, sweep_width)
cached_freqs = base_freqs_cache.get(sweep_width)
if cached_freqs is None:
cached_freqs = np.linspace(SWEEP_FREQ_MIN_GHZ, SWEEP_FREQ_MAX_GHZ, sweep.size, dtype=np.float64)
base_freqs_cache[sweep_width] = cached_freqs
base_freqs = cached_freqs
runtime.current_info = info
runtime.full_current_aux_curves = None
runtime.full_current_aux_curves_codes = None

4
rfg_adc_plotter/state/ring_buffer.py
View File

@ -7,7 +7,7 @@ from typing import Optional
import numpy as np
from rfg_adc_plotter.constants import FFT_LEN, SWEEP_FREQ_MAX_GHZ, SWEEP_FREQ_MIN_GHZ, WF_WIDTH
from rfg_adc_plotter.constants import FFT_LEN, SWEEP_FREQ_MAX_GHZ, SWEEP_FREQ_MIN_GHZ
from rfg_adc_plotter.processing.fft import compute_distance_axis, compute_fft_mag_row, fft_mag_to_db
@ -93,7 +93,7 @@ class RingBuffer:
def ensure_init(self, sweep_width: int) -> bool:
"""Allocate or resize buffers. Returns True when geometry changed."""
target_width = max(int(sweep_width), int(WF_WIDTH))
target_width = max(1, int(sweep_width))
changed = False
if self.ring is None or self.ring_time is None or self.ring_fft is None:
self.width = target_width

31
tests/test_processing.py
View File

@ -17,10 +17,12 @@ from rfg_adc_plotter.gui.pyqtgraph_backend import (
compute_aux_phase_curve,
convert_tty_i16_to_voltage,
decimate_curve_for_display,
is_short_sweep,
resolve_axis_bounds,
resolve_heavy_refresh_stride,
resolve_initial_window_size,
resolve_distance_cut_start,
update_expected_sweep_width,
sanitize_curve_data_for_display,
sanitize_image_for_display,
set_image_rect_if_ready,
@ -337,15 +339,15 @@ class ProcessingTests(unittest.TestCase):
def test_coalesce_packets_for_ui_keeps_newest_packets(self):
packets = [
(np.asarray([float(idx)], dtype=np.float32), {"sweep": idx}, None)
for idx in range(6)
for idx in range(12)
]
kept, skipped = coalesce_packets_for_ui(packets, max_packets=2)
kept, skipped = coalesce_packets_for_ui(packets, max_packets=8, backlog_packets=12)
self.assertEqual(skipped, 4)
self.assertEqual(skipped, 10)
self.assertEqual(len(kept), 2)
self.assertEqual(int(kept[0][1]["sweep"]), 4)
self.assertEqual(int(kept[1][1]["sweep"]), 5)
self.assertEqual(int(kept[0][1]["sweep"]), 10)
self.assertEqual(int(kept[1][1]["sweep"]), 11)
def test_coalesce_packets_for_ui_never_returns_empty_for_non_empty_input(self):
packets = [
@ -372,8 +374,23 @@ class ProcessingTests(unittest.TestCase):
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)
self.assertEqual(resolve_heavy_refresh_stride(8, max_packets=8), 2)
self.assertEqual(resolve_heavy_refresh_stride(16, max_packets=8), 4)
def test_update_expected_sweep_width_initializes_from_first_valid_sweep(self):
self.assertEqual(update_expected_sweep_width(0, 411), 411)
def test_update_expected_sweep_width_ignores_tiny_and_short_outliers(self):
expected = update_expected_sweep_width(0, 411)
self.assertEqual(update_expected_sweep_width(expected, 4), 411)
self.assertEqual(update_expected_sweep_width(expected, 180), 411)
def test_update_expected_sweep_width_applies_ema_for_normal_sweeps(self):
self.assertEqual(update_expected_sweep_width(411, 420), 412)
def test_is_short_sweep_compares_against_dynamic_expected_width(self):
self.assertFalse(is_short_sweep(411, 411))
self.assertTrue(is_short_sweep(180, 411))
def test_sanitize_curve_data_for_display_rejects_fully_nonfinite_series(self):
xs, ys = sanitize_curve_data_for_display(

10
tests/test_ring_buffer.py
View File

@ -20,7 +20,9 @@ class RingBufferTests(unittest.TestCase):
self.assertIsNotNone(ring.distance_axis)
self.assertIsNotNone(ring.get_last_fft_linear())
self.assertIsNotNone(ring.last_fft_db)
self.assertEqual(ring.width, 64)
self.assertEqual(ring.ring.shape[0], 4)
self.assertEqual(ring.ring.shape[1], 64)
self.assertEqual(ring.ring_fft.shape, (4, ring.fft_bins))
def test_ring_buffer_reallocates_when_sweep_width_grows(self):
@ -32,6 +34,14 @@ class RingBufferTests(unittest.TestCase):
self.assertIsNotNone(ring.ring)
self.assertEqual(ring.ring.shape, (3, ring.width))
def test_ring_buffer_reallocates_when_sweep_width_shrinks(self):
ring = RingBuffer(max_sweeps=3)
ring.push(np.ones((2048,), dtype=np.float32), np.linspace(3.3, 14.3, 2048))
ring.push(np.ones((256,), dtype=np.float32), np.linspace(3.3, 14.3, 256))
self.assertEqual(ring.width, 256)
self.assertIsNotNone(ring.ring)
self.assertEqual(ring.ring.shape, (3, 256))
def test_ring_buffer_tracks_latest_fft_and_display_arrays(self):
ring = RingBuffer(max_sweeps=2)
ring.push(np.linspace(0.0, 1.0, 64, dtype=np.float32), np.linspace(3.3, 14.3, 64))