服务器上使用wandb老是断开连接_服务器使用Wandb离线模式配置_

1、测试组新增存在BUG，新增的测试组还是测试组1，而且删除后图标中原本的测试组1选项（实际只删除了新增的假测试组1，原本的测试组1还在），也没了。说明新增测试组的这里，序号有问题。2、报错，严重BUG，阻塞运行Traceback (most recent call last): File "e:\VS Python\Iperf3 GUI Tool_DS.py", line 1095, in start_test server_port = group_widget.server_port_spin.value() ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ AttributeError: 'QGroupBox' object has no attribute 'server_port_spin' 源代码： import sys import os import ctypes import json import subprocess import threading import time import psutil import pythoncom from datetime import datetime, timedelta from PyQt5.QtWidgets import ( QApplication, QMainWindow, QTabWidget, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QPushButton, QCheckBox, QComboBox, QSpinBox, QGroupBox, QScrollArea, QTextEdit, QFileDialog, QMessageBox, QDoubleSpinBox, QSizePolicy, QListWidget, QAbstractItemView, QListWidgetItem ) from PyQt5.QtCore import Qt, QThread, pyqtSignal, QTimer from PyQt5.QtGui import QColor, QIcon import wmi # Matplotlib 导入 import matplotlib matplotlib.use('Qt5Agg') import matplotlib.pyplot as plt from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas from matplotlib.figure import Figure import matplotlib.dates as mdates from matplotlib import ticker """# 检查管理员权限并请求提升 def is_admin(): try: return ctypes.windll.shell32.IsUserAnAdmin() except: return False if not is_admin(): # 请求管理员权限 ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, " ".join(sys.argv), None, 1) sys.exit(0)""" # 设置中文字体支持 plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'SimSun', 'KaiTi'] # 设置中文字体 plt.rcParams['axes.unicode_minus'] = False # 解决负号显示问题 # 网卡事件监听线程 class NicMonitorThread(QThread): nic_event = pyqtSignal(dict) # 发送事件：{'name': 网卡名, 'event': 'up' or 'down', 'time': 时间字符串} error_signal = pyqtSignal(str) # 错误信号 def init(self): super().init() self.running = True self.c = None self.watcher = None self.last_status = {} # 记录每个网卡的最后状态 def init_wmi(self): """初始化WMI连接""" try: pythoncom.CoInitialize() # 添加COM初始化 self.c = wmi.WMI() return True except Exception as e: self.error_signal.emit(f"无法初始化WMI连接: {str(e)}") return False def run(self): # 创建事件监听 if not self.init_wmi(): self.error_signal.emit("网卡事件监听器初始化失败") return try: # 获取当前网卡状态作为初始状态 for adapter in self.c.Win32_NetworkAdapter(NetConnectionStatus=2): self.last_status[adapter.Name] = 'up' for adapter in self.c.Win32_NetworkAdapter(NetConnectionStatus=7): self.last_status[adapter.Name] = 'down' # 创建事件监听器 self.watcher = self.c.watch_for( notification_type="__InstanceModificationEvent", wmi_class="Win32_NetworkAdapter", delay_secs=0.1, # 更短的延迟以提高响应速度 fields=["NetConnectionStatus", "Name"] ) while self.running: try: adapter = self.watcher() if adapter: name = adapter.Name status = adapter.NetConnectionStatus event_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3] # 只处理状态变化 if status == 2 and self.last_status.get(name) != 'up': self.nic_event.emit({'name': name, 'event': 'up', 'time': event_time}) self.last_status[name] = 'up' elif status == 7 and self.last_status.get(name) != 'down': self.nic_event.emit({'name': name, 'event': 'down', 'time': event_time}) self.last_status[name] = 'down' except wmi.x_wmi_timed_out: # 超时是正常的，继续循环 pass except Exception as e: self.error_signal.emit(f"监听网卡事件出错: {str(e)}") # 尝试重新初始化WMI连接 time.sleep(1) if not self.init_wmi(): break except Exception as e: self.error_signal.emit(f"创建网卡监听器出错: {str(e)}") finally: pythoncom.CoUninitialize() # 清理COM资源 def stop(self): self.running = False if self.watcher: self.watcher.cancel() # iperf3 进程管理器 class IperfManager: def init(self, iperf_path=''): self.active_servers = {} self.active_clients = {} self.iperf_path = iperf_path or 'iperf3' def start_server(self, group_id, port, server_ip, nic_name=""): """启动iperf3服务器""" # 使用服务器IP地址作为绑定的IP地址 if not server_ip or not self.validate_ip(server_ip): raise ValueError(f"测试组{group_id}服务器IP地址无效: {server_ip}") command = f'"{self.iperf_path}" -s -p {port} -B {server_ip} -J --logfile server_{group_id}_{port}.log' proc = subprocess.Popen(command, shell=True) self.active_servers[(group_id, port)] = (proc, nic_name) # 保存网卡名称用于日志 return command def start_client(self, group_id, server_ip, port, duration, streams, client_ip, reverse=False, omit=5, nic_name=""): """启动iperf3客户端""" # 使用客户端IP地址作为绑定的IP地址 if not client_ip or not self.validate_ip(client_ip): raise ValueError(f"测试组{group_id}客户端IP地址无效: {client_ip}") if not server_ip or not self.validate_ip(server_ip): raise ValueError(f"测试组{group_id}服务器IP地址无效: {server_ip}") direction = "-R" if reverse else "" # 添加-B参数绑定客户端IP，-c参数连接服务器IP command = f'"{self.iperf_path}" -c {server_ip} -p {port} -t {duration} -P {streams} {direction} -O {omit} -B {client_ip} -J --logfile client_{group_id}_{port}.json' proc = subprocess.Popen(command, shell=True) self.active_clients[(group_id, port)] = (proc, nic_name) # 保存网卡名称用于日志 return command def validate_ip(self, ip_str): """验证IP地址格式""" if not ip_str: return False parts = ip_str.split('.') if len(parts) != 4: return False for part in parts: if not part.isdigit(): return False num = int(part) if num < 0 or num > 255: return False return True def stop_all(self): """停止所有iperf进程""" for key, (proc, _) in list(self.active_servers.items()) + list(self.active_clients.items()): try: parent = psutil.Process(proc.pid) for child in parent.children(recursive=True): child.kill() parent.kill() except Exception as e: print(f"停止进程时出错: {e}") self.active_servers = {} self.active_clients = {} # 实时数据收集线程 class DataCollectorThread(QThread): data_updated = pyqtSignal(dict) # {group_id: {timestamp: (up, down)}} test_completed = pyqtSignal(int, dict) # group_id, full_data test_timeout = pyqtSignal() # 测试超时信号 def init(self, manager, groups, parent=None): super().init(parent) self.manager = manager self.groups = groups self.running = True self.data = {} self.full_data = {} self.start_time = time.time() self.max_duration = 0 # 记录最长测试时间（秒） self.timeout_timer = None # 计算最长测试时间（用于超时检测） for config in groups.values(): if config['duration'] > self.max_duration: self.max_duration = config['duration'] def run(self): # 初始化数据结构 for group_id in self.groups: self.data[group_id] = {"up": {}, "down": {}} self.full_data[group_id] = {"up": {}, "down": {}} # 设置超时计时器（最长测试时间+5秒） self.timeout_timer = threading.Timer(self.max_duration + 5, self.handle_timeout) self.timeout_timer.start() # 主采集循环 while self.running: for group_id, config in self.groups.items(): # 处理上传日志 upload_log = f"client_{group_id}_{config['upload_port']}.json" if os.path.exists(upload_log): try: with open(upload_log, 'r') as f: data = json.load(f) # 解析上传数据 if 'intervals' in data: for interval in data['intervals']: ts = time.time() up = interval['sum']['bits_per_second'] / 1e6 # 转换为Mbps # 存储实时数据 self.data[group_id]["up"][ts] = up self.full_data[group_id]["up"][ts] = up except Exception as e: print(f"读取上传日志文件错误: {e}") # 处理下载日志（如果是双向测试） if config['direction'] in ["下载", "双向"]: download_log = f"client_{group_id}_{config['download_port']}.json" if os.path.exists(download_log): try: with open(download_log, 'r') as f: data = json.load(f) # 解析下载数据 if 'intervals' in data: for interval in data['intervals']: ts = time.time() down = interval['sum']['bits_per_second'] / 1e6 # 存储实时数据 self.data[group_id]["down"][ts] = down self.full_data[group_id]["down"][ts] = down except Exception as e: print(f"读取下载日志文件错误: {e}") # 发送更新信号 self.data_updated.emit(self.data) time.sleep(1) # 每秒更新一次 # 测试完成后发送完整数据 for group_id in self.groups: self.test_completed.emit(group_id, self.full_data[group_id]) def handle_timeout(self): """处理测试超时""" self.running = False self.test_timeout.emit() def stop(self): self.running = False if self.timeout_timer: self.timeout_timer.cancel() # Matplotlib 图表组件（支持中文） class MplCanvas(FigureCanvas): def init(self, parent=None, width=5, height=4, dpi=100): self.fig = Figure(figsize=(width, height), dpi=dpi, facecolor=(0.94, 0.97, 1.0)) # 淡蓝色背景 self.ax = self.fig.add_subplot(111) super().init(self.fig) self.setParent(parent) self.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) self.updateGeometry() # 初始化图表（中文） self.ax.set_xlabel('时间 (秒)') self.ax.set_ylabel('吞吐量 (Mbps)') self.ax.set_title('iPerf3 流量监控') self.ax.grid(True) self.ax.set_facecolor((0.98, 0.99, 1.0)) # 更淡的蓝色背景 self.relative_time = True # 默认使用相对时间 self.start_time = None self.end_time = None self.start_timestamp = None def update_plot(self, data, selected_groups=None): """更新图表数据""" self.ax.clear() # 颜色列表 colors = [ '#4169E1', # RoyalBlue '#4682B4', # SteelBlue '#6495ED', # CornflowerBlue '#1E90FF', # DodgerBlue '#00BFFF' # DeepSkyBlue ] max_value = 0 min_timestamp = float('inf') max_timestamp = float('-inf') for group_id, group_data in data.items(): if selected_groups is not None and group_id not in selected_groups: continue color = colors[(group_id-1) % len(colors)] # 上传数据（实线） if group_data["up"]: timestamps = sorted(group_data["up"].keys()) values = [group_data["up"][ts] for ts in timestamps] # 更新时间范围 if timestamps: min_timestamp = min(min_timestamp, min(timestamps)) max_timestamp = max(max_timestamp, max(timestamps)) if not self.relative_time and self.start_time: # 转换为绝对时间 abs_timestamps = [self.start_time + timedelta(seconds=ts - self.start_timestamp) for ts in timestamps] self.ax.plot(abs_timestamps, values, '-', color=color, label=f"组{group_id} 上传", linewidth=2) else: # 相对时间（从0开始） rel_timestamps = [ts - min_timestamp for ts in timestamps] self.ax.plot(rel_timestamps, values, '-', color=color, label=f"组{group_id} 上传", linewidth=2) if values: max_value = max(max_value, max(values)) # 下载数据（虚线） if group_data["down"]: timestamps = sorted(group_data["down"].keys()) values = [group_data["down"][ts] for ts in timestamps] # 更新时间范围 if timestamps: min_timestamp = min(min_timestamp, min(timestamps)) max_timestamp = max(max_timestamp, max(timestamps)) if not self.relative_time and self.start_time: abs_timestamps = [self.start_time + timedelta(seconds=ts - self.start_timestamp) for ts in timestamps] self.ax.plot(abs_timestamps, values, '--', color=color, label=f"组{group_id} 下载", linewidth=2) else: rel_timestamps = [ts - min_timestamp for ts in timestamps] self.ax.plot(rel_timestamps, values, '--', color=color, label=f"组{group_id} 下载", linewidth=2) if values: max_value = max(max_value, max(values)) # 设置坐标轴范围 if max_value > 0: self.ax.set_ylim(0, max_value * 1.2) # 添加图例（中文） self.ax.legend(loc='best', fontsize=9) # 设置标签和标题（中文） if self.relative_time: self.ax.set_xlabel('时间 (秒)') else: self.ax.set_xlabel('时间') # 设置时间格式 self.ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S')) self.ax.xaxis.set_major_locator(ticker.AutoLocator()) self.ax.set_ylabel('吞吐量 (Mbps)') self.ax.set_title('iPerf3 流量监控') self.ax.grid(True) self.draw() def switch_to_absolute_time(self, start_time, end_time): """切换到绝对时间模式""" self.relative_time = False self.start_time = start_time self.end_time = end_time # 记录测试开始时的第一个时间戳（作为相对时间的零点） if self.start_time: self.start_timestamp = time.time() - (datetime.now() - start_time).total_seconds() # 主应用窗口 class IperfVisualizer(QMainWindow): def init(self): super().init() self.setWindowTitle("iPerf3 流量可视化工具") self.setWindowIcon(QIcon('iperf_icon.ico')) # 设置软件图标 self.resize(1400, 900) # 应用配置 self.config = { 'iperf_path': '', 'auto_save_log': True, 'log_path': os.getcwd(), 'test_groups': {}, 'current_group_id': 1 } # 初始化管理器 self.manager = None self.data_collector = None self.nic_monitor = None self.nic_events = [] # 存储网卡事件 self.nic_summary = {} # 网卡事件汇总: {网卡名: {'down': 次数, 'up': 次数, 'last_status': 最后状态}} self.nic_map = {} # 网卡到测试组映射: {网卡名: [(group_id, port, role)]} # 设置UI self.init_ui() self.load_config() # 启动网卡事件监听 self.start_nic_monitor() def init_ui(self): # 主布局 main_widget = QWidget() main_layout = QVBoxLayout() main_widget.setLayout(main_layout) self.setCentralWidget(main_widget) # 标签页 self.tabs = QTabWidget() main_layout.addWidget(self.tabs) # 配置页 self.config_tab = QWidget() self.tabs.addTab(self.config_tab, "测试配置") self.setup_config_tab() # 图表页 self.chart_tab = QWidget() self.tabs.addTab(self.chart_tab, "流量图表") self.setup_chart_tab() # 信息页 self.info_tab = QWidget() self.tabs.addTab(self.info_tab, "测试信息") self.setup_info_tab() # 状态栏 self.statusBar().showMessage("就绪") def setup_config_tab(self): layout = QVBoxLayout() self.config_tab.setLayout(layout) # iperf路径配置 iperf_group = QGroupBox("iperf3 配置") iperf_layout = QVBoxLayout() path_layout = QHBoxLayout() path_layout.addWidget(QLabel("iperf3路径:")) self.iperf_path = QLineEdit(self.config['iperf_path']) path_layout.addWidget(self.iperf_path) self.iperf_browse_btn = QPushButton("浏览...") self.iperf_browse_btn.clicked.connect(self.browse_iperf_path) path_layout.addWidget(self.iperf_browse_btn) iperf_layout.addLayout(path_layout) # 测试按钮 test_btn_layout = QHBoxLayout() self.test_iperf_btn = QPushButton("测试路径") self.test_iperf_btn.clicked.connect(self.test_iperf_path) test_btn_layout.addWidget(self.test_iperf_btn) iperf_layout.addLayout(test_btn_layout) iperf_group.setLayout(iperf_layout) layout.addWidget(iperf_group) # 测试描述 desc_group = QGroupBox("测试描述") desc_layout = QVBoxLayout() self.test_desc = QLineEdit("iPerf3流量测试") desc_layout.addWidget(QLabel("测试描述:")) desc_layout.addWidget(self.test_desc) desc_group.setLayout(desc_layout) layout.addWidget(desc_group) # 日志设置 log_group = QGroupBox("日志设置") log_layout = QVBoxLayout() self.auto_save = QCheckBox("自动保存测试日志") self.auto_save.setChecked(True) log_layout.addWidget(self.auto_save) path_layout = QHBoxLayout() path_layout.addWidget(QLabel("保存路径:")) self.log_path = QLineEdit(os.getcwd()) path_layout.addWidget(self.log_path) self.browse_btn = QPushButton("浏览...") self.browse_btn.clicked.connect(self.browse_log_path) path_layout.addWidget(self.browse_btn) log_layout.addLayout(path_layout) log_group.setLayout(log_layout) layout.addWidget(log_group) # 测试组配置 self.test_groups_area = QScrollArea() self.test_groups_area.setWidgetResizable(True) self.test_groups_widget = QWidget() self.test_groups_layout = QVBoxLayout() self.test_groups_widget.setLayout(self.test_groups_layout) self.test_groups_area.setWidget(self.test_groups_widget) layout.addWidget(QLabel("测试组配置:")) layout.addWidget(self.test_groups_area) # 添加测试组按钮 self.add_group_btn = QPushButton("添加测试组") self.add_group_btn.clicked.connect(self.add_test_group) layout.addWidget(self.add_group_btn) # 操作按钮 btn_layout = QHBoxLayout() self.start_btn = QPushButton("开始测试") self.start_btn.clicked.connect(self.start_test) btn_layout.addWidget(self.start_btn) self.stop_btn = QPushButton("停止测试") self.stop_btn.clicked.connect(self.stop_test) self.stop_btn.setEnabled(False) btn_layout.addWidget(self.stop_btn) layout.addLayout(btn_layout) # 添加初始测试组 self.add_test_group() def setup_chart_tab(self): layout = QVBoxLayout() self.chart_tab.setLayout(layout) # 使用 Matplotlib 图表（支持中文） self.canvas = MplCanvas(self, width=10, height=6, dpi=100) layout.addWidget(self.canvas, 3) # 图表占3/4空间 # 测试组选择面板 group_select_layout = QHBoxLayout() # 测试组列表 group_list_layout = QVBoxLayout() group_list_layout.addWidget(QLabel("选择显示的测试组:")) self.group_list = QListWidget() self.group_list.setSelectionMode(QAbstractItemView.MultiSelection) # 多选 group_list_layout.addWidget(self.group_list) # 选择按钮 btn_layout = QHBoxLayout() self.select_all_btn = QPushButton("全选") self.select_all_btn.clicked.connect(self.select_all_groups) btn_layout.addWidget(self.select_all_btn) self.deselect_all_btn = QPushButton("全不选") self.deselect_all_btn.clicked.connect(self.deselect_all_groups) btn_layout.addWidget(self.deselect_all_btn) group_list_layout.addLayout(btn_layout) group_select_layout.addLayout(group_list_layout, 1) # 占1/4宽度 # 图表操作按钮 chart_btn_layout = QVBoxLayout() chart_btn_layout.addWidget(QLabel("图表操作:")) self.export_btn = QPushButton("导出图表") self.export_btn.clicked.connect(self.export_chart) chart_btn_layout.addWidget(self.export_btn) self.clear_btn = QPushButton("清除图表") self.clear_btn.clicked.connect(self.clear_chart) chart_btn_layout.addWidget(self.clear_btn) self.zoom_reset_btn = QPushButton("重置缩放") self.zoom_reset_btn.clicked.connect(self.reset_zoom) chart_btn_layout.addWidget(self.zoom_reset_btn) group_select_layout.addLayout(chart_btn_layout, 1) # 占1/4宽度 layout.addLayout(group_select_layout, 1) # 控制面板占1/4高度 # 初始化后更新组列表 self.update_group_list() def setup_info_tab(self): layout = QVBoxLayout() self.info_tab.setLayout(layout) # 执行日志 exec_group = QGroupBox("执行日志") exec_layout = QVBoxLayout() self.exec_log = QTextEdit() self.exec_log.setReadOnly(True) exec_layout.addWidget(self.exec_log) btn_layout = QHBoxLayout() self.clear_log_btn = QPushButton("清除日志") self.clear_log_btn.clicked.connect(lambda: self.exec_log.clear()) btn_layout.addWidget(self.clear_log_btn) self.export_log_btn = QPushButton("导出日志") self.export_log_btn.clicked.connect(self.export_log) btn_layout.addWidget(self.export_log_btn) exec_layout.addLayout(btn_layout) exec_group.setLayout(exec_layout) layout.addWidget(exec_group) # 网卡状态 nic_group = QGroupBox("网卡状态监控") nic_layout = QVBoxLayout() self.nic_status = QTextEdit() self.nic_status.setReadOnly(True) nic_layout.addWidget(self.nic_status) # 网卡事件摘要 self.nic_summary_text = QTextEdit() self.nic_summary_text.setReadOnly(True) nic_layout.addWidget(QLabel("网卡事件摘要:")) nic_layout.addWidget(self.nic_summary_text) # 网卡映射 self.nic_mapping_text = QTextEdit() self.nic_mapping_text.setReadOnly(True) nic_layout.addWidget(QLabel("网卡测试组映射:")) nic_layout.addWidget(self.nic_mapping_text) nic_btn_layout = QHBoxLayout() self.refresh_nic_btn = QPushButton("刷新状态") self.refresh_nic_btn.clicked.connect(self.update_nic_status) nic_btn_layout.addWidget(self.refresh_nic_btn) self.export_nic_btn = QPushButton("导出状态") self.export_nic_btn.clicked.connect(self.export_nic_status) nic_btn_layout.addWidget(self.export_nic_btn) self.clear_nic_btn = QPushButton("清除事件") self.clear_nic_btn.clicked.connect(self.clear_nic_events) nic_btn_layout.addWidget(self.clear_nic_btn) nic_layout.addLayout(nic_btn_layout) nic_group.setLayout(nic_layout) layout.addWidget(nic_group) # 初始更新网卡状态 self.update_nic_status(initial=True) def browse_iperf_path(self): """浏览并选择iperf3可执行文件""" file_path, _ = QFileDialog.getOpenFileName( self, "选择iperf3可执行文件", "", "可执行文件 (.exe);;所有文件 ()" ) if file_path: self.iperf_path.setText(file_path) self.config['iperf_path'] = file_path self.save_config() def test_iperf_path(self): """测试iperf3路径是否有效""" path = self.iperf_path.text() if not path: QMessageBox.warning(self, "警告", "请先选择iperf3路径") return try: result = subprocess.run( [path, '-v'], capture_output=True, text=True, creationflags=subprocess.CREATE_NO_WINDOW ) if 'iperf 3.' in result.stdout: QMessageBox.information(self, "测试成功", f"iperf3版本信息:\n{result.stdout.splitlines()[0]}") self.config['iperf_path'] = path self.save_config() else: QMessageBox.critical(self, "错误", "选择的文件不是有效的iperf3可执行文件") except Exception as e: QMessageBox.critical(self, "错误", f"测试失败: {str(e)}") def start_nic_monitor(self): """启动网卡事件监听线程""" self.nic_monitor = NicMonitorThread() self.nic_monitor.nic_event.connect(self.handle_nic_event) self.nic_monitor.error_signal.connect(self.handle_nic_error) # 连接错误信号 self.nic_monitor.start() def handle_nic_error(self, error_msg): """处理网卡监听错误""" self.exec_log.append(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {error_msg}") # 如果无法使用事件监听，回退到轮询方式 if "无法初始化WMI连接" in error_msg or "创建网卡监听器出错" in error_msg: self.exec_log.append("将使用轮询方式监控网卡状态") self.start_polling_nic_status() def start_polling_nic_status(self): """启动轮询方式监控网卡状态""" self.last_nic_status = {} # 初始化状态记录 self.nic_poll_timer = QTimer() self.nic_poll_timer.timeout.connect(self.poll_nic_status) self.nic_poll_timer.start(1000) # 每秒轮询一次 def poll_nic_status(self): """轮询网卡状态""" try: c = wmi.WMI() # 获取所有网卡状态 for adapter in c.Win32_NetworkAdapter(): status = 'unknown' if adapter.NetConnectionStatus == 2: status = 'up' elif adapter.NetConnectionStatus == 7: status = 'down' # 只处理状态变化的网卡 if adapter.Name in self.last_nic_status and self.last_nic_status[adapter.Name] != status: event_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3] self.handle_nic_event({'name': adapter.Name, 'event': status, 'time': event_time}) # 记录当前状态 self.last_nic_status[adapter.Name] = status except Exception as e: self.exec_log.append(f"轮询网卡状态出错: {str(e)}") def handle_nic_event(self, event): """处理网卡事件""" name = event['name'] event_type = event['event'] event_time = event['time'] # 检查网卡映射 affected_groups = [] if name in self.nic_map: for group_id, port, role in self.nic_map[name]: affected_groups.append(f"测试组{group_id}的{role}端口{port}") # 记录事件 event_msg = f"[{event_time}] 网卡 '{name}' 状态变化: {event_type}" if affected_groups: event_msg += f" (影响: {', '.join(affected_groups)})" self.nic_events.append(event_msg) # 更新汇总 if name not in self.nic_summary: self.nic_summary[name] = {'down': 0, 'up': 0, 'last_status': 'unknown'} if event_type == 'down': self.nic_summary[name]['down'] += 1 elif event_type == 'up': self.nic_summary[name]['up'] += 1 self.nic_summary[name]['last_status'] = event_type # 更新网卡状态显示 self.update_nic_status() def add_test_group(self): # 计算当前可用最小ID existing_ids = set(self.config['test_groups'].keys()) group_id = 1 while group_id in existing_ids: group_id += 1 if group_id > 30: QMessageBox.warning(self, "警告", "最多只能添加30个测试组") return group = QGroupBox(f"测试组 {group_id}") group.setProperty('group_id', group_id) # 设置属性用于标识 group_layout = QVBoxLayout() # 组名 name_layout = QHBoxLayout() name_layout.addWidget(QLabel("组名:")) group_name = QLineEdit(f"测试组{group_id}") name_layout.addWidget(group_name) group_layout.addLayout(name_layout) # Server配置 server_group = QGroupBox("Server配置") server_layout = QVBoxLayout() ip_layout = QHBoxLayout() ip_layout.addWidget(QLabel("IP地址:")) server_ip = QLineEdit() server_ip.setPlaceholderText("必须填写服务器IP") ip_layout.addWidget(server_ip) server_layout.addLayout(ip_layout) port_layout = QHBoxLayout() port_layout.addWidget(QLabel("端口号:")) server_port = QSpinBox() server_port.setRange(1024, 65535) server_port.setValue(5201 + group_id - 1) port_layout.addWidget(server_port) server_layout.addLayout(port_layout) # 添加网卡选择 nic_layout = QHBoxLayout() nic_layout.addWidget(QLabel("绑定网卡:")) server_nic = QComboBox() server_nic.addItem("") # 空选项 # 获取系统网卡列表 try: c = wmi.WMI() for adapter in c.Win32_NetworkAdapter(NetConnectionStatus=2): # 2表示已连接 server_nic.addItem(adapter.Name) except Exception as e: print(f"获取网卡列表失败: {e}") nic_layout.addWidget(server_nic) server_layout.addLayout(nic_layout) server_group.setLayout(server_layout) group_layout.addWidget(server_group) # Client配置 client_group = QGroupBox("Client配置") client_layout = QVBoxLayout() c_ip_layout = QHBoxLayout() c_ip_layout.addWidget(QLabel("IP地址:")) client_ip = QLineEdit() client_ip.setPlaceholderText("必须填写客户端IP") c_ip_layout.addWidget(client_ip) client_layout.addLayout(c_ip_layout) c_name_layout = QHBoxLayout() c_name_layout.addWidget(QLabel("客户端名称:")) client_name = QLineEdit(f"客户端{group_id}") c_name_layout.addWidget(client_name) client_layout.addLayout(c_name_layout) # 添加网卡选择 c_nic_layout = QHBoxLayout() c_nic_layout.addWidget(QLabel("绑定网卡:")) client_nic = QComboBox() client_nic.addItem("") # 空选项 # 获取系统网卡列表 try: c = wmi.WMI() for adapter in c.Win32_NetworkAdapter(NetConnectionStatus=2): # 2表示已连接 client_nic.addItem(adapter.Name) except Exception as e: print(f"获取网卡列表失败: {e}") c_nic_layout.addWidget(client_nic) client_layout.addLayout(c_nic_layout) client_group.setLayout(client_layout) group_layout.addWidget(client_group) # 测试参数 param_group = QGroupBox("测试参数") param_layout = QVBoxLayout() direction_layout = QHBoxLayout() direction_layout.addWidget(QLabel("测试方向:")) test_direction = QComboBox() test_direction.addItems(["上传", "下载", "双向"]) test_direction.currentIndexChanged.connect(lambda idx, gid=group_id: self.update_port_visibility(gid, idx)) direction_layout.addWidget(test_direction) param_layout.addLayout(direction_layout) # 下载端口配置（默认隐藏） download_port_layout = QHBoxLayout() download_port_layout.addWidget(QLabel("下载端口号:")) download_port = QSpinBox() download_port.setRange(1024, 65535) download_port.setValue(server_port.value() + 1) download_port_layout.addWidget(download_port) # 初始时隐藏下载端口配置（除非是双向测试） if test_direction.currentIndex() != 2: download_port_layout.itemAt(0).widget().hide() download_port_layout.itemAt(1).widget().hide() param_layout.addLayout(download_port_layout) time_layout = QHBoxLayout() time_layout.addWidget(QLabel("测试时长:")) test_time = QDoubleSpinBox() test_time.setRange(1, 10000) test_time.setValue(10) time_layout.addWidget(test_time) time_unit = QComboBox() time_unit.addItems(["秒", "分钟", "小时"]) time_layout.addWidget(time_unit) param_layout.addLayout(time_layout) omit_layout = QHBoxLayout() omit_layout.addWidget(QLabel("跳过前N秒:")) omit_seconds = QSpinBox() omit_seconds.setRange(0, 60) omit_seconds.setValue(5) omit_layout.addWidget(omit_seconds) param_layout.addLayout(omit_layout) streams_layout = QHBoxLayout() streams_layout.addWidget(QLabel("并行流数量:")) streams = QSpinBox() streams.setRange(1, 100) streams.setValue(1) streams_layout.addWidget(streams) param_layout.addLayout(streams_layout) auto_restart = QCheckBox("每6小时自动重跑") auto_restart.setChecked(True) param_layout.addWidget(auto_restart) error_restart = QCheckBox("异常自动重跑") error_restart.setChecked(True) param_layout.addWidget(error_restart) restart_count_layout = QHBoxLayout() restart_count_layout.addWidget(QLabel("重跑次数:")) restart_count = QSpinBox() restart_count.setRange(1, 10) restart_count.setValue(3) restart_count_layout.addWidget(restart_count) param_layout.addLayout(restart_count_layout) param_group.setLayout(param_layout) group_layout.addWidget(param_group) # 删除按钮 delete_btn = QPushButton("删除此组") delete_btn.clicked.connect(lambda: self.remove_test_group(group)) group_layout.addWidget(delete_btn) group.setLayout(group_layout) self.test_groups_layout.addWidget(group) # 将控件保存为group的属性，以便后续读取 group.server_ip_edit = server_ip group.client_ip_edit = client_ip # 保存配置（只保存控件值） self.config['test_groups'][group_id] = { 'name': group_name.text(), 'server_ip': server_ip.text(), # 确保保存的是文本内容 'server_port': server_port.value(), 'server_nic': server_nic.currentText(), # 保存网卡名称 'download_port': download_port.value(), 'client_ip': client_ip.text(), # 确保保存的是文本内容 'client_name': client_name.text(), 'client_nic': client_nic.currentText(), # 保存网卡名称 'direction': test_direction.currentIndex(), 'test_time': test_time.value(), 'time_unit': time_unit.currentIndex(), 'omit': omit_seconds.value(), 'streams': streams.value(), 'auto_restart': auto_restart.isChecked(), 'error_restart': error_restart.isChecked(), 'restart_count': restart_count.value() } # 更新组选择列表 if hasattr(self, 'group_list'): self.update_group_list() def update_port_visibility(self, group_id, direction_idx): """根据测试方向显示/隐藏下载端口配置""" # 找到对应的group_widget for child in self.test_groups_widget.children(): if isinstance(child, QGroupBox) and child.property('group_id') == group_id: # 找到下载端口控件 download_port_label = None download_port_widget = None for layout in child.findChildren(QHBoxLayout): if layout.itemAt(0) and layout.itemAt(0).widget() and layout.itemAt(0).widget().text() == "下载端口号:": download_port_label = layout.itemAt(0).widget() download_port_widget = layout.itemAt(1).widget() break if download_port_label and download_port_widget: # 双向测试时显示下载端口配置 if direction_idx == 2: # 双向 download_port_label.show() download_port_widget.show() else: download_port_label.hide() download_port_widget.hide() break def remove_test_group(self, group_widget): group_id = group_widget.property('group_id') if group_id: # 从布局中移除 self.test_groups_layout.removeWidget(group_widget) group_widget.deleteLater() # 从配置中移除 if group_id in self.config['test_groups']: del self.config['test_groups'][group_id] self.save_config() # 更新组选择列表 if hasattr(self, 'group_list'): self.update_group_list() def update_group_list(self): """更新测试组选择列表""" if not hasattr(self, 'group_list'): return self.group_list.clear() for group_id in self.config['test_groups'].keys(): item = QListWidgetItem(f"测试组 {group_id}") item.setData(Qt.UserRole, group_id) item.setFlags(item.flags() | Qt.ItemIsUserCheckable) item.setCheckState(Qt.Checked) self.group_list.addItem(item) def select_all_groups(self): """选择所有测试组""" if not hasattr(self, 'group_list'): return for i in range(self.group_list.count()): item = self.group_list.item(i) item.setCheckState(Qt.Checked) self.update_chart() def deselect_all_groups(self): """取消选择所有测试组""" if not hasattr(self, 'group_list'): return for i in range(self.group_list.count()): item = self.group_list.item(i) item.setCheckState(Qt.Unchecked) self.update_chart() def browse_log_path(self): path = QFileDialog.getExistingDirectory(self, "选择日志保存路径") if path: self.log_path.setText(path) self.config['log_path'] = path self.save_config() def start_test(self): # 验证配置 if not self.config['iperf_path']: QMessageBox.critical(self, "错误", "请先配置iperf3路径") return # 创建管理器（使用配置的iperf路径） self.manager = IperfManager(self.config['iperf_path']) # 禁用UI控件 self.start_btn.setEnabled(False) self.add_group_btn.setEnabled(False) self.stop_btn.setEnabled(True) # 清除图表 self.clear_chart() # 记录开始时间 self.start_time = datetime.now() self.canvas.start_time = self.start_time self.canvas.start_timestamp = time.time() self.exec_log.append(f"[{self.start_time.strftime('%Y-%m-%d %H:%M:%S')}] 测试开始") # 收集测试组配置 - 直接从界面控件读取最新值 test_groups = {} self.nic_map = {} # 重置网卡映射 # 遍历所有测试组控件 for i in range(self.test_groups_layout.count()): group_widget = self.test_groups_layout.itemAt(i).widget() if not group_widget or not isinstance(group_widget, QGroupBox): continue group_id = group_widget.property('group_id') if not group_id: continue # 直接从控件读取值 server_ip = group_widget.server_ip_edit.text().strip() server_port = group_widget.server_port_spin.value() server_nic = group_widget.server_nic_combo.currentText() download_port = group_widget.download_port_spin.value() client_ip = group_widget.client_ip_edit.text().strip() client_name = group_widget.client_name_edit.text() client_nic = group_widget.client_nic_combo.currentText() test_direction_idx = group_widget.test_direction_combo.currentIndex() test_time_val = group_widget.test_time_spin.value() time_unit_idx = group_widget.time_unit_combo.currentIndex() omit = group_widget.omit_spin.value() streams = group_widget.streams_spin.value() # 计算测试时长（秒）- 转换为整数秒 if time_unit_idx == 1: # 分钟 duration = test_time_val * 60 elif time_unit_idx == 2: # 小时 duration = test_time_val * 3600 else: # 秒 duration = test_time_val duration = max(1, int(round(duration))) test_direction = ["上传", "下载", "双向"][test_direction_idx] # 调试信息 self.exec_log.append(f"测试组{group_id}配置: server_ip={server_ip}, client_ip={client_ip}") test_groups[group_id] = { 'server_ip': server_ip, 'server_port': server_port, 'client_ip': client_ip, 'direction': test_direction, 'duration': duration, # 使用整数秒 'streams': streams, 'omit': omit, 'upload_port': server_port, # 上传端口 'download_port': download_port # 下载端口 } # 记录网卡映射 if server_nic: if server_nic not in self.nic_map: self.nic_map[server_nic] = [] self.nic_map[server_nic].append((group_id, server_port, "Server")) if client_nic: if client_nic not in self.nic_map: self.nic_map[client_nic] = [] self.nic_map[client_nic].append((group_id, server_port, "Client")) # 启动服务器 for group_id, config in test_groups.items(): try: # 获取网卡名称 server_nic = "" for i in range(self.test_groups_layout.count()): group_widget = self.test_groups_layout.itemAt(i).widget() if group_widget.property('group_id') == group_id: server_nic = group_widget.server_nic_combo.currentText() break # 启动服务器 command = self.manager.start_server( group_id, config['server_port'], config['server_ip'], server_nic ) self.exec_log.append(f"[测试组{group_id}] 启动服务器: {command}") except ValueError as e: self.exec_log.append(f"[错误] 测试组{group_id}服务器启动失败: {str(e)}") QMessageBox.critical(self, "错误", f"测试组{group_id}服务器启动失败: {str(e)}") self.stop_test() return # 等待服务器启动 time.sleep(1) # 启动客户端 for group_id, config in test_groups.items(): try: # 获取网卡名称 client_nic = "" for i in range(self.test_groups_layout.count()): group_widget = self.test_groups_layout.itemAt(i).widget() if group_widget.property('group_id') == group_id: client_nic = group_widget.client_nic_combo.currentText() break # 启动客户端 reverse = config['direction'] in ["下载", "双向"] command = self.manager.start_client( group_id, config['server_ip'], config['server_port'], config['duration'], config['streams'], config['client_ip'], reverse, config['omit'], client_nic ) self.exec_log.append(f"[测试组{group_id}] 启动客户端: {command}") # 如果是双向测试，启动下载客户端 if config['direction'] == "双向": command = self.manager.start_client( group_id, config['server_ip'], config['download_port'], config['duration'], config['streams'], config['client_ip'], False, # 上传方向 config['omit'], client_nic ) self.exec_log.append(f"[测试组{group_id}] 启动下载客户端: {command}") except ValueError as e: self.exec_log.append(f"[错误] 测试组{group_id}客户端启动失败: {str(e)}") QMessageBox.critical(self, "错误", f"测试组{group_id}客户端启动失败: {str(e)}") self.stop_test() return # 启动数据收集线程 self.data_collector = DataCollectorThread(self.manager, test_groups) self.data_collector.data_updated.connect(self.update_chart) self.data_collector.test_completed.connect(self.test_finished) self.data_collector.test_timeout.connect(self.test_timeout_handler) self.data_collector.start() def stop_test(self): """停止所有测试""" if self.manager: self.manager.stop_all() if self.data_collector: self.data_collector.stop() self.exec_log.append(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 测试停止") # 启用UI控件 self.start_btn.setEnabled(True) self.add_group_btn.setEnabled(True) self.stop_btn.setEnabled(False) def test_finished(self, group_id, full_data): """测试完成处理""" self.exec_log.append(f"[测试组{group_id}] 测试完成") # 自动保存日志 if self.auto_save.isChecked(): self.save_log(group_id, full_data) def test_timeout_handler(self): """测试超时处理""" self.exec_log.append(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 测试超时") self.stop_test() def update_chart(self, data=None): """更新图表""" if not data: # 如果没有数据，显示空图表 self.canvas.ax.clear() self.canvas.ax.set_xlabel('时间 (秒)') self.canvas.ax.set_ylabel('吞吐量 (Mbps)') self.canvas.ax.set_title('iPerf3 流量监控') self.canvas.ax.grid(True) self.canvas.draw() return # 获取选中的测试组 selected_groups = set() for i in range(self.group_list.count()): item = self.group_list.item(i) if item.checkState() == Qt.Checked: selected_groups.add(item.data(Qt.UserRole)) # 更新图表 self.canvas.update_plot(data, selected_groups) def clear_chart(self): """清除图表数据""" self.canvas.ax.clear() self.canvas.ax.set_xlabel('时间 (秒)') self.canvas.ax.set_ylabel('吞吐量 (Mbps)') self.canvas.ax.set_title('iPerf3 流量监控') self.canvas.ax.grid(True) self.canvas.draw() def reset_zoom(self): """重置图表缩放""" self.canvas.ax.relim() self.canvas.ax.autoscale_view() self.canvas.draw() def export_chart(self): """导出图表为图片""" file_path, _ = QFileDialog.getSaveFileName( self, "保存图表", "", "PNG图片 (.png);;所有文件 ()" ) if file_path: self.canvas.fig.savefig(file_path, dpi=300) self.exec_log.append(f"图表已保存到: {file_path}") def export_log(self): """导出执行日志""" file_path, _ = QFileDialog.getSaveFileName( self, "保存日志", "", "文本文件 (.txt);;所有文件 ()" ) if file_path: with open(file_path, 'w', encoding='utf-8') as f: f.write(self.exec_log.toPlainText()) self.exec_log.append(f"日志已保存到: {file_path}") def update_nic_status(self, initial=False): """更新网卡状态显示""" # 网卡状态 self.nic_status.clear() try: c = wmi.WMI() for adapter in c.Win32_NetworkAdapter(): status = "未知" if adapter.NetConnectionStatus == 2: status = "已连接" elif adapter.NetConnectionStatus == 7: status = "已断开" self.nic_status.append(f"{adapter.Name}: {status}") except Exception as e: self.nic_status.append(f"获取网卡状态失败: {str(e)}") # 网卡事件摘要 self.nic_summary_text.clear() for nic, summary in self.nic_summary.items(): self.nic_summary_text.append(f"{nic}: 断开次数={summary['down']}, 连接次数={summary['up']}, 当前状态={summary['last_status']}") # 网卡映射 self.nic_mapping_text.clear() for nic, mappings in self.nic_map.items(): for group_id, port, role in mappings: self.nic_mapping_text.append(f"{nic} -> 测试组{group_id} {role}端口{port}") def export_nic_status(self): """导出网卡状态""" file_path, _ = QFileDialog.getSaveFileName( self, "保存网卡状态", "", "文本文件 (.txt);;所有文件 ()" ) if file_path: with open(file_path, 'w', encoding='utf-8') as f: f.write("=== 网卡状态 ===\n") f.write(self.nic_status.toPlainText() + "\n\n") f.write("=== 网卡事件摘要 ===\n") f.write(self.nic_summary_text.toPlainText() + "\n\n") f.write("=== 网卡测试组映射 ===\n") f.write(self.nic_mapping_text.toPlainText()) self.exec_log.append(f"网卡状态已保存到: {file_path}") def clear_nic_events(self): """清除网卡事件记录""" self.nic_events = [] self.nic_summary = {} self.update_nic_status() def save_log(self, group_id, full_data): """保存测试日志""" if not self.auto_save.isChecked(): return log_dir = self.log_path.text() if not os.path.exists(log_dir): os.makedirs(log_dir) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") file_path = os.path.join(log_dir, f"iperf_test_{group_id}_{timestamp}.json") with open(file_path, 'w', encoding='utf-8') as f: json.dump(full_data, f, indent=4) self.exec_log.append(f"[测试组{group_id}] 日志已保存到: {file_path}") def save_config(self): """保存配置到文件""" # 更新配置 self.config['iperf_path'] = self.iperf_path.text() self.config['log_path'] = self.log_path.text() # 保存到文件 with open('iperf_visualizer_config.json', 'w', encoding='utf-8') as f: json.dump(self.config, f, indent=4) def load_config(self): """从文件加载配置""" try: if os.path.exists('iperf_visualizer_config.json'): with open('iperf_visualizer_config.json', 'r', encoding='utf-8') as f: self.config = json.load(f) # 更新UI self.iperf_path.setText(self.config['iperf_path']) self.log_path.setText(self.config['log_path']) # 创建测试组 for group_id, group in self.config['test_groups'].items(): self.add_test_group_from_config(group_id, group) except Exception as e: print(f"加载配置失败: {str(e)}") def add_test_group_from_config(self, group_id, group_config): """从配置添加测试组""" group = QGroupBox(f"测试组 {group_id}") group.setProperty('group_id', group_id) group_layout = QVBoxLayout() # 组名 name_layout = QHBoxLayout() name_layout.addWidget(QLabel("组名:")) group_name = QLineEdit(group_config['name']) name_layout.addWidget(group_name) group_layout.addLayout(name_layout) # Server配置 server_group = QGroupBox("Server配置") server_layout = QVBoxLayout() ip_layout = QHBoxLayout() ip_layout.addWidget(QLabel("IP地址:")) server_ip = QLineEdit(group_config['server_ip']) server_ip.setPlaceholderText("必须填写服务器IP") ip_layout.addWidget(server_ip) server_layout.addLayout(ip_layout) port_layout = QHBoxLayout() port_layout.addWidget(QLabel("端口号:")) server_port = QSpinBox() server_port.setRange(1024, 65535) server_port.setValue(group_config['server_port']) port_layout.addWidget(server_port) server_layout.addLayout(port_layout) # 添加网卡选择 nic_layout = QHBoxLayout() nic_layout.addWidget(QLabel("绑定网卡:")) server_nic = QComboBox() server_nic.addItem("") # 空选项 # 获取系统网卡列表 try: c = wmi.WMI() for adapter in c.Win32_NetworkAdapter(NetConnectionStatus=2): # 2表示已连接 server_nic.addItem(adapter.Name) if adapter.Name == group_config['server_nic']: server_nic.setCurrentText(adapter.Name) except Exception as e: print(f"获取网卡列表失败: {e}") nic_layout.addWidget(server_nic) server_layout.addLayout(nic_layout) server_group.setLayout(server_layout) group_layout.addWidget(server_group) # Client配置 client_group = QGroupBox("Client配置") client_layout = QVBoxLayout() c_ip_layout = QHBoxLayout() c_ip_layout.addWidget(QLabel("IP地址:")) client_ip = QLineEdit(group_config['client_ip']) client_ip.setPlaceholderText("必须填写客户端IP") c_ip_layout.addWidget(client_ip) client_layout.addLayout(c_ip_layout) c_name_layout = QHBoxLayout() c_name_layout.addWidget(QLabel("客户端名称:")) client_name = QLineEdit(group_config['client_name']) c_name_layout.addWidget(client_name) client_layout.addLayout(c_name_layout) # 添加网卡选择 c_nic_layout = QHBoxLayout() c_nic_layout.addWidget(QLabel("绑定网卡:")) client_nic = QComboBox() client_nic.addItem("") # 空选项 # 获取系统网卡列表 try: c = wmi.WMI() for adapter in c.Win32_NetworkAdapter(NetConnectionStatus=2): # 2表示已连接 client_nic.addItem(adapter.Name) if adapter.Name == group_config['client_nic']: client_nic.setCurrentText(adapter.Name) except Exception as e: print(f"获取网卡列表失败: {e}") c_nic_layout.addWidget(client_nic) client_layout.addLayout(c_nic_layout) client_group.setLayout(client_layout) group_layout.addWidget(client_group) # 测试参数 param_group = QGroupBox("测试参数") param_layout = QVBoxLayout() direction_layout = QHBoxLayout() direction_layout.addWidget(QLabel("测试方向:")) test_direction = QComboBox() test_direction.addItems(["上传", "下载", "双向"]) test_direction.setCurrentIndex(group_config['direction']) test_direction.currentIndexChanged.connect(lambda idx, gid=group_id: self.update_port_visibility(gid, idx)) direction_layout.addWidget(test_direction) param_layout.addLayout(direction_layout) # 下载端口配置 download_port_layout = QHBoxLayout() download_port_layout.addWidget(QLabel("下载端口号:")) download_port = QSpinBox() download_port.setRange(1024, 65535) download_port.setValue(group_config['download_port']) download_port_layout.addWidget(download_port) # 初始时隐藏下载端口配置（除非是双向测试） if group_config['direction'] != 2: download_port_layout.itemAt(0).widget().hide() download_port_layout.itemAt(1).widget().hide() param_layout.addLayout(download_port_layout) time_layout = QHBoxLayout() time_layout.addWidget(QLabel("测试时长:")) test_time = QDoubleSpinBox() test_time.setRange(1, 10000) test_time.setValue(group_config['test_time']) time_layout.addWidget(test_time) time_unit = QComboBox() time_unit.addItems(["秒", "分钟", "小时"]) time_unit.setCurrentIndex(group_config['time_unit']) time_layout.addWidget(time_unit) param_layout.addLayout(time_layout) omit_layout = QHBoxLayout() omit_layout.addWidget(QLabel("跳过前N秒:")) omit_seconds = QSpinBox() omit_seconds.setRange(0, 60) omit_seconds.setValue(group_config['omit']) omit_layout.addWidget(omit_seconds) param_layout.addLayout(omit_layout) streams_layout = QHBoxLayout() streams_layout.addWidget(QLabel("并行流数量:")) streams = QSpinBox() streams.setRange(1, 100) streams.setValue(group_config['streams']) streams_layout.addWidget(streams) param_layout.addLayout(streams_layout) auto_restart = QCheckBox("每6小时自动重跑") auto_restart.setChecked(group_config['auto_restart']) param_layout.addWidget(auto_restart) error_restart = QCheckBox("异常自动重跑") error_restart.setChecked(group_config['error_restart']) param_layout.addWidget(error_restart) restart_count_layout = QHBoxLayout() restart_count_layout.addWidget(QLabel("重跑次数:")) restart_count = QSpinBox() restart_count.setRange(1, 10) restart_count.setValue(group_config['restart_count']) restart_count_layout.addWidget(restart_count) param_layout.addLayout(restart_count_layout) param_group.setLayout(param_layout) group_layout.addWidget(param_group) # 删除按钮 delete_btn = QPushButton("删除此组") delete_btn.clicked.connect(lambda: self.remove_test_group(group)) group_layout.addWidget(delete_btn) group.setLayout(group_layout) self.test_groups_layout.addWidget(group) # 将控件保存为group的属性，以便后续读取 group.server_ip_edit = server_ip group.client_ip_edit = client_ip def closeEvent(self, event): """关闭应用时保存配置""" self.save_config() if self.nic_monitor: self.nic_monitor.stop() event.accept() if name == "main": app = QApplication(sys.argv) window = IperfVisualizer() window.show() sys.exit(app.exec_())

