import sys import time import struct import numpy as np import pandas as pd import serial import serial.tools.list_ports from PyQt6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QFileDialog, QTableView, QLabel, QRadioButton, QButtonGroup, QTabWidget, QMessageBox, QHeaderView, QCheckBox, QComboBox, QSplitter, QSlider) from PyQt6.QtCore import Qt, QAbstractTableModel, QThread, pyqtSignal, QTimer import pyqtgraph as pg # ========================================== # 1. 数据结构定义 # ========================================== # 同步头（小端序字节序列） MARKER_STANDARD = b'\xff\xff\xff\xff' # 0xFFFFFFFF - 标准包（无光泵） MARKER_OPTIC = b'\xfe\xff\xff\xff' # 0xFFFFFFFE - 扩展包（含光泵） def make_dtype(is_raw: bool, has_gps_pos: bool, has_optic: bool) -> np.dtype: """动态构建 numpy dtype，精确匹配固件 __attribute__((packed)) 结构体布局。 固件包结构: start_byte (u32) | adc1/corr_x | adc2/corr_y | adc3/corr_z | gps_time (u32) [| lat f32 | lon f32 | alt f32] [| optical_mag u32] """ fields = [('start_byte', ' 0: buffer.extend(self.serial_port.read(self.serial_port.in_waiting)) std_pkts = [] optic_pkts = [] while len(buffer) >= 4: m = struct.unpack_from('= std_size: std_pkts.append(bytes(buffer[:std_size])) del buffer[:std_size] elif mixed and m == MARKER_OPTIC_INT and len(buffer) >= optic_size: optic_pkts.append(bytes(buffer[:optic_size])) del buffer[:optic_size] elif m in (MARKER_STD_INT, MARKER_OPTIC_INT): break # 包还没收完，等下次 else: del buffer[0] # 重新同步 if std_pkts: self.data_received.emit( np.frombuffer(b''.join(std_pkts), dtype=self.dtype_std)) if optic_pkts: self.optic_received.emit( np.frombuffer(b''.join(optic_pkts), dtype=self.dtype_optic)) except Exception as e: if self.is_running: self.error_occurred.emit(str(e)) finally: self.stop() def stop(self): self.is_running = False if self.serial_port: try: self.serial_port.cancel_read() except Exception: pass try: if self.serial_port.is_open: self.serial_port.close() except Exception: pass finally: self.serial_port = None # ========================================== # 4. 后台线程：光泵磁力仪数据模拟输出 # ========================================== def _encode_optic_frame(value_thousandths_nT: int) -> bytes: """将 0.001 nT 为单位的整数编码为固件期望的 BCD 帧。 帧格式: 0x5A + 10字节BCD（每字节低nibble为一位十进制数字，高位在前）+ 0xA5 固件仅读取前9位，第10位被丢弃。 """ value = max(0, min(999_999_999, value_thousandths_nT)) # 提取10位十进制数字（高位在前） digits = [(value // (10 ** (9 - i))) % 10 for i in range(10)] bcd_bytes = bytes([d & 0x0F for d in digits]) return bytes([0x5A]) + bcd_bytes + bytes([0xA5]) class OpticMagSimulatorThread(QThread): """以指定频率向串口发送模拟光泵磁力仪 BCD 帧。 模拟量：基础值 + 慢漂移（正弦，周期60s，幅值±100 nT）+ 白噪声（±0.5 nT） """ error_occurred = pyqtSignal(str) value_updated = pyqtSignal(float) # 当前模拟值 (nT)，用于UI显示 def __init__(self, port, baudrate, base_nT=50000.0, rate_hz=10): super().__init__() self.port = port self.baudrate = baudrate self.base_nT = base_nT self.rate_hz = rate_hz self.is_running = False self.serial_port = None def run(self): self.is_running = True interval = 1.0 / self.rate_hz t0 = time.perf_counter() try: self.serial_port = serial.Serial(self.port, self.baudrate, timeout=1) while self.is_running: t = time.perf_counter() - t0 drift = 100.0 * np.sin(2 * np.pi * t / 60.0) noise = np.random.uniform(-0.5, 0.5) value_nT = self.base_nT + drift + noise value_thousandths = int(round(value_nT * 1000)) frame = _encode_optic_frame(value_thousandths) if self.serial_port and self.serial_port.is_open: self.serial_port.write(frame) self.value_updated.emit(value_nT) next_tick = t0 + (t // interval + 1) * interval sleep_time = next_tick - time.perf_counter() if sleep_time > 0: time.sleep(sleep_time) except Exception as e: if self.is_running: self.error_occurred.emit(str(e)) finally: self.stop() def stop(self): self.is_running = False if self.serial_port: try: self.serial_port.cancel_write() except Exception: pass try: if self.serial_port.is_open: self.serial_port.close() except Exception: pass finally: self.serial_port = None # ========================================== # 5. 后台线程：GPS 动态模拟输出 (10Hz) # ========================================== class GPSSimulatorThread(QThread): error_occurred = pyqtSignal(str) def __init__(self, port, baudrate): super().__init__() self.port = port self.baudrate = baudrate self.is_running = False self.serial_port = None def get_nmea_checksum(self, sentence): calc_cksum = 0 for char in sentence: calc_cksum ^= ord(char) return f"{calc_cksum:02X}" def run(self): self.is_running = True try: self.serial_port = serial.Serial(self.port, self.baudrate, timeout=1) current_lat = 39.9042 current_lon = 116.3972 current_alt = 43.0 lat_step = 0.00001 lon_step = 0.00001 alt_step = 0.05 while self.is_running: start_time = time.perf_counter() now = time.time() gm_time = time.gmtime(now) ms = int((now % 1) * 1000) time_str = time.strftime("%H%M%S", gm_time) + f".{ms:03d}" date_str = time.strftime("%d%m%y", gm_time) current_lat += lat_step current_lon += lon_step current_alt += alt_step lat_deg = int(abs(current_lat)) lat_min = (abs(current_lat) - lat_deg) * 60 lat_str = f"{lat_deg:02d}{lat_min:07.4f}" lat_dir = "N" if current_lat >= 0 else "S" lon_deg = int(abs(current_lon)) lon_min = (abs(current_lon) - lon_deg) * 60 lon_str = f"{lon_deg:03d}{lon_min:07.4f}" lon_dir = "E" if current_lon >= 0 else "W" speed_knots = "19.4" course_true = "45.0" gga_core = f"GPGGA,{time_str},{lat_str},{lat_dir},{lon_str},{lon_dir},1,08,1.0,{current_alt:.1f},M,0.0,M,," rmc_core = f"GPRMC,{time_str},A,{lat_str},{lat_dir},{lon_str},{lon_dir},{speed_knots},{course_true},{date_str},0.0,E" gga_sentence = f"${gga_core}*{self.get_nmea_checksum(gga_core)}\r\n" rmc_sentence = f"${rmc_core}*{self.get_nmea_checksum(rmc_core)}\r\n" if self.serial_port and self.serial_port.is_open: self.serial_port.write(gga_sentence.encode('ascii')) self.serial_port.write(rmc_sentence.encode('ascii')) elapsed = time.perf_counter() - start_time sleep_time = 0.1 - elapsed if sleep_time > 0: time.sleep(sleep_time) except Exception as e: if self.is_running: self.error_occurred.emit(str(e)) finally: self.stop() def stop(self): self.is_running = False if self.serial_port: try: self.serial_port.cancel_write() except Exception: pass try: if self.serial_port.is_open: self.serial_port.close() except Exception: pass finally: self.serial_port = None # ========================================== # 5. 主窗口 # ========================================== class DataAnalyzerUI(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("ATEM 数据分析工具") self.resize(1300, 900) self.df = pd.DataFrame() self.df_optic = pd.DataFrame() self.live_data_list = [] self.live_optic_list = [] self.is_live_mode = False self.curves = [] self.optic_curve = None self.last_fps_time = 0 self.frame_count = 0 self.last_packet_count = 0 self.current_packet_count = 0 self.view_offset = 0 self.is_sim_running = False self.sim_thread = None self.is_optic_sim_running = False self.optic_sim_thread = None pg.setConfigOptions(antialias=False) self.init_ui() self.plot_timer = QTimer() self.plot_timer.timeout.connect(self.update_live_plot) def init_ui(self): main_widget = QWidget() self.setCentralWidget(main_widget) layout = QVBoxLayout(main_widget) # --- 第一排工具栏：格式与模式选择 --- top_bar1 = QHBoxLayout() self.mode_group = QButtonGroup(self) self.rb_raw = QRadioButton("原始数据 (Raw)") self.rb_corr = QRadioButton("校正数据 (Corr)") self.rb_raw.setChecked(True) self.mode_group.addButton(self.rb_raw) self.mode_group.addButton(self.rb_corr) # 格式选项：GPS经纬度 / 光泵 self.chk_gps_pos = QCheckBox("含GPS经纬度 (lat/lon/alt)") self.chk_gps_pos.setChecked(True) self.chk_optic = QCheckBox("含光泵数据 (optical mag)") self.chk_optic.setChecked(False) self.chk_optic.stateChanged.connect(self._on_optic_toggled) btn_load = QPushButton("打开文件") btn_load.clicked.connect(self.load_file) btn_export = QPushButton("导出CSV") btn_export.clicked.connect(self.export_csv) self.chk_mouse_mode = QCheckBox("鼠标框选放大") self.chk_mouse_mode.stateChanged.connect(self.toggle_mouse_mode) btn_autoscale = QPushButton("复位视图") btn_autoscale.clicked.connect(self.reset_view) top_bar1.addWidget(QLabel("模式:")) top_bar1.addWidget(self.rb_raw) top_bar1.addWidget(self.rb_corr) top_bar1.addSpacing(15) top_bar1.addWidget(QLabel("格式:")) top_bar1.addWidget(self.chk_gps_pos) top_bar1.addWidget(self.chk_optic) top_bar1.addSpacing(15) top_bar1.addWidget(btn_load) top_bar1.addWidget(btn_export) top_bar1.addStretch() top_bar1.addWidget(self.chk_mouse_mode) top_bar1.addWidget(btn_autoscale) # --- 第二排工具栏：接收串口 --- top_bar2 = QHBoxLayout() self.cb_ports = QComboBox() self.cb_baudrate = QComboBox() self.cb_baudrate.addItems(["9600", "115200", "230400", "460800", "921600", "2000000"]) self.cb_baudrate.setCurrentText("2000000") btn_refresh_ports = QPushButton("刷新端口") btn_refresh_ports.clicked.connect(self.refresh_ports) self.btn_toggle_serial = QPushButton("打开接收串口") self.btn_toggle_serial.setStyleSheet("background-color: #4CAF50; color: white; font-weight: bold;") self.btn_toggle_serial.clicked.connect(self.toggle_serial) top_bar2.addWidget(QLabel("接收串口:")) top_bar2.addWidget(self.cb_ports) top_bar2.addWidget(btn_refresh_ports) top_bar2.addWidget(QLabel("波特率:")) top_bar2.addWidget(self.cb_baudrate) top_bar2.addWidget(self.btn_toggle_serial) top_bar2.addStretch() # --- 第三排工具栏：GPS 模拟器 --- top_bar3 = QHBoxLayout() self.cb_sim_ports = QComboBox() self.cb_sim_baudrate = QComboBox() self.cb_sim_baudrate.addItems(["9600", "115200", "230400", "460800", "921600"]) self.cb_sim_baudrate.setCurrentText("115200") self.btn_toggle_sim = QPushButton("开启GPS动态模拟 (10Hz)") self.btn_toggle_sim.setStyleSheet("background-color: #FF9800; color: white; font-weight: bold;") self.btn_toggle_sim.clicked.connect(self.toggle_gps_sim) top_bar3.addWidget(QLabel("输出串口:")) top_bar3.addWidget(self.cb_sim_ports) top_bar3.addWidget(QLabel("波特率:")) top_bar3.addWidget(self.cb_sim_baudrate) top_bar3.addWidget(self.btn_toggle_sim) top_bar3.addStretch() # --- 第四排工具栏：光泵模拟器 --- top_bar4 = QHBoxLayout() self.cb_optic_ports = QComboBox() self.cb_optic_baudrate = QComboBox() self.cb_optic_baudrate.addItems(["9600", "115200", "230400", "460800", "921600"]) self.cb_optic_baudrate.setCurrentText("115200") self.sb_optic_base = QComboBox() # 用ComboBox做快速基础值选择 self.sb_optic_base.setEditable(True) self.sb_optic_base.addItems(["50000", "55000", "60000", "65000"]) self.sb_optic_base.setCurrentText("50000") self.sb_optic_base.setFixedWidth(80) self.cb_optic_rate = QComboBox() self.cb_optic_rate.addItems(["1", "5", "10", "20", "50"]) self.cb_optic_rate.setCurrentText("10") self.cb_optic_rate.setFixedWidth(55) self.btn_toggle_optic_sim = QPushButton("开启光泵模拟") self.btn_toggle_optic_sim.setStyleSheet("background-color: #673AB7; color: white; font-weight: bold;") self.btn_toggle_optic_sim.clicked.connect(self.toggle_optic_sim) self.lbl_optic_sim_val = QLabel("-- nT") self.lbl_optic_sim_val.setStyleSheet("color: #673AB7; font-weight: bold; min-width: 120px;") top_bar4.addWidget(QLabel("光泵输出串口:")) top_bar4.addWidget(self.cb_optic_ports) top_bar4.addWidget(QLabel("波特率:")) top_bar4.addWidget(self.cb_optic_baudrate) top_bar4.addWidget(QLabel("基础值(nT):")) top_bar4.addWidget(self.sb_optic_base) top_bar4.addWidget(QLabel("频率(Hz):")) top_bar4.addWidget(self.cb_optic_rate) top_bar4.addWidget(self.btn_toggle_optic_sim) top_bar4.addWidget(self.lbl_optic_sim_val) top_bar4.addStretch() layout.addLayout(top_bar1) layout.addLayout(top_bar2) layout.addLayout(top_bar3) layout.addLayout(top_bar4) self.refresh_ports() # --- 监控栏 --- info_layout = QHBoxLayout() self.lbl_info = QLabel("请加载文件或打开串口接收数据...") self.lbl_info.setStyleSheet("color: blue; font-weight: bold;") self.lbl_fps = QLabel("绘图帧率: -- FPS | 接收率: -- 包/秒") self.lbl_fps.setStyleSheet("color: #E91E63; font-weight: bold;") info_layout.addWidget(self.lbl_info) info_layout.addStretch() info_layout.addWidget(self.lbl_fps) layout.addLayout(info_layout) # --- 视图窗口控制栏（仅文件模式有效）--- view_bar = QHBoxLayout() self.cb_page_size = QComboBox() self.cb_page_size.addItems(["1000", "5000", "10000", "50000", "全部"]) self.cb_page_size.setCurrentText("5000") self.cb_page_size.setFixedWidth(75) self.cb_page_size.currentTextChanged.connect(self._on_page_size_changed) self.btn_prev_page = QPushButton("◀ 上一页") self.btn_prev_page.setFixedWidth(80) self.btn_prev_page.clicked.connect(self._prev_page) self.btn_next_page = QPushButton("下一页 ▶") self.btn_next_page.setFixedWidth(80) self.btn_next_page.clicked.connect(self._next_page) self.sld_view = QSlider(Qt.Orientation.Horizontal) self.sld_view.setMinimum(0) self.sld_view.setMaximum(0) self.sld_view.setValue(0) self.sld_view.valueChanged.connect(self._on_slider_changed) self.lbl_view_range = QLabel("无数据") self.lbl_view_range.setMinimumWidth(220) view_bar.addWidget(QLabel("每页点数:")) view_bar.addWidget(self.cb_page_size) view_bar.addWidget(self.btn_prev_page) view_bar.addWidget(self.sld_view) view_bar.addWidget(self.btn_next_page) view_bar.addWidget(self.lbl_view_range) view_bar.addStretch() layout.addLayout(view_bar) # ========================================== # 多标签内容区域 # ========================================== self.tabs = QTabWidget() layout.addWidget(self.tabs) # [Tab 1] 主波形图（ADC / 校正值） self.plot_widget = pg.PlotWidget() self.plot_widget.setBackground('w') self.plot_widget.showGrid(x=True, y=True, alpha=0.3) self.plot_widget.addLegend() self.plot_widget.setLabel('bottom', 'Data Points (Index)') self.plot_widget.setLabel('left', 'Value') self.vb = self.plot_widget.plotItem.vb self.tabs.addTab(self.plot_widget, "波形图 (Plot)") # [Tab 2] 光泵磁场（仅含光泵时有意义） self.optic_plot_widget = pg.PlotWidget() self.optic_plot_widget.setBackground('w') self.optic_plot_widget.showGrid(x=True, y=True, alpha=0.3) self.optic_plot_widget.addLegend() self.optic_plot_widget.setLabel('bottom', 'Data Points (Index)') self.optic_plot_widget.setLabel('left', 'Magnetic Field (nT)') self.optic_curve = self.optic_plot_widget.plot( pen=pg.mkPen(color='#9C27B0', width=1.5), name='光泵磁场 (nT)') self.tabs.addTab(self.optic_plot_widget, "光泵磁场 (Optic Mag)") # [Tab 3] GPS 轨迹与海拔视图 traj_container = QWidget() traj_layout = QVBoxLayout(traj_container) traj_splitter = QSplitter(Qt.Orientation.Vertical) self.traj_plot = pg.PlotWidget(title="实时轨迹 (经度 vs 纬度)") self.traj_plot.setBackground('w') self.traj_plot.showGrid(x=True, y=True, alpha=0.5) self.traj_plot.setLabel('bottom', 'Longitude (经度)') self.traj_plot.setLabel('left', 'Latitude (纬度)') self.traj_curve = self.traj_plot.plot(pen=pg.mkPen('b', width=2), symbol='o', symbolSize=3, symbolBrush='b') self.alt_plot = pg.PlotWidget(title="海拔高度 (Altitude)") self.alt_plot.setBackground('w') self.alt_plot.showGrid(x=True, y=True, alpha=0.5) self.alt_plot.setLabel('bottom', 'Data Points (Index)') self.alt_plot.setLabel('left', 'Altitude (m)') self.alt_curve = self.alt_plot.plot(pen=pg.mkPen('g', width=2)) traj_splitter.addWidget(self.traj_plot) traj_splitter.addWidget(self.alt_plot) traj_layout.addWidget(traj_splitter) self.tabs.addTab(traj_container, "轨迹与海拔 (Trajectory)") # [Tab 4] 数据表格 self.table_view = QTableView() self.table_view.setAlternatingRowColors(True) self.table_view.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Interactive) self.tabs.addTab(self.table_view, "数据表 (Table)") self._update_optic_tab_visibility() def _on_optic_toggled(self): self._update_optic_tab_visibility() def _update_optic_tab_visibility(self): # Tab 1 是光泵标签页 tab_idx = 1 if self.chk_optic.isChecked(): self.tabs.setTabEnabled(tab_idx, True) self.tabs.setTabText(tab_idx, "光泵磁场 (Optic Mag)") else: self.tabs.setTabEnabled(tab_idx, False) self.tabs.setTabText(tab_idx, "光泵磁场 (未启用)") def _format_info(self): has_pos = self.chk_gps_pos.isChecked() has_optic = self.chk_optic.isChecked() mode = "Raw" if self.rb_raw.isChecked() else "Corr" parts = [mode] if has_pos: parts.append("含GPS经纬度") if has_optic: parts.append("含光泵") return " | ".join(parts) def get_current_dtype(self): return make_dtype( is_raw=self.rb_raw.isChecked(), has_gps_pos=self.chk_gps_pos.isChecked(), has_optic=self.chk_optic.isChecked() ) def get_start_marker(self): return MARKER_OPTIC if self.chk_optic.isChecked() else MARKER_STANDARD def get_column_config(self): is_raw = self.rb_raw.isChecked() has_gps_pos = self.chk_gps_pos.isChecked() has_optic = self.chk_optic.isChecked() if is_raw: data_cols = ['adc1', 'adc2', 'adc3'] labels = ['ADC 1', 'ADC 2', 'ADC 3'] else: data_cols = ['corr_x', 'corr_y', 'corr_z'] labels = ['X Axis', 'Y Axis', 'Z Axis'] cols = data_cols + ['gps_time'] if has_gps_pos: cols += ['gps_latitude', 'gps_longitude', 'gps_altitude'] if has_optic: cols.append('optical_mag') return cols, data_cols, labels def _get_page_size(self): txt = self.cb_page_size.currentText() if txt == "全部": return len(self.df) if not self.df.empty else 0 return int(txt) def _update_view_slider(self): total = len(self.df) page = self._get_page_size() max_offset = max(0, total - page) self.sld_view.blockSignals(True) self.sld_view.setMaximum(max_offset) self.sld_view.setSingleStep(max(1, page // 10)) self.sld_view.setPageStep(page) self.view_offset = min(self.view_offset, max_offset) self.sld_view.setValue(self.view_offset) self.sld_view.blockSignals(False) self._update_view_label(total, page) def _update_view_label(self, total, page): if total == 0: self.lbl_view_range.setText("无数据") return start = self.view_offset end = min(start + page, total) self.lbl_view_range.setText(f"显示 {start} – {end} / 共 {total} 点") def _on_slider_changed(self, value): self.view_offset = value total = len(self.df) page = self._get_page_size() self._update_view_label(total, page) if not self.df.empty and not self.is_live_mode: _, data_cols, labels = self.get_column_config() self.plot_static_data(data_cols, labels) def _on_page_size_changed(self): self.view_offset = 0 if not self.df.empty and not self.is_live_mode: self._update_view_slider() _, data_cols, labels = self.get_column_config() self.plot_static_data(data_cols, labels) def _prev_page(self): page = self._get_page_size() self.sld_view.setValue(max(0, self.view_offset - page)) def _next_page(self): page = self._get_page_size() self.sld_view.setValue(min(self.sld_view.maximum(), self.view_offset + page)) def load_file(self): if self.is_live_mode: QMessageBox.warning(self, "警告", "请先关闭串口后再加载文件。") return file_name, _ = QFileDialog.getOpenFileName(self, "选择文件", "", "Data (*.dat);;All (*)") if not file_name: return try: is_raw = self.rb_raw.isChecked() has_gps_pos = self.chk_gps_pos.isChecked() has_optic = self.chk_optic.isChecked() raw_bytes = open(file_name, 'rb').read() if len(raw_bytes) == 0: QMessageBox.warning(self, "提示", "文件为空。") return if has_optic: # 混流解析：文件中可能同时存在标准包和光泵包 self.df, self.df_optic = parse_mixed_stream(raw_bytes, is_raw, has_gps_pos) else: # 纯标准包，直接 frombuffer dtype = make_dtype(is_raw, has_gps_pos, False) arr = np.frombuffer(raw_bytes, dtype=dtype) self.df = pd.DataFrame(arr) self.df_optic = pd.DataFrame() if self.df.empty and self.df_optic.empty: QMessageBox.warning(self, "提示", "未能解析到有效数据包，请检查格式选项。") return cols, data_cols, labels = self.get_column_config() std_size = make_dtype(is_raw, has_gps_pos, False).itemsize optic_size = make_dtype(is_raw, has_gps_pos, True).itemsize self.lbl_info.setText( f"文件加载成功 | 标准包: {len(self.df)} 行 ({std_size}B)" + (f" | 光泵包: {len(self.df_optic)} 行 ({optic_size}B)" if has_optic else "") + f" | 格式: {self._format_info()}" ) self.lbl_fps.setText("绘图帧率: -- FPS | 接收率: -- 包/秒") self.view_offset = 0 self._update_view_slider() self.refresh_table(cols) self.plot_static_data(data_cols, labels) except Exception as e: QMessageBox.critical(self, "解析错误", str(e)) def refresh_table(self, cols): valid_cols = [c for c in cols if c in self.df.columns] display_df = self.df[valid_cols] if not self.df.empty else pd.DataFrame(columns=valid_cols) self.model = BigDataModel(display_df) self.table_view.setModel(self.model) self.table_view.resizeColumnsToContents() def plot_static_data(self, data_cols, labels): page = self._get_page_size() start = self.view_offset end = min(start + page, len(self.df)) if page > 0 else len(self.df) df_view = self.df.iloc[start:end] # 1. 主波形图 self.plot_widget.clear() self.curves.clear() colors = ['#FF0000', '#00AA00', '#0000FF'] x_data = np.arange(start, start + len(df_view)) for i, col in enumerate(data_cols): y_data = df_view[col].values curve = pg.PlotCurveItem(x=x_data, y=y_data, pen=pg.mkPen(color=colors[i], width=1.5), name=labels[i], skipFiniteCheck=True, autoDownsample=True, clipToView=True) self.plot_widget.addItem(curve) self.plot_widget.setLabel('bottom', 'Data Points (Index)') self.plot_widget.autoRange() # 2. 光泵磁场图（按比例切片 df_optic） if self.chk_optic.isChecked() and not self.df_optic.empty and 'optical_mag' in self.df_optic.columns: total_std = max(len(self.df), 1) total_optic = len(self.df_optic) o_start = int(start * total_optic / total_std) o_end = int(end * total_optic / total_std) optic_nT = self.df_optic['optical_mag'].values[o_start:o_end] / 1000.0 self.optic_curve.setData(x=np.arange(o_start, o_start + len(optic_nT)), y=optic_nT) self.optic_plot_widget.autoRange() else: self.optic_curve.setData([], []) # 3. 轨迹（显示全部）和海拔（随窗口切片） if self.chk_gps_pos.isChecked() and 'gps_longitude' in self.df.columns: lons_all = self.df['gps_longitude'].values lats_all = self.df['gps_latitude'].values valid_idx = (lons_all != 0.0) & (lats_all != 0.0) self.traj_curve.setData(lons_all[valid_idx], lats_all[valid_idx]) if np.any(valid_idx) else self.traj_curve.setData(lons_all, lats_all) self.traj_plot.autoRange() alts_view = df_view['gps_altitude'].values if 'gps_altitude' in df_view.columns else np.zeros(len(df_view)) self.alt_curve.setData(x=x_data, y=alts_view) self.alt_plot.autoRange() else: self.traj_curve.setData([], []) self.alt_curve.setData([]) def refresh_ports(self): self.cb_ports.clear() self.cb_sim_ports.clear() self.cb_optic_ports.clear() ports = serial.tools.list_ports.comports() for p in ports: port_name = f"{p.device} - {p.description}" self.cb_ports.addItem(port_name, p.device) self.cb_sim_ports.addItem(port_name, p.device) self.cb_optic_ports.addItem(port_name, p.device) def _set_format_controls_enabled(self, enabled: bool): self.rb_raw.setEnabled(enabled) self.rb_corr.setEnabled(enabled) self.chk_gps_pos.setEnabled(enabled) self.chk_optic.setEnabled(enabled) def toggle_serial(self): if not self.is_live_mode: port = self.cb_ports.currentData() if not port: QMessageBox.warning(self, "提示", "未找到有效串口") return baud = int(self.cb_baudrate.currentText()) is_raw = self.rb_raw.isChecked() has_gps_pos = self.chk_gps_pos.isChecked() has_optic = self.chk_optic.isChecked() dtype_std = make_dtype(is_raw, has_gps_pos, False) dtype_optic = make_dtype(is_raw, has_gps_pos, True) if has_optic else None self.df = pd.DataFrame() self.df_optic = pd.DataFrame() self.live_data_list = [] self.live_optic_list = [] self.plot_widget.clear() self.curves.clear() self.optic_curve.setData([], []) self.traj_curve.setData([], []) self.alt_curve.setData([]) self.plot_widget.setLabel('bottom', 'Latest Points (Index)') colors = ['#FF0000', '#00AA00', '#0000FF'] _, _, labels = self.get_column_config() for i in range(len(labels)): curve = pg.PlotCurveItem(pen=pg.mkPen(color=colors[i], width=1.5), name=labels[i]) self.plot_widget.addItem(curve) self.curves.append(curve) self.current_packet_count = 0 self.last_packet_count = 0 self.frame_count = 0 self.last_fps_time = time.perf_counter() self.lbl_fps.setText("绘图帧率: 计算中... | 接收率: 计算中...") self.serial_thread = SerialReaderThread(port, baud, dtype_std, dtype_optic) self.serial_thread.data_received.connect(self.on_live_data_received) self.serial_thread.optic_received.connect(self.on_live_optic_received) self.serial_thread.error_occurred.connect(self.on_serial_error) self.serial_thread.start() self.plot_timer.start(50) self.is_live_mode = True self.btn_toggle_serial.setText("关闭串口停止接收") self.btn_toggle_serial.setStyleSheet("background-color: #f44336; color: white; font-weight: bold;") self._set_format_controls_enabled(False) else: self.serial_thread.stop() self.serial_thread.wait() self.plot_timer.stop() self.is_live_mode = False self.btn_toggle_serial.setText("打开接收串口") self.btn_toggle_serial.setStyleSheet("background-color: #4CAF50; color: white; font-weight: bold;") self.lbl_fps.setText("绘图帧率: -- FPS | 接收率: -- 包/秒") self._set_format_controls_enabled(True) if self.live_data_list: self.df = pd.DataFrame(np.concatenate(self.live_data_list)) if self.live_optic_list: self.df_optic = pd.DataFrame(np.concatenate(self.live_optic_list)) cols, _, _ = self.get_column_config() self.refresh_table(cols) self.lbl_info.setText( f"串口采集完毕。标准包: {len(self.df)} 行" + (f" | 光泵包: {len(self.df_optic)} 行" if not self.df_optic.empty else "") ) def on_live_data_received(self, data_array): self.live_data_list.append(data_array) self.current_packet_count += len(data_array) self.lbl_info.setText(f"正在接收数据... 标准包: {self.current_packet_count} 帧 | 格式: {self._format_info()}") def on_live_optic_received(self, data_array): self.live_optic_list.append(data_array) def update_live_plot(self): self.frame_count += 1 current_time = time.perf_counter() elapsed = current_time - self.last_fps_time if elapsed >= 1.0: fps = self.frame_count / elapsed pps = (self.current_packet_count - self.last_packet_count) / elapsed self.lbl_fps.setText(f"绘图帧率: {fps:.1f} FPS | 接收率: {pps:.0f} 包/秒") self.last_fps_time = current_time self.frame_count = 0 self.last_packet_count = self.current_packet_count if not self.live_data_list: return MAX_POINTS = 5000 recent_chunks = [] point_count = 0 for arr in reversed(self.live_data_list): recent_chunks.append(arr) point_count += len(arr) if point_count >= MAX_POINTS: break recent_data = np.concatenate(recent_chunks[::-1]) if len(recent_data) > MAX_POINTS: recent_data = recent_data[-MAX_POINTS:] # 主波形图 _, data_cols, _ = self.get_column_config() for i, col in enumerate(data_cols): self.curves[i].setData(recent_data[col]) # 光泵磁场（实时，来自独立的 live_optic_list） if self.chk_optic.isChecked() and self.live_optic_list: MAX_OPTIC = 500 recent_optic = np.concatenate( list(reversed(self.live_optic_list))[:10][::-1])[-MAX_OPTIC:] self.optic_curve.setData(recent_optic['optical_mag'] / 1000.0) # 轨迹和海拔（实时） if self.chk_gps_pos.isChecked() and 'gps_longitude' in recent_data.dtype.names: lons = recent_data['gps_longitude'] lats = recent_data['gps_latitude'] alts = recent_data['gps_altitude'] if 'gps_altitude' in recent_data.dtype.names else np.zeros_like(lons) valid_idx = (lons != 0.0) & (lats != 0.0) self.traj_curve.setData(lons[valid_idx], lats[valid_idx]) if np.any(valid_idx) else self.traj_curve.setData(lons, lats) self.alt_curve.setData(alts) def on_serial_error(self, err_msg): self.toggle_serial() QMessageBox.critical(self, "接收串口错误", f"发生错误：\n{err_msg}") def toggle_gps_sim(self): if not self.is_sim_running: port = self.cb_sim_ports.currentData() if not port: QMessageBox.warning(self, "提示", "未找到有效的输出串口") return if self.is_live_mode and port == self.cb_ports.currentData(): reply = QMessageBox.question(self, "警告", "模拟输出端口与当前接收端口相同，可能会导致端口冲突。确定要继续吗？", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No) if reply == QMessageBox.StandardButton.No: return baud = int(self.cb_sim_baudrate.currentText()) self.sim_thread = GPSSimulatorThread(port, baud) self.sim_thread.error_occurred.connect(self.on_sim_error) self.sim_thread.start() self.is_sim_running = True self.btn_toggle_sim.setText("关闭GPS动态模拟") self.btn_toggle_sim.setStyleSheet("background-color: #f44336; color: white; font-weight: bold;") else: self.sim_thread.stop() self.sim_thread.wait() self.is_sim_running = False self.btn_toggle_sim.setText("开启GPS动态模拟 (10Hz)") self.btn_toggle_sim.setStyleSheet("background-color: #FF9800; color: white; font-weight: bold;") def on_sim_error(self, err_msg): self.toggle_gps_sim() QMessageBox.critical(self, "输出串口错误", f"模拟器串口发生错误：\n{err_msg}") def toggle_optic_sim(self): if not self.is_optic_sim_running: port = self.cb_optic_ports.currentData() if not port: QMessageBox.warning(self, "提示", "未找到有效的光泵输出串口") return try: base_nT = float(self.sb_optic_base.currentText()) except ValueError: QMessageBox.warning(self, "提示", "基础值请输入有效数字（单位 nT）") return rate_hz = int(self.cb_optic_rate.currentText()) baud = int(self.cb_optic_baudrate.currentText()) self.optic_sim_thread = OpticMagSimulatorThread(port, baud, base_nT, rate_hz) self.optic_sim_thread.error_occurred.connect(self.on_optic_sim_error) self.optic_sim_thread.value_updated.connect( lambda v: self.lbl_optic_sim_val.setText(f"{v:.3f} nT")) self.optic_sim_thread.start() self.is_optic_sim_running = True self.btn_toggle_optic_sim.setText("关闭光泵模拟") self.btn_toggle_optic_sim.setStyleSheet("background-color: #f44336; color: white; font-weight: bold;") self.sb_optic_base.setEnabled(False) self.cb_optic_rate.setEnabled(False) else: self.optic_sim_thread.stop() self.optic_sim_thread.wait() self.is_optic_sim_running = False self.btn_toggle_optic_sim.setText("开启光泵模拟") self.btn_toggle_optic_sim.setStyleSheet("background-color: #673AB7; color: white; font-weight: bold;") self.lbl_optic_sim_val.setText("-- nT") self.sb_optic_base.setEnabled(True) self.cb_optic_rate.setEnabled(True) def on_optic_sim_error(self, err_msg): self.toggle_optic_sim() QMessageBox.critical(self, "光泵串口错误", f"光泵模拟器串口发生错误：\n{err_msg}") def toggle_mouse_mode(self, state): mode = self.vb.RectMode if state == 2 else self.vb.PanMode self.vb.setMouseMode(mode) self.traj_plot.plotItem.vb.setMouseMode(mode) self.alt_plot.plotItem.vb.setMouseMode(mode) def reset_view(self): self.plot_widget.autoRange() self.optic_plot_widget.autoRange() self.traj_plot.autoRange() self.alt_plot.autoRange() def export_csv(self): if self.df.empty: return path, _ = QFileDialog.getSaveFileName(self, "保存", "export.csv", "CSV (*.csv)") if path: cols, _, _ = self.get_column_config() valid_cols = [c for c in cols if c in self.df.columns] self.df[valid_cols].to_csv(path, index=False) QMessageBox.information(self, "完成", "导出成功") if __name__ == "__main__": app = QApplication(sys.argv) w = DataAnalyzerUI() w.show() sys.exit(app.exec())