"""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