python3GUI--运维系统大屏 By:PyQt5（详细分享）

原创

懷淰

修改于 2025-07-23 07:26:01

1600

文章被收录于专栏：PyQt5PyQt5

@toc

文件大小120M，欢迎下载体验！https://pan.baidu.com/s/1GfAx35XnoaO_pXb_laFNdg?pwd=2333

一．前言

今天给大家带来我使用PyQt5开发的运维系统大屏，笔者定义系统名称为：「SysPulse」 - 智能系统监测中心，此系统用于展示当前操作系统相关指标参数，采用多种可视化方案对当前系统的指标进行可视化展示，请大家拭目以待！

本篇我将详细分享软件系统实现流程，也会粘贴具体的代码片段和大家分享！

二．预览

软件启动后进入到系统主屏，主屏包括三个区域分别是左侧、中间、右侧区域，这三个区域分别展示了：

通过折线图、地图、水球图、条形图、表格动态展示了系统真实数据，实时刷新。

系统内置7种背景图，支持动态调整！

左侧	中间	右侧
• 系统信息	• IP归属地	• CPU使用率
• 网关连通性	• 磁盘	• 内存使用率
• 网络情况		• TOP5进程

三．相关技术

1.ECharts

ECharts 是一款由百度开源的高性能 JavaScript 可视化库，提供丰富的图表类型（如折线图、柱状图、地图等）和强大的交互功能（缩放、拖拽、动态更新等），支持响应式设计并兼容多端设备。其灵活的配置选项和扩展性使开发者能够轻松创建专业的数据可视化应用，广泛应用于数据分析、实时监控及商业报表等领域。通过简洁的代码即可实现复杂图表，是前端数据可视化的热门选择。

2.PyQt5

PyQt5 是一个用于创建图形用户界面（GUI）的 Python 库，基于 Qt 框架开发。它提供了丰富的控件（如按钮、文本框、表格等）和强大的功能（多线程、网络通信、数据库交互等），支持跨平台运行（Windows、macOS、Linux）。开发者可以用 PyQt5 快速构建复杂的桌面应用程序，同时结合 Python 的简洁语法和 Qt 的高性能渲染，适用于数据分析工具、多媒体软件、自动化程序等开发。

3.PyQt5 WEB技术

我们本次的可视化方案大多数都是使用Echarts图展示的并且不依赖本地html文件，这是如何做到的呢？

PyQtWebEngine 是 PyQt5 的一个扩展模块，基于 Chromium 的 Qt WebEngine 框架，用于在 PyQt5 应用程序中嵌入现代网页浏览器功能。它支持 HTML5、CSS3、JavaScript 和 WebGL，允许开发者：内嵌网页渲染：在 PyQt5 窗口内显示网页内容，如加载在线地图、Web 应用或本地 HTML 文件。

交互控制：通过 Python 与网页 JavaScript 双向通信（如调用 JS 函数或监听网页事件）。

定制浏览器：构建带有导航栏、开发者工具等功能的完整浏览器应用。

适用于需要混合 Web 技术与桌面 GUI 的场景，如内嵌 Web 报表、在线文档查看器或基于 Web 的桌面应用。

首先我们重写了QWebEnginePage的contextMenuEvent禁用了浏览器右击事件，这样避免了用户误操作右击事件，然后我们重写了QWebEngineView的contextMenuEvent也禁用了鼠标右击事件，最后我们定义了一个BaseChart，继承自QWidget，所有子图表类都继承自这个基类，拥有相同的属性，大致的初始化流程见下图。

这里我们以水球图为例，水球图继承图表基类，通过generate_html方法生成整体html框架，最后使用self.view.setHtml(html, QUrl(""))函数设置内存中的html代码到webengintview里，这样就完成了整体页面的渲染，大家能看到下图的效果。

如何更新这个百分比数值呢？这里给出两种方案：

1.使用js，通过page的runJavaScript对图表数据进行更新，具体代码可以参考：

	def update_value(self, percent):
		self.percent = percent
		percent_display = f"{percent * 100:.2f}%"
		js = f"""
        if (window.chart) {{
            window.chart.setOption({{
                series: [{{
                    data: [{percent}],
                    label: {{
                        formatter: '{percent_display}'
                    }}
                }}]
            }});
        }}
        """
		self.view.page().runJavaScript(js)

