Show processes lifecycle in real time

live_processes.py

#!/usr/bin/env python3

# Simple monitoring script to print processes being created/stopped in real-time

import time
from typing import Optional
import psutil
import shutil
import sys
import argparse
import re
from dataclasses import dataclass

parser = argparse.ArgumentParser(description='Live process lifecycle monitoring')
parser.add_argument('-d', '--delay', type=float, default=1, help='Delay in seconds between probes')
parser.add_argument('-e', '--exclude', default=[], action='append', help='Exclude processes. Interpreted as regex for name and full text for PID')
parser.add_argument('--ppid', action='store_true', help='Show PPID')
args = parser.parse_args()

def set_cursor(y: int, x: int):
    sys.stdout.write("\033[%d;%dH" % (y+1, x+1))

@dataclass
class ProcessData:
    cmd: str
    pid: int
    ppid: Optional[int]

    def string(self) -> str:
        s = f"{self.pid} {self.cmd}"
        if args.ppid:
            if self.ppid:
                s = f"{self.ppid} {s}"
            else:
                s = f"None {s}"
        return s

def filter_pass(process: ProcessData) -> bool:
    for exclude in args.exclude:
        if str(process.pid) == exclude:
            return False
        if re.search(exclude, process.cmd):
            return False
    return True


def get_running_processes(filter: bool = True) -> list[ProcessData]:
    processes = []
    for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
        if proc.info['cmdline']:
            process = ProcessData(proc.info['cmdline'][0], proc.pid, proc.ppid())
            if filter and not filter_pass(process):
                continue
            processes.append(process)

    return processes

def draw_left(stopped_processes: list[ProcessData]):
    set_cursor(0, 0)
    print("Stopped Processes:")
    for process in stopped_processes:
        print(process.string())

RIGHT = shutil.get_terminal_size().columns /2
def draw_right(created_processes: list[ProcessData]):
    set_cursor(0, RIGHT)
    print("Created Processes:")
    for i, process in enumerate(created_processes, start=1):
        set_cursor(i, RIGHT)
        print(process.string())

def clear():
    print(chr(27) + "[2J")

print("Waiting for second probe..")
print("Delay between probe is %d seconds" % args.delay)
print("Show ppid:", args.ppid)
print("Excludes:", args.exclude)
running_processes = get_running_processes()
while True:
    # Wait before next processes probe
    time.sleep(args.delay)

    # Get the updated list of running processes
    updated_processes = get_running_processes()

    # Process to get the wanted data
    stopped_processes = [p for p in running_processes if p not in updated_processes]
    created_processes = [p for p in updated_processes if p not in running_processes]

    # Print to screen
    clear()
    draw_left(stopped_processes)
    draw_right(created_processes)

    # Update list
    running_processes = updated_processes