这段代码没有正确考虑已删除的测试组ID，导致新添加的测试组总是使用最小的可用ID（通常是1）。此外，删除逻辑中未更新组列表控件。 ### 问题2：属性错误 **错误信息**： AttributeError: 'QGroupBox' object ...

raise get_mysql_exception( mysql.connector.errors.DataError: 1292 (22007): Incorrect datetime value: '025-06-20 03:17:35' for column 'timestamp' at row 1 2025-06-20 03:17:44,355 - SensorServer - ERROR - 获取数据库连接失败: 1292 (22007): Incorrect datetime value: '025-06-20 03:17:35' for column 'timestamp' at row 1 2025-06-20 03:17:44,562 - SensorServer - ERROR - 数据库写入失败: 1292 (22007): Incorrect datetime value: '025-06-20 03:17:35' for column 'timestamp' at row 1 2025-06-20 03:17:44,564 - SensorServer - ERROR - 异常详情: Traceback (most recent call last): File "C:\Program Files\JetBrains\Python\lib\site-packages\mysql\connector\connection_cext.py", line 772, in cmd_query self._cmysql.query( _mysql_connector.MySQLInterfaceError: Incorrect datetime value: '025-06-20 03:17:35' for column 'timestamp' at row 1

例如，我们假设时间戳的格式总是YYYY-MM-DDHH:MM:SS，但年份可能不足4位？那么我们可以用正则表达式提取各个部分。由于时间戳字符串的格式可能不固定，我们可以使用正则表达式来提取日期时间的各个部分。修改...

智能水务管理平台_基于物联网技术的供水系统远程监控与数据分析平台_通过NB-lot智能水表智能消火栓智能阀门及各类传感器实时采集管道压力流量水位浊度余氯泵频等数据结合GPRS.zip

智能水务管理平台_基于物联网技术的供水系统远程监控与数据分析平台_通过NB-lot智能水表智能消火栓智能阀门及各类传感器实时采集管道压力流量水位浊度余氯泵频等数据结合GPRS.zip

使用C语言编程实现统计文件内容，资源为源代码

c语言编程实现对指定文件中的内容进行统计。具体可实现：输入要进行统计的文件的路径及名称，统计出该文件中的字符、空格、数字及其他字符的个数，并将统计结果存到指定的文件中。

人脸遮挡检测，人脸遮挡，tkinter实现简易的人脸五官有无遮挡标注小工具，simple face occlusion label tools

打开下面链接，直接免费下载资源： https://renmaiwang.cn/s/2htte (最新版、最全版本)人脸遮挡检测，人脸遮挡，tkinter实现简易的人脸五官有无遮挡标注小工具，simple face occlusion label tools

大学生就业平台项目_基于SpringBoot和Vuejs的全栈式高校人才招聘与职业发展服务系统_提供企业岗位发布学生简历投递在线面试安排就业数据分析职业规划指导技能培训.zip

大学生就业平台项目_基于SpringBoot和Vuejs的全栈式高校人才招聘与职业发展服务系统_提供企业岗位发布学生简历投递在线面试安排就业数据分析职业规划指导技能培训.zip

二自由度悬架传递函数-多阻尼对比

在车辆工程中，悬架系统的性能评估和优化一直是研究的热点。悬架不仅关乎车辆的乘坐舒适性，还直接影响到车辆的操控性和稳定性。为了深入理解悬架的动态行为，研究人员经常使用“二自由度悬架模型”来简化分析，并运用“传递函数”这一数学工具来描述悬架系统的动态特性。二自由度悬架模型将复杂的车辆系统简化为两个独立的部分：车轮和车身。这种简化模型能够较准确地模拟出车辆在垂直方向上的运动行为，同时忽略了侧向和纵向的动态影响，这使得工程师能够更加专注于分析与优化与垂直动态相关的性能指标。传递函数作为控制系统理论中的一种工具，能够描述系统输入和输出之间的关系。在悬架系统中，传递函数特别重要，因为它能够反映出路面不平度如何被悬架系统转化为车内乘员感受到的振动。通过传递函数，我们可以得到一个频率域上的表达式，从中分析出悬架系统的关键动态特性，如系统的振幅衰减特性和共振频率等。在实际应用中，工程师通过使用MATLAB这类数学软件，建立双质量悬架的数学模型。模型中的参数包括车轮质量、车身质量、弹簧刚度以及阻尼系数等。通过编程求解，工程师可以得到悬架系统的传递函数，并据此绘制出传递函数曲线。这为评估悬架性能提供了一个直观的工具，使工程师能够了解悬架在不同频率激励下的响应情况。

人脸关键点检测技术与化妆应用结合探讨

打开下面链接，直接免费下载资源： https://renmaiwang.cn/s/n7d3y (最新版、最全版本)人脸关键点检测技术与化妆应用结合探讨

一个简单的多线程 TCP 客户端-服务器和客户端之间的通信基于简单的文本协议

c 一个简单的多线程 TCP 客户端——服务器和客户端之间的通信基于简单的文本协议；代码说明客户端创建套接字：使用 socket 创建一个 TCP 套接字。连接服务器：使用 connect 连接到服务器。发送和接收数据：使用 write 发送数据，使用 read 接收数据。关闭连接：客户端断开连接后，关闭客户端套接字。

Verilog 编程 - 基础知识

本课程将为您介绍Verilog，这是数字设计中使用的必备硬件描述语言。通过理论和实践的结合，您将学习Verilog编码的基础知识，包括逻辑门、数据类型和过程语句。完成本课程后，您将能够使用Verilog设计、模拟和实现基本的数字电路，为高级数字设计项目奠定坚实的基础。

微信小程序开发学习源代码项目_包含天气首页swiper视图联系人首字母索引数据请求网易云API外卖小程序结课设计天气预报游戏行情工具计时码表倒数日计算器计时器等_用于帮助初学者掌握.zip

微信小程序开发学习源代码项目_包含天气首页swiper视图联系人首字母索引数据请求网易云API外卖小程序结课设计天气预报游戏行情工具计时码表倒数日计算器计时器等_用于帮助初学者掌握.zip

fetch与axios的对比

fetch缺陷 1.fetch默认不会带cookie，需要添加配置项, 要用crendentials:“include”, 2.fetch不支持abort，不支持超时控制， axios好处 1.提供了一些并发请求的接口 function getUserAccount() { return axios.get('/user/12345'); } function getUserPermissions() { return axios.get('/user/12345/permissions'); }

superstream-java-0.0.1-alpha4-javadoc.jar

服务器上使用wandb老是断开连接

相关推荐

mySQL服务器连接,断开及cmd使用操作

解决阿里云服务器短时间连接自动断开问题

浪潮服务器RAID配置：专家解答20个常见问题

【连接池技术揭秘】：数据库连接效率提升的必杀技

权威分析：MySQL连接超时的真正原因

命令行使用指南：从基础到进阶

Unscrambler 11外部数据库集成：打造数据连接的无缝桥梁

【网络打印优化速成】：一步解决LPR连接和延迟问题

【专业破解MySQL连接难题】：彻底解决2013-lost connection故障

智能水务管理平台_基于物联网技术的供水系统远程监控与数据分析平台_通过NB-lot智能水表智能消火栓智能阀门及各类传感器实时采集管道压力流量水位浊度余氯泵频等数据结合GPRS.zip

使用C语言编程实现统计文件内容，资源为源代码

人脸遮挡检测，人脸遮挡，tkinter实现简易的人脸五官有无遮挡标注小工具，simple face occlusion label tools

大学生就业平台项目_基于SpringBoot和Vuejs的全栈式高校人才招聘与职业发展服务系统_提供企业岗位发布学生简历投递在线面试安排就业数据分析职业规划指导技能培训.zip

二自由度悬架传递函数-多阻尼对比

人脸关键点检测技术与化妆应用结合探讨

一个简单的多线程 TCP 客户端-服务器和客户端之间的通信基于简单的文本协议

Verilog 编程 - 基础知识

微信小程序开发学习源代码项目_包含天气首页swiper视图联系人首字母索引数据请求网易云API外卖小程序结课设计天气预报游戏行情工具计时码表倒数日计算器计时器等_用于帮助初学者掌握.zip

fetch与axios的对比

superstream-java-0.0.1-alpha4-javadoc.jar

大家在看

天津大学逻辑与形式化方法复习资料.rar

MqttAndroidClient

SAP-SD 三方贸易

WANOS 全景声介绍的一份详细资料

STM32F10x_StdPeriph_Lib_V3.5.0.zip

最新推荐

VScode连接远程服务器上的jupyter notebook的实现

连接MySQL服务器的两种方式：服务器不在本地而是在远程主机上怎么办？

C#如何连接服务器共享文件夹

Java远程连接Linux服务器并执行命令及上传文件功能

浅谈使用nodejs搭建web服务器的过程

Docker环境下的弹性APM服务器搭建指南

游戏开发与部署全流程指南

初级运维面试题

构建Ikiwiki的Docker容器：简易部署与使用

Unity开发实用指南：快捷键、外部工具与模型创建