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

updated parsers to be more robust. No changes in functionality

Browse Source
This commit is contained in:
Theodor Chikin
2026-03-05 16:39:08 +03:00
parent a707bedc31
commit 6d32cd8712
1 changed files with 150 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

181
RFG_ADC_dataplotter.py
View File

@ -169,22 +169,12 @@ def _compute_auto_ylim(*series_list: Optional[np.ndarray]) -> Optional[Tuple[flo
def calibrate_freqs(sweep: Mapping[str, Any]) -> SweepData: def calibrate_freqs(sweep: Mapping[str, Any]) -> SweepData:
"""Вернуть копию свипа для будущей калибровки частотной оси.""" """Вернуть копию свипа с применённой калибровкой частотной оси."""
F = np.asarray(sweep["F"], dtype=np.float64).copy() F = np.asarray(sweep["F"], dtype=np.float64).copy()
I = np.asarray(sweep["I"], dtype=np.float64).copy() I = np.asarray(sweep["I"], dtype=np.float64).copy()
tmp = []
C = np.asarray(CALIBRATION_C, dtype=np.float64) C = np.asarray(CALIBRATION_C, dtype=np.float64)
if F.size > 0:
F = C[0] + C[1] * F + C[2] * (F * F)
for f in F:
val = C[0] + (f**1) * C[1] + (f**2) * C[2]
tmp.append(val)
F = np.asanyarray(tmp, dtype=np.float64)
if F.size >= 2: if F.size >= 2:
F_cal = np.linspace(float(F[0]), float(F[-1]), F.size, dtype=np.float64) F_cal = np.linspace(float(F[0]), float(F[-1]), F.size, dtype=np.float64)
@ -198,7 +188,6 @@ def calibrate_freqs(sweep: Mapping[str, Any]) -> SweepData:
"I": I_cal, "I": I_cal,
} }
def _prepare_fft_segment( def _prepare_fft_segment(
sweep: np.ndarray, sweep: np.ndarray,
freqs: Optional[np.ndarray], freqs: Optional[np.ndarray],
@ -906,6 +895,8 @@ class SweepReader(threading.Thread):
ys: list[int] = [] ys: list[int] = []
cur_channel: Optional[int] = None cur_channel: Optional[int] = None
cur_channels: set[int] = set() cur_channels: set[int] = set()
expected_step: Optional[int] = None
syncing = False
words = deque() words = deque()
buf = bytearray() buf = bytearray()
@ -927,9 +918,12 @@ class SweepReader(threading.Thread):
words.append(w) words.append(w)
i += 2 i += 2
# Бинарный протокол: # Бинарный протокол:
# старт свипа (актуальный): 0xFFFF, 0xFFFF, 0xFFFF, (ch<<8)|0x0A # старт свипа: 0xFFFF, 0xFFFF, 0xFFFF, (ch<<8)|0x0A
# старт свипа (legacy): 0xFFFF, 0xFFFF, channel, 0x0A0A
# точка: step, value_hi, value_lo, 0x000A # точка: step, value_hi, value_lo, 0x000A
while len(words) >= 4: while len(words) >= 4:
w0 = int(words[0]) w0 = int(words[0])
@ -938,33 +932,48 @@ class SweepReader(threading.Thread):
w3 = int(words[3]) w3 = int(words[3])
if w0 == 0xFFFF and w1 == 0xFFFF and w2 == 0xFFFF and (w3 & 0x00FF) == 0x000A: if w0 == 0xFFFF and w1 == 0xFFFF and w2 == 0xFFFF and (w3 & 0x00FF) == 0x000A:
if len(words) < 5:
break
if int(words[4]) != 1:
words.popleft()
continue
self._finalize_current(xs, ys, cur_channels) self._finalize_current(xs, ys, cur_channels)
xs.clear() xs.clear()
ys.clear() ys.clear()
cur_channels.clear() cur_channels.clear()
cur_channel = (w3 >> 8) & 0x00FF cur_channel = (w3 >> 8) & 0x00FF
cur_channels.add(cur_channel) cur_channels.add(cur_channel)
expected_step = 1
syncing = False
for _ in range(4): for _ in range(4):
words.popleft() words.popleft()
continue continue
if w0 == 0xFFFF and w1 == 0xFFFF and w3 == 0x0A0A: if syncing:
self._finalize_current(xs, ys, cur_channels) if w0 == 0x000A:
xs.clear() syncing = False
ys.clear() expected_step = None
cur_channels.clear()
cur_channel = w2
cur_channels.add(cur_channel)
for _ in range(4):
words.popleft() words.popleft()
continue continue
if w3 == 0x000A: if w3 != 0x000A:
syncing = True
continue
if w0 <= 0:
syncing = True
continue
if expected_step is not None and w0 < expected_step:
syncing = True
continue
if cur_channel is not None: if cur_channel is not None:
cur_channels.add(cur_channel) cur_channels.add(cur_channel)
xs.append(w0) xs.append(w0)
value_u32 = (w1 << 16) | w2 value_u32 = (w1 << 16) | w2
ys.append(self._u32_to_i32(value_u32)) ys.append(self._u32_to_i32(value_u32))
expected_step = w0 + 1
for _ in range(4): for _ in range(4):
words.popleft() words.popleft()
continue continue
@ -985,6 +994,8 @@ class SweepReader(threading.Thread):
avg_2_vals: list[int] = [] avg_2_vals: list[int] = []
cur_channel: Optional[int] = None cur_channel: Optional[int] = None
cur_channels: set[int] = set() cur_channels: set[int] = set()
expected_step: Optional[int] = None
syncing = False
words = deque() words = deque()
buf = bytearray() buf = bytearray()
@ -1025,6 +1036,11 @@ class SweepReader(threading.Thread):
and w4 == 0xFFFF and w4 == 0xFFFF
and (w5 & 0x00FF) == 0x000A and (w5 & 0x00FF) == 0x000A
): ):
if len(words) < 7:
break
if int(words[6]) != 1:
words.popleft()
continue
self._finalize_current( self._finalize_current(
xs, xs,
ys, ys,
@ -1039,11 +1055,31 @@ class SweepReader(threading.Thread):
cur_channels.clear() cur_channels.clear()
cur_channel = (w5 >> 8) & 0x00FF cur_channel = (w5 >> 8) & 0x00FF
cur_channels.add(cur_channel) cur_channels.add(cur_channel)
expected_step = 1
syncing = False
for _ in range(6): for _ in range(6):
words.popleft() words.popleft()
continue continue
if w5 == 0x000A: if syncing:
if w0 == 0x000A:
syncing = False
expected_step = None
words.popleft()
continue
if w5 != 0x000A:
syncing = True
continue
if w0 <= 0:
syncing = True
continue
if expected_step is not None and w0 < expected_step:
syncing = True
continue
if cur_channel is not None: if cur_channel is not None:
cur_channels.add(cur_channel) cur_channels.add(cur_channel)
avg_1 = self._u32_to_i32((w1 << 16) | w2) avg_1 = self._u32_to_i32((w1 << 16) | w2)
@ -1052,6 +1088,7 @@ class SweepReader(threading.Thread):
avg_1_vals.append(avg_1) avg_1_vals.append(avg_1)
avg_2_vals.append(avg_2) avg_2_vals.append(avg_2)
ys.append(_log_pair_to_sweep(avg_1, avg_2)) ys.append(_log_pair_to_sweep(avg_1, avg_2))
expected_step = w0 + 1
#ys.append(LOG_BASE**(avg_1/LOG_SCALER) - LOG_BASE**(avg_2/LOG_SCALER)) #ys.append(LOG_BASE**(avg_1/LOG_SCALER) - LOG_BASE**(avg_2/LOG_SCALER))
for _ in range(6): for _ in range(6):
words.popleft() words.popleft()
@ -1078,6 +1115,8 @@ class SweepReader(threading.Thread):
avg_2_vals: list[int] = [] avg_2_vals: list[int] = []
cur_channel: Optional[int] = None cur_channel: Optional[int] = None
cur_channels: set[int] = set() cur_channels: set[int] = set()
expected_step: Optional[int] = None
syncing = False
words = deque() words = deque()
buf = bytearray() buf = bytearray()
@ -1098,41 +1137,23 @@ class SweepReader(threading.Thread):
w = int(buf[i]) | (int(buf[i + 1]) << 8) w = int(buf[i]) | (int(buf[i + 1]) << 8)
words.append(w) words.append(w)
i += 2 i += 2
#print(i)
#print(words)
# Бинарный logscale-протокол (16-bit x2): # Бинарный logscale-протокол (16-bit x2):
# старт свипа (новый): 0xFFFF, 0xFFFF, (ch<<8)|0x0A # старт свипа: 0xFFFF, 0xFFFF, 0xFFFF, (ch<<8)|0x0A
# старт свипа (fallback): 0xFFFF, 0xFFFF, 0xFFFF, (ch<<8)|0x0A
# точка: step, avg1_lo16, avg2_lo16, 0x000A # точка: step, avg1_lo16, avg2_lo16, 0x000A
while len(words) >= 3: while len(words) >= 4:
w0 = int(words[0]) w0 = int(words[0])
w1 = int(words[1]) w1 = int(words[1])
w2 = int(words[2]) w2 = int(words[2])
if w0 == 0xFFFF and w1 == 0xFFFF and (w2 & 0x00FF) == 0x000A:
self._finalize_current(
xs,
ys,
cur_channels,
raw_curves=(avg_1_vals, avg_2_vals),
apply_inversion=False,
)
xs.clear()
ys.clear()
avg_1_vals.clear()
avg_2_vals.clear()
cur_channels.clear()
cur_channel = (w2 >> 8) & 0x00FF
cur_channels.add(cur_channel)
for _ in range(3):
words.popleft()
continue
if len(words) < 4:
break
w3 = int(words[3]) w3 = int(words[3])
if w0 == 0xFFFF and w1 == 0xFFFF and w2 == 0xFFFF and (w3 & 0x00FF) == 0x000A: if w0 == 0xFFFF and w1 == 0xFFFF and w2 == 0xFFFF and (w3 & 0x00FF) == 0x000A:
if len(words) < 5:
break
if int(words[4]) != 1:
words.popleft()
continue
self._finalize_current( self._finalize_current(
xs, xs,
ys, ys,
@ -1147,11 +1168,31 @@ class SweepReader(threading.Thread):
cur_channels.clear() cur_channels.clear()
cur_channel = (w3 >> 8) & 0x00FF cur_channel = (w3 >> 8) & 0x00FF
cur_channels.add(cur_channel) cur_channels.add(cur_channel)
expected_step = 1
syncing = False
for _ in range(4): for _ in range(4):
words.popleft() words.popleft()
continue continue
if w3 == 0x000A: if syncing:
if w0 == 0x000A:
syncing = False
expected_step = None
words.popleft()
continue
if w3 != 0x000A:
syncing = True
continue
if w0 <= 0:
syncing = True
continue
if expected_step is not None and w0 < expected_step:
syncing = True
continue
if cur_channel is not None: if cur_channel is not None:
cur_channels.add(cur_channel) cur_channels.add(cur_channel)
avg_1 = self._u16_to_i16(w1) avg_1 = self._u16_to_i16(w1)
@ -1160,6 +1201,7 @@ class SweepReader(threading.Thread):
avg_1_vals.append(avg_1) avg_1_vals.append(avg_1)
avg_2_vals.append(avg_2) avg_2_vals.append(avg_2)
ys.append(_log_pair_to_sweep(avg_1, avg_2)) ys.append(_log_pair_to_sweep(avg_1, avg_2))
expected_step = w0 + 1
for _ in range(4): for _ in range(4):
words.popleft() words.popleft()
continue continue
@ -1283,7 +1325,7 @@ def main():
action="store_true", action="store_true",
help=( help=(
"Бинарный logscale-протокол c парой int16 (avg_1, avg_2): " "Бинарный logscale-протокол c парой int16 (avg_1, avg_2): "
"старт 0xFFFF,0xFFFF,(CH<<8)|0x0A; точка step,avg1_lo16,avg2_lo16,0x000A" "старт 0xFFFF,0xFFFF,0xFFFF,(CH<<8)|0x0A; точка step,avg1_lo16,avg2_lo16,0x000A"
), ),
) )
parser.add_argument( parser.add_argument(
@ -1348,6 +1390,7 @@ def main():
current_sweep_raw: Optional[np.ndarray] = None current_sweep_raw: Optional[np.ndarray] = None
current_aux_curves: SweepAuxCurves = None current_aux_curves: SweepAuxCurves = None
current_sweep_norm: Optional[np.ndarray] = None current_sweep_norm: Optional[np.ndarray] = None
current_fft_db: Optional[np.ndarray] = None
last_calib_sweep: Optional[np.ndarray] = None last_calib_sweep: Optional[np.ndarray] = None
current_info: Optional[SweepInfo] = None current_info: Optional[SweepInfo] = None
x_shared: Optional[np.ndarray] = None x_shared: Optional[np.ndarray] = None
@ -2195,6 +2238,7 @@ def run_pyqtgraph(args):
current_sweep_raw: Optional[np.ndarray] = None current_sweep_raw: Optional[np.ndarray] = None
current_aux_curves: SweepAuxCurves = None current_aux_curves: SweepAuxCurves = None
current_sweep_norm: Optional[np.ndarray] = None current_sweep_norm: Optional[np.ndarray] = None
current_fft_db: Optional[np.ndarray] = None
last_calib_sweep: Optional[np.ndarray] = None last_calib_sweep: Optional[np.ndarray] = None
current_info: Optional[SweepInfo] = None current_info: Optional[SweepInfo] = None
# Авто-уровни цветовой шкалы водопада сырых данных пересчитываются по видимой области. # Авто-уровни цветовой шкалы водопада сырых данных пересчитываются по видимой области.
@ -2212,6 +2256,7 @@ def run_pyqtgraph(args):
norm_type = str(getattr(args, "norm_type", "projector")).strip().lower() norm_type = str(getattr(args, "norm_type", "projector")).strip().lower()
c_edits = [] c_edits = []
c_value_labels = [] c_value_labels = []
plot_dirty = False
# Диапазон по Y: авто по умолчанию (поддерживает отрицательные значения) # Диапазон по Y: авто по умолчанию (поддерживает отрицательные значения)
fixed_ylim: Optional[Tuple[float, float]] = None fixed_ylim: Optional[Tuple[float, float]] = None
if args.ylim: if args.ylim:
@ -2227,7 +2272,7 @@ def run_pyqtgraph(args):
return _normalize_by_calib(raw, calib, norm_type=norm_type) return _normalize_by_calib(raw, calib, norm_type=norm_type)
def _set_calib_enabled(): def _set_calib_enabled():
nonlocal calib_enabled, current_sweep_norm nonlocal calib_enabled, current_sweep_norm, plot_dirty
try: try:
calib_enabled = bool(calib_cb.isChecked()) calib_enabled = bool(calib_cb.isChecked())
except Exception: except Exception:
@ -2236,13 +2281,15 @@ def run_pyqtgraph(args):
current_sweep_norm = _normalize_sweep(current_sweep_raw, last_calib_sweep) current_sweep_norm = _normalize_sweep(current_sweep_raw, last_calib_sweep)
else: else:
current_sweep_norm = None current_sweep_norm = None
plot_dirty = True
def _set_bg_subtract_enabled(): def _set_bg_subtract_enabled():
nonlocal bg_subtract_enabled nonlocal bg_subtract_enabled, plot_dirty
try: try:
bg_subtract_enabled = bool(bg_subtract_cb.isChecked()) bg_subtract_enabled = bool(bg_subtract_cb.isChecked())
except Exception: except Exception:
bg_subtract_enabled = False bg_subtract_enabled = False
plot_dirty = True
try: try:
calib_cb.stateChanged.connect(lambda _v: _set_calib_enabled()) calib_cb.stateChanged.connect(lambda _v: _set_calib_enabled())
@ -2426,7 +2473,7 @@ def run_pyqtgraph(args):
return np.nan_to_num(bg_spec, nan=0.0).astype(np.float32, copy=False) return np.nan_to_num(bg_spec, nan=0.0).astype(np.float32, copy=False)
def push_sweep(s: np.ndarray, freqs: Optional[np.ndarray] = None): def push_sweep(s: np.ndarray, freqs: Optional[np.ndarray] = None):
nonlocal ring, ring_time, head, ring_fft, y_min_fft, y_max_fft nonlocal ring, ring_time, head, ring_fft, y_min_fft, y_max_fft, current_fft_db
if s is None or s.size == 0 or ring is None: if s is None or s.size == 0 or ring is None:
return return
w = ring.shape[1] w = ring.shape[1]
@ -2443,6 +2490,7 @@ def run_pyqtgraph(args):
fft_mag = _compute_fft_mag_row(s, freqs, bins) fft_mag = _compute_fft_mag_row(s, freqs, bins)
ring_fft[(head - 1) % ring_fft.shape[0], :] = fft_mag ring_fft[(head - 1) % ring_fft.shape[0], :] = fft_mag
fft_row = _fft_mag_to_db(fft_mag) fft_row = _fft_mag_to_db(fft_mag)
current_fft_db = fft_row
fr_min = np.nanmin(fft_row) fr_min = np.nanmin(fft_row)
fr_max = np.nanmax(fft_row) fr_max = np.nanmax(fft_row)
if y_min_fft is None or (not np.isnan(fr_min) and fr_min < y_min_fft): if y_min_fft is None or (not np.isnan(fr_min) and fr_min < y_min_fft):
@ -2488,8 +2536,9 @@ def run_pyqtgraph(args):
current_sweep_norm = None current_sweep_norm = None
sweep_for_proc = s sweep_for_proc = s
ensure_buffer(s.size) ensure_buffer(s.size)
_update_physical_axes()
push_sweep(sweep_for_proc, current_freqs) push_sweep(sweep_for_proc, current_freqs)
if drained > 0:
_update_physical_axes()
return drained return drained
# Попытка применить LUT из колормэпа (если доступен) # Попытка применить LUT из колормэпа (если доступен)
@ -2504,11 +2553,12 @@ def run_pyqtgraph(args):
pass pass
def update(): def update():
nonlocal current_peak_width nonlocal current_peak_width, plot_dirty, current_fft_db
if peak_calibrate_mode and any(edit.hasFocus() for edit in c_edits): if peak_calibrate_mode and any(edit.hasFocus() for edit in c_edits):
return return
changed = drain_queue() > 0 changed = drain_queue() > 0
if current_sweep_raw is not None and x_shared is not None: redraw_needed = changed or plot_dirty
if redraw_needed and current_sweep_raw is not None and x_shared is not None:
if current_freqs is not None and current_freqs.size == current_sweep_raw.size: if current_freqs is not None and current_freqs.size == current_sweep_raw.size:
xs = current_freqs xs = current_freqs
elif current_sweep_raw.size <= x_shared.size: elif current_sweep_raw.size <= x_shared.size:
@ -2546,7 +2596,13 @@ def run_pyqtgraph(args):
# Обновим спектр # Обновим спектр
sweep_for_fft = current_sweep_norm if current_sweep_norm is not None else current_sweep_raw sweep_for_fft = current_sweep_norm if current_sweep_norm is not None else current_sweep_raw
if sweep_for_fft.size > 0 and distance_shared is not None: if sweep_for_fft.size > 0 and distance_shared is not None:
fft_vals = _compute_fft_row(sweep_for_fft, current_freqs, distance_shared.size) if (
current_fft_db is None
or current_fft_db.size != distance_shared.size
or plot_dirty
):
current_fft_db = _compute_fft_row(sweep_for_fft, current_freqs, distance_shared.size)
fft_vals = current_fft_db
xs_fft = current_distances if current_distances is not None else distance_shared xs_fft = current_distances if current_distances is not None else distance_shared
if fft_vals.size > xs_fft.size: if fft_vals.size > xs_fft.size:
fft_vals = fft_vals[: xs_fft.size] fft_vals = fft_vals[: xs_fft.size]
@ -2583,6 +2639,7 @@ def run_pyqtgraph(args):
spec_left_line.setVisible(False) spec_left_line.setVisible(False)
spec_right_line.setVisible(False) spec_right_line.setVisible(False)
current_peak_width = None current_peak_width = None
plot_dirty = False
if changed and ring is not None: if changed and ring is not None:
disp = ring if head == 0 else np.roll(ring, -head, axis=0) disp = ring if head == 0 else np.roll(ring, -head, axis=0)