2.使用QWebChannel设置信号“桥”来进行通信，具体来说是定义一个类继承自QObject，在其中定义一个信号，通过这个信号来改变网页中的内容

class GaugeBridge(QObject):
	update_value = pyqtSignal(float, str)

	def send_value(self, value: float, label: str):
		self.update_value.emit(value, label)

	def get_html(self):
		title_block = (
			f"title: {{ text: '{self.chart_title}', textStyle: {{ color: '#ffffff' }} }},"
			if self.chart_title else ""
		)

		# 根据 mode 设置颜色
		if self.mode == 2:
			color_gradient = """
                color: [
                    [1, new echarts.graphic.LinearGradient(0, 0, 1, 1, [
                        { offset: 0, color: 'rgb(0,255,127)' },
                        { offset: 1, color: 'rgb(0,255,127)' }
                    ])]
                ]
            """
		else:
			color_gradient = """
                color: [
                    [1, new echarts.graphic.LinearGradient(0, 0, 1, 1, [
                        { offset: 0, color: 'rgb(0,228,200)' },
                        { offset: 1, color: 'rgb(0,220,222)' }
                    ])]
                ]
            """

		return f"""
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>仪表盘</title>
    <script src="https://cdn.jsdelivr.net/npm/echarts@5"></script>
    <script src="qrc:///qtwebchannel/qwebchannel.js"></script>
    <style>
        html, body, #main {{
            width: 100%;
            height: 100%;
            margin: 0;
            background: transparent;
        }}
    </style>
</head>
<body>
    <div id="main"></div>
    <script>
        var chart = echarts.init(document.getElementById('main'));

        window.addEventListener("resize", function () {{
            chart.resize();
        }});

        var option = {{
            {title_block}
            series: [{{
                type: 'gauge',
                min: 0,
                max: 100,
                axisLine: {{
                    lineStyle: {{
                        width: 20,
                        {color_gradient}
                    }}
                }},
                progress: {{
                    show: true,
                    width: 20
                }},
                pointer: {{
                    width: 6,
                    length: '80%',
                    itemStyle: {{
                        color: '#fff'
                    }}
                }},
                axisTick: {{ show: false }},
                splitLine: {{
                    length: 20,
                    lineStyle: {{ color: '#fff', width: 2 }}
                }},
                axisLabel: {{
                    color: '#ffffff',
                    fontSize: 12
                }},
                detail: {{
                    formatter: function(value) {{
                        return value;
                    }},
                    fontSize: 20,
                    offsetCenter: [0, '65%'],
                    color: '#ffffff'
                }},
                data: [{{ value: 0 }}]
            }}]
        }};

        chart.setOption(option);

        new QWebChannel(qt.webChannelTransport, function(channel) {{
            let bridge = channel.objects.bridge;
            bridge.update_value.connect(function(val, label) {{
                chart.setOption({{
                    series: [{{
                        data: [{{ value: val }}],
                        detail: {{
                            formatter: function() {{
                                return label;
                            }}
                        }}
                    }}]
                }});
            }});
        }});
    </script>
</body>
</html>
"""

更新效果见下图

4.系统核心

本小结为系统核心部分，这里把我代码里的main_utils.py代码贴出来，需要的朋友直接复制粘贴即可运行，界面上所有的数据都是取自系统，这个工具脚本可以帮你完成一切。

import getpass
import platform
import socket
import netifaces
from datetime import datetime
import requests
import psutil
import time
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from src.conf import test_data, system_conf
from src.utils.custom_utils import format_size


def sample_process(proc, num_cpus):
	try:
		cpu = proc.cpu_percent(None) / num_cpus
		mem = proc.memory_percent()
		info = proc.as_dict(attrs=['pid', 'name'])

		name = info['name']
		pid = info['pid']

		if name in ["System Idle Process", "", "System"] or pid == 0:
			return None

		return {
			'pid': pid,
			'name': name,
			'cpu_percent': cpu,
			'memory_percent': mem
		}
	except (psutil.NoSuchProcess, psutil.AccessDenied):
		return None


