Network Monitor Blighty widget

net.py

import socket
from time import sleep

import psutil
from attrdict import AttrDict
from blighty import CanvasGravity, TextAlign
from blighty.legacy import Graph
from blighty.x11 import Canvas, start_event_loop
from requests import get

from fonts import Fonts


def get_nic_addr(nic):
    ifaces = psutil.net_if_addrs()

    for addr in ifaces[nic]:
        if addr.family == socket.AF_INET and addr.address:
            return addr.address

    return None


def get_active_nics():
    ifaces = psutil.net_if_addrs()

    if_names = [
        iface for iface in ifaces
        if [
            addr for addr in ifaces[iface]
            if addr.family == socket.AF_INET and addr.address
        ]
        and iface != "lo"
    ]
    return if_names


def get_nics_stats():
    ifaces = psutil.net_if_addrs()
    rates = psutil.net_io_counters(pernic=True)

    nics = {}

    for iface in ifaces:
        if iface == "lo":
            continue
        for addr in ifaces[iface]:
            if addr.family == socket.AF_INET and addr.address:
                nics[iface] = {
                    "addr": addr.address,
                    "rates": rates[iface]
                }

    return nics


def unitise(n):
    if n > 2**30:
        return "{} GB/s".format(round(n / 2**30))
    elif n > 2**20:
        return "{} MB/s".format(round(n / 2**20))
    elif n > 2**10:
        return "{} KB/s".format(round(n / 2**10))

    return "{} B/s".format(round(n))


def get_nics(interval=1):
    old_stats = get_nics_stats()
    sleep(interval)
    new_stats = get_nics_stats()
    for nic in new_stats:
        delta = {}
        if nic in old_stats:
            old_rates = old_stats[nic]["rates"]._asdict()
            new_rates = new_stats[nic]["rates"]._asdict()
            for e in new_rates:
                delta[e] = (new_rates[e] - old_rates[e]) / interval

        new_stats[nic]["delta"] = delta

    return new_stats


class Network(Canvas):
    SIZE = (256, 512)
    CORE_POLYGON = AttrDict({"height": 30, "length": 20})

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.upgraphs = {}
        self.downgraphs = {}

    @staticmethod
    def build(x = 0, y = 0, gravity = CanvasGravity.NORTH_WEST):
        return Network(x, y, *Network.SIZE, gravity = gravity, interval=1000)

    def on_button_pressed(self, button, *args):
        self.dispose()

    def draw_nics(c):
        nics = get_nics()

        c.save()

        y = 0
        for nic in nics:
            c.set_source_rgb(1, 1, 1)
            c.set_font_size(16)
            c.write_text(0, 100 + y, nic)

            c.set_source_rgb(.8, .8, .8)
            c.set_font_size(12)
            c.write_text(0, 116 + y, "LOCAL")
            c.set_source_rgb(1, 1, 1)
            c.write_text(48, 116 + y, nics[nic]["addr"])

            if nic not in c.canvas.upgraphs:
                c.canvas.upgraphs[nic] = Graph(0, 124 + y, 64, 24, scale=None)
                c.canvas.upgraphs[nic].push_value(1)
                c.canvas.downgraphs[nic] = Graph(0, 124 + y + 24, 64, -24, scale=None)
                c.canvas.downgraphs[nic].push_value(1)

            c.set_font_size(18)
            c.write_text(
                70, 124 + y + 24 - 4,
                unitise(nics[nic]["delta"]["bytes_recv"]),
                align=TextAlign.TOP_LEFT
            )
            c.set_font_size(14)
            c.write_text(
                70, 124 + y + 24 + 4,
                unitise(nics[nic]["delta"]["bytes_sent"]),
                align=TextAlign.BOTTOM_LEFT
            )

            upgraph = c.canvas.upgraphs[nic]
            downgraph = c.canvas.downgraphs[nic]

            upgraph.push_value(nics[nic]["delta"]["bytes_recv"])
            upgraph.draw(c)

            c.set_source_rgba(.5, .5, .5, .8)
            downgraph.push_value(nics[nic]["delta"]["bytes_sent"])
            downgraph.draw(c)

            y += 100

        c.restore()

    def on_draw(self, c):
        # c.draw_grid()

        c.select_font_face(*Fonts.LAKSAMAN_NORMAL)
        c.set_font_size(36)
        c.set_source_rgb(1, 1, 1)

        w, h = Network.SIZE

        y_poly = (Network.CORE_POLYGON.height + Network.CORE_POLYGON.length)

        c.write_text(0, y_poly, "Network", align = TextAlign.TOP_LEFT)

        ip = get('https://api.ipify.org').text
        c.set_font_size(14)
        c.write_text(200, y_poly - 2, ip, align = TextAlign.CENTER_MIDDLE)
        c.set_source_rgb(.8, .8, .8)
        c.write_text(200, y_poly - 18, "PUBLIC", align = TextAlign.CENTER_MIDDLE)
        c.draw_nics()


if __name__ == "__main__":
    Network.build(720, 480, gravity = CanvasGravity.NORTH_WEST).show()
    start_event_loop()