生成时间: {{ timestamp }}
+ +| 测试名称 | +带宽(Mbps) | +延迟(ms) | +丢包率(%) | +状态 | +
|---|---|---|---|---|
| {{ result.test_name }} | +{{ result.bandwidth | default('N/A') }} | +{{ result.latency | default('N/A') }} | +{{ result.loss | default('N/A') }} | +{{ result.status }} | +
{{ result.raw_output }}
+ {% endfor %}
+
+
+ """
+
+ # 使用模板引擎渲染
+ # ...
+
+ return html_content
+```
+
+#### 6.3.2 测试场景定义文件
+```yaml
+# tests.yaml
+tests:
+ # 基础连通性测试
+ basic_connectivity:
+ description: "基础连通性测试"
+ server: "192.168.1.1"
+ port: 5201
+ duration: 30
+ parallel: 1
+ expected:
+ bandwidth_min: 800 # Mbps
+ latency_max: 10 # ms
+ loss_max: 0.1 # %
+
+ # 压力测试
+ stress_test:
+ description: "多流压力测试"
+ server: "192.168.1.1"
+ duration: 300
+ parallel: 16
+ bandwidth: "1G"
+ expected:
+ bandwidth_min: 900
+
+ # 双向测试
+ bidirectional_test:
+ description: "双向带宽测试"
+ server: "192.168.1.1"
+ duration: 60
+ parallel: 8
+ bidirectional: true
+
+ # 不同协议测试
+ protocol_tests:
+ - name: "tcp_test"
+ protocol: "tcp"
+ duration: 30
+ - name: "udp_test"
+ protocol: "udp"
+ duration: 30
+ bandwidth: "100M"
+
+schedule:
+ daily: "02:00"
+ weekly: "sunday 03:00"
+ monthly: "1 04:00"
+
+reporting:
+ formats: ["html", "pdf", "email"]
+ recipients: ["admin@example.com"]
+ thresholds:
+ warning: 80%
+ critical: 90%
+```
+
+### 6.4 网络质量评分系统
+
+#### 6.4.1 评分算法设计
+```python
+class NetworkQualityScorer:
+ def __init__(self, weights=None):
+ # 默认权重
+ self.weights = weights or {
+ "bandwidth": 0.4, # 带宽权重 40%
+ "latency": 0.3, # 延迟权重 30%
+ "loss": 0.2, # 丢包权重 20%
+ "jitter": 0.1, # 抖动权重 10%
+ }
+
+ # 基准值（可根据网络类型调整）
+ self.baselines = {
+ "lan": {
+ "bandwidth": 1000, # 1Gbps
+ "latency": 1, # 1ms
+ "loss": 0, # 0%
+ "jitter": 0.1, # 0.1ms
+ },
+ "wan": {
+ "bandwidth": 100, # 100Mbps
+ "latency": 50, # 50ms
+ "loss": 0.1, # 0.1%
+ "jitter": 5, # 5ms
+ },
+ "internet": {
+ "bandwidth": 50, # 50Mbps
+ "latency": 100, # 100ms
+ "loss": 0.5, # 0.5%
+ "jitter": 10, # 10ms
+ }
+ }
+
+ def calculate_score(self, results, network_type="wan"):
+ """计算网络质量分数（0-100）"""
+ baseline = self.baselines[network_type]
+
+ scores = {}
+
+ # 带宽得分（越高越好）
+ bandwidth_score = min(
+ 100,
+ (results.get("bandwidth", 0) / baseline["bandwidth"]) * 100
+ )
+
+ # 延迟得分（越低越好，需要转换）
+ latency = results.get("latency", float('inf'))
+ if latency <= baseline["latency"]:
+ latency_score = 100
+ else:
+ # 延迟加倍，分数减半
+ latency_score = max(
+ 0,
+ 100 * (baseline["latency"] / latency)
+ )
+
+ # 丢包得分（0丢包得100分）
+ loss = results.get("loss", 100) # 默认100%丢包
+ if loss <= baseline["loss"]:
+ loss_score = 100
+ else:
+ loss_score = max(
+ 0,
+ 100 * (1 - loss / 100) # 转换为比例
+ )
+
+ # 抖动得分
+ jitter = results.get("jitter", float('inf'))
+ if jitter <= baseline["jitter"]:
+ jitter_score = 100
+ else:
+ jitter_score = max(
+ 0,
+ 100 * (baseline["jitter"] / jitter)
+ )
+
+ # 加权总分
+ total_score = (
+ bandwidth_score * self.weights["bandwidth"] +
+ latency_score * self.weights["latency"] +
+ loss_score * self.weights["loss"] +
+ jitter_score * self.weights["jitter"]
+ )
+
+ return {
+ "total": round(total_score, 1),
+ "components": {
+ "bandwidth": round(bandwidth_score, 1),
+ "latency": round(latency_score, 1),
+ "loss": round(loss_score, 1),
+ "jitter": round(jitter_score, 1),
+ },
+ "details": {
+ "bandwidth_actual": results.get("bandwidth"),
+ "latency_actual": results.get("latency"),
+ "loss_actual": results.get("loss"),
+ "jitter_actual": results.get("jitter"),
+ }
+ }
+
+ def get_quality_level(self, score):
+ """根据分数获取质量等级"""
+ if score >= 90:
+ return {"level": "优秀", "color": "green", "emoji": "✅"}
+ elif score >= 80:
+ return {"level": "良好", "color": "blue", "emoji": "👍"}
+ elif score >= 60:
+ return {"level": "一般", "color": "yellow", "emoji": "⚠️"}
+ elif score >= 40:
+ return {"level": "较差", "color": "orange", "emoji": "🔶"}
+ else:
+ return {"level": "差", "color": "red", "emoji": "❌"}
+
+ def generate_quality_report(self, test_results):
+ """生成质量报告"""
+ report = {
+ "timestamp": datetime.now().isoformat(),
+ "tests": [],
+ "summary": {
+ "average_score": 0,
+ "best_test": None,
+ "worst_test": None,
+ "recommendations": []
+ }
+ }
+
+ total_score = 0
+ best_score = -1
+ worst_score = 101
+
+ for test in test_results:
+ score_result = self.calculate_score(test["metrics"])
+ quality = self.get_quality_level(score_result["total"])
+
+ test_report = {
+ "name": test["name"],
+ "score": score_result["total"],
+ "quality": quality,
+ "details": score_result
+ }
+
+ report["tests"].append(test_report)
+
+ # 更新统计
+ total_score += score_result["total"]
+
+ if score_result["total"] > best_score:
+ best_score = score_result["total"]
+ report["summary"]["best_test"] = test["name"]
+
+ if score_result["total"] < worst_score:
+ worst_score = score_result["total"]
+ report["summary"]["worst_test"] = test["name"]
+
+ # 计算平均分
+ if report["tests"]:
+ report["summary"]["average_score"] = total_score / len(report["tests"])
+
+ # 生成建议
+ report["summary"]["recommendations"] = self.generate_recommendations(report)
+
+ return report
+
+ def generate_recommendations(self, report):
+ """根据测试结果生成改进建议"""
+ recommendations = []
+
+ for test in report["tests"]:
+ details = test["details"]["details"]
+ components = test["details"]["components"]
+
+ # 带宽相关建议
+ if components["bandwidth"] < 80:
+ actual = details.get("bandwidth_actual", 0)
+ expected = self.baselines["wan"]["bandwidth"]
+ recommendations.append(
+ f"测试 '{test['name']}' 带宽较低 ({actual}Mbps vs 预期{expected}Mbps)。"
+ "建议检查：1.网线质量 2.网络设备规格 3.是否有带宽限制"
+ )
+
+ # 延迟相关建议
+ if components["latency"] < 80:
+ actual = details.get("latency_actual", 0)
+ recommendations.append(
+ f"测试 '{test['name']}' 延迟较高 ({actual}ms)。"
+ "建议：1.优化路由 2.减少网络跳数 3.检查网络设备负载"
+ )
+
+ # 丢包相关建议
+ if components["loss"] < 80:
+ actual = details.get("loss_actual", 0)
+ recommendations.append(
+ f"测试 '{test['name']}' 丢包率较高 ({actual}%)。"
+ "建议：1.检查网络线路 2.优化MTU设置 3.检查网络设备错误计数"
+ )
+
+ # 去重
+ return list(set(recommendations))
+```
+
+#### 6.4.2 可视化质量报告
+```python
+def create_quality_dashboard(report):
+ """创建质量仪表板"""
+ import matplotlib.pyplot as plt
+ import pandas as pd
+
+ # 准备数据
+ tests = [t["name"] for t in report["tests"]]
+ scores = [t["score"] for t in report["tests"]]
+ colors = [t["quality"]["color"] for t in report["tests"]]
+
+ # 创建图表
+ fig, axes = plt.subplots(2, 2, figsize=(15, 10))
+
+ # 1. 总分柱状图
+ axes[0, 0].bar(tests, scores, color=colors)
+ axes[0, 0].set_title("各测试总分")
+ axes[0, 0].set_ylabel("分数 (0-100)")
+ axes[0, 0].set_ylim(0, 100)
+ axes[0, 0].tick_params(axis='x', rotation=45)
+
+ # 2. 各指标雷达图
+ categories = ['带宽', '延迟', '丢包', '抖动']
+ N = len(categories)
+
+ for i, test in enumerate(report["tests"]):
+ values = [
+ test["details"]["components"]["bandwidth"],
+ test["details"]["components"]["latency"],
+ test["details"]["components"]["loss"],
+ test["details"]["components"]["jitter"]
+ ]
+ values += values[:1] # 闭合雷达图
+
+ angles = [n / float(N) * 2 * 3.14159 for n in range(N)]
+ angles += angles[:1]
+
+ ax = axes[0, 1]
+ ax.plot(angles, values, label=test["name"])
+ ax.fill(angles, values, alpha=0.1)
+
+ axes[0, 1].set_title("各测试指标雷达图")
+ axes[0, 1].legend()
+
+ # 3. 时间趋势图（如果有历史数据）
+ # ...
+
+ # 4. 质量分布饼图
+ quality_counts = {}
+ for test in report["tests"]:
+ level = test["quality"]["level"]
+ quality_counts[level] = quality_counts.get(level, 0) + 1
+
+ axes[1, 0].pie(
+ quality_counts.values(),
+ labels=quality_counts.keys(),
+ autopct='%1.1f%%',
+ colors=['green', 'blue', 'yellow', 'orange', 'red'][:len(quality_counts)]
+ )
+ axes[1, 0].set_title("质量等级分布")
+
+ # 5. 详细数据表格
+ axes[1, 1].axis('tight')
+ axes[1, 1].axis('off')
+
+ table_data = []
+ for test in report["tests"]:
+ table_data.append([
+ test["name"],
+ f"{test['score']:.1f}",
+ test["quality"]["level"],
+ f"{test['details']['details'].get('bandwidth_actual', 'N/A')}",
+ f"{test['details']['details'].get('latency_actual', 'N/A')}ms",
+ f"{test['details']['details'].get('loss_actual', 'N/A')}%"
+ ])
+
+ table = axes[1, 1].table(
+ cellText=table_data,
+ colLabels=["测试", "分数", "等级", "带宽", "延迟", "丢包"],
+ loc='center',
+ cellLoc='center'
+ )
+ table.auto_set_font_size(False)
+ table.set_fontsize(9)
+ table.scale(1, 1.5)
+
+ plt.tight_layout()
+ plt.savefig("network_quality_report.png", dpi=300, bbox_inches='tight')
+ plt.show()
+
+ return fig
+```
+
+---
+
+## 第七章：安全考虑与最佳实践
+
+### 7.1 安全配置建议
+
+#### 7.1.1 服务器安全
+**不要将iperf服务器长期暴露在公网**，除非：
+1. 有认证机制
+2. 有访问控制
+3. 有日志监控
+4. 有带宽限制
+
+**推荐的安全配置**：
+```bash
+# 使用认证
+iperf3 -s --username admin --rsa-private-key-path /path/to/key
+
+# 限制带宽
+iperf3 -s --server-bitrate-limit 100M
+
+# 限制连接时间
+iperf3 -s --idle-timeout 300 --server-max-duration 3600
+
+# 使用非标准端口
+iperf3 -s -p 15201
+
+# 绑定特定IP
+iperf3 -s -B 192.168.1.100
+```
+
+#### 7.1.2 防火墙配置
+**必要的最小规则**：
+```
+# 只允许特定IP访问
+iptables -A INPUT -p tcp --dport 5201 -s 192.168.1.0/24 -j ACCEPT
+iptables -A INPUT -p tcp --dport 5201 -j DROP
+
+# 限制连接速率
+iptables -A INPUT -p tcp --dport 5201 -m limit --limit 10/min -j ACCEPT
+iptables -A INPUT -p tcp --dport 5201 -j DROP
+```
+
+### 7.2 性能测试最佳实践
+
+#### 7.2.1 测试环境准备
+**测试前检查清单**：
+```
+□ 1. 确认网络拓扑
+□ 2. 记录设备型号和固件版本
+□ 3. 检查线缆质量（CAT5e以上）
+□ 4. 确认端口协商状态（千兆/万兆）
+□ 5. 关闭节能模式（EEE）
+□ 6. 更新网卡驱动
+□ 7. 调整TCP参数（如果需要）
+□ 8. 关闭无关应用程序
+□ 9. 清理ARP缓存
+□ 10. 记录测试前计数器
+```
+
+#### 7.2.2 测试方法论
+**科学的测试步骤**：
+1. **基线测试**：
+ - 最低负载测试
+ - 单流测试
+ - 记录基准值
+
+2. **增量测试**：
+ - 逐步增加负载
+ - 观察性能变化
+ - 找到拐点
+
+3. **压力测试**：
+ - 长时间满负载
+ - 测试稳定性
+ - 观察错误率
+
+4. **恢复测试**：
+ - 停止压力后
+ - 测试恢复时间
+ - 验证弹性
+
+#### 7.2.3 结果验证
+**确保结果可信的方法**：
+1. **重复测试**：至少3次，取平均值
+2. **交叉验证**：用不同工具验证
+3. **双向验证**：测试两个方向
+4. **分段验证**：分段测试定位问题
+5. **时间验证**：不同时间测试
+
+### 7.3 避免常见陷阱
+
+#### 7.3.1 技术陷阱
+**避免这些常见错误**：
+
+1. **忽视CPU性能**：
+ - iperf3是单线程的
+ - 高速测试需要多核CPU
+ - 使用`-A`参数绑定CPU
+
+2. **忘记窗口缩放**：
+ - 高速高延迟需要大窗口
+ - 确保窗口缩放启用
+ - 调整系统参数
+
+3. **MTU不匹配**：
+ - 检查路径MTU
+ - 避免分片
+ - 适当调整MSS
+
+4. **缓冲区不足**：
+ - 增加socket缓冲区
+ - 调整系统参数
+ - 验证实际使用
+
+#### 7.3.2 操作陷阱
+**测试时的注意事项**：
+
+1. **不要在生产高峰测试**：
+ - 选择非业务时间
+ - 提前通知相关人员
+ - 做好回滚准备
+
+2. **注意带宽占用**：
+ - 从低带宽开始
+ - 逐步增加
+ - 监控影响
+
+3. **保存原始数据**：
+ - 保存iperf原始输出
+ - 记录测试环境
+ - 保存配置参数
+
+4. **考虑测试影响**：
+ - 对网络设备的影响
+ - 对其他应用的影响
+ - 对监控系统的影响
+
+---
+
+## 第八章：故障排除手册
+
+### 8.1 常见错误与解决方案
+
+#### 8.1.1 连接问题
+**错误：`connect failed: Connection refused`**
+
+**可能原因**：
+1. 服务器未运行
+2. 端口错误
+3. 防火墙阻挡
+
+**解决步骤**：
+```
+1. 检查服务器是否运行：ps aux | grep iperf3
+2. 检查端口是否正确：netstat -tlnp | grep 5201
+3. 检查防火墙：iptables -L -n
+4. 尝试telnet测试：telnet 服务器IP 端口
+```
+
+**错误：`connect failed: Connection timed out`**
+
+**可能原因**：
+1. 网络不通
+2. 路由问题
+3. 中间设备丢弃
+
+**解决步骤**：
+```
+1. ping测试连通性
+2. traceroute查看路径
+3. 检查中间设备ACL
+4. 检查MTU问题
+```
+
+#### 8.1.2 性能问题
+**问题：速度远低于预期**
+
+**排查步骤**：
+```
+步骤1：测试本地回环 → 验证工具
+步骤2：测试同交换机 → 验证本地网络
+步骤3：逐跳测试 → 定位问题段
+步骤4：检查设备计数器 → 查找错误
+```
+
+**常见原因**：
+1. **网卡协商问题**：
+ ```bash
+ # 检查协商状态
+ ethtool eth0
+
+ # 强制千兆
+ ethtool -s eth0 speed 1000 duplex full autoneg off
+ ```
+
+2. **TCP参数问题**：
+ ```bash
+ # 检查窗口大小
+ sysctl net.ipv4.tcp_window_scaling
+
+ # 检查缓冲区
+ sysctl net.core.rmem_max
+ sysctl net.core.wmem_max
+ ```
+
+3. **系统限制问题**：
+ ```bash
+ # 检查文件描述符限制
+ ulimit -n
+
+ # 检查CPU频率
+ cpupower frequency-info
+ ```
+
+#### 8.1.3 稳定性问题
+**问题：测试结果波动大**
+
+**可能原因**：
+1. 网络拥塞
+2. 无线干扰
+3. 设备负载
+4. 其他应用影响
+
+**诊断方法**：
+1. **时间分析**：
+ - 不同时间测试
+ - 寻找规律
+
+2. **对比分析**：
+ - 与其他链路对比
+ - 与历史数据对比
+
+3. **监控分析**：
+ - 实时监控网络流量
+ - 监控设备负载
+
+### 8.2 高级诊断技巧
+
+#### 8.2.1 使用tcpdump分析
+**捕获iperf流量分析**：
+```bash
+# 在服务器端捕获
+tcpdump -i eth0 -w iperf.pcap port 5201
+
+# 分析TCP序列号
+tcpdump -r iperf.pcap -n 'tcp port 5201' | head -20
+
+# 查看重传
+tcpdump -r iperf.pcap -n 'tcp port 5201 and tcp[tcpflags] & (tcp-syn|tcp-ack) == tcp-ack'
+```
+
+#### 8.2.2 使用系统工具分析
+**综合性能分析**：
+```bash
+# 实时监控
+nethogs eth0 # 查看每个连接带宽
+iftop -i eth0 # 查看实时流量
+iotop # 查看磁盘IO
+htop # 查看CPU/内存
+
+# 网络统计
+sar -n DEV 1 # 网络设备统计
+sar -n TCP 1 # TCP统计
+sar -n EDEV 1 # 错误统计
+```
+
+#### 8.2.3 使用专业工具分析
+**深入分析建议**：
+1. **Wireshark**：详细协议分析
+2. **perf**：系统性能分析
+3. **systemtap**：内核级分析
+4. **bpftrace**：现代性能分析
+
+### 8.3 应急处理方案
+
+#### 8.3.1 测试导致网络故障
+**立即停止测试**：
+```bash
+# 快速停止所有iperf进程
+pkill -9 iperf3
+
+# 清理连接
+ss -K dst 服务器IP dport = 5201
+```
+
+**恢复网络**：
+```bash
+# 刷新ARP
+ip neigh flush dev eth0
+
+# 清理conntrack
+conntrack -F
+
+# 重启网络服务
+systemctl restart network
+```
+
+#### 8.3.2 数据采集与保存
+**故障时收集信息**：
+```bash
+# 收集系统信息
+dmesg -T > dmesg.log
+journalctl -xe > journal.log
+
+# 收集网络信息
+ip addr > ip_addr.log
+ip route > ip_route.log
+ss -tulnp > ss.log
+netstat -s > netstat_s.log
+
+# 收集设备信息
+ethtool -S eth0 > ethtool_S.log
+ethtool -k eth0 > ethtool_k.log
+
+# 打包所有日志
+tar czf debug_$(date +%Y%m%d_%H%M%S).tar.gz *.log
+```
+
+---
+
+## 第九章：扩展与定制开发
+
+### 9.1 插件系统设计
+
+#### 9.1.1 插件架构
+```python
+# plugin_base.py
+import abc
+from typing import Dict, Any
+
+class IperfPlugin(abc.ABC):
+ """插件基类"""
+
+ def __init__(self, name, version):
+ self.name = name
+ self.version = version
+
+ @abc.abstractmethod
+ def before_test(self, config: Dict[str, Any]) -> Dict[str, Any]:
+ """测试前调用，可以修改配置"""
+ pass
+
+ @abc.abstractmethod
+ def during_test(self, output_line: str) -> None:
+ """测试过程中每行输出调用"""
+ pass
+
+ @abc.abstractmethod
+ def after_test(self, results: Dict[str, Any]) -> Dict[str, Any]:
+ """测试后调用，可以处理结果"""
+ pass
+
+ @abc.abstractmethod
+ def get_ui_widget(self):
+ """返回插件的UI部件"""
+ pass
+
+# plugin_manager.py
+class PluginManager:
+ def __init__(self):
+ self.plugins = {}
+ self.load_plugins()
+
+ def load_plugins(self):
+ """动态加载插件"""
+ import importlib
+ import pkgutil
+
+ plugin_packages = ['plugins']
+
+ for package_name in plugin_packages:
+ try:
+ package = importlib.import_module(package_name)
+ for _, module_name, _ in pkgutil.iter_modules(package.__path__):
+ module = importlib.import_module(f"{package_name}.{module_name}")
+
+ for attr_name in dir(module):
+ attr = getattr(module, attr_name)
+
+ if (isinstance(attr, type) and
+ issubclass(attr, IperfPlugin) and
+ attr != IperfPlugin):
+
+ plugin_instance = attr()
+ self.plugins[plugin_instance.name] = plugin_instance
+
+ except ImportError:
+ continue
+
+ def run_before_hooks(self, config):
+ """运行所有before_test钩子"""
+ for plugin in self.plugins.values():
+ config = plugin.before_test(config)
+ return config
+
+ def run_during_hooks(self, output_line):
+ """运行所有during_test钩子"""
+ for plugin in self.plugins.values():
+ plugin.during_test(output_line)
+
+ def run_after_hooks(self, results):
+ """运行所有after_test钩子"""
+ for plugin in self.plugins.values():
+ results = plugin.after_test(results)
+ return results
+```
+
+#### 9.1.2 示例插件
+```python
+# plugins/bandwidth_monitor.py
+import time
+from typing import Dict, Any
+from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel, QProgressBar
+
+class BandwidthMonitorPlugin(IperfPlugin):
+ """带宽监控插件"""
+
+ def __init__(self):
+ super().__init__("带宽监控", "1.0")
+ self.start_time = None
+ self.total_bytes = 0
+ self.max_bandwidth = 0
+ self.ui_widget = None
+
+ def before_test(self, config: Dict[str, Any]) -> Dict[str, Any]:
+ self.start_time = time.time()
+ self.total_bytes = 0
+ self.max_bandwidth = 0
+ return config
+
+ def during_test(self, output_line: str) -> None:
+ # 解析输出行，提取带宽信息
+ if "Mbits/sec" in output_line or "Gbits/sec" in output_line:
+ # 这里添加解析逻辑
+ pass
+
+ def after_test(self, results: Dict[str, Any]) -> Dict[str, Any]:
+ if self.start_time:
+ duration = time.time() - self.start_time
+ avg_bandwidth = (self.total_bytes * 8) / duration / 1e6 # Mbps
+
+ results["plugin_stats"] = {
+ "total_bytes": self.total_bytes,
+ "average_bandwidth_mbps": avg_bandwidth,
+ "max_bandwidth_mbps": self.max_bandwidth,
+ "duration": duration
+ }
+
+ return results
+
+ def get_ui_widget(self):
+ if not self.ui_widget:
+ self.ui_widget = QWidget()
+ layout = QVBoxLayout()
+
+ self.current_label = QLabel("当前带宽: --")
+ self.average_label = QLabel("平均带宽: --")
+ self.max_label = QLabel("最大带宽: --")
+ self.progress_bar = QProgressBar()
+
+ layout.addWidget(self.current_label)
+ layout.addWidget(self.average_label)
+ layout.addWidget(self.max_label)
+ layout.addWidget(self.progress_bar)
+
+ self.ui_widget.setLayout(layout)
+
+ return self.ui_widget
+```
+
+### 9.2 API接口设计
+
+#### 9.2.1 REST API服务
+```python
+# api_server.py
+from flask import Flask, request, jsonify
+import threading
+import time
+
+app = Flask(__name__)
+
+class APIServer:
+ def __init__(self, iperf_gui):
+ self.gui = iperf_gui
+ self.test_thread = None
+ self.current_test = None
+
+ def run(self, host='0.0.0.0', port=5000):
+ """启动API服务器"""
+ app.run(host=host, port=port, threaded=True)
+
+ @app.route('/api/test/start', methods=['POST'])
+ def start_test():
+ """启动测试"""
+ data = request.json
+
+ if self.current_test and self.current_test.is_alive():
+ return jsonify({"error": "测试正在进行中"}), 409
+
+ # 准备测试参数
+ params = self.prepare_params(data)
+
+ # 在新线程中运行测试
+ self.test_thread = threading.Thread(
+ target=self.run_test_in_thread,
+ args=(params,)
+ )
+ self.test_thread.start()
+
+ return jsonify({
+ "status": "started",
+ "test_id": str(int(time.time()))
+ })
+
+ @app.route('/api/test/stop', methods=['POST'])
+ def stop_test():
+ """停止测试"""
+ if self.current_test:
+ self.current_test.stop()
+ return jsonify({"status": "stopped"})
+ else:
+ return jsonify({"error": "没有正在运行的测试"}), 404
+
+ @app.route('/api/test/status', methods=['GET'])
+ def get_status():
+ """获取测试状态"""
+ if self.current_test and self.current_test.is_alive():
+ return jsonify({
+ "status": "running",
+ "progress": self.current_test.get_progress(),
+ "results": self.current_test.get_partial_results()
+ })
+ else:
+ return jsonify({"status": "stopped"})
+
+ @app.route('/api/config', methods=['GET', 'POST'])
+ def config():
+ """配置管理"""
+ if request.method == 'GET':
+ return jsonify(self.gui.get_current_config())
+ else:
+ config = request.json
+ self.gui.apply_config(config)
+ return jsonify({"status": "applied"})
+
+ def run_test_in_thread(self, params):
+ """在线程中运行测试"""
+ self.current_test = self.gui.start_test_with_params(params)
+
+ while self.current_test.is_alive():
+ time.sleep(0.1)
+
+ self.current_test = None
+```
+
+#### 9.2.2 WebSocket实时数据
+```python
+# websocket_server.py
+from flask_socketio import SocketIO, emit
+import json
+
+socketio = SocketIO(app, cors_allowed_origins="*")
+
+@socketio.on('connect')
+def handle_connect():
+ print('客户端连接')
+ emit('connected', {'data': '连接成功'})
+
+@socketio.on('start_test')
+def handle_start_test(data):
+ """通过WebSocket启动测试"""
+ test_id = start_test_async(data)
+ emit('test_started', {'test_id': test_id})
+
+ # 实时推送结果
+ def send_updates():
+ while test_is_running(test_id):
+ results = get_latest_results(test_id)
+ emit('test_update', results)
+ socketio.sleep(0.5)
+
+ socketio.start_background_task(send_updates)
+
+@socketio.on('stop_test')
+def handle_stop_test(data):
+ test_id = data.get('test_id')
+ stop_test(test_id)
+ emit('test_stopped', {'test_id': test_id})
+```
+
+### 9.3 自定义报告生成
+
+#### 9.3.1 报告模板系统
+```python
+# report_generator.py
+from jinja2 import Environment, FileSystemLoader
+import pdfkit
+import pandas as pd
+
+class ReportGenerator:
+ def __init__(self, template_dir='templates'):
+ self.env = Environment(
+ loader=FileSystemLoader(template_dir),
+ extensions=['jinja2.ext.do']
+ )
+
+ # 注册自定义过滤器
+ self.env.filters['format_bandwidth'] = self.format_bandwidth
+ self.env.filters['format_time'] = self.format_time
+ self.env.filters['quality_color'] = self.quality_color
+
+ def generate_report(self, data, template_name, output_format='html'):
+ """生成报告"""
+ template = self.env.get_template(template_name)
+
+ # 渲染模板
+ html_content = template.render(**data)
+
+ if output_format == 'html':
+ return html_content
+ elif output_format == 'pdf':
+ return self.html_to_pdf(html_content)
+ elif output_format == 'docx':
+ return self.html_to_docx(html_content)
+ else:
+ raise ValueError(f"不支持的格式: {output_format}")
+
+ def format_bandwidth(self, value, unit='auto'):
+ """格式化带宽显示"""
+ if unit == 'auto':
+ if value >= 1000:
+ return f"{value/1000:.2f} Gbps"
+ elif value >= 1:
+ return f"{value:.2f} Mbps"
+ else:
+ return f"{value*1000:.2f} Kbps"
+ elif unit == 'G':
+ return f"{value/1000:.2f} Gbps"
+ elif unit == 'M':
+ return f"{value:.2f} Mbps"
+ elif unit == 'K':
+ return f"{value*1000:.2f} Kbps"
+
+ def format_time(self, seconds):
+ """格式化时间显示"""
+ if seconds < 60:
+ return f"{seconds:.1f} 秒"
+ elif seconds < 3600:
+ minutes = int(seconds // 60)
+ secs = seconds % 60
+ return f"{minutes} 分 {secs:.0f} 秒"
+ else:
+ hours = int(seconds // 3600)
+ minutes = int((seconds % 3600) // 60)
+ return f"{hours} 小时 {minutes} 分"
+
+ def quality_color(self, score):
+ """根据分数返回颜色"""
+ if score >= 90:
+ return "#28a745" # 绿色
+ elif score >= 80:
+ return "#17a2b8" # 蓝色
+ elif score >= 60:
+ return "#ffc107" # 黄色
+ elif score >= 40:
+ return "#fd7e14" # 橙色
+ else:
+ return "#dc3545" # 红色
+
+ def html_to_pdf(self, html_content):
+ """HTML转PDF"""
+ options = {
+ 'page-size': 'A4',
+ 'margin-top': '0.75in',
+ 'margin-right': '0.75in',
+ 'margin-bottom': '0.75in',
+ 'margin-left': '0.75in',
+ 'encoding': "UTF-8",
+ 'no-outline': None,
+ 'enable-local-file-access': None
+ }
+
+ pdf = pdfkit.from_string(html_content, False, options=options)
+ return pdf
+
+ def generate_comparison_report(self, results_list, baseline=None):
+ """生成对比报告"""
+ # 创建DataFrame
+ df = pd.DataFrame(results_list)
+
+ # 计算统计
+ stats = {
+ '测试次数': len(df),
+ '平均带宽': df['bandwidth'].mean(),
+ '最大带宽': df['bandwidth'].max(),
+ '最小带宽': df['bandwidth'].min(),
+ '带宽标准差': df['bandwidth'].std(),
+ '平均延迟': df['latency'].mean(),
+ '平均丢包': df['loss'].mean(),
+ }
+
+ if baseline:
+ stats['与基线差异'] = {
+ '带宽': ((df['bandwidth'].mean() - baseline['bandwidth']) / baseline['bandwidth']) * 100,
+ '延迟': ((df['latency'].mean() - baseline['latency']) / baseline['latency']) * 100,
+ }
+
+ # 生成报告数据
+ report_data = {
+ 'statistics': stats,
+ 'results': df.to_dict('records'),
+ 'charts': {
+ 'bandwidth_trend': self.create_bandwidth_chart(df),
+ 'quality_distribution': self.create_quality_distribution(df),
+ 'comparison_chart': self.create_comparison_chart(df, baseline) if baseline else None
+ }
+ }
+
+ return self.generate_report(report_data, 'comparison_report.html')
+
+ def create_bandwidth_chart(self, df):
+ """创建带宽趋势图数据"""
+ # 这里生成图表数据，可以是Plotly、Chart.js等格式
+ chart_data = {
+ 'labels': df['timestamp'].tolist(),
+ 'datasets': [{
+ 'label': '带宽 (Mbps)',
+ 'data': df['bandwidth'].tolist(),
+ 'borderColor': 'rgb(75, 192, 192)',
+ 'backgroundColor': 'rgba(75, 192, 192, 0.2)',
+ }]
+ }
+ return chart_data
+```
+
+#### 9.3.2 报告模板示例
+```html
+
+
+
+
+
+
+ 生成时间: {{ timestamp }}
+测试次数: {{ statistics.测试次数 }} 次
+范围: {{ statistics.最小带宽|format_bandwidth }} - {{ statistics.最大带宽|format_bandwidth }}
+丢包率: {{ statistics.平均丢包|round(3) }}%
+延迟变化: {{ statistics.与基线差异.延迟|round(2) }}%
+| 时间 | +带宽 | +延迟 | +丢包 | +质量评分 | +备注 | +
|---|---|---|---|---|---|
| {{ result.timestamp }} | +{{ result.bandwidth|format_bandwidth }} | +{{ result.latency|round(2) }} ms | +{{ result.loss|round(3) }}% | ++ + {{ result.quality_score|round(1) }} + | +{{ result.notes or '' }} | +