Python Scripts for evaluation & diagnosing : nettop.py & pmap.py

kitty terminal detached window for nettop.py :

 

#!/bin/bash

kitty -o font_size=11 -o initial_window_width=65c -o initial_window_height=26c --directory /home/c0ntrol/apps/psutil_scripts python3 nettop.py --detach

nettop.py full code review:

 

#!/usr/bin/env python3
# nettop.py
# bryce_polymorph@proton.me $


import sys
import time


try:
    import curses
except ImportError:
    sys.exit('platform not supported')

import psutil
from psutil._common import bytes2human


lineno = 0
win = curses.initscr()


def printl(line, highlight=False):
    """A thin wrapper around curses's addstr()."""
    global lineno
    try:
        if highlight:
            line += " " * (win.getmaxyx()[1] - len(line))
            win.addstr(lineno, 0, line, curses.A_REVERSE)
        else:
            win.addstr(lineno, 0, line, 0)
    except curses.error:
        lineno = 0
        win.refresh()
        raise
    else:
        lineno += 1


def poll(interval):
    """Retrieve raw stats within an interval window."""
    tot_before = psutil.net_io_counters()
    pnic_before = psutil.net_io_counters(pernic=True)
    # sleep some time
    time.sleep(interval)
    tot_after = psutil.net_io_counters()
    pnic_after = psutil.net_io_counters(pernic=True)
    return (tot_before, tot_after, pnic_before, pnic_after)


def refresh_window(tot_before, tot_after, pnic_before, pnic_after):
    """Print stats on screen."""
    global lineno

    # totals
    printl(
        "total bytes:           sent: %-10s   received: %s"
        % (
            bytes2human(tot_after.bytes_sent),
            bytes2human(tot_after.bytes_recv),
        )
    )

    # per-network interface details: let's sort network interfaces so
    # that the ones which generated more traffic are shown first
    printl("")
    nic_names = list(pnic_after.keys())
    nic_names.sort(key=lambda x: sum(pnic_after[x]), reverse=True)
    for name in nic_names:
        stats_before = pnic_before[name]
        stats_after = pnic_after[name]
        templ = "%-15s %15s %15s"
        # fmt: off
        printl(templ % (name, "TOTAL", "PER-SEC"), highlight=True)
        printl(templ % (
            "bytes-sent",
            bytes2human(stats_after.bytes_sent),
            bytes2human(
                stats_after.bytes_sent - stats_before.bytes_sent) + '/s',
        ))
        printl(templ % (
            "bytes-recv",
            bytes2human(stats_after.bytes_recv),
            bytes2human(
                stats_after.bytes_recv - stats_before.bytes_recv) + '/s',
        ))
        printl(templ % (
            "pkts-sent",
            stats_after.packets_sent,
            stats_after.packets_sent - stats_before.packets_sent,
        ))
        printl(templ % (
            "pkts-recv",
            stats_after.packets_recv,
            stats_after.packets_recv - stats_before.packets_recv,
        ))
        printl("")
        # fmt: on
    win.refresh()
    lineno = 0


def setup():
    curses.start_color()
    curses.use_default_colors()
    for i in range(curses.COLORS):
        curses.init_pair(i + 1, i, -1)
    curses.endwin()
    win.nodelay(1)


def tear_down():
    win.keypad(0)
    curses.nocbreak()
    curses.echo()
    curses.endwin()


def main():
    setup()
    try:
        interval = 0
        while True:
            if win.getch() == ord('q'):
                break
            args = poll(interval)
            refresh_window(*args)
            interval = 0.5
    except (KeyboardInterrupt, SystemExit):
        pass
    finally:
        tear_down()


if __name__ == '__main__':
    main() 

 

 

 

Full Code Review : pmap.py

 

#!/usr/bin/env python3 

import sys

import psutil
from psutil._common import bytes2human
from psutil._compat import get_terminal_size


def safe_print(s):
    s = s[: get_terminal_size()[0]]
    try:
        print(s)
    except UnicodeEncodeError:
        print(s.encode('ascii', 'ignore').decode())


def main():
    if len(sys.argv) != 2:
        sys.exit('usage: pmap <pid>')
    p = psutil.Process(int(sys.argv[1]))
    templ = "%-20s %10s  %-7s %s"
    print(templ % ("Address", "RSS", "Mode", "Mapping"))
    total_rss = 0
    for m in p.memory_maps(grouped=False):
        total_rss += m.rss
        line = templ % (
            m.addr.split('-')[0].zfill(16),
            bytes2human(m.rss),
            m.perms,
            m.path,
        )
        safe_print(line)
    print("-" * 31)
    print(templ % ("Total", bytes2human(total_rss), '', ''))
    safe_print("PID = %s, name = %s" % (p.pid, p.name()))


if __name__ == '__main__':
    main()