Today, we’re crafting a handy tool that’s perfect for anyone curious about what’s going on behind the scenes of their computer. Imagine being able to peek at every process running on your machine and understanding how much of your computer’s resources they’re gobbling up.
In this tutorial, you’ll learn how to build your own Process Monitor in Python. This practical guide will take you through setting up a graphical user interface (GUI) that displays the processes currently running on your system, along with their CPU and memory usage and more. We’ll utilize key Python libraries such as psutil for accessing process details, pandas for managing the data efficiently, and tkinter for crafting the user interface.
Let’s get started!
Table of Contents
Final Result
Once you execute this code, you’ll see the result as shown in the image below. Simply click the refresh button after you run the code to see the results immediately.
I’ve run this code on Windows:
and also on Linux:
Necessary Libraries
Let’s get everything set up before we dive into the code part, so make sure to install these libraries using the terminal or your command prompt:
$ pip install tk
$ pip install psutil
$ pip install pandas
Imports
We start by importing psutil, which allows us to gather process information. Then we import datetime to handle date and time objects. Next, we import pandas for data manipulation and analysis.
After that, we import tkinter to create a graphical user interface that makes our program user-friendly. Finally, we import threading to run multiple tasks without overloading the main window.
import psutil
from datetime import datetime
import pandas as pd
import tkinter as tk
from tkinter import ttk
import threading
Process Monitor Functions
Now, let’s define our functions:
format_size Function
The objective of this function is to convert a size in bytes into a human-readable format such as kB, MB, GB, etc.
def format_size(bytes, suffix='B'):
"""
Scale bytes to its proper format
e.g:
1253656 => '1.20MB'
1253656678 => '1.17GB'
"""
factor = 1024
for unit in ["", "K", "M", "G", "T", "P"]:
if bytes < factor:
return f"{bytes:.2f}{unit}{suffix}"
bytes /= factor
collect_process_info Function
This function uses process.oneshot() to efficiently extract multiple attributes of a given process, such as the creation date with datetime, CPU usage with process.cpu_percent(), number of cores with len(), and memory usage with process.memory_full_info(). It returns all this information in a dictionary and also handles any exceptions that may occur during the process.
def collect_process_info(process):
"""
Collect detailed information about a given process.
"""
try:
with process.oneshot():
pid = process.pid
if pid == 0:
return None
name = process.name()
create_time = datetime.fromtimestamp(process.create_time())
cores = len(process.cpu_affinity())
cpu_usage = process.cpu_percent(interval=0.1)
status = process.status()
nice = process.nice()
memory_usage = process.memory_full_info().uss
io_counters = process.io_counters()
read_bytes, write_bytes = io_counters.read_bytes, io_counters.write_bytes
n_threads = process.num_threads()
username = process.username()
return {
'pid': pid, 'name': name, 'create_time': create_time,
'cores': cores, 'cpu_usage': cpu_usage, 'status': status, 'nice': nice,
'memory_usage': memory_usage, 'read_bytes': read_bytes, 'write_bytes': write_bytes,
'n_threads': n_threads, 'username': username
}
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
return None
get_process_info Function
This one calls the previous function for each process retrieved by psutil.process_iter() to gather information about all currently running processes on the system. It only includes entries in the returned list where information has been successfully obtained, omitting any processes that do not return valid data.
def get_processes_info():
"""
Get a list of all processes info.
"""
return [collect_process_info(p) for p in psutil.process_iter() if collect_process_info(p)]
construct_dataframe Function
The purpose of this one is to create a DataFrame from the processed information using pandas, and then sort the DataFrame by the appropriate column while ensuring that the format is readable.
def construct_dataframe(processes, sort_by='memory_usage', descending=True, columns=None):
"""
Construct dataframe from processes information.
"""
df = pd.DataFrame(processes)
df.set_index('pid', inplace=True)
df.sort_values(by=sort_by, inplace=True, ascending=not descending)
if columns:
df = df[columns]
df['memory_usage'] = df['memory_usage'].apply(format_size)
df['read_bytes'] = df['read_bytes'].apply(format_size)
df['write_bytes'] = df['write_bytes'].apply(format_size)
df['create_time'] = df['create_time'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))
return df
create_widgets Function
This function initializes and arranges user interface widgets within a provided root window. It begins by adding a Treeview widget (tree), configuring it to expand and fill both horizontal and vertical space within its container. It then sets up the headings for each column based on the column_config list, using the column names converted to title case for the headings.
Additionally, it creates a “Refresh” button positioned at the bottom of the window. This button is configured to call a refresh() function, which updates the Treeview contents, whenever clicked. The button stretches across the full width of the window.
def create_widgets(root, tree, column_config):
tree.pack(expand=True, fill=tk.BOTH)
for col in column_config:
tree.heading(col, text=col.title())
refresh_button = ttk.Button(root, text="Refresh", command=lambda: refresh(tree, column_config, refresh_button))
refresh_button.pack(side=tk.BOTTOM, fill=tk.X)
refresh Function
It starts a new thread to execute the update_processes() function and disables the refresh button to prevent multiple simultaneous refresh attempts.
def refresh(tree, column_config, button):
button.config(state="disabled")
threading.Thread(target=update_processes, args=(tree, column_config, button)).start()
update_processes Function
This function retrieves new process information using the get_process_info() function, constructs a new DataFrame using pandas, and then updates the Treeview widget to display this new information.
def update_processes(tree, column_config, button):
processes = get_processes_info()
df = construct_dataframe(processes, columns=column_config)
# Update the treeview in the main thread
tree.after(0, populate_treeview, tree, df, button)
populate_treeview Function
The last one processes all the data currently displayed on the treeview widget and clears it using tree.delete(). Then, it inserts entries from the newly retrieved DataFrame using tree.insert(), and after these updates, it returns the refresh button to its normal state from its previously disabled state.
def populate_treeview(tree, df, button):
# Clear existing entries
for i in tree.get_children():
tree.delete(i)
# Insert new entries
for row in df.itertuples():
tree.insert("", tk.END, values=row[1:]) # Skip the index
button.config(state="normal")
Main Block
Lastly, this part of the script ensures that it can only be run directly and not imported as a module. It initializes the main window, sets its title, and defines its geometry. Additionally, it specifies the column configuration, which serves as headings for the columns in the treeview widget. The treeview widget is then created, and the column headings are displayed using show="headings".
After setting up these GUI elements using the create_widgets() function, the script starts the main event loop to keep the main window running and responsive to user interactions until it is exited willingly.
if __name__ == "__main__":
root = tk.Tk()
root.title("Process Monitor - The Pycodes")
root.geometry("1400x600")
column_config = "name,cpu_usage,memory_usage,read_bytes,write_bytes,status,create_time,nice,n_threads,cores".split(
",")
tree = ttk.Treeview(root, columns=column_config, show="headings")
create_widgets(root, tree, column_config)
root.mainloop()
Full Code
import psutil
from datetime import datetime
import pandas as pd
import tkinter as tk
from tkinter import ttk
import threading
def format_size(bytes, suffix='B'):
"""
Scale bytes to its proper format
e.g:
1253656 => '1.20MB'
1253656678 => '1.17GB'
"""
factor = 1024
for unit in ["", "K", "M", "G", "T", "P"]:
if bytes < factor:
return f"{bytes:.2f}{unit}{suffix}"
bytes /= factor
def collect_process_info(process):
"""
Collect detailed information about a given process.
"""
try:
with process.oneshot():
pid = process.pid
if pid == 0:
return None
name = process.name()
create_time = datetime.fromtimestamp(process.create_time())
cores = len(process.cpu_affinity())
cpu_usage = process.cpu_percent(interval=0.1)
status = process.status()
nice = process.nice()
memory_usage = process.memory_full_info().uss
io_counters = process.io_counters()
read_bytes, write_bytes = io_counters.read_bytes, io_counters.write_bytes
n_threads = process.num_threads()
username = process.username()
return {
'pid': pid, 'name': name, 'create_time': create_time,
'cores': cores, 'cpu_usage': cpu_usage, 'status': status, 'nice': nice,
'memory_usage': memory_usage, 'read_bytes': read_bytes, 'write_bytes': write_bytes,
'n_threads': n_threads, 'username': username
}
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
return None
def get_processes_info():
"""
Get a list of all processes info.
"""
return [collect_process_info(p) for p in psutil.process_iter() if collect_process_info(p)]
def construct_dataframe(processes, sort_by='memory_usage', descending=True, columns=None):
"""
Construct dataframe from processes information.
"""
df = pd.DataFrame(processes)
df.set_index('pid', inplace=True)
df.sort_values(by=sort_by, inplace=True, ascending=not descending)
if columns:
df = df[columns]
df['memory_usage'] = df['memory_usage'].apply(format_size)
df['read_bytes'] = df['read_bytes'].apply(format_size)
df['write_bytes'] = df['write_bytes'].apply(format_size)
df['create_time'] = df['create_time'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))
return df
def create_widgets(root, tree, column_config):
tree.pack(expand=True, fill=tk.BOTH)
for col in column_config:
tree.heading(col, text=col.title())
refresh_button = ttk.Button(root, text="Refresh", command=lambda: refresh(tree, column_config, refresh_button))
refresh_button.pack(side=tk.BOTTOM, fill=tk.X)
def refresh(tree, column_config, button):
button.config(state="disabled")
threading.Thread(target=update_processes, args=(tree, column_config, button)).start()
def update_processes(tree, column_config, button):
processes = get_processes_info()
df = construct_dataframe(processes, columns=column_config)
# Update the treeview in the main thread
tree.after(0, populate_treeview, tree, df, button)
def populate_treeview(tree, df, button):
# Clear existing entries
for i in tree.get_children():
tree.delete(i)
# Insert new entries
for row in df.itertuples():
tree.insert("", tk.END, values=row[1:]) # Skip the index
button.config(state="normal")
if __name__ == "__main__":
root = tk.Tk()
root.title("Process Monitor - The Pycodes")
root.geometry("1400x600")
column_config = "name,cpu_usage,memory_usage,read_bytes,write_bytes,status,create_time,nice,n_threads,cores".split(
",")
tree = ttk.Treeview(root, columns=column_config, show="headings")
create_widgets(root, tree, column_config)
root.mainloop()
Happy Coding!
Subscribe for Top Free Python Tutorials!
Receive the best directly. Elevate Your Coding Journey!
