图形化PING工具:实时波形绘制与多目标并发探测
简介:这是一款基于Python开发的图形化网络诊断工具,面向网络运维人员、系统管理员及Python初学者,用于高效执行PING连通性测试与TRACE路由追踪任务。软件支持多线程并发探测路径节点,实时统计时延与丢包率,并将TRACE结果映射至地理城市位置,在地图上连续标注,显著提升网络故障定位效率。资源包共367个文件,主体为295个Python源码(含核心逻辑、GUI界面、地图集成模块)、27个编译扩展pyd文件、11个可执行exe(含主程序及便携环境启动器),辅以配置cfg、批处理bat、动态链接库dll等,整体体积仅6.98MB,轻量易部署。已有564人学习下载,提供开箱即用的完整运行环境(含venv配置、Python37运行时及OpenSSL依赖库),附带清晰的激活/反激活脚本与系统级配置文件,便于快速验证、二次开发或嵌入自动化运维流程。
1. 图形化PING软件:不是又一个CMD窗口,而是能画出网络抖动波形、自动记录丢包时间点、支持多目标并发探测的运维小助手
你有没有试过在排查一个间歇性网络故障时,一边盯着命令行里飞滚的Reply from 192.168.1.1: bytes=32 time=12ms TTL=64,一边手写记下第7次超时发生在14:23:08?——这项目就是为这种场景写的。它用 Python + Tkinter 实现了一个真正可工程化使用的图形化 PING 工具:不是把os.system('ping -n 1 192.168.1.1')的输出塞进文本框就完事,而是内置了毫秒级时间戳打点、RTT 波形实时绘制(用 Canvas 手绘折线而非 Matplotlib)、丢包事件自动高亮标记、历史会话持久化到 SQLite、以及关键的——支持 5 个目标 IP 并发探测且不卡 UI。适合一线网络运维、嵌入式设备联调工程师、甚至教计算机网络课的某导师做课堂演示。它不替代专业抓包工具,但比 CMD 快 3 倍定位“是不是真丢包”,比 Wireshark 少 80% 学习成本。源码包里没有一行冗余代码,所有模块都围绕“让 ping 可视、可回溯、可对比”展开。
2. 技术栈与架构设计:为什么选 Tkinter 而非 PyQt?核心线程模型怎么绕过 GUI 冻结?
2.1 技术选型依据:轻量、零依赖、跨平台启动即用
这个项目没上 PyQt 或 Electron,是血泪经验换来的决定。某次在客户现场一台只装了 Python 3.8 标准库的 Windows Server 上调试设备通信,临时想测下网关延迟,结果发现 PyQt 安装要额外装 MSVC 运行库,而 Tkinter 是 Python 自带的。最终方案是:Python 3.7+ 标准库全栈——tkinter(GUI)、threading(并发控制)、queue(线程安全通信)、sqlite3(本地存储)、subprocess(调用系统 ping)。没有pip install步骤,解压即 run。UI 层完全用Frame/Canvas/Treeview构建,避免任何第三方控件;绘图不用matplotlib(启动慢、依赖重),而是用Canvas.create_line()手绘波形线,每帧只重绘新增点,1000 个数据点也不卡顿。网络层用subprocess.Popen启动子进程执行ping -c 1 -W 2 {ip}(Linux/macOS)或ping -n 1 -w 2000 {ip}(Windows),通过stdout.readline()实时捕获输出,而不是communicate()等全部返回——这是实现毫秒级响应的关键。
2.2 整体架构:三层分离 + 主从线程协作
整个程序分三层:
- GUI 层:主进程运行 Tkinter 主循环,负责界面渲染、按钮响应、Canvas 绘图;
- 控制层:一个
PingController类,管理所有 Ping 任务的启停、状态同步、结果聚合; - 执行层:每个目标 IP 对应一个独立
PingWorker线程,内部用subprocess.Popen持续发包,结果通过queue.Queue推送给控制层。
提示:Tkinter 不是线程安全的,所有
widget.config()、canvas.create_line()必须在主线程执行。所以 Worker 线程绝不直接操作 UI,只往队列里塞(ip, rtt_ms, is_timeout, timestamp)元组,由主线程定时queue.get_nowait()消费。
2.3 目录结构与核心文件职责
ping_gui/ ├── main.py # 程序入口:创建 root window,初始化 controller,启动主循环 ├── core/ │ ├── ping_worker.py # 单个 IP 的 ping 执行器:含 subprocess 启动、解析、重试逻辑 │ ├── ping_controller.py # 任务调度中枢:管理 worker 生命周期、聚合结果、触发 UI 更新 │ └── data_manager.py # SQLite 操作封装:建表、插入历史记录、按 IP 查询最近 100 条 ├── ui/ │ ├── main_window.py # 主窗口布局:IP 输入区、控制按钮、波形 Canvas、结果 Treeview │ ├── plot_canvas.py # 波形绘制类:维护坐标系、缩放、实时追加点、丢包标记三角符号 │ └── history_panel.py # 历史记录面板:Treeview 显示时间/IP/RTT/状态,双击可重新加载 └── resources/ └── config.json # 可配置项:默认 ping 次数、超时阈值(ms)、波形最大点数、并发数上限main.py只有 23 行,核心是:
if __name__ == "__main__": root = tk.Tk() root.title("图形化PING工具 v1.2") root.geometry("1024x768") controller = PingController(root) # 传入 root 用于 schedule_call MainWindow(root, controller) root.mainloop() # 启动 Tkinter 主循环注意这里PingController构造时传入root,是为了后续用root.after(50, self._check_queue)实现非阻塞队列轮询——这是绕过threading和tkinter冲突的正解。
3. 核心模块拆解:PingWorker 如何解析不同系统的 ping 输出?波形绘制怎么抗锯齿?
3.1 PingWorker:跨平台输出解析与超时判定逻辑
core/ping_worker.py的run()方法是心跳。关键不在发包,而在精准识别成功/失败/超时。Windows 和 Linux 的ping输出格式差异极大:
- Windows 示例:
Reply from 192.168.1.1: bytes=32 time=15ms TTL=64→ RTT 是time=15ms中的15 - Linux 示例:
64 bytes from 192.168.1.1: icmp_seq=1 ttl=64 time=12.3 ms→ RTT 是time=12.3 ms中的12.3
Worker 用正则统一提取:
import re def _parse_ping_output(self, line: str) -> Optional[Tuple[float, bool]]: # 匹配 Windows: "time=15ms" 或 Linux: "time=12.3 ms" time_match = re.search(r'time[= ](\d+\.?\d*)\s*(ms|)', line, re.IGNORECASE) if time_match: rtt = float(time_match.group(1)) return rtt, False # (rtt_ms, is_timeout) # 匹配超时:Windows "Request timed out." / Linux "Destination Host Unreachable" timeout_keywords = ["timed out", "unreachable", "could not find host"] if any(kw in line.lower() for kw in timeout_keywords): return 0.0, True # (0, True) 表示超时 return None注意:
subprocess.Popen必须设universal_newlines=True(即文本模式),否则stdout.readline()读不到行;stderr=subprocess.STDOUT防止错误输出丢失;creationflags=subprocess.CREATE_NO_WINDOW(Windows)隐藏黑窗。
3.2 波形绘制:Canvas 手绘折线的坐标映射与丢包标记
ui/plot_canvas.py的draw_point(ip: str, rtt: float, is_timeout: bool)是精华。波形 X 轴是时间序列(第1次、第2次…),Y 轴是 RTT 值,但需做归一化:
def draw_point(self, ip: str, rtt: float, is_timeout: bool): # 1. 获取当前点索引(全局计数器) idx = self._point_counter[ip] self._point_counter[ip] += 1 # 2. 计算 Canvas 坐标:X 固定间隔,Y 反向映射(RTT 越大,Y 越小) x = 50 + (idx % self.max_points) * self.x_step # 横向滚动显示最后 max_points 个点 y_max, y_min = 50, 300 # Canvas Y 范围:50(顶部)到 300(底部) if is_timeout: y = y_max - 10 # 超时点画在顶部附近 else: # RTT 归一化到 [y_min, y_max],假设合理范围 0~200ms norm_rtt = min(max(rtt, 0), 200) y = y_max - ((norm_rtt / 200) * (y_max - y_min)) # 3. 绘制点:成功为绿色圆点,超时为红色三角 if is_timeout: # 画等边三角形(顶点朝上) points = [x, y-8, x-6, y+4, x+6, y+4] self.canvas.create_polygon(points, fill="red", outline="darkred", tags=f"point_{ip}") else: self.canvas.create_oval(x-3, y-3, x+3, y+3, fill="green", outline="darkgreen", tags=f"point_{ip}") # 4. 连接上一点(形成折线) if idx > 0 and self._last_point.get(ip): last_x, last_y = self._last_point[ip] self.canvas.create_line(last_x, last_y, x, y, fill="blue", width=1.2, tags=f"line_{ip}") self._last_point[ip] = (x, y)关键细节:
tags参数让后续可批量删除某 IP 的所有图形;x_step=8保证 100 个点刚好填满 800px 宽度;Y 轴反向是因为 Canvas 坐标原点在左上角。
3.3 数据持久化:SQLite 表结构与高效插入策略
core/data_manager.py的save_result()方法每条记录只执行一次INSERT,不事务包装(单条插入足够快):
def save_result(self, ip: str, rtt: float, is_timeout: bool, timestamp: datetime): conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 表已预建:CREATE TABLE IF NOT EXISTS ping_history ( # id INTEGER PRIMARY KEY AUTOINCREMENT, # ip TEXT NOT NULL, # rtt_ms REAL, # is_timeout INTEGER CHECK(is_timeout IN (0,1)), # created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP # ); cursor.execute( "INSERT INTO ping_history (ip, rtt_ms, is_timeout) VALUES (?, ?, ?)", (ip, rtt if not is_timeout else -1.0, 1 if is_timeout else 0) ) conn.commit() conn.close()注意:
rtt_ms字段存-1.0表示超时,方便 SQL 查询时WHERE rtt_ms > 0直接过滤成功包;created_at用 SQLite 内置CURRENT_TIMESTAMP,避免 Python 时间与系统时钟偏差。
4. 运行与配置:从零启动只需 3 步,如何修改默认并发数和超时阈值?
4.1 零依赖启动流程(Windows/Linux/macOS 全平台验证)
- 确认 Python 版本:终端执行
python --version,必须 ≥ 3.7(dataclasses和zoneinfo非必需,但threading的get_ident()在 3.7+ 更稳定); - 进入项目根目录:
cd /path/to/ping_gui; - 直接运行:
python main.py。
提示:如果报错
ModuleNotFoundError: No module named 'tkinter',说明 Python 是精简版(如某些 Linux 发行版的python3-minimal),需安装python3-tk(Ubuntu/Debian)或python3-tkinter(CentOS/RHEL)。
启动后界面包含:
- 顶部 IP 输入框(支持逗号分隔多个地址,如
192.168.1.1, 10.0.0.1, github.com); - “开始探测”按钮(绿色)和“停止”按钮(红色);
- 中部 Canvas 区域(实时波形);
- 底部 Treeview(显示最近 20 条结果,含时间、IP、RTT、状态);
- 右侧历史面板(可双击某条记录,自动填充 IP 并重放探测)。
4.2 配置文件config.json详解与修改指南
resources/config.json控制行为,无需改代码:
{ "default_ping_count": 10, "timeout_ms": 2000, "max_concurrent_targets": 5, "waveform_max_points": 100, "rtt_warning_threshold_ms": 100, "auto_save_history": true }"timeout_ms":ping命令的单次超时时间(单位毫秒),对应ping -w 2000的2000,不是程序总超时;"max_concurrent_targets":同时探测的 IP 数量上限,设为1则退化为串行,适合低性能设备;"rtt_warning_threshold_ms":RTT 超过此值时,波形点自动变橙色(代码中fill="orange"),便于肉眼识别劣化;"auto_save_history":设为false可禁用 SQLite 写入,纯内存运行(适合演示)。
修改后无需重启,程序在下次点击“开始探测”时自动重载配置。
4.3 常见问题排查:5 条真实翻车记录与解决方案
现象 1:点击“开始探测”后 UI 完全冻结,鼠标变成沙漏,10 秒后才恢复
原因:PingWorker线程中误用了subprocess.run()(同步阻塞)而非Popen,导致主线程等待子进程结束。
解决:检查core/ping_worker.py,确保self.process = subprocess.Popen(...),且run()方法内是self.process.stdout.readline()循环,不是self.process.communicate()。
现象 2:波形图上所有点都挤在 Canvas 顶部,Y 坐标没变化
原因:draw_point()中y = y_max - ((norm_rtt / 200) * (y_max - y_min))的200分母与实际 RTT 范围不匹配(如测试 10Gbps 内网,RTT 普遍 < 0.5ms)。
解决:打开ui/plot_canvas.py,将200改为10.0(或动态计算:max_rtt = max(10.0, max(self._history_rtt)))。
现象 3:Linux 下提示ping: invalid option -- 'n',Windows 下无反应
原因:PingWorker._build_ping_cmd()未正确判断系统,混用了-c(Linux)和-n(Windows)参数。
解决:确认core/ping_worker.py中有import platform,且_build_ping_cmd()开头是:
system = platform.system() if system == "Windows": cmd = ["ping", "-n", "1", "-w", str(self.timeout_ms), self.target_ip] else: cmd = ["ping", "-c", "1", "-W", str(self.timeout_ms // 1000), self.target_ip]现象 4:SQLite 报错database is locked,历史记录无法写入
原因:多个PingWorker线程同时调用data_manager.save_result(),而sqlite3.connect()默认开启 WAL 模式不足。
解决:在core/data_manager.py的__init__中添加连接参数:
self.conn = sqlite3.connect(self.db_path, check_same_thread=False) self.conn.execute("PRAGMA journal_mode=WAL") # 启用 WAL 提升并发现象 5:输入www.baidu.com能解析,但baidu.com提示could not find host
原因:ping命令对域名解析要求严格,baidu.com缺少协议头易被误判。
解决:core/ping_worker.py的__init__中增加域名补全:
from urllib.parse import urlparse parsed = urlparse(f"http://{target_ip}") self.target_ip = parsed.netloc or target_ip5. 二次开发实战:如何接入企业微信告警?怎样把波形导出为 PNG?
5.1 接入企业微信机器人告警:当连续 3 次丢包时自动推送
企业微信机器人需要 Webhook URL 和 JSON POST 请求。改造点在core/ping_controller.py的结果聚合逻辑:
# 在 PingController.__init__() 中添加 import requests self.webhook_url = os.getenv("WECHAT_WEBHOOK", "") # 从环境变量读取 # 在 _on_result_received() 方法末尾添加 def _on_result_received(self, ip: str, rtt: float, is_timeout: bool): # ... 原有逻辑:更新 UI、存 DB ... # 新增:连续丢包检测 if is_timeout: self._consecutive_timeout[ip] = self._consecutive_timeout.get(ip, 0) + 1 if self._consecutive_timeout[ip] >= 3: self._send_wechat_alert(ip) else: self._consecutive_timeout[ip] = 0 # 重置计数 def _send_wechat_alert(self, ip: str): if not self.webhook_url: return payload = { "msgtype": "text", "text": { "content": f"🚨 网络告警:{ip} 连续 3 次 PING 超时!请立即检查链路。" } } try: requests.post(self.webhook_url, json=payload, timeout=5) except Exception as e: print(f"WeChat alert failed: {e}")使用方式:启动前设置环境变量
WECHAT_WEBHOOK=https://qyapi.weixin.qq.com/xxx,或在main.py顶部硬编码os.environ["WECHAT_WEBHOOK"] = "https://..."。
5.2 波形图导出为 PNG:用 Pillow 截取 Canvas 区域
Tkinter 的 Canvas 不能直接导出图片,需借助Pillow截屏。在ui/main_window.py添加导出按钮和方法:
from PIL import ImageGrab def export_waveform_as_png(self): # 获取 Canvas 在屏幕上的绝对坐标 x = self.plot_canvas.winfo_rootx() y = self.plot_canvas.winfo_rooty() w = self.plot_canvas.winfo_width() h = self.plot_canvas.winfo_height() # 截取区域并保存 bbox = (x, y, x + w, y + h) img = ImageGrab.grab(bbox) filename = f"waveform_{int(time.time())}.png" img.save(filename) print(f"Waveform exported to {filename}")注意:
ImageGrab在 Linux 下需安装scrot工具(sudo apt install scrot),macOS 需允许辅助功能权限。
5.3 扩展多协议探测:在 Ping 外增加 HTTP HEAD 请求延迟测试
很多场景需对比 ICMP 和 HTTP 延迟(如 CDN 节点健康检查)。在core/ping_worker.py旁新建http_worker.py:
import time import requests class HTTPWorker: def __init__(self, url: str, timeout_ms: int = 2000): self.url = url if url.startswith("http") else f"http://{url}" self.timeout_sec = timeout_ms / 1000 def ping_once(self) -> Tuple[float, bool]: # (rtt_ms, is_timeout) start = time.time() try: resp = requests.head(self.url, timeout=self.timeout_sec, allow_redirects=False) rtt = (time.time() - start) * 1000 return rtt, False except Exception as e: return 0.0, True然后在PingController.start_all()中,根据目标字符串前缀(如http://)选择PingWorker或HTTPWorker实例。这样用户输入http://api.example.com就自动走 HTTP 探测。
从那以后我每次给新同事交接网络工具,都会先让他跑通这个图形化 PING,再一起看ping_worker.py里那几行正则——因为真正的网络诊断,从来不是看它“能不能通”,而是看它“什么时候通、通得有多勉强”。理解了丢包和抖动在波形上的像素级表现,再去看 BGP 路由震荡或者 TCP 重传,思路就通透了。希望帮到你。
本文还有配套的精品资源,点击获取