150 lines
4.5 KiB
Python
150 lines
4.5 KiB
Python
import os
|
|
from time import sleep
|
|
from neurotron.logbus import logbus
|
|
|
|
|
|
class Perception:
|
|
"""
|
|
Sensores internos via /proc:
|
|
- CPU % por delta de /proc/stat
|
|
- Memória % via /proc/meminfo
|
|
- Carga média /proc/loadavg ou os.getloadavg()
|
|
Totalmente silencioso (sem prints).
|
|
"""
|
|
|
|
# ----------------------------------------------------------------------
|
|
# CPU
|
|
# ----------------------------------------------------------------------
|
|
|
|
def _read_proc_stat(self):
|
|
"""Lê a linha 'cpu ' de /proc/stat. Retorna dict ou None."""
|
|
try:
|
|
with open("/proc/stat", "r") as f:
|
|
line = f.readline()
|
|
if not line.startswith("cpu "):
|
|
return None
|
|
|
|
parts = line.strip().split()[1:]
|
|
vals = list(map(int, parts[:10]))
|
|
return {
|
|
"user": vals[0], "nice": vals[1], "system": vals[2], "idle": vals[3],
|
|
"iowait": vals[4], "irq": vals[5], "softirq": vals[6], "steal": vals[7],
|
|
"guest": vals[8], "guest_nice": vals[9],
|
|
}
|
|
|
|
except Exception:
|
|
return None
|
|
|
|
def _cpu_percent(self, interval=0.05):
|
|
"""Computa CPU% entre duas leituras."""
|
|
a = self._read_proc_stat()
|
|
if not a:
|
|
return "?"
|
|
|
|
sleep(interval)
|
|
|
|
b = self._read_proc_stat()
|
|
if not b:
|
|
return "?"
|
|
|
|
# Totais
|
|
idle_a = a["idle"] + a["iowait"]
|
|
idle_b = b["idle"] + b["iowait"]
|
|
|
|
non_a = sum(a.values()) - idle_a
|
|
non_b = sum(b.values()) - idle_b
|
|
|
|
total_a = idle_a + non_a
|
|
total_b = idle_b + non_b
|
|
|
|
totald = total_b - total_a
|
|
idled = idle_b - idle_a
|
|
|
|
if totald <= 0:
|
|
return "?"
|
|
|
|
usage = (totald - idled) * 100.0 / totald
|
|
return round(usage, 1)
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Memória
|
|
# ----------------------------------------------------------------------
|
|
|
|
def _mem_percent(self):
|
|
try:
|
|
info = {}
|
|
with open("/proc/meminfo", "r") as f:
|
|
for line in f:
|
|
k, v = line.split(":", 1)
|
|
info[k.strip()] = v.strip()
|
|
|
|
def kB(key):
|
|
return float(info[key].split()[0]) if key in info else None
|
|
|
|
mem_total = kB("MemTotal")
|
|
mem_avail = kB("MemAvailable")
|
|
|
|
if mem_total is None or mem_avail is None:
|
|
return "?"
|
|
|
|
used = mem_total - mem_avail
|
|
return round(used * 100.0 / mem_total, 1)
|
|
|
|
except Exception:
|
|
return "?"
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Loadavg
|
|
# ----------------------------------------------------------------------
|
|
|
|
def _loadavg(self):
|
|
try:
|
|
if hasattr(os, "getloadavg"):
|
|
l1, l5, l15 = os.getloadavg()
|
|
return [round(l1, 2), round(l5, 2), round(l15, 2)]
|
|
|
|
with open("/proc/loadavg", "r") as f:
|
|
parts = f.read().strip().split()
|
|
l1, l5, l15 = map(float, parts[:3])
|
|
return [round(l1, 2), round(l5, 2), round(l15, 2)]
|
|
|
|
except Exception:
|
|
return ["?", "?", "?"]
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Snapshot principal
|
|
# ----------------------------------------------------------------------
|
|
|
|
def snapshot(self) -> dict:
|
|
"""
|
|
Retorna snapshot interno:
|
|
{
|
|
"env_user": "...",
|
|
"env_term": "...",
|
|
"cpu_percent": n | "?",
|
|
"mem_percent": n | "?",
|
|
"loadavg": [l1, l5, l15]
|
|
}
|
|
Sempre seguro. Nunca lança exceção.
|
|
"""
|
|
try:
|
|
return {
|
|
"env_user": os.environ.get("USER") or "root",
|
|
"env_term": os.environ.get("TERM") or "unknown",
|
|
"cpu_percent": self._cpu_percent(),
|
|
"mem_percent": self._mem_percent(),
|
|
"loadavg": self._loadavg(),
|
|
}
|
|
|
|
except Exception as e:
|
|
# fallback extremo — nunca quebrar o Neurotron
|
|
logbus.emit(f"[warn] Perception.snapshot falhou: {e}")
|
|
return {
|
|
"env_user": "unknown",
|
|
"env_term": "unknown",
|
|
"cpu_percent": "?",
|
|
"mem_percent": "?",
|
|
"loadavg": ["?", "?", "?"],
|
|
}
|
|
|