本文主要是介绍Python教程:使用psutil和rich库实现一个终端资源监控小工具(附完整代码),希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
在本文中,我们将介绍如何使用 Python 的 psutil 和 rich 库来实现一个终端资源监控小工具。psutil 是一个跨平台的 Python 库,用于获取关于系统进程和系统利用率的信息,而 rich 是一个 Python 库,用于在终端中创建美观的输出。
1.psutil 库介绍
psutil(Python System and Process Utilities)是一个强大的跨平台库,用于获取关于系统进程和系统利用率的信息,例如 CPU、内存、磁盘、网络等。它提供了简单易用的接口,可以方便地监控和管理系统资源。
在本项目中,我们主要使用 psutil 来获取关于指定进程的详细信息,例如进程的用户名、创建时间、线程数、CPU 占比、内存占比等。
2.rich 库介绍
rich 是一个 Python 库,用于在终端中创建美观的输出。它提供了丰富的样式和布局选项,使得在终端中输出信息更加清晰和易读。
在本项目中,我们将使用 rich 库来创建一个美观的表格,显示获取到的进程信息。
3.设计思路
-
导入所需的库:我们首先需要导入 psutil 和 rich 库,以及一些其他必要的模块。
-
编写函数获取目标进程:编写一个函数,通过进程名称获取指定进程的详细信息。
-
编写函数显示信息:编写一个函数,使用 rich 库创建一个表格,并显示获取到的进程信息。
-
主程序逻辑:在主程序中,设置要监控的进程名称和更新时间间隔,并调用上述函数实现监控功能。
4.实现原理
-
使用 psutil 库的
psutil.Process()
方法获取指定进程的详细信息,例如用户名、创建时间、线程数、CPU 占比、内存占比等。 -
使用 rich 库的
Table
类创建一个表格,使用add_column()
方法添加列,使用add_row()
方法添加行,并使用console.log()
方法在终端中输出表格。 -
在主程序中,设置一个循环,每隔一定时间获取一次指定进程的信息,并使用 rich 库在终端中显示。
5.工具整体功能介绍
-
进程信息获取功能:
- 使用
psutil.pids()
获取当前系统所有进程的 PID 列表。 - 遍历每个 PID,通过
psutil.Process(pid)
获取进程对象p
。 - 判断进程对象
p
的名称是否包含指定的进程名procName
,以确定目标进程。
- 使用
-
进程信息展示功能:
- 使用
rich
库中的Table
类创建一个表格,用于展示进程的详细信息。 - 将进程的关键信息(如用户名、创建时间、线程数、CPU 占比、内存占比等)添加到表格中的不同行中。
- 使用
console.log()
方法将表格和其他相关信息以美观的格式输出到终端。
- 使用
-
循环监控功能:
- 在
showMsg
函数中使用while True
循环,持续监控目标进程的状态。 - 每次循环中,重新获取目标进程的信息并更新展示在终端上的表格。
- 使用
time.sleep(timeInterval)
控制每次更新的时间间隔,避免过度频繁地查询和展示信息。
- 在
-
异常处理:
- 使用
try
和except
块捕获psutil.Process(pid)
可能抛出的异常,确保程序在获取进程信息时不会意外中断。
- 使用
-
主程序逻辑:
- 在主程序中,通过读取配置文件(例如
Config.py
)获取要监控的进程名称和更新时间间隔。 - 初始化
rich
库中的Console
对象,用于在终端上输出信息。 - 调用
showMsg
函数,传入要监控的进程名称和时间间隔,实现进程信息的持续监控和展示。
- 在主程序中,通过读取配置文件(例如
6.完整代码及配置如下
monitor.py
# -*- coding: gbk -*-
'''
Created on 2022年11月24日@author: DanMo
@status: done
'''
import time
import psutil
from rich.console import Console
from rich.table import Column, Table
from rich.progress import track
from Config import configdef getTargetProc(procName):pidList = psutil.pids()for pid in track(pidList, description="获取本机所有进程..."):try:p = psutil.Process(pid)except:continueif procName in p.name():return pdef showMsg(procName, timeInterval):while True:table = Table(show_header=True, header_style="bold magenta")table.add_column("项目", style='dim')table.add_column("详情", style='dim')p = getTargetProc(procName)table.add_row('用户', f'{p.username()}')table.add_row('进程创建时间', f'{p.create_time()}')table.add_row('线程数', f'{p.num_threads()}')table.add_row('父进程', f'{p.parent()}')table.add_row('当前进程状态', f'{p.status()}')table.add_row('被调用的命令行', f'{p.cmdline()}')table.add_row('exe', f'{p.exe()}')table.add_row('io', f'{p.io_counters()}')table.add_row('CPU占比', f'{p.cpu_percent()}')table.add_row('内存占比', f'{p.memory_percent()}')table.add_row('当前进程打开的句柄数', f'{p.num_handles()}')console.log('进程信息总览:\n', p.__dict__, style='dim')console.log('当前进程创建的所有线程:\n', p.threads(), style='dim')console.log('打开的文件:\n', p.open_files(), style='dim')console.log('子进程:\n', p.children(True), style='dim')console.log('进程所有连接信息[IPv4、IPv6、TCP、TCP over IPv4、TCP over IPv6、UDP、UDP over IPv4、UDP over IPv6、UNIX socket (both UDP and TCP protocols)]:\n',p.connections('all'), style='dim')if config.get_conf('envPath'):console.log('环境变量:\n', p.environ(), style='dim')console.log('内存信息:\n', p.memory_full_info(), style='dim')if config.get_conf('memorymap'):console.log('内存映射:\n', p.memory_maps(), style='dim')console.log('其他关于进程的信息:\n', style='dim')console.log(table)time.sleep(timeInterval)if __name__ == '__main__':console = Console()procName = config.get_conf('processname')timeInterval = config.get_conf('timeinterval')showMsg(procName, timeInterval)
Config.py
# -*- coding: gbk -*-
'''
Created on 2022年11月24日@author: DanMo
@status: done
'''
import os
import json
import threading
import configparserworkpath = os.path.dirname(os.path.abspath(__file__))
configFile = os.path.join(workpath, 'config.txt')
_lock = threading.Lock()class CConfig(object):'''实现对配置的操作本类中,配置文件为config.txt对外部提供的结构为:set_confget_confclean_confcleanall_conf'''def __init__(self, filename='config.txt'):self.filename = os.path.abspath(filename)self.config = configparser.ConfigParser(allow_no_value=True)# TODO:xiehs config逻辑,可以采取预先读取所有的方式,进行性能提高def set_conf(self, option, value=None, section='Monitor'):try:_lock.acquire()if not os.path.exists(self.filename):tmpf = open(self.filename, 'w')tmpf.close()if not self._is_exist(self.filename):raise Exception('CONFIG_File_Not_File')self.config.read(self.filename)if not self.config.has_section(section):self.config.add_section(section)self.config.set(section, option, json.dumps(value))with open(self.filename, 'w') as configfile:self.config.write(configfile)finally:_lock.release()def get_conf(self, option, section='Monitor'):if not os.path.exists(self.filename):return Noneif not self._is_exist(self.filename):raise Exception('CONFIG_File_Not_File')self.config.read(self.filename)if not self.config.has_section(section) or not self.config.has_option(section, option):return Noneval = self.config.get(section, option)val = json.loads(val)return valdef _is_exist(self, path):return os.path.isfile(path)config = CConfig(configFile)
config.txt
[Monitor]
processname = "YourAppName"
timeinterval = 4
memorymap = false
envPath = false
这篇关于Python教程:使用psutil和rich库实现一个终端资源监控小工具(附完整代码)的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!