Python Qt : network tool with PyQt6-Charts.

Today, this python script will create a tool for network with PyQt6-Charts.

You need to install the PyQt6-Charts:

python.exe -m pip install PyQt6-Charts
Collecting PyQt6-Charts
...
Installing collected packages: PyQt6-Charts-Qt6, PyQt6-Charts
Successfully installed PyQt6-Charts-6.11.0 PyQt6-Charts-Qt6-6.11.1

Let's see the python script.

import sys
import psutil
from pathlib import Path

from PyQt6.QtWidgets import (
    QApplication, QWidget, QVBoxLayout, QHBoxLayout, QLabel,
    QTableWidget, QTableWidgetItem, QHeaderView
)
from PyQt6.QtCore import QTimer, Qt
from PyQt6.QtCharts import QChart, QChartView, QLineSeries, QValueAxis


def bytes_to_human(n: int) -> str:
    symbols = ('B', 'KB', 'MB', 'GB', 'TB')
    prefix = {}
    for i, s in enumerate(symbols[1:], 1):
        prefix[s] = 1 << (i * 10)
    for s in reversed(symbols[1:]):
        if n >= prefix[s]:
            value = float(n) / prefix[s]
            return f"{value:.2f} {s}"
    return f"{n} B"


class NetworkMonitor(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Network Monitor – PyQt6 (Traffic + Ports)")
        self.resize(1000, 650)

        main_layout = QVBoxLayout(self)

        # ---------------- TOP LABELS ----------------
        self.label_up = QLabel("Upload: 0 B/s")
        self.label_down = QLabel("Download: 0 B/s")

        top_layout = QHBoxLayout()
        top_layout.addWidget(self.label_up)
        top_layout.addWidget(self.label_down)
        main_layout.addLayout(top_layout)

        # ---------------- CHART ----------------
        self.series_up = QLineSeries()
        self.series_down = QLineSeries()
        self.series_up.setName("Upload")
        self.series_down.setName("Download")

        self.chart = QChart()
        self.chart.addSeries(self.series_up)
        self.chart.addSeries(self.series_down)
        self.chart.setTitle("Network traffic (bytes/sec)")

        # Axes (Qt6 style)
        self.axis_x = QValueAxis()
        self.axis_y = QValueAxis()

        self.axis_x.setRange(0, 60)
        self.axis_y.setRange(0, 1024 * 1024)  # 1 MB/s default
        self.axis_x.setTitleText("Time (s)")
        self.axis_y.setTitleText("Bytes / second")

        self.chart.addAxis(self.axis_x, Qt.AlignmentFlag.AlignBottom)
        self.chart.addAxis(self.axis_y, Qt.AlignmentFlag.AlignLeft)

        self.series_up.attachAxis(self.axis_x)
        self.series_up.attachAxis(self.axis_y)
        self.series_down.attachAxis(self.axis_x)
        self.series_down.attachAxis(self.axis_y)

        self.chart_view = QChartView(self.chart)
        main_layout.addWidget(self.chart_view)

        # ---------------- PORTS TABLE ----------------
        self.table = QTableWidget(0, 5)
        self.table.setHorizontalHeaderLabels(
            ["Local IP:Port", "Remote IP:Port", "Status", "PID", "Process"]
        )
        self.table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch)

        # 🔥 Sortare activată
        self.table.setSortingEnabled(True)

        main_layout.addWidget(self.table)

        # ---------------- STATE ----------------
        self.old_stats = psutil.net_io_counters()
        self.x_pos = 0

        # ---------------- TIMER ----------------
        self.timer = QTimer(self)
        self.timer.timeout.connect(self.update_stats)
        self.timer.start(1000)  # 1s

    # ---------------------------------------------------------
    def update_stats(self):
        # --- traffic ---
        new_stats = psutil.net_io_counters()
        sent = new_stats.bytes_sent - self.old_stats.bytes_sent
        recv = new_stats.bytes_recv - self.old_stats.bytes_recv
        self.old_stats = new_stats

        self.label_up.setText(f"Upload: {bytes_to_human(sent)}/s")
        self.label_down.setText(f"Download: {bytes_to_human(recv)}/s")

        # --- chart data ---
        self.series_up.append(self.x_pos, sent)
        self.series_down.append(self.x_pos, recv)
        self.x_pos += 1

        # keep last 60 seconds visible
        if self.x_pos > 60:
            self.axis_x.setRange(self.x_pos - 60, self.x_pos)

        # auto-scale Y a bit
        max_val = max(sent, recv, 1)
        current_max = self.axis_y.max()
        if max_val > current_max * 0.9:
            self.axis_y.setRange(0, max_val * 1.5)

        # --- ports ---
        self.update_ports()

    # ---------------------------------------------------------
    def update_ports(self):
        conns = psutil.net_connections(kind="tcp")

        self.table.setSortingEnabled(False)  # prevenim flicker
        self.table.setRowCount(0)

        for c in conns:
            if not c.laddr:
                continue

            row = self.table.rowCount()
            self.table.insertRow(row)

            # Local
            local = f"{c.laddr.ip}:{c.laddr.port}"
            item_local = QTableWidgetItem(local)
            item_local.setData(Qt.ItemDataRole.UserRole, c.laddr.port)

            # Remote
            if c.raddr:
                remote = f"{c.raddr.ip}:{c.raddr.port}"
                remote_port = c.raddr.port
            else:
                remote = "-"
                remote_port = -1

            item_remote = QTableWidgetItem(remote)
            item_remote.setData(Qt.ItemDataRole.UserRole, remote_port)

            # Status
            item_status = QTableWidgetItem(c.status)

            # PID
            pid = c.pid if c.pid else -1
            item_pid = QTableWidgetItem(str(pid))
            item_pid.setData(Qt.ItemDataRole.UserRole, pid)

            # Process name
            proc_name = "-"
            if c.pid:
                try:
                    proc_name = psutil.Process(c.pid).name()
                except:
                    proc_name = "?"

            item_proc = QTableWidgetItem(proc_name)

            self.table.setItem(row, 0, item_local)
            self.table.setItem(row, 1, item_remote)
            self.table.setItem(row, 2, item_status)
            self.table.setItem(row, 3, item_pid)
            self.table.setItem(row, 4, item_proc)

        self.table.setSortingEnabled(True)  # reactivăm sortarea


if __name__ == "__main__":
    app = QApplication(sys.argv)
    w = NetworkMonitor()
    w.show()
    sys.exit(app.exec())