def get_top_processes():
	num_cpus = psutil.cpu_count(logical=True)

	procs = []
	for proc in psutil.process_iter(['pid', 'name']):
		try:
			proc.cpu_percent(None)  # 初始化
			procs.append(proc)
		except (psutil.NoSuchProcess, psutil.AccessDenied):
			continue

	time.sleep(0.1)  # 等待 CPU 使用率更新

	temp = defaultdict(lambda: {'pid': None, 'name': '', 'cpu_percent': 0.0, 'memory_percent': 0.0})

	with ThreadPoolExecutor(max_workers=32) as executor:
		futures = [executor.submit(sample_process, proc, num_cpus) for proc in procs]
		for future in as_completed(futures):
			result = future.result()
			if result:
				name = result['name']
				if temp[name]['pid'] is None:
					temp[name]['pid'] = result['pid']
					temp[name]['name'] = name
				temp[name]['cpu_percent'] += result['cpu_percent']
				temp[name]['memory_percent'] += result['memory_percent']

	merged_processes = list(temp.values())
	top_processes = sorted(merged_processes, key=lambda x: x['cpu_percent'], reverse=True)[:5]

	return top_processes


def get_disk_usage():
	"""
	获取硬盘使用数据
	:return: List[Dict] 每个字典包含一个分区的信息
	"""
	partitions = psutil.disk_partitions()
	disk_info_list = []

	for part in partitions:
		try:
			usage = psutil.disk_usage(part.mountpoint)
			disk_info = {
				"device": part.device,
				"mountpoint": part.mountpoint,
				"fstype": part.fstype,
				"total": format_size(usage.total),
				"used": format_size(usage.used),
				"free": format_size(usage.free),
				"percent": f"{usage.percent:.1f}%"
			}
			disk_info_list.append(disk_info)
		except PermissionError:
			continue  # 忽略无权限访问的分区

	return disk_info_list


def get_default_gateway():
	"""
	获取网关地址
	:return:
	"""
	gateways = netifaces.gateways()
	default_gateway = gateways.get('default')
	if default_gateway:
		return default_gateway.get(netifaces.AF_INET, [None])[0]
	return None


def get_system_info():
	"""
	获取系统数据
	:return:
	"""
	uname = platform.uname()
	boot_time = psutil.boot_time()
	uptime_str = str(datetime.now() - datetime.fromtimestamp(boot_time)).split('.')[0]
	info = {
		"os_version": f"{uname.system} {uname.release}",
		"cpu_model": uname.processor or platform.processor(),
		"physical_cores": psutil.cpu_count(logical=False),
		"logical_cores": psutil.cpu_count(logical=True),
		"architecture": platform.machine(),
		"internal_ip": socket.gethostbyname(socket.gethostname()),
		"uptime": uptime_str,
		"username": getpass.getuser()
	}
	return info


def get_cpu_memory_usage():
	"""
	获取cpu和内存数据
	:return:
	"""
	cpu_usage = psutil.cpu_percent(interval=1)  # 1 秒采样时间
	memory_usage = psutil.virtual_memory().percent
	return float(cpu_usage), float(memory_usage)


def get_network_speed(interval=1.0):
	"""
	获取当前网络上下行速度，单位 Mbps。
	:param interval: 采样间隔，单位秒
	:return: (upload_mbps, download_mbps)
	"""
	net1 = psutil.net_io_counters()
	time.sleep(interval)
	net2 = psutil.net_io_counters()

	bytes_sent = net2.bytes_sent - net1.bytes_sent
	bytes_recv = net2.bytes_recv - net1.bytes_recv

	# Bytes → bits → megabits (除以 1e6)
	upload_mbps = (bytes_sent * 8) / (interval * 1e6)
	download_mbps = (bytes_recv * 8) / (interval * 1e6)

	return round(upload_mbps, 3), round(download_mbps, 3)


if __name__ == '__main__':
	print(get_location_by_ip())

5.为什么系统不卡？

有的同学可能会好奇，为什么系统1秒更新一次数据，界面并没有明显卡顿呢？

这里要多亏了多线程，具体来说我们定义了多个线程类，这些类继承自QThread，实现了其中的run方法，这样我们可以使用信号和槽对线程之间的数据进行管理，使用线程进行耗时操作不阻塞UI线程，所有数据更新都是在子线程中进行的，当数据处理完成通过信号的方式发射回到主线程，主线程操作UI，更新展示数据，通过上面的操作就避免了系统卡顿，这里我贴一段代码吧！