import csv import os import sys import time from time import sleep from typing import Tuple, Optional import matplotlib import matplotlib.pyplot as plt from PyQt5.QtCore import QTimer from PyQt5.QtWidgets import QFrame, QStatusBar, QSizePolicy, QFileDialog, QApplication, QMainWindow, QWidget from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas, NavigationToolbar2QT from scipy.fft import fft, fftfreq from SamplingDialog import * from feauture_calculate import * from batch_dialog import BatchOperationDialog # 启用高DPI支持 QApplication.setAttribute(Qt.AA_EnableHighDpiScaling) QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps) # 强制使用 Qt5Agg 后端 matplotlib.use('Qt5Agg') plt.rcParams['font.sans-serif'] = ['SimHei'] # Windows 中文字体 plt.rcParams['axes.unicode_minus'] = False # 解决负号 '-' 显示问题 HEADER_MAGIC = bytes([0xAA, 0x55, 0xAA]) HEADER_SIZE = 6 # PackgeHead除去data字段的大小(3+1+1+1) class Eigenvalue: def __init__(self, temp1, temp2, offset1, offset2): self.temp1 = temp1 self.temp2 = temp2 self.offset1 = offset1 self.offset2 = offset2 class MatplotlibCanvas(FigureCanvas): """ 用于在 Qt 界面中嵌入 Matplotlib 绘图 """ def __init__(self, parent=None): self.fig, self.axs = plt.subplots(2, 2, figsize=(10, 5)) # 2 个子图（时域 + 频域） self.fig.subplots_adjust(hspace=0.6) # 增大时域图和频域图的间距 super().__init__(self.fig) self.setParent(parent) self.init_plot() def init_plot(self): """ 初始化默认图像（占位提示） """ self.axs[0, 0].set_title("加速度时域信号", fontsize=12, fontweight='bold') self.axs[0, 0].set_xlabel("采样点") self.axs[0, 0].set_ylabel("幅度(m/s2)") self.axs[0, 0].text(0.5, 0.5, "暂无数据", fontsize=12, ha='center', va='center', transform=self.axs[0, 0].transAxes) self.axs[0, 1].set_title("加速度频域信号", fontsize=12, fontweight='bold') self.axs[0, 1].set_xlabel("频率 (Hz)") self.axs[0, 1].set_ylabel("幅度(m/s2)") self.axs[0, 1].text(0.5, 0.5, "暂无数据", fontsize=12, ha='center', va='center', transform=self.axs[0, 1].transAxes) self.axs[1, 0].set_title("速度时域信号", fontsize=12, fontweight='bold') self.axs[1, 0].set_xlabel("采样点") self.axs[1, 0].set_ylabel("幅度(mm/s)") self.axs[1, 0].text(0.5, 0.5, "暂无数据", fontsize=12, ha='center', va='center', transform=self.axs[1, 0].transAxes) self.axs[1, 1].set_title("速度频域信号", fontsize=12, fontweight='bold') self.axs[1, 1].set_xlabel("频率 (Hz)") self.axs[1, 1].set_ylabel("幅度(mm/s)") self.axs[1, 1].text(0.5, 0.5, "暂无数据", fontsize=12, ha='center', va='center', transform=self.axs[1, 1].transAxes) self.draw() # 更新绘图 def plot_data(self, data, sample): """ 绘制时域和频域图 """ self.axs[0, 0].clear() self.axs[0, 1].clear() # 时域信号 self.axs[0, 0].plot(data, color='blue') self.axs[0, 0].set_title("加速度时域信号", fontsize=12, fontweight='bold') self.axs[0, 0].set_xlabel("采样点") self.axs[0, 0].set_ylabel("幅度(m/s2)") # 频域信号 N = len(data) T = 1.0 / sample yf = fft(data)[:N // 2] xf = fftfreq(N, T)[:N // 2] # self.axs[0,1].plot(xf, 2.0 / N * np.abs(yf[:N // 2]), color='red') # yf = yf[:4000] # xf = xf[:4000] self.axs[0, 1].plot(xf, 2.0 / N * np.abs(yf), lw=0.2, color='black') self.axs[0, 1].plot(xf, 2.0 / N * np.abs(yf), '.', lw=0.3, color='red') self.axs[0, 1].set_title("加速度频域信号", fontsize=12, fontweight='bold') self.axs[0, 1].set_xlabel("频率 (Hz)") self.axs[0, 1].set_ylabel("幅度(m/s2)") self.draw() # 更新绘图 def plot_data_vel(self, data, sample): """ 绘制时域和频域图 """ self.axs[1, 0].clear() self.axs[1, 1].clear() # 时域信号 self.axs[1, 0].plot(data, color='blue') self.axs[1, 0].set_title("速度时域信号", fontsize=12, fontweight='bold') self.axs[1, 0].set_xlabel("采样点") self.axs[1, 0].set_ylabel("幅度(mm/s)") # 频域信号 N = len(data) T = 1.0 / sample yf = fft(data)[:N // 2] xf = fftfreq(N, T)[:N // 2] # self.axs[1,1].plot(xf, 2.0 / N * np.abs(yf[:N // 2]), color='red') yf = yf[:1000] xf = xf[:1000] self.axs[1, 1].plot(xf, 2.0 / N * np.abs(yf), lw=0.2, color='black') self.axs[1, 1].plot(xf, 2.0 / N * np.abs(yf), '.', lw=0.3, color='red') self.axs[1, 1].set_title("速度频域信号", fontsize=12, fontweight='bold') self.axs[1, 1].set_xlabel("频率 (Hz)") self.axs[1, 1].set_ylabel("幅度(mm/s)") self.draw() # 更新绘图 class SocketClientApp(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("Socket Client & Data Plotter") self.setGeometry(100, 100, 900, 700) # 设置初始尺寸 self.rate_code = 4 self.time_str = '1' self.recv_state = '' self.current_cmd = None self.expected_length = 0 self.partial_data = QByteArray() self.scaled_data = '' self.sampling_rate = 0 self.buffer = QByteArray() self.reconnect_timer = QTimer(self) self.reconnect_timer.timeout.connect(self.attempt_reconnect) self.reconnect_interval = 5000 self.max_reconnect_attempts = 10 self.current_reconnect_attempts = 0 self.socket = QTcpSocket(self) self.socket.readyRead.connect(self.on_ready_read) self.socket.disconnected.connect(self.on_socket_disconnected) self.socket.errorOccurred.connect(self.on_socket_error) self.socket.connected.connect(self.on_socket_connected) # 组件初始化 self.ip_label = QLabel("IP 地址:") self.port_label = QLabel("端口号:") self.ip_input = QLineEdit(self) self.port_input = QLineEdit(self) self.connect_button = QPushButton("连接") self.get_data_button = QPushButton("获取数据1") self.get_data_button2 = QPushButton("获取数据2") self.get_temp_button = QPushButton("温度、偏置") self.upgrade_button = QPushButton("更新固件") self.get_mac_button = QPushButton("获取MAC地址") self.mac_config_button = QPushButton("mac配置") self.ipv4_config_button = QPushButton("网络配置") self.batch_process_button = QPushButton("批量操作") self.calibration_config_button = QPushButton("校准") self.get_version_button = QPushButton("版本") self.upgrade_button_sampling = QPushButton("采样率/时间") self.save_button_csv = QPushButton("存储到csv") self.get_data_button.setEnabled(False) # 初始状态不可点击 self.get_data_button2.setEnabled(False) # 初始状态不可点击 self.upgrade_button.setEnabled(False) # 初始状态不可点击 self.get_temp_button.setEnabled(False) self.upgrade_button_sampling.setEnabled(False) self.save_button_csv.setEnabled(False) self.mac_config_button.setEnabled(False) self.get_mac_button.setEnabled(False) self.ipv4_config_button.setEnabled(False) self.calibration_config_button.setEnabled(False) self.get_version_button.setEnabled(False) self.ip_input.setText("192.168.0.191") self.port_input.setText("12345") # 预留绘图区域 self.canvas = MatplotlibCanvas(self) self.toolbar = NavigationToolbar2QT(self.canvas, self) # 添加工具栏（放大、缩小、拖拽） # 预留特征值显示区域 self.feature_label = QLabel("特征值:") self.acceleration_label_rms = QLabel("加速度有效值: -") self.acceleration_label_pp = QLabel("加速度峰值: -") self.velocity_label_rms = QLabel("速度有效值: -") self.velocity_label_pp = QLabel("速度峰值: -") self.temp1_label = QLabel("温度1: -") self.temp2_label = QLabel("温度2：-") self.offset1_label = QLabel("偏置电压1：-") self.offset2_label = QLabel("偏置电压2：-") # 设置分割线 self.line = QFrame() self.line.setFrameShape(QFrame.HLine) self.line.setFrameShadow(QFrame.Sunken) # 布局管理 main_layout = QVBoxLayout() # 输入区域（IP & 端口） input_layout = QHBoxLayout() input_layout.addWidget(self.ip_label) input_layout.addWidget(self.ip_input) input_layout.addWidget(self.port_label) input_layout.addWidget(self.port_input) main_layout.addLayout(input_layout) # 按钮区域 button_layout = QHBoxLayout() button_layout.addWidget(self.connect_button) button_layout.addWidget(self.get_data_button) button_layout.addWidget(self.get_data_button2) button_layout.addWidget(self.get_temp_button) button_layout.addWidget(self.upgrade_button) button_layout.addWidget(self.upgrade_button_sampling) button_layout.addWidget(self.get_mac_button) button_layout.addWidget(self.mac_config_button) button_layout.addWidget(self.ipv4_config_button) button_layout.addWidget(self.batch_process_button) button_layout.addWidget(self.calibration_config_button) button_layout.addWidget(self.get_version_button) button_layout.addWidget(self.save_button_csv) main_layout.addLayout(button_layout) # 添加分割线 main_layout.addWidget(self.line) # 工具栏（放大、缩小、拖拽） main_layout.addWidget(self.toolbar) # 绘图 + 特征值区域 graph_layout = QVBoxLayout() graph_layout.addWidget(self.canvas) # 特征值显示区域 feature_layout = QGridLayout() feature_layout.setVerticalSpacing(10) feature_layout.addWidget(self.feature_label, 0, 0) feature_layout.addWidget(self.acceleration_label_rms, 0, 1) feature_layout.addWidget(self.acceleration_label_pp, 0, 2) feature_layout.addWidget(self.velocity_label_rms, 1, 1) feature_layout.addWidget(self.velocity_label_pp, 1, 2) feature_layout.addWidget(self.temp1_label, 2, 1) feature_layout.addWidget(self.temp2_label, 2, 2) feature_layout.addWidget(self.offset1_label, 3, 1) feature_layout.addWidget(self.offset2_label, 3, 2) graph_layout.addLayout(feature_layout) main_layout.addLayout(graph_layout) # 增大特征值区域与状态栏的间距 main_layout.addSpacing(20) # 创建状态栏 self.status_bar = QStatusBar() self.setStatusBar(self.status_bar) # 状态栏显示初始消息 self.status_bar.showMessage("状态: 请填写 IP 和端口号") # 主窗口的中央部件 central_widget = QWidget() central_widget.setLayout(main_layout) self.setCentralWidget(central_widget) # 事件绑定 self.connect_button.clicked.connect(self.connect_to_server) self.get_data_button.clicked.connect(self.on_button_clicked) self.get_data_button2.clicked.connect(self.on_button_clicked2) self.upgrade_button.clicked.connect(self.on_button_upgrade) self.get_temp_button.clicked.connect(self.on_button_temp) self.upgrade_button_sampling.clicked.connect(self.on_button_samping_set) self.save_button_csv.clicked.connect(self.on_button_save_csv) self.mac_config_button.clicked.connect(self.mac_config_dialog) self.get_mac_button.clicked.connect(self.on_button_get_mac) self.ipv4_config_button.clicked.connect(self.ipv4_config_dialog) self.calibration_config_button.clicked.connect(self.calibration_dialog) self.get_version_button.clicked.connect(self.on_button_get_version) self.batch_process_button.clicked.connect(self.on_button_batch) # 设置布局策略，确保控件大小随窗口调整 self.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) def on_button_batch(self): dialog = BatchOperationDialog() if dialog.exec_() == QDialog.Accepted: dialog.close() def get_extremes(self, data): """获取最大和最小的10个值（排序法）""" sorted_data = np.sort(data) return { 'top10_max': sorted_data[-10:][::-1].tolist(), # 降序排列 'top10_min': sorted_data[:10].tolist() } def mean_without_max_optimized(self, data): """优化内存使用的版本""" arr = np.array(data) if len(arr) < 2: return np.nan max_val = np.max(arr) # 计算总和和计数时直接排除最大值 total = np.sum(arr) - max_val count = len(arr) - 1 return total / count def mean_without_min_optimized(self, data): """优化内存使用的版本""" arr = np.array(data) if len(arr) < 2: return np.nan min_val = np.min(arr) # 计算总和和计数时直接排除最大值 total = np.sum(arr) - min_val count = len(arr) - 1 return total / count def calculate_crc(self, data: bytes) -> int: """计算数据的累加和CRC""" crc = 0 for byte in data: crc += byte return crc & 0xFF # 只保留最低字节 def connect_to_server(self): # self.process_wave_packet('') if self.connect_button.text() == "断开": self.connect_button.setText("连接") self.socket.disconnectFromHost() elif self.connect_button.text() == "连接": ip = self.ip_input.text() port = int(self.port_input.text()) self.socket.abort() self.socket.connectToHost(ip, port) def on_socket_connected(self): self.status_bar.showMessage("状态: 连接成功") self.get_data_button.setEnabled(True) self.get_data_button2.setEnabled(True) self.upgrade_button.setEnabled(True) self.get_temp_button.setEnabled(True) self.upgrade_button_sampling.setEnabled(True) self.save_button_csv.setEnabled(True) self.get_mac_button.setEnabled(True) self.mac_config_button.setEnabled(True) self.ipv4_config_button.setEnabled(True) self.calibration_config_button.setEnabled(True) self.get_version_button.setEnabled(True) self.connect_button.setText("断开") def on_socket_disconnected(self): print("on_socket_disconnected") self.connect_button.setText("连接") self.status_bar.showMessage("状态: 连接断开") self.get_data_button.setEnabled(False) self.get_data_button2.setEnabled(False) self.upgrade_button.setEnabled(False) self.get_temp_button.setEnabled(False) self.upgrade_button_sampling.setEnabled(False) self.save_button_csv.setEnabled(False) self.get_mac_button.setEnabled(False) self.mac_config_button.setEnabled(False) self.ipv4_config_button.setEnabled(False) self.calibration_config_button.setEnabled(False) self.get_version_button.setEnabled(False) # self.current_reconnect_attempts = 0 # self.reconnect_timer.start(self.reconnect_interval) def on_socket_error(self, error): self.status_bar.showMessage(f"状态: 错误 - {self.socket.errorString()}") def attempt_reconnect(self): if self.current_reconnect_attempts >= self.max_reconnect_attempts: self.reconnect_timer.stop() self.status_bar.showMessage("状态: 多次重连失败") return self.current_reconnect_attempts += 1 print(f"第 {self.current_reconnect_attempts} 次尝试重连...") self.socket.abort() self.connect_to_server() if self.socket.waitForConnected(1000): self.status_bar.showMessage("状态: 重新连接成功") self.get_data_button.setEnabled(True) self.reconnect_timer.stop() def on_button_save_csv(self): file_path = QFileDialog.getExistingDirectory( None, "选择保存目录", "", QFileDialog.ShowDirsOnly ) if not file_path: return file_path = os.path.join(file_path, "wave.csv") try: # 将 QByteArray 转换为字符串并写入文件 # 写入 CSV 文件 with open(file_path, 'w', newline='') as csvfile: writer = csv.writer(csvfile) for val in self.scaled_data: writer.writerow([val]) print(f"已保存 {len(self.scaled_data)} 个 float 到 {file_path}") QMessageBox.information( self, "保存成功", f"文件已成功保存到:\n{file_path}" ) except Exception as e: QMessageBox.critical( self, "保存失败", f"保存文件时出错:\n{str(e)}" ) def on_button_samping_set(self): dialog = SamplingDialog(self) if dialog.exec_() == QDialog.Accepted: self.rate_code, self.time_str = dialog.get_values() def on_ready_read(self): while self.socket.bytesAvailable(): if self.recv_state == 'WAIT_HEADER': self.start_time = time.time() print(f"开始时间戳: {self.start_time}") if self.socket.bytesAvailable() >= HEADER_SIZE: self.header = self.socket.read(HEADER_SIZE) if self.header[:3] != HEADER_MAGIC: print("无效 header magic，跳过") continue self.cmd = self.header[3] if self.cmd == 0x01: self.recv_state = 'WAIT_WAVE_HEADER' else: print(f"未知或暂不处理的 cmd: {self.cmd}") self.recv_state = 'WAIT_HEADER' return else: return elif self.recv_state == 'WAIT_WAVE_HEADER': if self.socket.bytesAvailable() >= 5: wave_header = self.socket.read(5) self.sampling_rate = struct.unpack('= self.expected_length: end_time = time.time() execution_time = end_time - self.start_time print(f"结束时间戳: {end_time}") print(f"代码执行时间: {round(execution_time, 3)} 秒") print(f"speed :{round((self.sampling_rate * 4 * 8) / execution_time, 3)} Kbps") self.process_wave_packet(bytes(self.partial_data)) self.recv_state = 'WAIT_HEADER' self.partial_data.clear() else: if self.socket.bytesAvailable() >= HEADER_SIZE: recv_data = self.socket.read(1340) if recv_data[:3] != HEADER_MAGIC: print("无效 header magic，跳过") continue cmd = recv_data[3] if cmd == 0x0C: body_format = '< H H H H' body_size = struct.calcsize(body_format) body_data = recv_data[HEADER_SIZE:HEADER_SIZE + body_size] unpacked_data = struct.unpack(body_format, body_data) value = Eigenvalue(*unpacked_data) print( f"温度1：{value.temp1 / 1000},温度2：{value.temp2 / 1000},偏置电压1：{value.offset1 / 100},偏置电压2：{value.offset2 / 100}") self.temp1_label.setText(f"温度1：{value.temp1 / 1000} v") self.temp2_label.setText(f"温度2：{value.temp2 / 1000} v") self.offset1_label.setText(f"偏置电压1：{value.offset1 / 100} v") self.offset2_label.setText(f"偏置电压2：{value.offset2 / 100} v") self.status_bar.showMessage("状态: 温度数据获取完成", 5000) elif cmd == 0x0F: body_format = '< B B' body_size = struct.calcsize(body_format) body_data = recv_data[HEADER_SIZE:HEADER_SIZE + body_size] unpacked_data = struct.unpack(body_format, body_data) print(f"version{unpacked_data}") QMessageBox.information(self, "版本", f"{unpacked_data[0]}.{unpacked_data[1]}") elif cmd == 0x12: body_format = '< B B' body_size = struct.calcsize(body_format) body_data = recv_data[HEADER_SIZE:HEADER_SIZE + body_size] unpacked_data = struct.unpack(body_format, body_data) print(f"version{unpacked_data}") QMessageBox.information(self, "MAC地址", f"50-29-4D-20-{format(unpacked_data[0], '02x')}-{format(unpacked_data[1], '02x')}") def receive_data(self, length: int): while len(self.buffer) < length: if not self.socket.waitForReadyRead(1000): raise ConnectionError("等待数据超时") data = self.buffer[:length] self.buffer = self.buffer[length:] return bytes(data) def parse_package_head(self, data: bytes) -> Tuple[Optional[dict], Optional[bytes]]: """ 解析包头部 返回: (header_dict, remaining_data) """ if len(data) < HEADER_SIZE: return None, None # 检查魔数 if data[:3] != HEADER_MAGIC: print(f"Invalid header magic: {data[:3].hex()}") return None, None # 解析头部 cmd, version, result_code = struct.unpack_from('