网络状态实时监测工具
下面是一个基于Python的图形化网络状态实时监测工具,使用Ping命令检查目标主机的连通性并显示实时结果。
import subprocess
import time
import platform
import tkinter as tk
from tkinter import ttk, messagebox
import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
import threading
import queue
class PingMonitor:
def __init__(self, root):
self.root = root
self.root.title("网络状态实时监测工具")
self.root.geometry("900x700")
self.root.resizable(True, True)
# 创建队列用于线程间通信
self.data_queue = queue.Queue()
# 初始化变量
self.target_host = tk.StringVar(value="www.google.com")
self.interval = tk.DoubleVar(value=1.0) # 默认1秒
self.ping_count = tk.IntVar(value=0)
self.success_count = tk.IntVar(value=0)
self.fail_count = tk.IntVar(value=0)
self.running = False
self.ping_thread = None
# 创建UI
self.create_widgets()
# 初始化图表
self.setup_chart()
# 初始化数据列表
self.timestamps = []
self.latencies = []
self.statuses = [] # 0表示失败，1表示成功
def create_widgets(self):
# 控制面板
control_frame = ttk.LabelFrame(self.root, text="控制面板")
control_frame.pack(fill=tk.X, padx=10, pady=10)
# 目标主机输入
ttk.Label(control_frame, text="目标主机:").grid(row=0, column=0, padx=5, pady=5, sticky=tk.W)
host_entry = ttk.Entry(control_frame, textvariable=self.target_host, width=30)
host_entry.grid(row=0, column=1, padx=5, pady=5, sticky=tk.W)
# 间隔时间
ttk.Label(control_frame, text="检测间隔(秒):").grid(row=0, column=2, padx=5, pady=5, sticky=tk.W)
interval_spin = ttk.Spinbox(control_frame, from_=0.1, to=10, increment=0.1,
textvariable=self.interval, width=5)
interval_spin.grid(row=0, column=3, padx=5, pady=5, sticky=tk.W)
# 按钮
self.start_button = ttk.Button(control_frame, text="开始监测", command=self.start_monitoring)
self.start_button.grid(row=0, column=4, padx=5, pady=5)
self.stop_button = ttk.Button(control_frame, text="停止监测", command=self.stop_monitoring, state=tk.DISABLED)
self.stop_button.grid(row=0, column=5, padx=5, pady=5)
# 统计信息
stats_frame = ttk.LabelFrame(self.root, text="统计信息")
stats_frame.pack(fill=tk.X, padx=10, pady=5)
ttk.Label(stats_frame, text="已发送请求:").grid(row=0, column=0, padx=10, pady=5, sticky=tk.W)
ttk.Label(stats_frame, textvariable=self.ping_count).grid(row=0, column=1, padx=10, pady=5, sticky=tk.W)
ttk.Label(stats_frame, text="成功次数:").grid(row=0, column=2, padx=10, pady=5, sticky=tk.W)
ttk.Label(stats_frame, textvariable=self.success_count, foreground="green").grid(row=0, column=3, padx=10, pady=5, sticky=tk.W)
ttk.Label(stats_frame, text="失败次数:").grid(row=0, column=4, padx=10, pady=5, sticky=tk.W)
ttk.Label(stats_frame, textvariable=self.fail_count, foreground="red").grid(row=0, column=5, padx=10, pady=5, sticky=tk.W)
# 日志区域
log_frame = ttk.LabelFrame(self.root, text="实时日志")
log_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
self.log_text = tk.Text(log_frame, height=8)
self.log_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
self.log_text.config(state=tk.DISABLED)
# 图表区域
chart_frame = ttk.LabelFrame(self.root, text="延迟图表")
chart_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
self.figure = plt.Figure(figsize=(8, 4), dpi=100)
self.ax = self.figure.add_subplot(111)
self.canvas = FigureCanvasTkAgg(self.figure, master=chart_frame)
self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True)
# 状态栏
self.status_var = tk.StringVar(value="就绪")
status_bar = ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W)
status_bar.pack(side=tk.BOTTOM, fill=tk.X)
def setup_chart(self):
self.ax.clear()
self.ax.set_title('网络延迟监测')
self.ax.set_xlabel('时间')
self.ax.set_ylabel('延迟 (ms)')
self.ax.grid(True, linestyle='--', alpha=0.7)
self.canvas.draw()
def update_chart(self):
if not self.timestamps or not self.latencies:
return
self.ax.clear()
# 绘制成功点
success_times = [self.timestamps[i] for i in range(len(self.statuses)) if self.statuses[i] == 1]
success_latencies = [self.latencies[i] for i in range(len(self.statuses)) if self.statuses[i] == 1]
if success_times:
self.ax.plot(success_times, success_latencies, 'go-', label='成功', linewidth=1, markersize=4)
# 绘制失败点
fail_times = [self.timestamps[i] for i in range(len(self.statuses)) if self.statuses[i] == 0]
fail_latencies = [self.latencies[i] for i in range(len(self.statuses)) if self.statuses[i] == 0]
if fail_times:
self.ax.plot(fail_times, fail_latencies, 'ro', label='失败', markersize=6)
# 设置图表属性
self.ax.set_title(f'目标: {self.target_host.get()} 延迟监测')
self.ax.set_xlabel('时间')
self.ax.set_ylabel('延迟 (ms)')
self.ax.grid(True, linestyle='--', alpha=0.7)
self.ax.legend()
# 自动调整Y轴范围
if success_latencies:
max_val = max(success_latencies) * 1.2 if max(success_latencies) > 0 else 100
self.ax.set_ylim(0, max_val)
self.canvas.draw()
def log_message(self, message, color="black"):
self.log_text.config(state=tk.NORMAL)
self.log_text.insert(tk.END, message + "n", color)
self.log_text.tag_config("red", foreground="red")
self.log_text.tag_config("green", foreground="green")
self.log_text.tag_config("blue", foreground="blue")
self.log_text.see(tk.END) # 滚动到最新消息
self.log_text.config(state=tk.DISABLED)
def ping_host(self):
"""执行Ping命令并返回结果"""
host = self.target_host.get()
# 根据操作系统选择不同的ping参数
param = "-n" if platform.system().lower() == "windows" else "-c"
count = "1" # 只发送一个包
try:
# 执行ping命令
output = subprocess.check_output(
["ping", param, count, host],
stderr=subprocess.STDOUT,
universal_newlines=True,
timeout=self.interval.get()
)
# 解析ping结果
if "time=" in output:
# 提取延迟时间
time_str = [line for line in output.split('n') if "time=" in line][0]
latency = float(time_str.split("time=")[1].split(" ")[0])
status = 1 # 成功
else:
latency = 0
status = 0 # 失败
return latency, status, output
except subprocess.CalledProcessError:
return 0, 0, "请求超时或目标不可达"
except subprocess.TimeoutExpired:
return 0, 0, "命令执行超时"
except Exception as e:
return 0, 0, f"错误: {str(e)}"
def monitoring_loop(self):
"""监控循环"""
while self.running:
start_time = time.time()
# 执行ping
latency, status, output = self.ping_host()
timestamp = time.strftime("%H:%M:%S")
# 更新统计数据
self.ping_count.set(self.ping_count.get() + 1)
if status == 1:
self.success_count.set(self.success_count.get() + 1)
log_color = "green"
log_message = f"[{timestamp}] 来自 {self.target_host.get()} 的回复: 时间={latency:.2f}ms"
else:
self.fail_count.set(self.fail_count.get() + 1)
log_color = "red"
log_message = f"[{timestamp}] 请求超时"
# 将数据放入队列
self.data_queue.put((timestamp, latency, status, log_message, log_color))
# 计算剩余等待时间
elapsed = time.time() - start_time
wait_time = max(0, self.interval.get() - elapsed)
time.sleep(wait_time)
def start_monitoring(self):
"""开始监控"""
if not self.target_host.get():
messagebox.showerror("错误", "请输入目标主机地址")
return
if self.running:
return
# 重置状态
self.running = True
self.ping_count.set(0)
self.success_count.set(0)
self.fail_count.set(0)
self.timestamps = []
self.latencies = []
self.statuses = []
# 清空日志
self.log_text.config(state=tk.NORMAL)
self.log_text.delete(1.0, tk.END)
self.log_text.config(state=tk.DISABLED)
# 更新按钮状态
self.start_button.config(state=tk.DISABLED)
self.stop_button.config(state=tk.NORMAL)
# 更新状态栏
self.status_var.set(f"正在监测 {self.target_host.get()} ...")
# 启动监控线程
self.ping_thread = threading.Thread(target=self.monitoring_loop, daemon=True)
self.ping_thread.start()
# 启动UI更新循环
self.update_ui()
self.log_message(f"开始监测 {self.target_host.get()}，间隔 {self.interval.get()} 秒", "blue")
def stop_monitoring(self):
"""停止监控"""
if self.running:
self.running = False
if self.ping_thread and self.ping_thread.is_alive():
self.ping_thread.join(timeout=1.0)
self.start_button.config(state=tk.NORMAL)
self.stop_button.config(state=tk.DISABLED)
self.status_var.set("监测已停止")
self.log_message("监测已停止", "blue")
def update_ui(self):
"""更新UI元素"""
# 处理队列中的所有消息
while not self.data_queue.empty():
timestamp, latency, status, log_message, log_color = self.data_queue.get()
# 更新日志
self.log_message(log_message, log_color)
# 更新数据列表
self.timestamps.append(timestamp)
self.latencies.append(latency)
self.statuses.append(status)
# 更新图表
if len(self.timestamps) % 3 == 0 or status == 0: # 每3个点更新一次图表，或者有失败时更新
self.update_chart()
# 如果仍在运行，则继续调度更新
if self.running:
self.root.after(100, self.update_ui)
if __name__ == "__main__":
root = tk.Tk()
app = PingMonitor(root)
root.mainloop()
功能说明
这个网络状态实时监测工具提供以下功能：
- 目标主机监测：可以输入任何域名或IP地址进行监测
- 实时日志：显示每次Ping请求的结果（成功或失败）
- 统计信息：
- 已发送请求总数
- 成功次数（绿色显示）
- 失败次数（红色显示）
- 延迟图表：以折线图形式展示网络延迟变化，绿色点表示成功，红色点表示失败
- 控制选项：
- 可调节检测间隔（0.1-10秒）
- 开始/停止监测按钮
使用说明
- 在”目标主机”输入框中输入要监测的域名或IP地址（如www.google.com）
- 设置检测间隔（默认1秒）
- 点击”开始监测”按钮启动监控
- 实时日志将显示每次Ping的结果
- 图表区域将显示网络延迟的变化情况
- 点击”停止监测”按钮结束监控
运行要求
- Python 3.x
- 需要安装以下库：
pip install matplotlib tk
这个工具可以帮助您实时监测网络连接状态，快速发现网络中断或延迟增加的问题,适用于网络维护和故障排查场景。
图片来源于AI模型，如侵权请联系管理员。作者：酷小编，如若转载，请注明出处：https://www.kufanyun.com/ask/285238.html
