На этой неделе я столкнулся с интересным проектом на 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 = user self.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: pass self.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 from threading import Thread from wx.lib.pubsub import 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 = user self.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.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") 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 from threading import Thread from wx.lib.pubsub import 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 = user self.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 = 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 print #---------------------------------------------------------------------- def setProcs(self): """""" 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() ProcThread() #---------------------------------------------------------------------- def updateDisplay(self, msg): """""" 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) 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 from threading import Thread from wx.lib.pubsub import 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 = 0 for 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 = user self.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!
спасибо все получилось!
ОтветитьУдалить)) пожалуйста
УдалитьКак скачать библиотеку wx? При попытке установки в pip выдаёт ERROR: Could not find a version that satisfies the requirement wx (from versions: none)
ОтветитьУдалитьERROR: No matching distribution found for wx
ну так она же называется не wx, a wxpython
Удалитьpip install wxPython
https://pypi.org/project/wxPython/
если конечно я ничего не путаю