253 lines
8.6 KiB
Python
253 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
# Lancement: python cpu_monitor_native.py
|
|
# Intégrable dans GNOME via /etc/xdg/autostart/resource-monitor.desktop
|
|
|
|
from collections import deque
|
|
import time
|
|
|
|
from gi.repository import Gtk, GLib
|
|
import psutil
|
|
from matplotlib.figure import Figure
|
|
from matplotlib.backends.backend_gtk4agg import FigureCanvasGTK4Agg as FigureCanvas
|
|
|
|
MAX_POINTS = 60
|
|
UPDATE_INTERVAL = 1000 # ms
|
|
|
|
|
|
class Tracker:
|
|
def __init__(self):
|
|
self.cpu_data = deque(maxlen=MAX_POINTS)
|
|
self.ram_data = deque(maxlen=MAX_POINTS)
|
|
self.disk_data = deque(maxlen=MAX_POINTS)
|
|
self.disk_read_data = deque(maxlen=MAX_POINTS)
|
|
self.disk_write_data = deque(maxlen=MAX_POINTS)
|
|
self.net_sent_data = deque(maxlen=MAX_POINTS)
|
|
self.net_recv_data = deque(maxlen=MAX_POINTS)
|
|
self._last_cpu = None
|
|
self._last_net_sent = None
|
|
self._last_net_recv = None
|
|
self._last_disk_read = None
|
|
self._last_disk_write = None
|
|
self.time_data = deque(maxlen=MAX_POINTS)
|
|
self.time_data.append(time.time())
|
|
|
|
def _rate(self, curr, last):
|
|
delta = curr - last
|
|
return max(int(delta / 2.0), 0)
|
|
|
|
def update(self):
|
|
cpu = psutil.cpu_percent(interval=None)
|
|
ram = psutil.virtual_memory().percent
|
|
disk = psutil.disk_usage('/').percent
|
|
self._last_cpu = cpu
|
|
t = time.time()
|
|
self.time_data.append(t)
|
|
|
|
self.cpu_data.append(cpu)
|
|
self.ram_data.append(ram)
|
|
self.disk_data.append(disk)
|
|
|
|
d_io = psutil.disk_io_counters()
|
|
if d_io is not None:
|
|
d_read = d_io.read_bytes
|
|
d_write = d_io.write_bytes
|
|
if self._last_disk_read is not None:
|
|
self.disk_read_data.append(self._rate(d_read, self._last_disk_read))
|
|
self.disk_write_data.append(self._rate(d_write, self._last_disk_write))
|
|
self._last_disk_read = d_read
|
|
self._last_disk_write = d_write
|
|
|
|
net = psutil.net_io_counters()
|
|
sent = net.bytes_sent
|
|
recv = net.bytes_recv
|
|
if self._last_net_sent is not None:
|
|
self.net_sent_data.append(self._rate(sent, self._last_net_sent))
|
|
self.net_recv_data.append(self._rate(recv, self._last_net_recv))
|
|
self._last_net_sent = sent
|
|
self._last_net_recv = recv
|
|
|
|
def get_latest(self):
|
|
if self._last_cpu is None:
|
|
self.update()
|
|
return self._last_cpu, self.ram_data[-1]
|
|
|
|
def get_disk(self):
|
|
if not self.disk_data:
|
|
return 0
|
|
return self.disk_data[-1]
|
|
|
|
def get_net_rates(self):
|
|
if not self.net_sent_data or not self.net_recv_data:
|
|
return 0, 0
|
|
return self.net_sent_data[-1], self.net_recv_data[-1]
|
|
|
|
|
|
class MonitorWindow(Gtk.ApplicationWindow):
|
|
def __init__(self, app):
|
|
super().__init__(application=app, title="Resource Monitor")
|
|
self.set_default_size(1100, 360)
|
|
self.set_icon_name("org.gnome.SystemMonitor")
|
|
|
|
self.tracker = Tracker()
|
|
self._running = True
|
|
|
|
# Dark palette
|
|
self.DARK_BG = "#1e1e2e"
|
|
self.DARK_SURFACE = "#313244"
|
|
self.DARK_BORDER = "#45475a"
|
|
self.TEXT = "#cdd6f4"
|
|
self.TEXT_DIM = "#a6adc8"
|
|
self.CPU_COLOR = "#f38ba8"
|
|
self.RAM_COLOR = "#a6e3a1"
|
|
self.DISK_COLOR = "#cba6f7"
|
|
self.NET_UP_COLOR = "#fab387"
|
|
self.NET_DOWN_COLOR = "#89b4fa"
|
|
self.ALPHA_40 = 0.4
|
|
|
|
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0)
|
|
self.add(box)
|
|
|
|
# --- Top bar with values ---
|
|
header = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=0)
|
|
header.set_css_classes(["sourcing-frame"])
|
|
header.get_style_context().add_class("card")
|
|
header.add_to_css_styleclass("monitor-header")
|
|
|
|
self.cpu_label = Gtk.Label()
|
|
self.cpu_label.set_css_classes(["stat-label"])
|
|
header.append(self.cpu_label)
|
|
|
|
self.ram_label = Gtk.Label()
|
|
self.ram_label.set_css_classes(["stat-label"])
|
|
header.append(self.ram_label)
|
|
|
|
self.disk_label = Gtk.Label()
|
|
self.disk_label.set_css_classes(["stat-label"])
|
|
header.append(self.disk_label)
|
|
|
|
self.net_label = Gtk.Label()
|
|
self.net_label.set_css_classes(["stat-label"])
|
|
header.append(self.net_label)
|
|
|
|
box.append(header)
|
|
|
|
# --- Plots ---
|
|
fig = Figure(figsize=(13, 3.2), facecolor=self.DARK_BG)
|
|
fig.set_tight_layout({"pad": 8})
|
|
|
|
axes = []
|
|
titles = ["CPU %", "RAM %", "Disk %", "Réseau (B/s)"]
|
|
for i, title in enumerate(titles):
|
|
ax = fig.add_subplot(1, 4, i + 1)
|
|
ax.set_facecolor(self.DARK_BG)
|
|
ax.set_title(title, color=self.TEXT, fontsize=11, fontweight="medium")
|
|
ax.set_ylim(0, 100)
|
|
ax.set_ylabel("%" if i < 3 else "B/s", color=self.TEXT_DIM, fontsize=9)
|
|
ax.tick_params(colors=self.TEXT_DIM)
|
|
ax.spines["bottom"].set_color(self.DARK_BORDER)
|
|
ax.spines["top"].set_color(self.DARK_BORDER)
|
|
ax.spines["left"].set_color(self.DARK_BORDER)
|
|
ax.spines["right"].set_color(self.DARK_BORDER)
|
|
axes.append(ax)
|
|
|
|
self.cpu_line, = axes[0].plot([], [], color=self.CPU_COLOR, linewidth=1.5)
|
|
self.ram_line, = axes[1].plot([], [], color=self.RAM_COLOR, linewidth=1.5)
|
|
self.disk_line, = axes[2].plot([], [], color=self.DISK_COLOR, linewidth=1.5)
|
|
self.net_up_line, = axes[3].plot([], [], color=self.NET_UP_COLOR, linewidth=1.5, label="↑")
|
|
self.net_down_line, = axes[3].plot([], [], color=self.NET_DOWN_COLOR, linewidth=1.5, label="↓")
|
|
|
|
for ax in axes[:3]:
|
|
ax.axhline(y=80, color=ax.get_lines()[0].get_color(), linestyle="--", alpha=self.ALPHA_40)
|
|
|
|
canvas = FigureCanvas(fig)
|
|
canvas.set_css_classes(["plot-canvas"])
|
|
box.append(canvas)
|
|
|
|
# --- Footer ---
|
|
footer = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=0)
|
|
footer.set_valign(Gtk.Align.END)
|
|
footer.get_style_context().add_class("linked")
|
|
|
|
self.status_label = Gtk.Label(label="Monitoring...")
|
|
self.status_label.set_css_classes(["dim-label"])
|
|
footer.append(self.status_label)
|
|
|
|
self.stop_btn = Gtk.Button(label="Arrêter")
|
|
self.stop_btn.set_css_classes(["destructive-action"])
|
|
self.stop_btn.connect("clicked", lambda *_: self._stop())
|
|
footer.append(self.stop_btn)
|
|
|
|
box.append(footer)
|
|
|
|
self.fig = fig
|
|
self.axes = axes
|
|
self.canvas = canvas
|
|
|
|
GLib.timeout_add(UPDATE_INTERVAL, self._refresh)
|
|
|
|
def _update_plot(self):
|
|
cpu_arr = list(self.tracker.cpu_data)
|
|
ram_arr = list(self.tracker.ram_data)
|
|
disk_arr = list(self.tracker.disk_data)
|
|
sent_arr = list(self.tracker.net_sent_data)
|
|
recv_arr = list(self.tracker.net_recv_data)
|
|
|
|
self.cpu_line.set_data(range(len(cpu_arr)), cpu_arr)
|
|
self.ram_line.set_data(range(len(ram_arr)), ram_arr)
|
|
self.disk_line.set_data(range(len(disk_arr)), disk_arr)
|
|
self.net_up_line.set_data(range(len(sent_arr)), sent_arr)
|
|
self.net_down_line.set_data(range(len(recv_arr)), recv_arr)
|
|
|
|
for i, ax in enumerate(self.axes[:3]):
|
|
ax.set_xlim(0, MAX_POINTS)
|
|
|
|
max_net = max(max(sent_arr) if sent_arr else 0, max(recv_arr) if recv_arr else 0)
|
|
self.axes[3].set_xlim(0, MAX_POINTS)
|
|
if max_net > 0:
|
|
self.axes[3].set_ylim(0, max_net * 1.2)
|
|
|
|
self.fig.canvas.draw()
|
|
|
|
def _refresh(self):
|
|
if not self._running:
|
|
return GLib.SOURCE_REMOVE
|
|
|
|
self.tracker.update()
|
|
cpu, ram = self.tracker.get_latest()
|
|
disk = self.tracker.get_disk()
|
|
send_rate, recv_rate = self.tracker.get_net_rates()
|
|
|
|
self.cpu_label.set_label(f"CPU: {cpu:5.1f}%")
|
|
self.ram_label.set_label(f"RAM: {ram:5.1f}%")
|
|
self.disk_label.set_label(f"Disk: {disk:5.1f}%")
|
|
|
|
mb_up = send_rate / (1024 * 1024)
|
|
mb_dn = recv_rate / (1024 * 1024)
|
|
self.net_label.set_label(f"Net: ↑{mb_up:.2f} ↓{mb_dn:.2f} MB/s")
|
|
|
|
self._update_plot()
|
|
return GLib.SOURCE_CONTINUE
|
|
|
|
def _stop(self, *_):
|
|
self._running = False
|
|
self.status_label.set_label("Arrêté.")
|
|
self.stop_btn.set_sensitive(False)
|
|
|
|
|
|
class MonitorApplication(Gtk.Application):
|
|
def __init__(self):
|
|
super().__init__(application_id="com.local.resource-monitor")
|
|
|
|
def do_activate(self):
|
|
win = MonitorWindow(self)
|
|
win.present()
|
|
|
|
|
|
def main():
|
|
app = MonitorApplication()
|
|
app.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|