Python-блог: wxPython: Создаём ваш собственный кросс-платформенный монитор процессов при помощи psutil (Перевод)

На этой неделе я столкнулся с интересным проектом на Python под названием psutil на Google Code. Он работает на Linux, Windows, OSX и FreeBSD. Что он делает? Он собирает все работающие процессы и выдаёт Вам информацию о них, предоставляя так же возможность их завершения. Неплохо, подумал я, сделать для него GUI и получить собственный диспетчер задач / монитор приложений на wxPython. Если у Вас есть время - я приглашаю Вас в путешествие по 4 итерациям моего кода.

Прототип

Моя первая версия всего лишь показывает что именно запущенно на данный момент и использует wx.Timer для обновления информации каждые пять секунд. Я использовал виджет ObjectListView для отображения данных, который на данный момент не включён в wxPython, так что Вам необходимо его установить, если Вы хотите запустить мой код.

import psutil
import wx
from ObjectListView import ObjectListView, ColumnDefn
 
########################################################################class Process(object):
 """ """#----------------------------------------------------------------------def__init__(self, name, pid, exe, user, cpu, mem, desc=None):
 """Конструктор"""self.name = name
 self.pid = pid
 self.exe = exe
 self.user = userself.cpu = cpu
 self.mem = mem
 #self.desc = desc########################################################################class MainPanel(wx.Panel):
 """"""#----------------------------------------------------------------------def__init__(self, parent):
 """Конструктор"""
 wx.Panel.__init__(self, parent=parent)self.procs = []self.procmonOlv = ObjectListView(self, style=wx.LC_REPORT|wx.SUNKEN_BORDER)self.setProcs()
 
 mainSizer = wx.BoxSizer(wx.VERTICAL)
 mainSizer.Add(self.procmonOlv, 1, wx.EXPAND|wx.ALL, 5)self.SetSizer(mainSizer)self.updateDisplay()# обновлять информацию каждые 5 секундself.timer = wx.Timer(self)self.Bind(wx.EVT_TIMER, self.update, self.timer)self.timer.Start(5000)#----------------------------------------------------------------------def setProcs(self):
 """"""
 cols = [
 ColumnDefn("name", "left", 150, "name"),
 ColumnDefn("pid", "left", 50, "pid"),
 ColumnDefn("exe location", "left", 100, "exe"),
 ColumnDefn("username", "left", 75, "user"),
 ColumnDefn("cpu", "left", 75, "cpu"),
 ColumnDefn("mem", "left", 75, "mem"),
 #ColumnDefn("description", "left", 200, "desc")]self.procmonOlv.SetColumns(cols)self.procmonOlv.SetObjects(self.procs)#----------------------------------------------------------------------def update(self, event):
 """"""self.updateDisplay()#----------------------------------------------------------------------def updateDisplay(self):
 """"""
 pids = psutil.get_pid_list()for pid in pids:
 
 try:
 p = psutil.Process(pid)
 new_proc = Process(p.name,
 str(p.pid),
 p.exe,
 p.username,
 str(p.get_cpu_percent()),
 str(p.get_memory_percent()))self.procs.append(new_proc)except:
 passself.setProcs()########################################################################class MainFrame(wx.Frame):
 """"""#----------------------------------------------------------------------def__init__(self):
 """Конструктор"""
 wx.Frame.__init__(self, None, title="PyProcMon")
 panel = MainPanel(self)self.Show()if __name__ == "__main__":
 app = wx.App(False)
 frame = MainFrame()
 app.MainLoop()

Тут всё понятно и без моих объяснений. Но вот получение списка процессов каждые пять секунд приостанавливает работу GUI, что несколько нервирует. Поэтому давайте добавим сюда нити, для выполнения этого в фоне.

Добавление нитей (Threading) - Alpha 2

В этой версии мы добавили нити и pubsub для облегчения передачи информации из нити в GUI. Обратите внимание, что мы так же должны использовать wx.CallAfter для вызова pubsub, так как pubsub не thread-safe.

import psutil # http://code.google.com/p/psutil/import wx
 
from ObjectListView import ObjectListView, ColumnDefn
fromthreadingimport Thread
from wx.lib.pubsubimport Publisher
 
########################################################################class ProcThread(Thread):
 """
 Получает всю информацию о процессах, которая нам нужна, так как psutil не очень скор
 """#----------------------------------------------------------------------def__init__(self):
 """Constructor"""
 Thread.__init__(self)self.start()#----------------------------------------------------------------------def run(self):
 """"""
 pids = psutil.get_pid_list()
 procs = []for pid in pids:
 try:
 p = psutil.Process(pid)
 new_proc = Process(p.name,
 str(p.pid),
 p.exe,
 p.username,
 str(p.get_cpu_percent()),
 str(p.get_memory_percent()))
 procs.append(new_proc)except:
 print"Error getting pid #%s information"% pid
 
 # посылаем pid'ы в GUI
 wx.CallAfter(Publisher().sendMessage, "update", procs)########################################################################class Process(object):
 """
 Определение модели Process для ObjectListView
 """#----------------------------------------------------------------------def__init__(self, name, pid, exe, user, cpu, mem, desc=None):
 """Constructor"""self.name = name
 self.pid = pid
 self.exe = exe
 self.user = userself.cpu = cpu
 self.mem = mem
 #self.desc = desc########################################################################class MainPanel(wx.Panel):
 """"""#----------------------------------------------------------------------def__init__(self, parent):
 """Constructor"""
 wx.Panel.__init__(self, parent=parent)self.procs = []self.procmonOlv = ObjectListView(self, style=wx.LC_REPORT|wx.SUNKEN_BORDER)self.setProcs()
 
 mainSizer = wx.BoxSizer(wx.VERTICAL)
 mainSizer.Add(self.procmonOlv, 1, wx.EXPAND|wx.ALL, 5)self.SetSizer(mainSizer)# проверяем обновления каждые пять минутself.timer = wx.Timer(self)self.Bind(wx.EVT_TIMER, self.update, self.timer)self.timer.Start(15000)self.setProcs()# создаём получателя для pubsub
 Publisher().subscribe(self.updateDisplay, "update")#----------------------------------------------------------------------def setProcs(self):
 """"""
 cols = [
 ColumnDefn("name", "left", 150, "name"),
 ColumnDefn("pid", "left", 50, "pid"),
 ColumnDefn("exe location", "left", 100, "exe"),
 ColumnDefn("username", "left", 75, "user"),
 ColumnDefn("cpu", "left", 75, "cpu"),
 ColumnDefn("mem", "left", 75, "mem"),
 #ColumnDefn("description", "left", 200, "desc")]self.procmonOlv.SetColumns(cols)self.procmonOlv.SetObjects(self.procs)self.procmonOlv.sortAscending = True#----------------------------------------------------------------------def update(self, event):
 """
 Запускаем поток для получения pid информации
 """self.timer.Stop()
 ProcThread()#----------------------------------------------------------------------def updateDisplay(self, msg):
 """"""self.procs = msg.dataself.setProcs()ifnotself.timer.IsRunning():
 self.timer.Start(15000)########################################################################class MainFrame(wx.Frame):
 """"""#----------------------------------------------------------------------def__init__(self):
 """Constructor"""
 wx.Frame.__init__(self, None, title="PyProcMon")
 panel = MainPanel(self)self.Show()if __name__ == "__main__":
 app = wx.App(False)
 frame = MainFrame()
 app.MainLoop()

Кроме того, мы увеличили интервал обновления до 15. Я сделал это, потому что иначе информация обновлялась слишком быстро и я не успевал хорошо рассмотреть список до того, как он обновлялся. На этом месте я заметил, что я не могу изменить размер колонок один раз и до конца, так как он каждый раз возвращался к исходному размеру после обновления. Кроме того, мне захотелось, чтобы приложение следило за тем, как я отсортировал колонки и какой процесс был выбран последним. И, наконец, мне нужна возможность убивать процессы (xD - прим. пер.)

Шаг 3: Добавляем базовую функциональность

Итак, на нашей третьей итерации мы добавили все эти возможности:

import psutil # http://code.google.com/p/psutil/import wx
 
from ObjectListView import ObjectListView, ColumnDefn
fromthreadingimport Thread
from wx.lib.pubsubimport Publisher
 
########################################################################class ProcThread(Thread):
 """
 Получает всю нужную информацию о процессах, так как psutil не быстр
 """#----------------------------------------------------------------------def__init__(self):
 """Constructor"""
 Thread.__init__(self)self.start()#----------------------------------------------------------------------def run(self):
 """"""
 pids = psutil.get_pid_list()
 procs = []for pid in pids:
 try:
 p = psutil.Process(pid)
 new_proc = Process(p.name,
 str(p.pid),
 p.exe,
 p.username,
 str(p.get_cpu_percent()),
 str(p.get_memory_percent()))
 procs.append(new_proc)except:
 pass# посылаем pid'ы в GUI
 wx.CallAfter(Publisher().sendMessage, "update", procs)########################################################################class Process(object):
 """
 Определение модели Process для ObjectListView
 """#----------------------------------------------------------------------def__init__(self, name, pid, exe, user, cpu, mem, desc=None):
 """Constructor"""self.name = name
 self.pid = pid
 self.exe = exe
 self.user = userself.cpu = cpu
 self.mem = mem
 #self.desc = desc########################################################################class MainPanel(wx.Panel):
 """"""#----------------------------------------------------------------------def__init__(self, parent):
 """Constructor"""
 wx.Panel.__init__(self, parent=parent)self.currentSelection = Noneself.gui_shown = Falseself.procs = []self.sort_col = 0self.col_w = {"name":175,
 "pid":50,
 "exe":300,
 "user":175,
 "cpu":60,
 "mem":75}self.procmonOlv = ObjectListView(self, style=wx.LC_REPORT|wx.SUNKEN_BORDER)self.procmonOlv.Bind(wx.EVT_LIST_COL_CLICK, self.onColClick)self.procmonOlv.Bind(wx.EVT_LIST_ITEM_SELECTED, self.onSelect)#self.procmonOlv.Selectself.setProcs()
 
 endProcBtn = wx.Button(self, label="End Process")
 endProcBtn.Bind(wx.EVT_BUTTON, self.onKillProc)
 
 mainSizer = wx.BoxSizer(wx.VERTICAL)
 mainSizer.Add(self.procmonOlv, 1, wx.EXPAND|wx.ALL, 5)
 mainSizer.Add(endProcBtn, 0, wx.ALIGN_RIGHT|wx.ALL, 5)self.SetSizer(mainSizer)# обновляем информацию каждые 15 секундself.timer = wx.Timer(self)self.Bind(wx.EVT_TIMER, self.update, self.timer)self.update("")self.setProcs()# создаём получателя для pubsub
 Publisher().subscribe(self.updateDisplay, "update")#----------------------------------------------------------------------def onColClick(self, event):
 """
 Запоминаем, по какой колонке была сортировка, пока только по возрастанию
 """self.sort_col = event.GetColumn()#----------------------------------------------------------------------def onKillProc(self, event):
 """
 Убиваем выбранный процесс по pid
 """
 obj = self.procmonOlv.GetSelectedObject()print
 pid = int(obj.pid)try:
 p = psutil.Process(pid)
 p.terminate()self.update("")exceptException, e:
 print"Error: " + e
 
 #----------------------------------------------------------------------def onSelect(self, event):
 """"""
 item = event.GetItem()
 itemId = item.GetId()self.currentSelection = itemId
 print#----------------------------------------------------------------------def setProcs(self):
 """"""
 cw = self.col_w# изменяем ширину колонки, если необходимоifself.gui_shown:
 cw["name"] = self.procmonOlv.GetColumnWidth(0)
 cw["pid"] = self.procmonOlv.GetColumnWidth(1)
 cw["exe"] = self.procmonOlv.GetColumnWidth(2)
 cw["user"] = self.procmonOlv.GetColumnWidth(3)
 cw["cpu"] = self.procmonOlv.GetColumnWidth(4)
 cw["mem"] = self.procmonOlv.GetColumnWidth(5)
 
 cols = [
 ColumnDefn("name", "left", cw["name"], "name"),
 ColumnDefn("pid", "left", cw["pid"], "pid"),
 ColumnDefn("exe location", "left", cw["exe"], "exe"),
 ColumnDefn("username", "left", cw["user"], "user"),
 ColumnDefn("cpu", "left", cw["cpu"], "cpu"),
 ColumnDefn("mem", "left", cw["mem"], "mem"),
 #ColumnDefn("description", "left", 200, "desc")]self.procmonOlv.SetColumns(cols)self.procmonOlv.SetObjects(self.procs)self.procmonOlv.SortBy(self.sort_col)ifself.currentSelection:
 self.procmonOlv.Select(self.currentSelection)self.procmonOlv.SetFocus()self.gui_shown = True#----------------------------------------------------------------------def update(self, event):
 """
 Запускаем поток для получения pid информации
 """print"update thread started!"self.timer.Stop()
 ProcThread()#----------------------------------------------------------------------def updateDisplay(self, msg):
 """"""print"thread done, updating display!"self.procs = msg.dataself.setProcs()ifnotself.timer.IsRunning():
 self.timer.Start(15000)########################################################################class MainFrame(wx.Frame):
 """"""#----------------------------------------------------------------------def__init__(self):
 """Constructor"""
 wx.Frame.__init__(self, None, title="PyProcMon", size=(1024, 768))
 panel = MainPanel(self)self.Show()#----------------------------------------------------------------------if __name__ == "__main__":
 app = wx.App(False)
 frame = MainFrame()
 app.MainLoop()

Вы можете заметить что мы перехватываем несколько событий для того, чтобы отслеживать сортировку по колонкам и выделение процесса. Я пока не придумал, как выяснить направление сортировки или как его изменить, так что это остаётся в моём TODO. И всё же, есть ещё одна вещь, которую я хотел бы добавить - панель статуса с информацией о количестве процессов, загрузке CPU и использованию памяти.

Итог: PyProcMon

Для нашей итоговой версии (по крайней мере, на данный момент), мы добавили панель статуса, разделённую на три части, и ещё один получатель / отправитель pubsub. Кроме того, мы выделили некоторые куски программы в отдельные модули. Код потока выделен в controller.py, класс Process в model.py. Начнём с контроллера:

# controller.py########################################################################import psutil
import wx
 
from model import Process
fromthreading import Thread
from wx.lib.pubsubimport Publisher
 
########################################################################class ProcThread(Thread):
 """
 Получает информацию о процессах, так как psutil не быстр
 """#----------------------------------------------------------------------def__init__(self):
 """Constructor"""
 Thread.__init__(self)self.start()#----------------------------------------------------------------------def run(self):
 """"""
 pids = psutil.get_pid_list()
 procs = []
 cpu_percent = 0
 mem_percent = 0for pid in pids:
 try:
 p = psutil.Process(pid)
 cpu = p.get_cpu_percent()
 mem = p.get_memory_percent()
 new_proc = Process(p.name,
 str(p.pid),
 p.exe,
 p.username,
 str(cpu),
 str(mem))
 procs.append(new_proc)
 cpu_percent += cpu
 mem_percent += mem
 except:
 pass# посылаем pid'ы в GUI
 wx.CallAfter(Publisher().sendMessage, "update", procs)
 
 number_of_procs = len(procs)
 wx.CallAfter(Publisher().sendMessage, "update_status",
 (number_of_procs, cpu_percent, mem_percent))

You’ve already seen this, so let’s move on to the model:

# model.py########################################################################class Process(object):
 """
 Определение модели Process для ObjectListView
 """#----------------------------------------------------------------------def__init__(self, name, pid, exe, user, cpu, mem, desc=None):
 """Constructor"""self.name = name
 self.pid = pid
 self.exe = exe
 self.user = userself.cpu = cpu
 self.mem = mem

Очень просто! Обратите внимание, что нам не нужно ничего импортировать в этот модуль. Теперь посмотрим на основной код:

# pyProcMon.py
import controller
import psutil # http://code.google.com/p/psutil/
import wx
 
from ObjectListView import ObjectListView, ColumnDefn
from wx.lib.pubsub import Publisher
 
########################################################################
class MainPanel(wx.Panel):
 """"""
 
 #----------------------------------------------------------------------
 def __init__(self, parent):
 """Constructor"""
 wx.Panel.__init__(self, parent=parent)
 self.currentSelection = None
 self.gui_shown = False
 self.procs = []
 self.sort_col = 0
 
 self.col_w = {"name":175,
 "pid":50,
 "exe":300,
 "user":175,
 "cpu":60,
 "mem":75}
 
 self.procmonOlv = ObjectListView(self, style=wx.LC_REPORT|wx.SUNKEN_BORDER)
 self.procmonOlv.Bind(wx.EVT_LIST_COL_CLICK, self.onColClick)
 self.procmonOlv.Bind(wx.EVT_LIST_ITEM_SELECTED, self.onSelect)
 #self.procmonOlv.Select
 self.setProcs()
 
 endProcBtn = wx.Button(self, label="End Process")
 endProcBtn.Bind(wx.EVT_BUTTON, self.onKillProc)
 
 mainSizer = wx.BoxSizer(wx.VERTICAL)
 mainSizer.Add(self.procmonOlv, 1, wx.EXPAND|wx.ALL, 5)
 mainSizer.Add(endProcBtn, 0, wx.ALIGN_RIGHT|wx.ALL, 5)
 self.SetSizer(mainSizer)
 
 # обновляем информацию каждые 15 секунд
 self.timer = wx.Timer(self)
 self.Bind(wx.EVT_TIMER, self.update, self.timer)
 self.update("")
 self.setProcs()
 
 # создаём получатель для pubsub
 Publisher().subscribe(self.updateDisplay, "update")
 
 #----------------------------------------------------------------------
 def onColClick(self, event):
 """
 Запоминаем, какая колонка была отсортирована. Пока только по возрастанию
 """
 self.sort_col = event.GetColumn()
 
 #----------------------------------------------------------------------
 def onKillProc(self, event):
 """
 Убиваем выбранный процесс по pid
 """
 obj = self.procmonOlv.GetSelectedObject()
 print
 pid = int(obj.pid)
 try:
 p = psutil.Process(pid)
 p.terminate()
 self.update("")
 except Exception, e:
 print "Error: " + e
 
 #----------------------------------------------------------------------
 def onSelect(self, event):
 """
 Вызывается при выборе элемента и помогает его отслеживать
 """
 item = event.GetItem()
 itemId = item.GetId()
 self.currentSelection = itemId
 
 #----------------------------------------------------------------------
 def setProcs(self):
 """
 Обновляет виджет ObjectListView
 """
 cw = self.col_w
 # изменяем ширину колонок на выбранную
 if self.gui_shown:
 cw["name"] = self.procmonOlv.GetColumnWidth(0)
 cw["pid"] = self.procmonOlv.GetColumnWidth(1)
 cw["exe"] = self.procmonOlv.GetColumnWidth(2)
 cw["user"] = self.procmonOlv.GetColumnWidth(3)
 cw["cpu"] = self.procmonOlv.GetColumnWidth(4)
 cw["mem"] = self.procmonOlv.GetColumnWidth(5)
 
 cols = [
 ColumnDefn("name", "left", cw["name"], "name"),
 ColumnDefn("pid", "left", cw["pid"], "pid"),
 ColumnDefn("exe location", "left", cw["exe"], "exe"),
 ColumnDefn("username", "left", cw["user"], "user"),
 ColumnDefn("cpu", "left", cw["cpu"], "cpu"),
 ColumnDefn("mem", "left", cw["mem"], "mem"),
 #ColumnDefn("description", "left", 200, "desc")
 ]
 self.procmonOlv.SetColumns(cols)
 self.procmonOlv.SetObjects(self.procs)
 self.procmonOlv.SortBy(self.sort_col)
 if self.currentSelection:
 self.procmonOlv.Select(self.currentSelection)
 self.procmonOlv.SetFocus()
 self.gui_shown = True
 
 #----------------------------------------------------------------------
 def update(self, event):
 """
 Запускаем поток для получения информации pid
 """
 print "update thread started!"
 self.timer.Stop()
 controller.ProcThread()
 
 #----------------------------------------------------------------------
 def updateDisplay(self, msg):
 """
 Ловим сообщения pubsub из потока и обновляем информацию на экране
 """
 print "thread done, updating display!"
 self.procs = msg.data
 self.setProcs()
 if not self.timer.IsRunning():
 self.timer.Start(15000)
 
########################################################################
class MainFrame(wx.Frame):
 """"""
 
 #----------------------------------------------------------------------
 def __init__(self):
 """Constructor"""
 wx.Frame.__init__(self, None, title="PyProcMon", size=(1024, 768))
 panel = MainPanel(self)
 
 # set up the statusbar
 self.CreateStatusBar()
 self.StatusBar.SetFieldsCount(3)
 self.StatusBar.SetStatusWidths([200, 200, 200])
 
 # создаём получателя pubsub
 Publisher().subscribe(self.updateStatusbar, "update_status")
 
 self.Show()
 
 #----------------------------------------------------------------------
 def updateStatusbar(self, msg):
 """"""
 procs, cpu, mem = msg.data
 self.SetStatusText("Processes: %s" % procs, 0)
 self.SetStatusText("CPU Usage: %s" % cpu, 1)
 self.SetStatusText("Physical Memory: %s" % mem, 2)
 
 
#----------------------------------------------------------------------
if __name__ == "__main__":
 app = wx.App(False)
 frame = MainFrame()
 app.MainLoop()

Главное, что сюда добавлено - это панель статуса и механизм её обновления. Пришлось с ней повозиться, но в итоге она обновляется вместе с экраном.

Итоги

Вы может быть думаете, почему информация о процессах собирается в выражении try/except. Ну, некоторые процессы не особо хотят делиться своей информацией или могут попытаться пропасть между тем, как я получаю список процессов и тем, как я получаю о нём информацию, так что лучше перестраховаться. На самом деле, таких процессов МНОГО. Кроме того, я так же обернул попытку убить процесс обработчиком исключений, так как не все процессы могут быть убиты. А так, всё работает замечательно. Вот ещё несколько вещей, которые стоило бы добавить: контекстное меню в ответ на клик правой кнопкой мыши, диалог подтверждения, меню с некоторыми опциями (закрыть, запустить, о программе).

Я надеюсь, Вы получили удовольствие в процессе нашего путешествия и узнали что-то новое. Happy hacking!

Исходники

psutil_gui.zip

Источник

UPD есть похожий проект тут

4 комментария:

postcoder 27 августа 2012 г. в 21:50
спасибо все получилось!

Ответить Удалить
Ответы
Hiropter 21 мая 2023 г. в 14:39
Как скачать библиотеку wx? При попытке установки в pip выдаёт ERROR: Could not find a version that satisfies the requirement wx (from versions: none)
ERROR: No matching distribution found for wx
Ответить Удалить
Ответы

Добавить комментарий

Ещё

[フレーム]

понедельник, 16 июля 2012 г.

wxPython: Создаём ваш собственный кросс-платформенный монитор процессов при помощи psutil (Перевод)