diff --git a/().npy b/().npy new file mode 100644 index 0000000..60cc29d Binary files /dev/null and b/().npy differ diff --git a/calib_envelope.npy b/calib_envelope.npy index 7805967..27a2e64 100644 Binary files a/calib_envelope.npy and b/calib_envelope.npy differ diff --git a/rfg_adc_plotter/gui/matplotlib_backend.py b/rfg_adc_plotter/gui/matplotlib_backend.py index a671074..110c7e6 100644 --- a/rfg_adc_plotter/gui/matplotlib_backend.py +++ b/rfg_adc_plotter/gui/matplotlib_backend.py @@ -12,7 +12,7 @@ from rfg_adc_plotter.constants import FFT_LEN, FREQ_SPAN_GHZ, IFFT_LEN _IFFT_T_MAX_NS = float((IFFT_LEN - 1) / (FREQ_SPAN_GHZ * 1e9) * 1e9) from rfg_adc_plotter.io.sweep_reader import SweepReader from rfg_adc_plotter.processing.normalizer import build_calib_envelopes -from rfg_adc_plotter.state.app_state import BACKGROUND_PATH, CALIB_ENVELOPE_PATH, AppState, format_status +from rfg_adc_plotter.state.app_state import AppState, format_status from rfg_adc_plotter.state.ring_buffer import RingBuffer from rfg_adc_plotter.types import SweepPacket @@ -82,7 +82,8 @@ def run_matplotlib(args): import matplotlib import matplotlib.pyplot as plt from matplotlib.animation import FuncAnimation - from matplotlib.widgets import CheckButtons, Slider + from matplotlib.widgets import Button as MplButton + from matplotlib.widgets import CheckButtons, RadioButtons, Slider, TextBox except Exception as e: sys.stderr.write(f"[error] Нужны matplotlib и её зависимости: {e}\n") sys.exit(1) @@ -111,6 +112,7 @@ def run_matplotlib(args): logscale_enabled = bool(getattr(args, "logscale", False)) state = AppState(norm_type=norm_type) + state.configure_capture_import(fancy=bool(args.fancy), logscale=bool(getattr(args, "logscale", False))) ring = RingBuffer(max_sweeps) # --- Создание фигуры --- @@ -118,10 +120,12 @@ def run_matplotlib(args): (ax_line, ax_img), (ax_fft, ax_spec) = axs if hasattr(fig.canvas.manager, "set_window_title"): fig.canvas.manager.set_window_title(args.title) - fig.subplots_adjust(wspace=0.25, hspace=0.35, left=0.07, right=0.90, top=0.92, bottom=0.08) + fig.subplots_adjust(wspace=0.25, hspace=0.35, left=0.07, right=0.90, top=0.92, bottom=0.22) # Статусная строка status_text = fig.text(0.01, 0.01, "", ha="left", va="bottom", fontsize=8, family="monospace") + pipeline_text = fig.text(0.01, 0.03, "", ha="left", va="bottom", fontsize=8, family="monospace") + ref_text = fig.text(0.01, 0.05, "", ha="left", va="bottom", fontsize=8, family="monospace") # График последнего свипа line_obj, = ax_line.plot([], [], lw=1, color="tab:blue") @@ -178,15 +182,144 @@ def run_matplotlib(args): ax_sctr = fig.add_axes([0.98, 0.55, 0.02, 0.35]) ax_cb = fig.add_axes([0.92, 0.45, 0.08, 0.08]) ax_cb_file = fig.add_axes([0.92, 0.36, 0.08, 0.08]) + ax_line_mode = fig.add_axes([0.92, 0.10, 0.08, 0.08]) ymin_slider = Slider(ax_smin, "Y min", 0, max(1, fft_bins - 1), valinit=0, valstep=1, orientation="vertical") ymax_slider = Slider(ax_smax, "Y max", 0, max(1, fft_bins - 1), valinit=max(1, fft_bins - 1), valstep=1, orientation="vertical") contrast_slider = Slider(ax_sctr, "Int max", 0, 100, valinit=100, valstep=1, orientation="vertical") calib_cb = CheckButtons(ax_cb, ["калибровка"], [False]) calib_file_cb = CheckButtons(ax_cb_file, ["из файла"], [False]) + line_mode_rb = RadioButtons(ax_line_mode, ("raw", "processed"), active=0) + try: + ax_line_mode.set_title("Линия", fontsize=8, pad=2) + except Exception: + pass + line_mode_state = {"value": "raw"} import os as _os - if not _os.path.isfile(CALIB_ENVELOPE_PATH): - ax_cb_file.set_visible(False) + try: + import tkinter as _tk + from tkinter import filedialog as _tk_filedialog + _tk_available = True + except Exception: + _tk = None + _tk_filedialog = None + _tk_available = False + + # Нижняя панель путей и кнопок (работает без Qt; выбор файла через tkinter опционален). + ax_calib_path = fig.add_axes([0.07, 0.14, 0.40, 0.04]) + ax_calib_load = fig.add_axes([0.48, 0.14, 0.07, 0.04]) + ax_calib_pick = fig.add_axes([0.56, 0.14, 0.06, 0.04]) + ax_calib_sample = fig.add_axes([0.63, 0.14, 0.09, 0.04]) + ax_calib_save = fig.add_axes([0.73, 0.14, 0.10, 0.04]) + + ax_bg_path = fig.add_axes([0.07, 0.09, 0.40, 0.04]) + ax_bg_load = fig.add_axes([0.48, 0.09, 0.07, 0.04]) + ax_bg_pick = fig.add_axes([0.56, 0.09, 0.06, 0.04]) + ax_bg_sample = fig.add_axes([0.63, 0.09, 0.09, 0.04]) + ax_bg_save2 = fig.add_axes([0.73, 0.09, 0.10, 0.04]) + + calib_path_box = TextBox(ax_calib_path, "Калибр", initial=state.calib_envelope_path) + bg_path_box = TextBox(ax_bg_path, "Фон", initial=state.background_path) + + calib_load_btn2 = MplButton(ax_calib_load, "Загруз.") + calib_pick_btn2 = MplButton(ax_calib_pick, "Файл") + calib_sample_btn2 = MplButton(ax_calib_sample, "sample") + calib_save_btn2 = MplButton(ax_calib_save, "Сохр env") + + bg_load_btn2 = MplButton(ax_bg_load, "Загруз.") + bg_pick_btn2 = MplButton(ax_bg_pick, "Файл") + bg_sample_btn2 = MplButton(ax_bg_sample, "sample") + bg_save_btn2 = MplButton(ax_bg_save2, "Сохр фон") + + if not _tk_available: + try: + calib_pick_btn2.label.set_text("Файл-") + bg_pick_btn2.label.set_text("Файл-") + except Exception: + pass + + def _tb_text(tb): + try: + return str(tb.text).strip() + except Exception: + return "" + + def _pick_file_dialog(initial_path: str) -> str: + if not _tk_available or _tk is None or _tk_filedialog is None: + return "" + root = None + try: + root = _tk.Tk() + root.withdraw() + root.attributes("-topmost", True) + except Exception: + root = None + try: + return str( + _tk_filedialog.askopenfilename( + initialdir=_os.path.dirname(initial_path) or ".", + initialfile=_os.path.basename(initial_path) or "", + title="Выбрать файл эталона (.npy или capture)", + ) + ) + finally: + try: + if root is not None: + root.destroy() + except Exception: + pass + + def _sync_path_boxes(): + try: + if _tb_text(calib_path_box) != state.calib_envelope_path: + calib_path_box.set_val(state.calib_envelope_path) + except Exception: + pass + try: + if _tb_text(bg_path_box) != state.background_path: + bg_path_box.set_val(state.background_path) + except Exception: + pass + + def _refresh_status_texts(): + pipeline_text.set_text(state.format_pipeline_status()) + ref_text.set_text(state.format_reference_status()) + try: + fig.canvas.draw_idle() + except Exception: + pass + + def _line_mode() -> str: + return str(line_mode_state.get("value", "raw")) + + def _refresh_checkboxes(): + try: + # file-mode чекбокс показываем всегда; он активен при наличии пути/данных. + ax_cb_file.set_visible(True) + except Exception: + pass + + def _load_calib_from_ui(): + p = _tb_text(calib_path_box) + if p: + state.set_calib_envelope_path(p) + ok = state.load_calib_reference() + if ok and bool(calib_file_cb.get_status()[0]): + state.set_calib_mode("file") + state.set_calib_enabled(bool(calib_cb.get_status()[0])) + _sync_path_boxes() + _refresh_checkboxes() + _refresh_status_texts() + return ok + + def _load_bg_from_ui(): + p = _tb_text(bg_path_box) + if p: + state.set_background_path(p) + ok = state.load_background_reference() + _sync_path_boxes() + _refresh_status_texts() + return ok def _on_ylim_change(_val): try: @@ -200,7 +333,7 @@ def run_matplotlib(args): def _on_calib_file_clicked(_v): use_file = bool(calib_file_cb.get_status()[0]) if use_file: - ok = state.load_calib_envelope(CALIB_ENVELOPE_PATH) + ok = _load_calib_from_ui() if ok: state.set_calib_mode("file") else: @@ -208,17 +341,15 @@ def run_matplotlib(args): else: state.set_calib_mode("live") state.set_calib_enabled(bool(calib_cb.get_status()[0])) + _refresh_status_texts() def _on_calib_clicked(_v): - import os as _os2 - if _os2.path.isfile(CALIB_ENVELOPE_PATH): - ax_cb_file.set_visible(True) state.set_calib_enabled(bool(calib_cb.get_status()[0])) - fig.canvas.draw_idle() + _refresh_checkboxes() + _refresh_status_texts() ax_btn_bg = fig.add_axes([0.92, 0.27, 0.08, 0.05]) ax_cb_bg = fig.add_axes([0.92, 0.20, 0.08, 0.06]) - from matplotlib.widgets import Button as MplButton save_bg_btn = MplButton(ax_btn_bg, "Сохр. фон") bg_cb = CheckButtons(ax_cb_bg, ["вычет фона"], [False]) @@ -226,21 +357,88 @@ def run_matplotlib(args): ok = state.save_background() if ok: state.load_background() - fig.canvas.draw_idle() + _sync_path_boxes() + _refresh_status_texts() def _on_bg_clicked(_v): state.set_background_enabled(bool(bg_cb.get_status()[0])) + _refresh_status_texts() + + def _on_calib_load_btn(_event): + _load_calib_from_ui() + + def _on_calib_pick_btn(_event): + path = _pick_file_dialog(_tb_text(calib_path_box) or state.calib_envelope_path) + if not path: + return + state.set_calib_envelope_path(path) + _sync_path_boxes() + _refresh_status_texts() + + def _on_calib_sample_btn(_event): + state.set_calib_envelope_path(_os.path.join("sample_data", "no_antennas_35dB_attenuators")) + _sync_path_boxes() + if _load_calib_from_ui() and not bool(calib_file_cb.get_status()[0]): + calib_file_cb.set_active(0) + + def _on_calib_save_btn(_event): + state.save_calib_envelope() + _sync_path_boxes() + _refresh_status_texts() + + def _on_bg_load_btn(_event): + _load_bg_from_ui() + + def _on_bg_pick_btn(_event): + path = _pick_file_dialog(_tb_text(bg_path_box) or state.background_path) + if not path: + return + state.set_background_path(path) + _sync_path_boxes() + _refresh_status_texts() + + def _on_bg_sample_btn(_event): + state.set_background_path(_os.path.join("sample_data", "empty")) + _sync_path_boxes() + _load_bg_from_ui() + + def _on_bg_save_btn2(_event): + ok = state.save_background() + if ok: + state.load_background() + _sync_path_boxes() + _refresh_status_texts() + + def _on_line_mode_clicked(label): + line_mode_state["value"] = str(label) + try: + fig.canvas.draw_idle() + except Exception: + pass save_bg_btn.on_clicked(_on_save_bg) bg_cb.on_clicked(_on_bg_clicked) + calib_load_btn2.on_clicked(_on_calib_load_btn) + calib_pick_btn2.on_clicked(_on_calib_pick_btn) + calib_sample_btn2.on_clicked(_on_calib_sample_btn) + calib_save_btn2.on_clicked(_on_calib_save_btn) + bg_load_btn2.on_clicked(_on_bg_load_btn) + bg_pick_btn2.on_clicked(_on_bg_pick_btn) + bg_sample_btn2.on_clicked(_on_bg_sample_btn) + bg_save_btn2.on_clicked(_on_bg_save_btn2) + line_mode_rb.on_clicked(_on_line_mode_clicked) ymin_slider.on_changed(_on_ylim_change) ymax_slider.on_changed(_on_ylim_change) contrast_slider.on_changed(lambda _v: fig.canvas.draw_idle()) calib_cb.on_clicked(_on_calib_clicked) calib_file_cb.on_clicked(_on_calib_file_clicked) + _sync_path_boxes() + _refresh_checkboxes() + _refresh_status_texts() except Exception: calib_cb = None + line_mode_state = {"value": "raw"} FREQ_MIN = 3.323 FREQ_MAX = 14.323 @@ -276,23 +474,37 @@ def run_matplotlib(args): xs = ring.x_shared[: raw.size] else: xs = np.arange(raw.size, dtype=np.int32) - line_obj.set_data(xs, raw) - if state.calib_mode == "file" and state.calib_file_envelope is not None: - upper = state.calib_file_envelope - lower = -upper - m_env = float(np.nanmax(np.abs(upper))) - if m_env <= 0.0: - m_env = 1.0 - line_env_lo.set_data(xs[: upper.size], lower / m_env) - line_env_hi.set_data(xs[: upper.size], upper / m_env) - elif state.last_calib_sweep is not None: - calib = state.last_calib_sweep - m_calib = float(np.nanmax(np.abs(calib))) - if m_calib <= 0.0: - m_calib = 1.0 - lower, upper = build_calib_envelopes(calib) - line_env_lo.set_data(xs[: calib.size], lower / m_calib) - line_env_hi.set_data(xs[: calib.size], upper / m_calib) + line_mode = str(line_mode_state.get("value", "raw")) + main = state.current_sweep_processed if line_mode == "processed" else raw + if main is not None: + line_obj.set_data(xs[: main.size], main) + else: + line_obj.set_data([], []) + if line_mode == "raw": + if state.calib_mode == "file" and state.calib_file_envelope is not None: + upper = np.asarray(state.calib_file_envelope, dtype=np.float32) + n_env = min(xs.size, upper.size) + if n_env > 0: + x_env = xs[:n_env] + y_env = upper[:n_env] + line_env_lo.set_data(x_env, -y_env) + line_env_hi.set_data(x_env, y_env) + else: + line_env_lo.set_data([], []) + line_env_hi.set_data([], []) + elif state.last_calib_sweep is not None: + calib = np.asarray(state.last_calib_sweep, dtype=np.float32) + lower, upper = build_calib_envelopes(calib) + n_env = min(xs.size, lower.size, upper.size) + if n_env > 0: + line_env_lo.set_data(xs[:n_env], lower[:n_env]) + line_env_hi.set_data(xs[:n_env], upper[:n_env]) + else: + line_env_lo.set_data([], []) + line_env_hi.set_data([], []) + else: + line_env_lo.set_data([], []) + line_env_hi.set_data([], []) else: line_env_lo.set_data([], []) line_env_hi.set_data([], []) @@ -306,16 +518,19 @@ def run_matplotlib(args): post = state.current_sweep_post_exp if state.current_sweep_post_exp is not None else raw line_post_exp_obj.set_data(xs[: post.size], post) - if state.current_sweep_processed is not None: - proc = state.current_sweep_processed - line_obj.set_data(xs[: proc.size], proc) + if line_mode == "processed": + if state.current_sweep_processed is not None: + proc = state.current_sweep_processed + line_obj.set_data(xs[: proc.size], proc) + else: + line_obj.set_data([], []) else: - line_obj.set_data([], []) + line_obj.set_data(xs[: raw.size], raw) line_norm_obj.set_data([], []) else: line_pre_exp_obj.set_data([], []) line_post_exp_obj.set_data([], []) - if state.current_sweep_norm is not None: + if line_mode == "raw" and state.current_sweep_norm is not None: line_norm_obj.set_data( xs[: state.current_sweep_norm.size], state.current_sweep_norm ) @@ -370,6 +585,11 @@ def run_matplotlib(args): if changed and state.current_info: status_text.set_text(format_status(state.current_info)) channel_text.set_text(state.format_channel_label()) + pipeline_text.set_text(state.format_pipeline_status()) + ref_text.set_text(state.format_reference_status()) + elif changed: + pipeline_text.set_text(state.format_pipeline_status()) + ref_text.set_text(state.format_reference_status()) return ( line_obj, @@ -382,6 +602,8 @@ def run_matplotlib(args): fft_line_obj, img_fft_obj, status_text, + pipeline_text, + ref_text, channel_text, ) diff --git a/rfg_adc_plotter/gui/pyqtgraph_backend.py b/rfg_adc_plotter/gui/pyqtgraph_backend.py index 09b7ca3..b2246ed 100644 --- a/rfg_adc_plotter/gui/pyqtgraph_backend.py +++ b/rfg_adc_plotter/gui/pyqtgraph_backend.py @@ -1,5 +1,6 @@ """PyQtGraph-бэкенд реалтайм-плоттера свипов.""" +import os import sys import threading from queue import Queue @@ -10,7 +11,7 @@ import numpy as np from rfg_adc_plotter.constants import FREQ_SPAN_GHZ, IFFT_LEN from rfg_adc_plotter.io.sweep_reader import SweepReader from rfg_adc_plotter.processing.normalizer import build_calib_envelopes -from rfg_adc_plotter.state.app_state import BACKGROUND_PATH, CALIB_ENVELOPE_PATH, AppState, format_status +from rfg_adc_plotter.state.app_state import AppState, format_status from rfg_adc_plotter.state.ring_buffer import RingBuffer from rfg_adc_plotter.types import SweepPacket @@ -90,6 +91,18 @@ def _visible_levels( return (vmin, vmax) +def _short_path(path: str, max_len: int = 48) -> str: + p = str(path or "").strip() + if not p: + return "(не задан)" + if len(p) <= max_len: + return p + base = os.path.basename(p) + if len(base) <= max_len: + return f".../{base}" + return "..." + p[-(max_len - 3) :] + + def run_pyqtgraph(args): """Быстрый GUI на PyQtGraph. Требует pyqtgraph и PyQt5/PySide6.""" try: @@ -128,13 +141,22 @@ def run_pyqtgraph(args): logscale_enabled = bool(getattr(args, "logscale", False)) state = AppState(norm_type=norm_type) + state.configure_capture_import(fancy=bool(args.fancy), logscale=bool(getattr(args, "logscale", False))) ring = RingBuffer(max_sweeps) + try: + _qt_text_selectable = QtCore.Qt.TextSelectableByMouse + except Exception: + try: + _qt_text_selectable = QtCore.Qt.TextInteractionFlag.TextSelectableByMouse + except Exception: + _qt_text_selectable = None + # --- Создание окна --- pg.setConfigOptions(useOpenGL=True, antialias=False) app = pg.mkQApp(args.title) win = pg.GraphicsLayoutWidget(show=True, title=args.title) - win.resize(1200, 600) + win.resize(1280, 760) # График последнего свипа (слева-сверху) p_line = win.addPlot(row=0, col=0, title="Сырые данные") @@ -196,32 +218,58 @@ def run_pyqtgraph(args): img_fft = pg.ImageItem() p_spec.addItem(img_fft) - # Чекбоксы калибровки — в одном контейнере + # Блок управления калибровкой calib_widget = QtWidgets.QWidget() - calib_layout = QtWidgets.QHBoxLayout(calib_widget) + calib_layout = QtWidgets.QVBoxLayout(calib_widget) calib_layout.setContentsMargins(2, 2, 2, 2) - calib_layout.setSpacing(8) + calib_layout.setSpacing(4) + + calib_row_1 = QtWidgets.QHBoxLayout() + calib_row_1.setSpacing(8) + calib_row_2 = QtWidgets.QHBoxLayout() + calib_row_2.setSpacing(6) calib_cb = QtWidgets.QCheckBox("калибровка") calib_file_cb = QtWidgets.QCheckBox("из файла") - calib_file_cb.setEnabled(False) # активируется только если файл существует + calib_file_cb.setEnabled(False) + calib_path_label = QtWidgets.QLabel() + calib_path_label.setMinimumWidth(260) + if _qt_text_selectable is not None: + calib_path_label.setTextInteractionFlags(_qt_text_selectable) + calib_pick_btn = QtWidgets.QPushButton("Файл…") + calib_load_btn = QtWidgets.QPushButton("Загрузить") + calib_save_btn = QtWidgets.QPushButton("Сохранить env") + calib_sample_btn = QtWidgets.QPushButton("sample calib") - calib_layout.addWidget(calib_cb) - calib_layout.addWidget(calib_file_cb) + calib_row_1.addWidget(calib_cb) + calib_row_1.addWidget(calib_file_cb) + calib_row_1.addStretch(1) + + calib_row_2.addWidget(QtWidgets.QLabel("Калибр:")) + calib_row_2.addWidget(calib_path_label, 1) + calib_row_2.addWidget(calib_pick_btn) + calib_row_2.addWidget(calib_load_btn) + calib_row_2.addWidget(calib_save_btn) + calib_row_2.addWidget(calib_sample_btn) + + calib_layout.addLayout(calib_row_1) + calib_layout.addLayout(calib_row_2) cb_container_proxy = QtWidgets.QGraphicsProxyWidget() cb_container_proxy.setWidget(calib_widget) win.addItem(cb_container_proxy, row=2, col=1) - def _check_file_cb_available(): - import os - calib_file_cb.setEnabled(os.path.isfile(CALIB_ENVELOPE_PATH)) - - _check_file_cb_available() + def _refresh_calib_controls(): + calib_path_label.setText(_short_path(state.calib_envelope_path)) + calib_path_label.setToolTip(state.calib_envelope_path) + calib_load_btn.setEnabled(bool(state.calib_envelope_path) and os.path.isfile(state.calib_envelope_path)) + calib_save_btn.setEnabled(state.last_calib_sweep is not None) + # Переключатель file-mode доступен, если файл существует или уже загружен в память. + calib_file_cb.setEnabled(state.has_calib_envelope_file() or state.calib_file_envelope is not None) def _on_calib_file_toggled(checked): if checked: - ok = state.load_calib_envelope(CALIB_ENVELOPE_PATH) + ok = state.load_calib_reference() if ok: state.set_calib_mode("file") else: @@ -229,43 +277,196 @@ def run_pyqtgraph(args): else: state.set_calib_mode("live") state.set_calib_enabled(calib_cb.isChecked()) + _refresh_calib_controls() + _refresh_pipeline_label() def _on_calib_toggled(_v): - _check_file_cb_available() state.set_calib_enabled(calib_cb.isChecked()) + _refresh_calib_controls() + _refresh_pipeline_label() + + def _on_pick_calib_path(): + path, _ = QtWidgets.QFileDialog.getOpenFileName( + win, + "Выбрать источник калибровки (.npy или capture)", + state.calib_envelope_path, + "Все файлы (*);;NumPy (*.npy)", + ) + if not path: + return + state.set_calib_envelope_path(path) + if calib_file_cb.isChecked(): + if state.load_calib_reference(): + state.set_calib_mode("file") + state.set_calib_enabled(calib_cb.isChecked()) + else: + calib_file_cb.setChecked(False) + _refresh_calib_controls() + _refresh_pipeline_label() + + def _on_load_calib(): + if state.load_calib_reference(): + if calib_file_cb.isChecked(): + state.set_calib_mode("file") + state.set_calib_enabled(calib_cb.isChecked()) + _refresh_calib_controls() + _refresh_pipeline_label() + + def _on_save_calib(): + if state.save_calib_envelope(): + if calib_file_cb.isChecked(): + state.load_calib_envelope() + state.set_calib_mode("file") + state.set_calib_enabled(calib_cb.isChecked()) + _refresh_calib_controls() + _refresh_pipeline_label() + + def _on_sample_calib(): + sample_path = os.path.join("sample_data", "no_antennas_35dB_attenuators") + state.set_calib_envelope_path(sample_path) + if state.load_calib_reference(): + calib_file_cb.setChecked(True) + state.set_calib_mode("file") + state.set_calib_enabled(calib_cb.isChecked()) + _refresh_calib_controls() + _refresh_pipeline_label() calib_cb.stateChanged.connect(_on_calib_toggled) calib_file_cb.stateChanged.connect(lambda _v: _on_calib_file_toggled(calib_file_cb.isChecked())) + calib_pick_btn.clicked.connect(_on_pick_calib_path) + calib_load_btn.clicked.connect(_on_load_calib) + calib_save_btn.clicked.connect(_on_save_calib) + calib_sample_btn.clicked.connect(_on_sample_calib) - # Кнопка сохранения фона + чекбокс вычета фона + # Блок управления фоном bg_widget = QtWidgets.QWidget() - bg_layout = QtWidgets.QHBoxLayout(bg_widget) + bg_layout = QtWidgets.QVBoxLayout(bg_widget) bg_layout.setContentsMargins(2, 2, 2, 2) - bg_layout.setSpacing(8) + bg_layout.setSpacing(4) - save_bg_btn = QtWidgets.QPushButton("Сохр. фон") + bg_row_1 = QtWidgets.QHBoxLayout() + bg_row_1.setSpacing(8) + bg_row_2 = QtWidgets.QHBoxLayout() + bg_row_2.setSpacing(6) + + save_bg_btn = QtWidgets.QPushButton("Сохранить фон") + load_bg_btn = QtWidgets.QPushButton("Загрузить") + bg_pick_btn = QtWidgets.QPushButton("Файл…") + bg_sample_btn = QtWidgets.QPushButton("sample bg") bg_cb = QtWidgets.QCheckBox("вычет фона") - bg_cb.setEnabled(False) + bg_cb.setEnabled(False) # активируется при успешной загрузке/сохранении + bg_path_label = QtWidgets.QLabel() + bg_path_label.setMinimumWidth(260) + if _qt_text_selectable is not None: + bg_path_label.setTextInteractionFlags(_qt_text_selectable) - bg_layout.addWidget(save_bg_btn) - bg_layout.addWidget(bg_cb) + bg_row_1.addWidget(bg_cb) + bg_row_1.addStretch(1) + + bg_row_2.addWidget(QtWidgets.QLabel("Фон:")) + bg_row_2.addWidget(bg_path_label, 1) + bg_row_2.addWidget(bg_pick_btn) + bg_row_2.addWidget(load_bg_btn) + bg_row_2.addWidget(save_bg_btn) + bg_row_2.addWidget(bg_sample_btn) + + bg_layout.addLayout(bg_row_1) + bg_layout.addLayout(bg_row_2) bg_container_proxy = QtWidgets.QGraphicsProxyWidget() bg_container_proxy.setWidget(bg_widget) win.addItem(bg_container_proxy, row=2, col=0) + def _refresh_bg_controls(): + bg_path_label.setText(_short_path(state.background_path)) + bg_path_label.setToolTip(state.background_path) + load_bg_btn.setEnabled(bool(state.background_path) and os.path.isfile(state.background_path)) + bg_cb.setEnabled(state.background is not None or state.background_source_type == "capture_raw") + + def _on_pick_bg_path(): + path, _ = QtWidgets.QFileDialog.getOpenFileName( + win, + "Выбрать источник фона (.npy или capture)", + state.background_path, + "Все файлы (*);;NumPy (*.npy)", + ) + if not path: + return + state.set_background_path(path) + if bg_cb.isChecked(): + if not state.load_background_reference(): + bg_cb.setChecked(False) + _refresh_bg_controls() + _refresh_pipeline_label() + + def _on_load_bg(): + state.load_background_reference() + _refresh_bg_controls() + _refresh_pipeline_label() + def _on_save_bg(): ok = state.save_background() if ok: state.load_background() - bg_cb.setEnabled(True) + _refresh_bg_controls() + _refresh_pipeline_label() + def _on_bg_toggled(_v): + state.set_background_enabled(bg_cb.isChecked()) + _refresh_pipeline_label() + + def _on_sample_bg(): + sample_path = os.path.join("sample_data", "empty") + state.set_background_path(sample_path) + if state.load_background_reference(): + bg_cb.setEnabled(True) + _refresh_bg_controls() + _refresh_pipeline_label() + + bg_pick_btn.clicked.connect(_on_pick_bg_path) + load_bg_btn.clicked.connect(_on_load_bg) save_bg_btn.clicked.connect(_on_save_bg) - bg_cb.stateChanged.connect(lambda _v: state.set_background_enabled(bg_cb.isChecked())) + bg_cb.stateChanged.connect(_on_bg_toggled) + bg_sample_btn.clicked.connect(_on_sample_bg) + + # Переключатель отображения верхнего линейного графика + line_mode_widget = QtWidgets.QWidget() + line_mode_layout = QtWidgets.QHBoxLayout(line_mode_widget) + line_mode_layout.setContentsMargins(2, 2, 2, 2) + line_mode_layout.setSpacing(8) + line_mode_layout.addWidget(QtWidgets.QLabel("Линия:")) + line_mode_raw_rb = QtWidgets.QRadioButton("raw") + line_mode_proc_rb = QtWidgets.QRadioButton("processed") + line_mode_raw_rb.setChecked(True) + line_mode_layout.addWidget(line_mode_raw_rb) + line_mode_layout.addWidget(line_mode_proc_rb) + line_mode_layout.addStretch(1) + line_mode_proxy = QtWidgets.QGraphicsProxyWidget() + line_mode_proxy.setWidget(line_mode_widget) + win.addItem(line_mode_proxy, row=6, col=0, colspan=2) + + def _line_mode() -> str: + return "processed" if line_mode_proc_rb.isChecked() else "raw" # Статусная строка status = pg.LabelItem(justify="left") win.addItem(status, row=3, col=0, colspan=2) + pipeline_status = pg.LabelItem(justify="left") + win.addItem(pipeline_status, row=4, col=0, colspan=2) + ref_status = pg.LabelItem(justify="left") + win.addItem(ref_status, row=5, col=0, colspan=2) + + def _refresh_pipeline_label(): + txt = state.format_pipeline_status() + trace = state.format_stage_trace() + if trace: + txt = f"{txt} | trace: {trace}" + pipeline_status.setText(txt) + ref_status.setText(state.format_reference_status()) + + _refresh_calib_controls() + _refresh_bg_controls() + _refresh_pipeline_label() _imshow_initialized = [False] @@ -293,28 +494,46 @@ def run_pyqtgraph(args): if changed and not _imshow_initialized[0] and ring.is_ready: _init_imshow_extents() _imshow_initialized[0] = True + if changed: + _refresh_calib_controls() + _refresh_bg_controls() + _refresh_pipeline_label() # Линейный график свипа if state.current_sweep_raw is not None and ring.x_shared is not None: raw = state.current_sweep_raw xs = ring.x_shared[: raw.size] if raw.size <= ring.x_shared.size else np.arange(raw.size) - curve.setData(xs, raw, autoDownsample=True) - if state.calib_mode == "file" and state.calib_file_envelope is not None: - upper = state.calib_file_envelope - lower = -upper - m_env = float(np.nanmax(np.abs(upper))) - if m_env <= 0.0: - m_env = 1.0 - curve_env_lo.setData(xs[: upper.size], lower / m_env, autoDownsample=True) - curve_env_hi.setData(xs[: upper.size], upper / m_env, autoDownsample=True) - elif state.last_calib_sweep is not None: - calib = state.last_calib_sweep - m_calib = float(np.nanmax(np.abs(calib))) - if m_calib <= 0.0: - m_calib = 1.0 - lower, upper = build_calib_envelopes(calib) - curve_env_lo.setData(xs[: calib.size], lower / m_calib, autoDownsample=True) - curve_env_hi.setData(xs[: calib.size], upper / m_calib, autoDownsample=True) + line_mode = _line_mode() + main = state.current_sweep_processed if line_mode == "processed" else raw + if main is not None: + curve.setData(xs[: main.size], main, autoDownsample=True) + else: + curve.setData([], []) + if line_mode == "raw": + if state.calib_mode == "file" and state.calib_file_envelope is not None: + upper = np.asarray(state.calib_file_envelope, dtype=np.float32) + n_env = min(xs.size, upper.size) + if n_env > 0: + x_env = xs[:n_env] + y_env = upper[:n_env] + curve_env_lo.setData(x_env, -y_env, autoDownsample=True) + curve_env_hi.setData(x_env, y_env, autoDownsample=True) + else: + curve_env_lo.setData([], []) + curve_env_hi.setData([], []) + elif state.last_calib_sweep is not None: + calib = np.asarray(state.last_calib_sweep, dtype=np.float32) + lower, upper = build_calib_envelopes(calib) + n_env = min(xs.size, upper.size, lower.size) + if n_env > 0: + curve_env_lo.setData(xs[:n_env], lower[:n_env], autoDownsample=True) + curve_env_hi.setData(xs[:n_env], upper[:n_env], autoDownsample=True) + else: + curve_env_lo.setData([], []) + curve_env_hi.setData([], []) + else: + curve_env_lo.setData([], []) + curve_env_hi.setData([], []) else: curve_env_lo.setData([], []) curve_env_hi.setData([], []) @@ -328,16 +547,19 @@ def run_pyqtgraph(args): post = state.current_sweep_post_exp if state.current_sweep_post_exp is not None else raw curve_post_exp.setData(xs[: post.size], post, autoDownsample=True) - if state.current_sweep_processed is not None: - proc = state.current_sweep_processed - curve.setData(xs[: proc.size], proc, autoDownsample=True) + if line_mode == "processed": + if state.current_sweep_processed is not None: + proc = state.current_sweep_processed + curve.setData(xs[: proc.size], proc, autoDownsample=True) + else: + curve.setData([], []) else: - curve.setData([], []) + curve.setData(xs[: raw.size], raw, autoDownsample=True) curve_norm.setData([], []) else: curve_pre_exp.setData([], []) curve_post_exp.setData([], []) - if state.current_sweep_norm is not None: + if line_mode == "raw" and state.current_sweep_norm is not None: curve_norm.setData( xs[: state.current_sweep_norm.size], state.current_sweep_norm, @@ -385,6 +607,8 @@ def run_pyqtgraph(args): except Exception: pass ch_text.setText(state.format_channel_label()) + elif changed: + _refresh_pipeline_label() # Водопад спектров — новые данные справа (без реверса) if changed and ring.is_ready: diff --git a/rfg_adc_plotter/io/capture_reference_loader.py b/rfg_adc_plotter/io/capture_reference_loader.py new file mode 100644 index 0000000..5fc519b --- /dev/null +++ b/rfg_adc_plotter/io/capture_reference_loader.py @@ -0,0 +1,227 @@ +"""Загрузка эталонов (калибровка/фон) из .npy или бинарных capture-файлов.""" + +from __future__ import annotations + +from collections import Counter +from dataclasses import dataclass +import os +from typing import Iterable, List, Optional, Tuple + +import numpy as np + +from rfg_adc_plotter.io.sweep_parser_core import BinaryRecordStreamParser, SweepAssembler +from rfg_adc_plotter.types import SweepPacket + + +@dataclass(frozen=True) +class CaptureParseSummary: + path: str + format: str # "npy" | "bin_capture" + sweeps_total: int + sweeps_valid: int + channels_seen: Tuple[int, ...] + dominant_width: Optional[int] + dominant_n_valid: Optional[int] + aggregation: str + warnings: Tuple[str, ...] + + +@dataclass(frozen=True) +class ReferenceLoadResult: + vector: np.ndarray + summary: CaptureParseSummary + kind: str # "calibration_envelope" | "background_raw" | "background_processed" + source_type: str # "npy" | "capture" + + +def detect_reference_file_format(path: str) -> Optional[str]: + """Определить формат файла эталона: .npy или бинарный capture.""" + p = str(path).strip() + if not p or not os.path.isfile(p): + return None + if p.lower().endswith(".npy"): + return "npy" + + try: + size = os.path.getsize(p) + except Exception: + return None + if size <= 0 or (size % 8) != 0: + return None + + try: + with open(p, "rb") as f: + sample = f.read(min(size, 8 * 2048)) + except Exception: + return None + + if len(sample) < 8: + return None + + # Быстрый sniff aligned-записей: в валидных записях байт 6 == 0x0A. + recs = len(sample) // 8 + if recs <= 0: + return None + marker_hits = 0 + start_hits = 0 + for i in range(0, recs * 8, 8): + b = sample[i : i + 8] + if b[6] == 0x0A: + marker_hits += 1 + if b[:6] == b"\xff\xff\xff\xff\xff\xff": + start_hits += 1 + if marker_hits >= max(4, int(recs * 0.8)) and start_hits >= 1: + return "bin_capture" + return None + + +def load_capture_sweeps(path: str, *, fancy: bool = False, logscale: bool = False) -> List[SweepPacket]: + """Загрузить свипы из бинарного capture-файла в формате --bin.""" + parser = BinaryRecordStreamParser() + assembler = SweepAssembler(fancy=fancy, logscale=logscale, debug=False) + sweeps: List[SweepPacket] = [] + + with open(path, "rb") as f: + while True: + chunk = f.read(65536) + if not chunk: + break + events = parser.feed(chunk) + for ev in events: + packets = assembler.consume_binary_event(ev) + if packets: + sweeps.extend(packets) + tail = assembler.finalize_current() + if tail is not None: + sweeps.append(tail) + + return sweeps + + +def _mode_int(values: Iterable[int]) -> Optional[int]: + vals = [int(v) for v in values] + if not vals: + return None + ctr = Counter(vals) + return int(max(ctr.items(), key=lambda kv: (kv[1], kv[0]))[0]) + + +def aggregate_capture_reference( + sweeps: List[SweepPacket], + *, + channel: int = 0, + method: str = "median", + path: str = "", +) -> Tuple[np.ndarray, CaptureParseSummary]: + """Отфильтровать и агрегировать свипы из capture в один эталонный вектор.""" + ch_target = int(channel) + meth = str(method).strip().lower() or "median" + warnings: list[str] = [] + + if meth != "median": + warnings.append(f"aggregation '{meth}' не поддерживается, использую median") + meth = "median" + + channels_seen: set[int] = set() + candidate_rows: list[np.ndarray] = [] + widths: list[int] = [] + n_valids: list[int] = [] + + for sweep, info in sweeps: + chs = info.get("chs") if isinstance(info, dict) else None + ch_set: set[int] = set() + if isinstance(chs, (list, tuple, set)): + for v in chs: + try: + ch_set.add(int(v)) + except Exception: + pass + else: + try: + ch_set.add(int(info.get("ch", 0))) # type: ignore[union-attr] + except Exception: + pass + channels_seen.update(ch_set) + if ch_target not in ch_set: + continue + + row = np.asarray(sweep, dtype=np.float32).reshape(-1) + candidate_rows.append(row) + widths.append(int(row.size)) + n_valids.append(int(np.count_nonzero(np.isfinite(row)))) + + sweeps_total = len(sweeps) + if not candidate_rows: + summary = CaptureParseSummary( + path=path, + format="bin_capture", + sweeps_total=sweeps_total, + sweeps_valid=0, + channels_seen=tuple(sorted(channels_seen)), + dominant_width=None, + dominant_n_valid=None, + aggregation=meth, + warnings=tuple(warnings + [f"канал ch{ch_target} не найден"]), + ) + raise ValueError(summary.warnings[-1]) + + dominant_width = _mode_int(widths) + dominant_n_valid = _mode_int(n_valids) + if dominant_width is None or dominant_n_valid is None: + summary = CaptureParseSummary( + path=path, + format="bin_capture", + sweeps_total=sweeps_total, + sweeps_valid=0, + channels_seen=tuple(sorted(channels_seen)), + dominant_width=dominant_width, + dominant_n_valid=dominant_n_valid, + aggregation=meth, + warnings=tuple(warnings + ["не удалось определить доминирующие параметры свипа"]), + ) + raise ValueError(summary.warnings[-1]) + + valid_rows: list[np.ndarray] = [] + n_valid_threshold = max(1, int(np.floor(0.95 * dominant_n_valid))) + for row in candidate_rows: + if row.size != dominant_width: + continue + n_valid = int(np.count_nonzero(np.isfinite(row))) + if n_valid < n_valid_threshold: + continue + valid_rows.append(row) + + if not valid_rows: + warnings.append("после фильтрации не осталось валидных свипов") + summary = CaptureParseSummary( + path=path, + format="bin_capture", + sweeps_total=sweeps_total, + sweeps_valid=0, + channels_seen=tuple(sorted(channels_seen)), + dominant_width=dominant_width, + dominant_n_valid=dominant_n_valid, + aggregation=meth, + warnings=tuple(warnings), + ) + raise ValueError(summary.warnings[-1]) + + # Детерминированная агрегация: медиана по валидным свипам. + stack = np.stack(valid_rows, axis=0).astype(np.float32, copy=False) + vector = np.nanmedian(stack, axis=0).astype(np.float32, copy=False) + + if len(valid_rows) < len(candidate_rows): + warnings.append(f"отфильтровано {len(candidate_rows) - len(valid_rows)} неполных/нестандартных свипов") + + summary = CaptureParseSummary( + path=path, + format="bin_capture", + sweeps_total=sweeps_total, + sweeps_valid=len(valid_rows), + channels_seen=tuple(sorted(channels_seen)), + dominant_width=dominant_width, + dominant_n_valid=dominant_n_valid, + aggregation=meth, + warnings=tuple(warnings), + ) + return vector, summary diff --git a/rfg_adc_plotter/io/sweep_parser_core.py b/rfg_adc_plotter/io/sweep_parser_core.py new file mode 100644 index 0000000..1ae8dc0 --- /dev/null +++ b/rfg_adc_plotter/io/sweep_parser_core.py @@ -0,0 +1,247 @@ +"""Переиспользуемые компоненты парсинга бинарных свипов и сборки SweepPacket.""" + +from __future__ import annotations + +from collections import deque +import time +from typing import Iterable, List, Optional, Sequence, Set, Tuple + +import numpy as np + +from rfg_adc_plotter.constants import DATA_INVERSION_THRESHOLD, LOG_EXP +from rfg_adc_plotter.types import SweepInfo, SweepPacket + +# Binary parser events: +# ("start", ch) +# ("point", ch, x, y) +BinaryEvent = Tuple[str, int] | Tuple[str, int, int, int] + + +def u32_to_i32(v: int) -> int: + """Преобразование 32-bit слова в знаковое значение.""" + return v - 0x1_0000_0000 if (v & 0x8000_0000) else v + + +class BinaryRecordStreamParser: + """Инкрементальный парсер бинарных записей протокола (по 8 байт).""" + + def __init__(self): + self._buf = bytearray() + self.bytes_consumed: int = 0 + self.start_count: int = 0 + self.point_count: int = 0 + self.desync_count: int = 0 + + def feed(self, data: bytes) -> List[BinaryEvent]: + if data: + self._buf += data + events: List[BinaryEvent] = [] + buf = self._buf + + while len(buf) >= 8: + w0 = int(buf[0]) | (int(buf[1]) << 8) + w1 = int(buf[2]) | (int(buf[3]) << 8) + w2 = int(buf[4]) | (int(buf[5]) << 8) + + if w0 == 0xFFFF and w1 == 0xFFFF and w2 == 0xFFFF and buf[6] == 0x0A: + ch = int(buf[7]) + events.append(("start", ch)) + del buf[:8] + self.bytes_consumed += 8 + self.start_count += 1 + continue + + if buf[6] == 0x0A: + ch = int(buf[7]) + value_u32 = (w1 << 16) | w2 + events.append(("point", ch, int(w0), u32_to_i32(value_u32))) + del buf[:8] + self.bytes_consumed += 8 + self.point_count += 1 + continue + + del buf[:1] + self.bytes_consumed += 1 + self.desync_count += 1 + + return events + + def buffered_size(self) -> int: + return len(self._buf) + + def clear_buffer_keep_tail(self, max_tail: int = 262_144): + if len(self._buf) > max_tail: + del self._buf[:-max_tail] + + +class SweepAssembler: + """Собирает точки в свип и применяет ту же постобработку, что realtime parser.""" + + def __init__(self, fancy: bool = False, logscale: bool = False, debug: bool = False): + self._fancy = bool(fancy) + self._logscale = bool(logscale) + self._debug = bool(debug) + + self._max_width: int = 0 + self._sweep_idx: int = 0 + self._last_sweep_ts: Optional[float] = None + self._n_valid_hist = deque() + + self._xs: list[int] = [] + self._ys: list[int] = [] + self._cur_channel: Optional[int] = None + self._cur_channels: set[int] = set() + + def reset_current(self): + self._xs.clear() + self._ys.clear() + self._cur_channel = None + self._cur_channels.clear() + + def add_point(self, ch: int, x: int, y: int): + if self._cur_channel is None: + self._cur_channel = int(ch) + self._cur_channels.add(int(ch)) + self._xs.append(int(x)) + self._ys.append(int(y)) + + def start_new_sweep(self, ch: int, now_ts: Optional[float] = None) -> Optional[SweepPacket]: + packet = self.finalize_current(now_ts=now_ts) + self.reset_current() + self._cur_channel = int(ch) + self._cur_channels.add(int(ch)) + return packet + + def consume_binary_event(self, event: BinaryEvent, now_ts: Optional[float] = None) -> List[SweepPacket]: + out: List[SweepPacket] = [] + tag = event[0] + if tag == "start": + packet = self.start_new_sweep(int(event[1]), now_ts=now_ts) + if packet is not None: + out.append(packet) + return out + # point + _tag, ch, x, y = event # type: ignore[misc] + self.add_point(int(ch), int(x), int(y)) + return out + + def finalize_arrays( + self, + xs: Sequence[int], + ys: Sequence[int], + channels: Optional[Set[int]], + now_ts: Optional[float] = None, + ) -> Optional[SweepPacket]: + if self._debug: + if not xs: + import sys + sys.stderr.write("[debug] _finalize_current: xs пуст — свип пропущен\n") + else: + import sys + sys.stderr.write( + f"[debug] _finalize_current: {len(xs)} точек → свип #{self._sweep_idx + 1}\n" + ) + if not xs: + return None + + ch_list = sorted(channels) if channels else [0] + ch_primary = ch_list[0] if ch_list else 0 + max_x = max(int(v) for v in xs) + width = max_x + 1 + self._max_width = max(self._max_width, width) + target_width = self._max_width if self._fancy else width + + sweep = np.full((target_width,), np.nan, dtype=np.float32) + try: + idx = np.asarray(xs, dtype=np.int64) + vals = np.asarray(ys, dtype=np.float32) + sweep[idx] = vals + except Exception: + for x, y in zip(xs, ys): + xi = int(x) + if 0 <= xi < target_width: + sweep[xi] = float(y) + + n_valid_cur = int(np.count_nonzero(np.isfinite(sweep))) + + if self._fancy: + try: + known = ~np.isnan(sweep) + if np.any(known): + known_idx = np.nonzero(known)[0] + for i0, i1 in zip(known_idx[:-1], known_idx[1:]): + if i1 - i0 > 1: + avg = (sweep[i0] + sweep[i1]) * 0.5 + sweep[i0 + 1 : i1] = avg + first_idx = int(known_idx[0]) + last_idx = int(known_idx[-1]) + if first_idx > 0: + sweep[:first_idx] = sweep[first_idx] + if last_idx < sweep.size - 1: + sweep[last_idx + 1 :] = sweep[last_idx] + except Exception: + pass + + try: + m = float(np.nanmean(sweep)) + if np.isfinite(m) and m < DATA_INVERSION_THRESHOLD: + sweep *= -1.0 + except Exception: + pass + + pre_exp_sweep = None + if self._logscale: + try: + pre_exp_sweep = sweep.copy() + with np.errstate(over="ignore", invalid="ignore"): + sweep = np.power(LOG_EXP, np.asarray(sweep, dtype=np.float64)).astype(np.float32) + sweep[~np.isfinite(sweep)] = np.nan + except Exception: + pass + + self._sweep_idx += 1 + if len(ch_list) > 1: + import sys + sys.stderr.write(f"[warn] Sweep {self._sweep_idx}: изменялся номер канала: {ch_list}\n") + + now = float(time.time() if now_ts is None else now_ts) + if self._last_sweep_ts is None: + dt_ms = float("nan") + else: + dt_ms = (now - self._last_sweep_ts) * 1000.0 + self._last_sweep_ts = now + + self._n_valid_hist.append((now, n_valid_cur)) + while self._n_valid_hist and (now - self._n_valid_hist[0][0]) > 1.0: + self._n_valid_hist.popleft() + if self._n_valid_hist: + n_valid = float(sum(v for _t, v in self._n_valid_hist) / len(self._n_valid_hist)) + else: + n_valid = float(n_valid_cur) + + if n_valid_cur > 0: + vmin = float(np.nanmin(sweep)) + vmax = float(np.nanmax(sweep)) + mean = float(np.nanmean(sweep)) + std = float(np.nanstd(sweep)) + else: + vmin = vmax = mean = std = float("nan") + + info: SweepInfo = { + "sweep": self._sweep_idx, + "ch": ch_primary, + "chs": ch_list, + "n_valid": n_valid, + "min": vmin, + "max": vmax, + "mean": mean, + "std": std, + "dt_ms": dt_ms, + } + if pre_exp_sweep is not None: + info["pre_exp_sweep"] = pre_exp_sweep + + return (sweep, info) + + def finalize_current(self, now_ts: Optional[float] = None) -> Optional[SweepPacket]: + return self.finalize_arrays(self._xs, self._ys, self._cur_channels, now_ts=now_ts) diff --git a/rfg_adc_plotter/io/sweep_reader.py b/rfg_adc_plotter/io/sweep_reader.py index d187e0a..bdc8c0f 100644 --- a/rfg_adc_plotter/io/sweep_reader.py +++ b/rfg_adc_plotter/io/sweep_reader.py @@ -3,15 +3,12 @@ import sys import threading import time -from collections import deque from queue import Full, Queue from typing import Optional -import numpy as np - -from rfg_adc_plotter.constants import DATA_INVERSION_THRESHOLD, LOG_EXP +from rfg_adc_plotter.io.sweep_parser_core import BinaryRecordStreamParser, SweepAssembler from rfg_adc_plotter.io.serial_source import SerialChunkReader, SerialLineSource -from rfg_adc_plotter.types import SweepInfo, SweepPacket +from rfg_adc_plotter.types import SweepPacket class SweepReader(threading.Thread): @@ -38,119 +35,13 @@ class SweepReader(threading.Thread): self._bin_mode = bool(bin_mode) self._logscale = bool(logscale) self._debug = bool(debug) - self._max_width: int = 0 - self._sweep_idx: int = 0 - self._last_sweep_ts: Optional[float] = None - self._n_valid_hist = deque() - - @staticmethod - def _u32_to_i32(v: int) -> int: - """Преобразование 32-bit слова в знаковое значение.""" - return v - 0x1_0000_0000 if (v & 0x8000_0000) else v + self._assembler = SweepAssembler(fancy=self._fancy, logscale=self._logscale, debug=self._debug) def _finalize_current(self, xs, ys, channels: Optional[set]): - if self._debug: - if not xs: - sys.stderr.write("[debug] _finalize_current: xs пуст — свип пропущен\n") - else: - sys.stderr.write(f"[debug] _finalize_current: {len(xs)} точек → свип #{self._sweep_idx + 1}\n") - if not xs: + packet = self._assembler.finalize_arrays(xs, ys, channels) + if packet is None: return - ch_list = sorted(channels) if channels else [0] - ch_primary = ch_list[0] if ch_list else 0 - max_x = max(xs) - width = max_x + 1 - self._max_width = max(self._max_width, width) - target_width = self._max_width if self._fancy else width - - sweep = np.full((target_width,), np.nan, dtype=np.float32) - try: - idx = np.asarray(xs, dtype=np.int64) - vals = np.asarray(ys, dtype=np.float32) - sweep[idx] = vals - except Exception: - for x, y in zip(xs, ys): - if 0 <= x < target_width: - sweep[x] = float(y) - - finite_pre = np.isfinite(sweep) - n_valid_cur = int(np.count_nonzero(finite_pre)) - - if self._fancy: - try: - known = ~np.isnan(sweep) - if np.any(known): - known_idx = np.nonzero(known)[0] - for i0, i1 in zip(known_idx[:-1], known_idx[1:]): - if i1 - i0 > 1: - avg = (sweep[i0] + sweep[i1]) * 0.5 - sweep[i0 + 1 : i1] = avg - first_idx = int(known_idx[0]) - last_idx = int(known_idx[-1]) - if first_idx > 0: - sweep[:first_idx] = sweep[first_idx] - if last_idx < sweep.size - 1: - sweep[last_idx + 1 :] = sweep[last_idx] - except Exception: - pass - - try: - m = float(np.nanmean(sweep)) - if np.isfinite(m) and m < DATA_INVERSION_THRESHOLD: - sweep *= -1.0 - except Exception: - pass - - pre_exp_sweep = None - if self._logscale: - try: - pre_exp_sweep = sweep.copy() - with np.errstate(over="ignore", invalid="ignore"): - sweep = np.power(LOG_EXP, np.asarray(sweep, dtype=np.float64)).astype(np.float32) - sweep[~np.isfinite(sweep)] = np.nan - except Exception: - pass - - self._sweep_idx += 1 - if len(ch_list) > 1: - sys.stderr.write( - f"[warn] Sweep {self._sweep_idx}: изменялся номер канала: {ch_list}\n" - ) - now = time.time() - if self._last_sweep_ts is None: - dt_ms = float("nan") - else: - dt_ms = (now - self._last_sweep_ts) * 1000.0 - self._last_sweep_ts = now - self._n_valid_hist.append((now, n_valid_cur)) - while self._n_valid_hist and (now - self._n_valid_hist[0][0]) > 1.0: - self._n_valid_hist.popleft() - if self._n_valid_hist: - n_valid = float(sum(v for _t, v in self._n_valid_hist) / len(self._n_valid_hist)) - else: - n_valid = float(n_valid_cur) - - if n_valid_cur > 0: - vmin = float(np.nanmin(sweep)) - vmax = float(np.nanmax(sweep)) - mean = float(np.nanmean(sweep)) - std = float(np.nanstd(sweep)) - else: - vmin = vmax = mean = std = float("nan") - info: SweepInfo = { - "sweep": self._sweep_idx, - "ch": ch_primary, - "chs": ch_list, - "n_valid": n_valid, - "min": vmin, - "max": vmax, - "mean": mean, - "std": std, - "dt_ms": dt_ms, - } - if pre_exp_sweep is not None: - info["pre_exp_sweep"] = pre_exp_sweep - + sweep, info = packet try: self._q.put_nowait((sweep, info)) except Full: @@ -263,6 +154,7 @@ class SweepReader(threading.Thread): ys: list[int] = [] cur_channel: Optional[int] = None cur_channels: set[int] = set() + parser = BinaryRecordStreamParser() # Бинарный протокол (4 слова LE u16 = 8 байт на запись): # старт свипа: 0xFFFF, 0xFFFF, 0xFFFF, (ch<<8)|0x0A @@ -274,7 +166,6 @@ class SweepReader(threading.Thread): # Признак записи: байт 6 == 0x0A, байт 7 — номер канала. # При десинхронизации сдвигаемся на 1 БАЙТ (не слово) для самосинхронизации. - buf = bytearray() _dbg_byte_count = 0 _dbg_desync_count = 0 _dbg_sweep_count = 0 @@ -282,20 +173,15 @@ class SweepReader(threading.Thread): while not self._stop.is_set(): data = chunk_reader.read_available() if data: - buf += data + events = parser.feed(data) else: time.sleep(0.0005) continue - while len(buf) >= 8: - # Читаем 4 LE u16 слова прямо из байтового буфера - w0 = int(buf[0]) | (int(buf[1]) << 8) - w1 = int(buf[2]) | (int(buf[3]) << 8) - w2 = int(buf[4]) | (int(buf[5]) << 8) - - # Старт свипа: три слова 0xFFFF + маркер 0x0A в байте 6, канал в байте 7 - if w0 == 0xFFFF and w1 == 0xFFFF and w2 == 0xFFFF and buf[6] == 0x0A: - ch_new = buf[7] + for ev in events: + tag = ev[0] + if tag == "start": + ch_new = int(ev[1]) if self._debug: sys.stderr.write(f"[debug] BIN: старт свипа, ch={ch_new}\n") _dbg_sweep_count += 1 @@ -305,41 +191,22 @@ class SweepReader(threading.Thread): cur_channels.clear() cur_channel = ch_new cur_channels.add(cur_channel) - del buf[:8] - _dbg_byte_count += 8 continue - # Точка данных: маркер 0x0A в байте 6, канал в байте 7 - if buf[6] == 0x0A: - ch_from_term = buf[7] - if cur_channel is None: - cur_channel = ch_from_term - cur_channels.add(cur_channel) - xs.append(w0) - value_u32 = (w1 << 16) | w2 - ys.append(self._u32_to_i32(value_u32)) - del buf[:8] - _dbg_byte_count += 8 - _dbg_point_count += 1 - if self._debug and _dbg_point_count <= 3: - sys.stderr.write( - f"[debug] BIN точка: step={w0} hi={w1:#06x} lo={w2:#06x} " - f"ch={ch_from_term} → value={self._u32_to_i32((w1 << 16) | w2)}\n" - ) - continue - - # Поток не выровнен; сдвигаемся на 1 байт до ресинхронизации. - _dbg_desync_count += 1 - _dbg_byte_count += 1 - if self._debug and _dbg_desync_count <= 8: - hex6 = " ".join(f"{buf[k]:02x}" for k in range(min(8, len(buf)))) + _tag, ch_from_term, step, value_i32 = ev # type: ignore[misc] + if cur_channel is None: + cur_channel = int(ch_from_term) + cur_channels.add(int(cur_channel)) + xs.append(int(step)) + ys.append(int(value_i32)) + _dbg_point_count += 1 + if self._debug and _dbg_point_count <= 3: sys.stderr.write( - f"[debug] BIN десинхронизация #{_dbg_desync_count}: " - f"байты [{hex6}] не совпадают ни с одним шаблоном\n" + f"[debug] BIN точка: step={int(step)} ch={int(ch_from_term)} → value={int(value_i32)}\n" ) - if self._debug and _dbg_desync_count == 9: - sys.stderr.write("[debug] BIN: дальнейшие десинхронизации не выводятся (слишком много)\n") - del buf[:1] + + _dbg_byte_count = parser.bytes_consumed + _dbg_desync_count = parser.desync_count if self._debug and _dbg_byte_count > 0 and _dbg_byte_count % 4000 < 8: sys.stderr.write( @@ -347,8 +214,8 @@ class SweepReader(threading.Thread): f"десинхронизаций={_dbg_desync_count}, точек={_dbg_point_count}, свипов={_dbg_sweep_count}\n" ) - if len(buf) > 1_000_000: - del buf[:-262144] + if parser.buffered_size() > 1_000_000: + parser.clear_buffer_keep_tail(262_144) self._finalize_current(xs, ys, cur_channels) diff --git a/rfg_adc_plotter/processing/fourier.py b/rfg_adc_plotter/processing/fourier.py new file mode 100644 index 0000000..54a7f8f --- /dev/null +++ b/rfg_adc_plotter/processing/fourier.py @@ -0,0 +1,43 @@ +"""Преобразование свипа в IFFT-временной профиль (дБ).""" + +from typing import Optional + +import numpy as np + +from rfg_adc_plotter.constants import FREQ_SPAN_GHZ, IFFT_LEN, SWEEP_LEN, ZEROS_LOW, ZEROS_MID + + +def build_ifft_time_axis_ns() -> np.ndarray: + """Временная ось IFFT в наносекундах.""" + return ( + np.arange(IFFT_LEN, dtype=np.float64) / (FREQ_SPAN_GHZ * 1e9) * 1e9 + ).astype(np.float32) + + +def compute_ifft_db_profile(sweep: Optional[np.ndarray]) -> np.ndarray: + """Построить IFFT-профиль свипа в дБ. + + Цепочка: + raw/processed sweep -> двусторонний спектр (заполнение нулями) -> + ifftshift -> ifft -> |x| -> 20log10. + """ + bins = IFFT_LEN + if sweep is None: + return np.full((bins,), np.nan, dtype=np.float32) + + s = np.asarray(sweep) + if s.size == 0: + return np.full((bins,), np.nan, dtype=np.float32) + + sig = np.zeros(SWEEP_LEN, dtype=np.float32) + take = min(int(s.size), SWEEP_LEN) + seg = np.nan_to_num(s[:take], nan=0.0).astype(np.float32, copy=False) + sig[:take] = seg + + data = np.zeros(IFFT_LEN, dtype=np.complex64) + data[ZEROS_LOW + ZEROS_MID :] = sig + + spec = np.fft.ifftshift(data) + result = np.fft.ifft(spec) + mag = np.abs(result).astype(np.float32) + return (mag + 1e-9).astype(np.float32) diff --git a/rfg_adc_plotter/processing/pipeline.py b/rfg_adc_plotter/processing/pipeline.py new file mode 100644 index 0000000..02dfbfe --- /dev/null +++ b/rfg_adc_plotter/processing/pipeline.py @@ -0,0 +1,415 @@ +"""Явный pipeline предобработки свипов перед помещением в RingBuffer.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +from typing import Optional, Tuple + +import numpy as np + +from rfg_adc_plotter.io.capture_reference_loader import ( + CaptureParseSummary, + aggregate_capture_reference, + detect_reference_file_format, + load_capture_sweeps, +) +from rfg_adc_plotter.processing.normalizer import ( + build_calib_envelopes, + normalize_by_calib, + normalize_by_envelope, +) + +DEFAULT_CALIB_ENVELOPE_PATH = "calib_envelope.npy" +DEFAULT_BACKGROUND_PATH = "background.npy" + + +def _normalize_path(path: str) -> str: + return str(path).strip() + + +def _normalize_save_npy_path(path: str) -> str: + p = _normalize_path(path) + if not p: + return p + _root, ext = os.path.splitext(p) + if ext: + return p + return f"{p}.npy" + + +def _summary_for_npy(path: str) -> CaptureParseSummary: + return CaptureParseSummary( + path=path, + format="npy", + sweeps_total=0, + sweeps_valid=0, + channels_seen=tuple(), + dominant_width=None, + dominant_n_valid=None, + aggregation="median", + warnings=tuple(), + ) + + +@dataclass(frozen=True) +class SweepProcessingResult: + """Результат предобработки одного свипа.""" + + processed_sweep: np.ndarray + normalized_sweep: Optional[np.ndarray] + calibration_applied: bool + background_applied: bool + calibration_source: str # off|live|npy|capture + background_source: str # off|npy|capture(raw)|capture(raw->calib) + is_calibration_reference: bool + stage_trace: Tuple[str, ...] + + +class SweepPreprocessor: + """Управляет калибровкой/фоном и применяет их к входному свипу.""" + + def __init__( + self, + norm_type: str = "projector", + calib_envelope_path: str = DEFAULT_CALIB_ENVELOPE_PATH, + background_path: str = DEFAULT_BACKGROUND_PATH, + auto_save_live_calib_envelope: bool = True, + ): + self.norm_type = str(norm_type).strip().lower() or "projector" + self.calib_enabled = False + self.calib_mode = "live" # live | file + self.background_enabled = False + self.auto_save_live_calib_envelope = bool(auto_save_live_calib_envelope) + + self.calib_envelope_path = _normalize_path(calib_envelope_path) + self.background_path = _normalize_path(background_path) + + self.last_calib_sweep: Optional[np.ndarray] = None + self.calib_file_envelope: Optional[np.ndarray] = None + + # background — в текущем домене вычитания (raw или normalized), UI использует для preview/state + self.background: Optional[np.ndarray] = None + # raw background loaded from capture file; преобразуется на лету при активной калибровке + self.background_raw_capture: Optional[np.ndarray] = None + + # Источники и метаданные загрузки + self.calib_external_source_type: str = "none" # none|npy|capture + self.background_source_type: str = "none" # none|npy_processed|capture_raw + self.calib_reference_summary: Optional[CaptureParseSummary] = None + self.background_reference_summary: Optional[CaptureParseSummary] = None + self.last_reference_error: str = "" + + # Параметры офлайн-парсинга capture (должны совпадать с live parser по настройке UI) + self.capture_fancy: bool = False + self.capture_logscale: bool = False + self.reference_aggregation_method: str = "median" + + # ---- Конфигурация ---- + def set_calib_mode(self, mode: str): + m = str(mode).strip().lower() + self.calib_mode = "file" if m == "file" else "live" + + def set_calib_enabled(self, enabled: bool): + self.calib_enabled = bool(enabled) + + def set_background_enabled(self, enabled: bool): + self.background_enabled = bool(enabled) + + def set_capture_parse_options(self, *, fancy: Optional[bool] = None, logscale: Optional[bool] = None): + if fancy is not None: + self.capture_fancy = bool(fancy) + if logscale is not None: + self.capture_logscale = bool(logscale) + + def set_calib_envelope_path(self, path: str): + p = _normalize_path(path) + if p: + if p != self.calib_envelope_path: + self.calib_file_envelope = None + if self.calib_external_source_type in ("npy", "capture"): + self.calib_external_source_type = "none" + self.calib_reference_summary = None + self.calib_envelope_path = p + + def set_background_path(self, path: str): + p = _normalize_path(path) + if p: + if p != self.background_path: + self.background = None + self.background_raw_capture = None + self.background_source_type = "none" + self.background_reference_summary = None + self.background_path = p + + def has_calib_envelope_file(self) -> bool: + return bool(self.calib_envelope_path) and os.path.isfile(self.calib_envelope_path) + + def has_background_file(self) -> bool: + return bool(self.background_path) and os.path.isfile(self.background_path) + + # ---- Загрузка/сохранение .npy ---- + def _save_array(self, arr: np.ndarray, current_path: str, path: Optional[str]) -> str: + target = _normalize_save_npy_path(path if path is not None else current_path) + if not target: + raise ValueError("Пустой путь сохранения") + np.save(target, arr) + return target + + def save_calib_envelope(self, path: Optional[str] = None) -> bool: + """Сохранить огибающую из последнего live-калибровочного свипа (экспорт .npy).""" + if self.last_calib_sweep is None: + return False + try: + _lower, upper = build_calib_envelopes(self.last_calib_sweep) + self.calib_envelope_path = self._save_array(upper, self.calib_envelope_path, path) + self.last_reference_error = "" + return True + except Exception as exc: + self.last_reference_error = f"save calib envelope failed: {exc}" + return False + + def save_background(self, sweep_for_ring: Optional[np.ndarray], path: Optional[str] = None) -> bool: + """Сохранить текущий свип (в текущем домене обработки) как .npy-фон.""" + if sweep_for_ring is None: + return False + try: + bg = np.asarray(sweep_for_ring, dtype=np.float32).copy() + self.background_path = self._save_array(bg, self.background_path, path) + self.background = bg + self.background_raw_capture = None + self.background_source_type = "npy_processed" + self.background_reference_summary = _summary_for_npy(self.background_path) + self.last_reference_error = "" + return True + except Exception as exc: + self.last_reference_error = f"save background failed: {exc}" + return False + + # ---- Загрузка эталонов (.npy или capture) ---- + def _detect_source_kind(self, path: str, source_kind: str) -> Optional[str]: + sk = str(source_kind).strip().lower() or "auto" + if sk == "auto": + return detect_reference_file_format(path) + if sk in ("npy", "bin_capture", "capture"): + return "bin_capture" if sk == "capture" else sk + return None + + def _load_npy_vector(self, path: str) -> np.ndarray: + arr = np.load(path) + return np.asarray(arr, dtype=np.float32).reshape(-1) + + def load_calib_reference( + self, + path: Optional[str] = None, + *, + source_kind: str = "auto", + method: str = "median", + ) -> bool: + """Загрузить калибровку из .npy (огибающая) или raw capture файла.""" + if path is not None: + self.set_calib_envelope_path(path) + p = self.calib_envelope_path + if not p or not os.path.isfile(p): + self.last_reference_error = f"Файл калибровки не найден: {p}" + return False + + fmt = self._detect_source_kind(p, source_kind) + if fmt is None: + self.last_reference_error = f"Неизвестный формат файла калибровки: {p}" + return False + + try: + if fmt == "npy": + env = self._load_npy_vector(p) + self.calib_file_envelope = env + self.calib_external_source_type = "npy" + self.calib_reference_summary = _summary_for_npy(p) + self.last_reference_error = "" + return True + + sweeps = load_capture_sweeps(p, fancy=self.capture_fancy, logscale=self.capture_logscale) + vec, summary = aggregate_capture_reference( + sweeps, + channel=0, + method=method or self.reference_aggregation_method, + path=p, + ) + _lower, upper = build_calib_envelopes(vec) + self.calib_file_envelope = np.asarray(upper, dtype=np.float32) + self.calib_external_source_type = "capture" + self.calib_reference_summary = summary + self.last_reference_error = "" + return True + except Exception as exc: + self.last_reference_error = f"Ошибка загрузки калибровки: {exc}" + return False + + def load_background_reference( + self, + path: Optional[str] = None, + *, + source_kind: str = "auto", + method: str = "median", + ) -> bool: + """Загрузить фон из .npy (готовый домен) или raw capture файла.""" + if path is not None: + self.set_background_path(path) + p = self.background_path + if not p or not os.path.isfile(p): + self.last_reference_error = f"Файл фона не найден: {p}" + return False + + fmt = self._detect_source_kind(p, source_kind) + if fmt is None: + self.last_reference_error = f"Неизвестный формат файла фона: {p}" + return False + + try: + if fmt == "npy": + bg = self._load_npy_vector(p) + self.background = bg + self.background_raw_capture = None + self.background_source_type = "npy_processed" + self.background_reference_summary = _summary_for_npy(p) + self.last_reference_error = "" + return True + + sweeps = load_capture_sweeps(p, fancy=self.capture_fancy, logscale=self.capture_logscale) + vec, summary = aggregate_capture_reference( + sweeps, + channel=0, + method=method or self.reference_aggregation_method, + path=p, + ) + self.background_raw_capture = np.asarray(vec, dtype=np.float32) + # Для UI/preview текущий background отражает текущий домен (пока raw по умолчанию). + self.background = self.background_raw_capture + self.background_source_type = "capture_raw" + self.background_reference_summary = summary + self.last_reference_error = "" + return True + except Exception as exc: + self.last_reference_error = f"Ошибка загрузки фона: {exc}" + return False + + # Совместимые обертки для старого API (строго .npy) + def load_calib_envelope(self, path: Optional[str] = None) -> bool: + target = path if path is not None else self.calib_envelope_path + return self.load_calib_reference(target, source_kind="npy") + + def load_background(self, path: Optional[str] = None) -> bool: + target = path if path is not None else self.background_path + return self.load_background_reference(target, source_kind="npy") + + # ---- Нормировка / фон ---- + def _normalize_against_active_reference(self, raw: np.ndarray) -> Tuple[Optional[np.ndarray], bool, str]: + if not self.calib_enabled: + return None, False, "off" + + if self.calib_mode == "file": + if self.calib_file_envelope is None: + return None, False, "off" + src = "capture" if self.calib_external_source_type == "capture" else "npy" + return normalize_by_envelope(raw, self.calib_file_envelope), True, src + + if self.last_calib_sweep is None: + return None, False, "off" + return normalize_by_calib(raw, self.last_calib_sweep, self.norm_type), True, "live" + + def _transform_raw_background_for_current_domain(self, calib_applied: bool) -> Optional[np.ndarray]: + if self.background_raw_capture is None: + return None + if not calib_applied: + return self.background_raw_capture + + # Порядок pipeline фиксирован: raw -> calibration -> background -> IFFT. + # Поэтому raw-фон из capture нужно привести в тот же домен, что и текущий sweep_for_ring. + if self.calib_mode == "file" and self.calib_file_envelope is not None: + return normalize_by_envelope(self.background_raw_capture, self.calib_file_envelope) + if self.calib_mode == "live" and self.last_calib_sweep is not None: + return normalize_by_calib(self.background_raw_capture, self.last_calib_sweep, self.norm_type) + return None + + def _effective_background(self, calib_applied: bool) -> Tuple[Optional[np.ndarray], str]: + if self.background_source_type == "capture_raw": + bg = self._transform_raw_background_for_current_domain(calib_applied) + if bg is None: + return None, "capture(raw->calib:missing-calib)" + self.background = np.asarray(bg, dtype=np.float32) + return self.background, ("capture(raw->calib)" if calib_applied else "capture(raw)") + if self.background_source_type == "npy_processed" and self.background is not None: + return self.background, "npy" + if self.background is not None: + return self.background, "unknown" + return None, "off" + + def _subtract_background(self, sweep: np.ndarray, calib_applied: bool) -> Tuple[np.ndarray, bool, str]: + if not self.background_enabled: + return sweep, False, "off" + bg, bg_src = self._effective_background(calib_applied) + if bg is None: + return sweep, False, f"{bg_src}:missing" + out = np.asarray(sweep, dtype=np.float32).copy() + w = min(out.size, bg.size) + if w > 0: + out[:w] -= bg[:w] + return out, True, bg_src + + def process(self, sweep: np.ndarray, channel: int, update_references: bool = True) -> SweepProcessingResult: + """Применить к свипу калибровку/фон и вернуть явные этапы обработки.""" + raw = np.asarray(sweep, dtype=np.float32) + ch = int(channel) + + if ch == 0: + if update_references: + self.last_calib_sweep = raw + if self.auto_save_live_calib_envelope: + self.save_calib_envelope() + + # ch0 всегда остаётся live-калибровочной ссылкой (raw), но при file-калибровке + # можем применять её и к ch0 для отображения/обработки независимо от канала. + calib_applied = False + calib_source = "off" + normalized: Optional[np.ndarray] = None + if self.calib_enabled and self.calib_mode == "file": + normalized, calib_applied, calib_source = self._normalize_against_active_reference(raw) + + base = normalized if normalized is not None else raw + processed, bg_applied, bg_source = self._subtract_background(base, calib_applied=calib_applied) + + stages = ["parsed_sweep", "raw_sweep", "ch0_live_calibration_reference"] + stages.append(f"calibration_{calib_source}" if calib_applied else "calibration_off") + stages.append(f"background_{bg_source}" if bg_applied else "background_off") + stages.extend(["ring_buffer", "ifft_db"]) + + return SweepProcessingResult( + processed_sweep=processed, + normalized_sweep=normalized, + calibration_applied=calib_applied, + background_applied=bg_applied, + calibration_source=calib_source if calib_applied else "off", + background_source=bg_source if bg_applied else "off", + is_calibration_reference=True, + stage_trace=tuple(stages), + ) + + normalized, calib_applied, calib_source = self._normalize_against_active_reference(raw) + base = normalized if normalized is not None else raw + processed, bg_applied, bg_source = self._subtract_background(base, calib_applied) + + stages = ["parsed_sweep", "raw_sweep"] + stages.append(f"calibration_{calib_source}" if calib_applied else "calibration_off") + stages.append(f"background_{bg_source}" if bg_applied else "background_off") + stages.extend(["ring_buffer", "ifft_db"]) + + return SweepProcessingResult( + processed_sweep=processed, + normalized_sweep=normalized, + calibration_applied=calib_applied, + background_applied=bg_applied, + calibration_source=calib_source if calib_applied else "off", + background_source=bg_source if bg_applied else "off", + is_calibration_reference=False, + stage_trace=tuple(stages), + ) diff --git a/rfg_adc_plotter/state/app_state.py b/rfg_adc_plotter/state/app_state.py index 10c339a..4adab5e 100644 --- a/rfg_adc_plotter/state/app_state.py +++ b/rfg_adc_plotter/state/app_state.py @@ -1,21 +1,20 @@ """Состояние приложения: текущие свипы и настройки калибровки/нормировки.""" -import os from queue import Empty, Queue -from typing import Any, Dict, Mapping, Optional +from typing import Any, Mapping, Optional import numpy as np -from rfg_adc_plotter.processing.normalizer import ( - build_calib_envelopes, - normalize_by_calib, - normalize_by_envelope, +from rfg_adc_plotter.processing.pipeline import ( + DEFAULT_BACKGROUND_PATH, + DEFAULT_CALIB_ENVELOPE_PATH, + SweepPreprocessor, ) from rfg_adc_plotter.state.ring_buffer import RingBuffer from rfg_adc_plotter.types import SweepInfo, SweepPacket -CALIB_ENVELOPE_PATH = "calib_envelope.npy" -BACKGROUND_PATH = "background.npy" +CALIB_ENVELOPE_PATH = DEFAULT_CALIB_ENVELOPE_PATH +BACKGROUND_PATH = DEFAULT_BACKGROUND_PATH def format_status(data: Mapping[str, Any]) -> str: @@ -39,11 +38,7 @@ def format_status(data: Mapping[str, Any]) -> str: class AppState: - """Весь изменяемый GUI-state: текущие данные, калибровка, настройки. - - Методы drain_queue и set_calib_enabled заменяют одноимённые closures - с nonlocal из оригинального кода. - """ + """Весь изменяемый GUI-state: текущие данные + pipeline предобработки.""" def __init__(self, norm_type: str = "projector"): self.current_sweep_pre_exp: Optional[np.ndarray] = None @@ -51,116 +46,265 @@ class AppState: self.current_sweep_processed: Optional[np.ndarray] = None self.current_sweep_raw: Optional[np.ndarray] = None self.current_sweep_norm: Optional[np.ndarray] = None - self.last_calib_sweep: Optional[np.ndarray] = None self.current_info: Optional[SweepInfo] = None - self.calib_enabled: bool = False - self.norm_type: str = norm_type - # "live" — нормировка по текущему ch0-свипу; "file" — по огибающей из файла - self.calib_mode: str = "live" - self.calib_file_envelope: Optional[np.ndarray] = None - # Вычет фона - self.background: Optional[np.ndarray] = None - self.background_enabled: bool = False + self.norm_type: str = str(norm_type).strip().lower() + self.preprocessor = SweepPreprocessor(norm_type=self.norm_type) self._last_sweep_for_ring: Optional[np.ndarray] = None + self._last_stage_trace: tuple[str, ...] = tuple() - def _normalize(self, raw: np.ndarray, calib: np.ndarray) -> np.ndarray: - if self.calib_mode == "file" and self.calib_file_envelope is not None: - return normalize_by_envelope(raw, self.calib_file_envelope) - return normalize_by_calib(raw, calib, self.norm_type) + def configure_capture_import(self, *, fancy: Optional[bool] = None, logscale: Optional[bool] = None): + self.preprocessor.set_capture_parse_options(fancy=fancy, logscale=logscale) - def save_calib_envelope(self, path: str = CALIB_ENVELOPE_PATH) -> bool: - """Вычислить огибающую из last_calib_sweep и сохранить в файл. + # ---- Свойства pipeline (для совместимости с GUI) ---- + @property + def calib_enabled(self) -> bool: + return self.preprocessor.calib_enabled - Возвращает True при успехе. - """ - if self.last_calib_sweep is None: - return False - try: - _lower, upper = build_calib_envelopes(self.last_calib_sweep) - np.save(path, upper) - return True - except Exception as exc: - import sys - sys.stderr.write(f"[warn] Не удалось сохранить огибающую: {exc}\n") - return False + @property + def calib_mode(self) -> str: + return self.preprocessor.calib_mode - def load_calib_envelope(self, path: str = CALIB_ENVELOPE_PATH) -> bool: - """Загрузить огибающую из файла. + @property + def calib_file_envelope(self) -> Optional[np.ndarray]: + return self.preprocessor.calib_file_envelope - Возвращает True при успехе. - """ - if not os.path.isfile(path): - return False - try: - env = np.load(path) - self.calib_file_envelope = np.asarray(env, dtype=np.float32) - return True - except Exception as exc: - import sys - sys.stderr.write(f"[warn] Не удалось загрузить огибающую: {exc}\n") - return False + @property + def last_calib_sweep(self) -> Optional[np.ndarray]: + return self.preprocessor.last_calib_sweep + + @property + def background(self) -> Optional[np.ndarray]: + return self.preprocessor.background + + @property + def background_enabled(self) -> bool: + return self.preprocessor.background_enabled + + @property + def calib_source_type(self) -> str: + return self.preprocessor.calib_external_source_type + + @property + def background_source_type(self) -> str: + return self.preprocessor.background_source_type + + @property + def calib_reference_summary(self): + return self.preprocessor.calib_reference_summary + + @property + def background_reference_summary(self): + return self.preprocessor.background_reference_summary + + @property + def last_reference_error(self) -> str: + return self.preprocessor.last_reference_error + + @property + def calib_envelope_path(self) -> str: + return self.preprocessor.calib_envelope_path + + @property + def background_path(self) -> str: + return self.preprocessor.background_path + + # ---- Управление файлами калибровки/фона ---- + def set_calib_envelope_path(self, path: str): + self.preprocessor.set_calib_envelope_path(path) + self._refresh_current_processed() + + def set_background_path(self, path: str): + self.preprocessor.set_background_path(path) + self._refresh_current_processed() + + def has_calib_envelope_file(self) -> bool: + return self.preprocessor.has_calib_envelope_file() + + def has_background_file(self) -> bool: + return self.preprocessor.has_background_file() + + def save_calib_envelope(self, path: Optional[str] = None) -> bool: + return self.preprocessor.save_calib_envelope(path) + + def load_calib_reference(self, path: Optional[str] = None) -> bool: + ok = self.preprocessor.load_calib_reference(path) + if ok: + self._refresh_current_processed() + return ok + + def load_calib_envelope(self, path: Optional[str] = None) -> bool: + return self.load_calib_reference(path) def set_calib_mode(self, mode: str): - """Переключить режим калибровки: 'live' или 'file'.""" - self.calib_mode = mode + self.preprocessor.set_calib_mode(mode) + self._refresh_current_processed() - def save_background(self, path: str = BACKGROUND_PATH) -> bool: - """Сохранить текущий sweep_for_ring как фоновый спектр. + def save_background(self, path: Optional[str] = None) -> bool: + return self.preprocessor.save_background(self._last_sweep_for_ring, path) - Сохраняет последний свип, который был записан в ринг-буфер - (нормированный, если калибровка включена, иначе сырой). - Возвращает True при успехе. - """ - if self._last_sweep_for_ring is None: - return False - try: - np.save(path, self._last_sweep_for_ring) - return True - except Exception as exc: - import sys - sys.stderr.write(f"[warn] Не удалось сохранить фон: {exc}\n") - return False + def load_background_reference(self, path: Optional[str] = None) -> bool: + ok = self.preprocessor.load_background_reference(path) + if ok: + self._refresh_current_processed() + return ok - def load_background(self, path: str = BACKGROUND_PATH) -> bool: - """Загрузить фоновый спектр из файла. - - Возвращает True при успехе. - """ - if not os.path.isfile(path): - return False - try: - bg = np.load(path) - self.background = np.asarray(bg, dtype=np.float32) - return True - except Exception as exc: - import sys - sys.stderr.write(f"[warn] Не удалось загрузить фон: {exc}\n") - return False + def load_background(self, path: Optional[str] = None) -> bool: + return self.load_background_reference(path) def set_background_enabled(self, enabled: bool): - """Включить/выключить вычет фона.""" - self.background_enabled = enabled + self.preprocessor.set_background_enabled(enabled) + self._refresh_current_processed() def set_calib_enabled(self, enabled: bool): - """Включить/выключить режим калибровки, пересчитать norm-свип.""" - self.calib_enabled = enabled - if self.calib_enabled and self.current_sweep_raw is not None: - if self.calib_mode == "file" and self.calib_file_envelope is not None: - self.current_sweep_norm = normalize_by_envelope( - self.current_sweep_raw, self.calib_file_envelope - ) - elif self.calib_mode == "live" and self.last_calib_sweep is not None: - self.current_sweep_norm = self._normalize( - self.current_sweep_raw, self.last_calib_sweep - ) - else: - self.current_sweep_norm = None + self.preprocessor.set_calib_enabled(enabled) + self._refresh_current_processed() + + # ---- Вспомогательные методы для UI ---- + def _current_channel(self) -> Optional[int]: + if not isinstance(self.current_info, dict): + return None + try: + return int(self.current_info.get("ch", 0)) + except Exception: + return 0 + + def _apply_result_to_current(self, result) -> None: + self._last_stage_trace = tuple(result.stage_trace) + if result.is_calibration_reference: + self.current_sweep_norm = None + elif result.calibration_applied or result.background_applied: + self.current_sweep_norm = result.processed_sweep else: self.current_sweep_norm = None - self.current_sweep_processed = ( - self.current_sweep_norm if self.current_sweep_norm is not None else self.current_sweep_raw + self.current_sweep_processed = result.processed_sweep + self._last_sweep_for_ring = result.processed_sweep + + def _refresh_current_processed(self): + if self.current_sweep_raw is None: + self.current_sweep_norm = None + self.current_sweep_processed = None + self._last_stage_trace = tuple() + return + ch = self._current_channel() or 0 + result = self.preprocessor.process(self.current_sweep_raw, ch, update_references=False) + self._apply_result_to_current(result) + + def format_pipeline_status(self) -> str: + """Краткое описание pipeline для UI: от распарсенного свипа до IFFT.""" + ch = self._current_channel() + if ch is None: + ch_txt = "?" + else: + ch_txt = str(ch) + + reader_stage = "log-exp" if self.current_sweep_pre_exp is not None else "linear" + + if ch == 0: + file_calib_applies = ( + self.calib_enabled + and self.calib_mode == "file" + and self.calib_file_envelope is not None + ) + if self.calib_enabled and self.calib_mode == "file": + calib_stage = self.format_calib_source_status() + else: + calib_stage = "calib[off]" + + if not self.background_enabled: + bg_stage = "bg[off]" + elif self.background_source_type == "capture_raw": + if self.background is None: + bg_stage = ( + "bg[capture(raw->calib):missing]" + if file_calib_applies + else "bg[capture(raw):missing]" + ) + else: + bg_stage = "bg[capture(raw->calib)]" if file_calib_applies else "bg[capture(raw)]" + elif self.background_source_type == "npy_processed": + bg_stage = "bg[npy]" if self.background is not None else "bg[npy:missing]" + else: + bg_stage = "bg[sub]" if self.background is not None else "bg[missing]" + return ( + f"pipeline ch{ch_txt}: parsed -> {reader_stage} -> raw -> " + f"live-calib-ref -> {calib_stage} -> {bg_stage} -> ring -> IFFT(dB)" + ) + + calib_stage = self.format_calib_source_status() + bg_stage = self.format_background_source_status() + + return ( + f"pipeline ch{ch_txt}: parsed -> {reader_stage} -> raw -> " + f"{calib_stage} -> {bg_stage} -> ring -> IFFT(dB)" ) + def _format_summary(self, summary) -> str: + if summary is None: + return "" + parts: list[str] = [] + if getattr(summary, "sweeps_valid", 0) or getattr(summary, "sweeps_total", 0): + parts.append(f"valid:{summary.sweeps_valid}/{summary.sweeps_total}") + if getattr(summary, "dominant_width", None) is not None: + parts.append(f"w:{summary.dominant_width}") + chs = getattr(summary, "channels_seen", tuple()) + if chs: + parts.append("chs:" + ",".join(str(v) for v in chs)) + warns = getattr(summary, "warnings", tuple()) + if warns: + parts.append(f"warn:{warns[0]}") + return " ".join(parts) + + def format_calib_source_status(self) -> str: + if not self.calib_enabled: + return "calib[off]" + if self.calib_mode == "live": + return "calib[live]" if self.last_calib_sweep is not None else "calib[live:no-ref]" + if self.calib_file_envelope is None: + return "calib[file:missing]" + if self.calib_source_type == "capture": + return "calib[capture]" + if self.calib_source_type == "npy": + return "calib[npy]" + return "calib[file]" + + def format_background_source_status(self) -> str: + if not self.background_enabled: + return "bg[off]" + src = self.background_source_type + if src == "capture_raw": + if self.calib_enabled: + can_map = ( + (self.calib_mode == "file" and self.calib_file_envelope is not None) + or (self.calib_mode == "live" and self.last_calib_sweep is not None) + ) + if not can_map: + return "bg[capture(raw->calib):missing]" + if self.background is None: + return "bg[capture(raw->calib):missing]" + return "bg[capture(raw->calib)]" if self.calib_enabled else "bg[capture(raw)]" + if src == "npy_processed": + return "bg[npy]" if self.background is not None else "bg[npy:missing]" + if self.background is not None: + return "bg[sub]" + return "bg[missing]" + + def format_reference_status(self) -> str: + parts: list[str] = [] + calib_s = self._format_summary(self.calib_reference_summary) + if calib_s: + parts.append(f"calib[{calib_s}]") + bg_s = self._format_summary(self.background_reference_summary) + if bg_s: + parts.append(f"bg[{bg_s}]") + if self.last_reference_error: + parts.append(f"err:{self.last_reference_error}") + return " | ".join(parts) + + def format_stage_trace(self) -> str: + if not self._last_stage_trace: + return "" + return " -> ".join(self._last_stage_trace) + def drain_queue(self, q: "Queue[SweepPacket]", ring: RingBuffer) -> int: """Вытащить все ожидающие свипы из очереди, обновить state и ring. @@ -173,49 +317,23 @@ class AppState: except Empty: break drained += 1 + self.current_sweep_raw = s self.current_sweep_post_exp = s self.current_info = info pre_exp = info.get("pre_exp_sweep") if isinstance(info, dict) else None self.current_sweep_pre_exp = pre_exp if isinstance(pre_exp, np.ndarray) else None - ch = 0 try: ch = int(info.get("ch", 0)) if isinstance(info, dict) else 0 except Exception: ch = 0 - # Канал 0 — опорный (калибровочный) свип - if ch == 0: - self.last_calib_sweep = s - self.save_calib_envelope() - self.current_sweep_norm = None - sweep_for_ring = s - self._last_sweep_for_ring = sweep_for_ring - else: - can_normalize = self.calib_enabled and ( - (self.calib_mode == "file" and self.calib_file_envelope is not None) - or (self.calib_mode == "live" and self.last_calib_sweep is not None) - ) - if can_normalize: - calib_ref = self.last_calib_sweep if self.last_calib_sweep is not None else s - self.current_sweep_norm = self._normalize(s, calib_ref) - sweep_for_ring = self.current_sweep_norm - else: - self.current_sweep_norm = None - sweep_for_ring = s + result = self.preprocessor.process(s, ch, update_references=True) + self._apply_result_to_current(result) - # Вычет фона (в том же домене что и sweep_for_ring) - if self.background_enabled and self.background is not None and ch != 0: - w = min(sweep_for_ring.size, self.background.size) - sweep_for_ring = sweep_for_ring.copy() - sweep_for_ring[:w] -= self.background[:w] - self.current_sweep_norm = sweep_for_ring - - self._last_sweep_for_ring = sweep_for_ring - self.current_sweep_processed = sweep_for_ring ring.ensure_init(s.size) - ring.push(sweep_for_ring) + ring.push(result.processed_sweep) return drained def format_channel_label(self) -> str: diff --git a/rfg_adc_plotter/state/ring_buffer.py b/rfg_adc_plotter/state/ring_buffer.py index 11f9f09..81e8355 100644 --- a/rfg_adc_plotter/state/ring_buffer.py +++ b/rfg_adc_plotter/state/ring_buffer.py @@ -6,14 +6,10 @@ from typing import Optional, Tuple import numpy as np from rfg_adc_plotter.constants import ( - FFT_LEN, - FREQ_SPAN_GHZ, IFFT_LEN, - SWEEP_LEN, WF_WIDTH, - ZEROS_LOW, - ZEROS_MID, ) +from rfg_adc_plotter.processing.fourier import build_ifft_time_axis_ns, compute_ifft_db_profile class RingBuffer: @@ -51,10 +47,8 @@ class RingBuffer: self.ring = np.full((self.max_sweeps, self.width), np.nan, dtype=np.float32) self.ring_time = np.full((self.max_sweeps,), np.nan, dtype=np.float64) self.ring_fft = np.full((self.max_sweeps, self.fft_bins), np.nan, dtype=np.float32) - # Временная ось IFFT: шаг dt = 1/(FREQ_SPAN_GHZ*1e9), переведём в нс - self.fft_time_axis = ( - np.arange(IFFT_LEN, dtype=np.float64) / (FREQ_SPAN_GHZ * 1e9) * 1e9 - ).astype(np.float32) + # Временная ось IFFT вынесена в processing.fourier для явного pipeline. + self.fft_time_axis = build_ifft_time_axis_ns() self.head = 0 # Обновляем x_shared если пришёл свип большего размера if self.x_shared is None or sweep_width > self.x_shared.size: @@ -75,29 +69,7 @@ class RingBuffer: self._push_fft(s) def _push_fft(self, s: np.ndarray): - bins = self.ring_fft.shape[1] # = IFFT_LEN = 1953 - if s is None or s.size == 0: - fft_row = np.full((bins,), np.nan, dtype=np.float32) - else: - # 1. Взять первые SWEEP_LEN отсчётов (остаток — нули если свип короче) - sig = np.zeros(SWEEP_LEN, dtype=np.float32) - take = min(int(s.size), SWEEP_LEN) - seg = np.nan_to_num(s[:take], nan=0.0).astype(np.float32, copy=False) - sig[:take] = seg - - # 2. Собрать двусторонний спектр: - # [ZEROS_LOW нулей | ZEROS_MID нулей | SWEEP_LEN данных] - # = [-14.3..-3.2 ГГц | -3.2..+3.2 ГГц | +3.2..+14.3 ГГц] - data = np.zeros(IFFT_LEN, dtype=np.complex64) - data[ZEROS_LOW + ZEROS_MID:] = sig - - # 3. ifftshift + ifft → временной профиль - spec = np.fft.ifftshift(data) - result = np.fft.ifft(spec) - - # 4. Амплитуда в дБ - mag = np.abs(result).astype(np.float32) - fft_row = (20.0 * np.log10(mag + 1e-9)).astype(np.float32) + fft_row = compute_ifft_db_profile(s) prev_head = (self.head - 1) % self.ring_fft.shape[0] self.ring_fft[prev_head, :] = fft_row diff --git a/tests/test_capture_reference_loader.py b/tests/test_capture_reference_loader.py new file mode 100644 index 0000000..ff510b4 --- /dev/null +++ b/tests/test_capture_reference_loader.py @@ -0,0 +1,84 @@ +from pathlib import Path + +import numpy as np + +from rfg_adc_plotter.io.capture_reference_loader import ( + aggregate_capture_reference, + detect_reference_file_format, + load_capture_sweeps, +) +from rfg_adc_plotter.processing.pipeline import SweepPreprocessor + + +ROOT = Path(__file__).resolve().parents[1] +SAMPLE_BG = ROOT / "sample_data" / "empty" +SAMPLE_CALIB = ROOT / "sample_data" / "no_antennas_35dB_attenuators" + + +def test_detect_reference_file_format_for_sample_capture(): + assert detect_reference_file_format(str(SAMPLE_BG)) == "bin_capture" + assert detect_reference_file_format(str(SAMPLE_CALIB)) == "bin_capture" + + +def test_load_capture_sweeps_parses_binary_capture(): + sweeps = load_capture_sweeps(str(SAMPLE_BG), fancy=False, logscale=False) + assert len(sweeps) > 100 + sweep0, info0 = sweeps[0] + assert isinstance(sweep0, np.ndarray) + assert "ch" in info0 + channels = set() + for _s, info in sweeps: + chs = info.get("chs", [info.get("ch", 0)]) + channels.update(int(v) for v in chs) + assert channels == {0} + + +def test_aggregate_capture_reference_filters_incomplete_sweeps(): + sweeps = load_capture_sweeps(str(SAMPLE_BG), fancy=False, logscale=False) + vector, summary = aggregate_capture_reference(sweeps, channel=0, method="median", path=str(SAMPLE_BG)) + assert isinstance(vector, np.ndarray) + assert vector.dtype == np.float32 + assert summary.sweeps_total == len(sweeps) + assert summary.sweeps_valid > 0 + assert summary.sweeps_valid < summary.sweeps_total + assert summary.dominant_width in (759, 758) # sample_data starts at x=1..758 => width=759 + + +def test_preprocessor_can_load_capture_calib_and_background_and_apply(): + p = SweepPreprocessor(norm_type="projector", auto_save_live_calib_envelope=False) + p.set_capture_parse_options(fancy=False, logscale=False) + + assert p.load_calib_reference(str(SAMPLE_CALIB)) + p.set_calib_mode("file") + p.set_calib_enabled(True) + assert p.calib_file_envelope is not None + assert p.calib_external_source_type == "capture" + + assert p.load_background_reference(str(SAMPLE_BG)) + p.set_background_enabled(True) + assert p.background_source_type == "capture_raw" + + n = min(758, int(p.calib_file_envelope.size)) + sweep = np.linspace(-100.0, 100.0, n, dtype=np.float32) + res = p.process(sweep, channel=1, update_references=False) + assert res.calibration_applied is True + assert res.background_applied is True + assert res.calibration_source == "capture" + assert "background_capture(raw->calib)" in res.stage_trace + + +def test_preprocessor_applies_background_for_ch0_reference_too(): + p = SweepPreprocessor(norm_type="projector", auto_save_live_calib_envelope=False) + p.set_capture_parse_options(fancy=False, logscale=False) + assert p.load_background_reference(str(SAMPLE_BG)) + p.set_background_enabled(True) + + n = min(758, int(p.background.size)) if p.background is not None else 758 + raw = np.linspace(-10.0, 10.0, n, dtype=np.float32) + res = p.process(raw, channel=0, update_references=True) + + assert res.is_calibration_reference is True + assert res.background_applied is True + assert np.any(np.abs(res.processed_sweep - raw) > 0) + assert p.last_calib_sweep is not None + assert np.allclose(p.last_calib_sweep[:n], raw[:n], equal_nan=True)