61 lines
3.0 KiB
Python
61 lines
3.0 KiB
Python
|
"""Read-only local telemetry; no dependency on the monitoring application."""
|
||
|
|
import re
|
||
|
|
import shutil
|
||
|
|
import subprocess
|
||
|
|
import time
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
|
||
|
|
class LocalState:
|
||
|
|
def __init__(self):
|
||
|
|
self.previous_cpu=None
|
||
|
|
self.network_at=float('-inf')
|
||
|
|
self.network={'kind':'NET','metric':'--','ip_label':'IP4','ip':'NO IP'}
|
||
|
|
|
||
|
|
def cpu(self):
|
||
|
|
values=list(map(int,Path('/proc/stat').read_text().splitlines()[0].split()[1:9]))
|
||
|
|
current=(sum(values),values[3]+values[4]);previous=self.previous_cpu;self.previous_cpu=current
|
||
|
|
if previous is None:return None
|
||
|
|
total=current[0]-previous[0]
|
||
|
|
return max(0,min(100,100*(1-(current[1]-previous[1])/total))) if total>0 else None
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def memory():
|
||
|
|
values={line.split(':')[0]:int(line.split()[1]) for line in Path('/proc/meminfo').read_text().splitlines()}
|
||
|
|
return 100*(1-values['MemAvailable']/values['MemTotal'])
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def disk():
|
||
|
|
usage=shutil.disk_usage('/')
|
||
|
|
return 100*usage.used/usage.total
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def read_network():
|
||
|
|
for family,address,label in [('-4','1.1.1.1','IP4'),('-6','2606:4700:4700::1111','IP6')]:
|
||
|
|
# Route lookup only: does not send packets or change network configuration.
|
||
|
|
result=subprocess.run(['ip',family,'route','get',address],capture_output=True,text=True,timeout=1)
|
||
|
|
dev=re.search(r'\bdev\s+(\S+)',result.stdout);src=re.search(r'\bsrc\s+(\S+)',result.stdout)
|
||
|
|
if not dev:continue
|
||
|
|
iface=dev.group(1);wireless=Path('/sys/class/net',iface,'wireless').exists();metric='--'
|
||
|
|
if wireless:
|
||
|
|
for line in Path('/proc/net/wireless').read_text().splitlines():
|
||
|
|
if line.lstrip().startswith(iface+':'):metric=str(float(line.split()[3].rstrip('.')));break
|
||
|
|
else:
|
||
|
|
try:metric=Path('/sys/class/net',iface,'speed').read_text().strip()
|
||
|
|
except OSError:metric='UP'
|
||
|
|
return {'kind':'WIF' if wireless else 'ETH','metric':metric,'ip_label':label,'ip':src.group(1) if src else 'NO IP'}
|
||
|
|
return {'kind':'NET','metric':'DOWN','ip_label':'IP4','ip':'NO IP'}
|
||
|
|
|
||
|
|
def sample(self):
|
||
|
|
system={};errors={}
|
||
|
|
for name,reader in [('cpu_percent',self.cpu),('memory_percent',self.memory),('disk_percent',self.disk),
|
||
|
|
('temperature_c',lambda:float(Path('/sys/class/thermal/thermal_zone0/temp').read_text())/1000)]:
|
||
|
|
try:system[name]=reader()
|
||
|
|
except (OSError,ValueError,KeyError,IndexError,ZeroDivisionError) as exc:system[name]=None;errors[name]=str(exc)
|
||
|
|
if time.monotonic()-self.network_at>=5:
|
||
|
|
try:self.network=self.read_network()
|
||
|
|
except (OSError,ValueError,subprocess.SubprocessError) as exc:
|
||
|
|
self.network={'kind':'NET','metric':'--','ip_label':'IP4','ip':'UNKNOWN'};errors['network']=str(exc)
|
||
|
|
self.network_at=time.monotonic()
|
||
|
|
return {'system':system,'network':dict(self.network)},errors
|