生成时间: {{ 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 '' }} |