#!/usr/bin/env python3 import time,socket,subprocess,configparser,signal,re,argparse,json,os,platform from pathlib import Path APP_VERSION="3.7.0" import sys sys.path.append("/usr/local/share/pigway-pi-control") from shutil import disk_usage from plugin_api import Publisher,registered,discover CFG="/etc/pigway-pi-control.conf" STATUS_PATH="/run/pigway-pi-control/status.json" running=True def stop(*_): global running; running=False signal.signal(signal.SIGTERM,stop); signal.signal(signal.SIGINT,stop) def log(level,event,**fields): parts=[f"level={level}",f"event={event}"] for k,v in fields.items(): parts.append(f"{k}={str(v).replace(chr(32),'_')}") print(" ".join(parts),flush=True) def uptime_seconds(): try:return int(float(Path("/proc/uptime").read_text().split()[0])) except Exception:return 0 def load_average(): try: x=Path("/proc/loadavg").read_text().split();return (float(x[0]),float(x[1]),float(x[2])) except Exception:return (0.0,0.0,0.0) def memory_usage(): try: values={} for line in Path("/proc/meminfo").read_text().splitlines(): key,value=line.split(":",1); values[key]=int(value.split()[0])*1024 total=values["MemTotal"]; available=values.get("MemAvailable",values.get("MemFree",0)) used=max(0,total-available) return total,used,available,(used*100.0/total if total else 0.0) except Exception:return 0,0,0,0.0 def wifi_percent(metric): try:return max(0,min(100,round((float(metric)+100.0)*2.0))) except Exception:return None def device_details(): info={} try: for line in Path("/proc/cpuinfo").read_text().splitlines(): if ":" in line: key,value=(x.strip() for x in line.split(":",1)) if key in ("Serial","Revision","Hardware","Model"): info[key.lower()]=value except Exception: pass try: info["model"]=Path("/proc/device-tree/model").read_bytes().rstrip(b"\0").decode() except Exception: pass try: os_release={} for line in Path("/etc/os-release").read_text().splitlines(): if "=" in line: key,value=line.split("=",1); os_release[key]=value.strip().strip('"') info["os"]=os_release.get("PRETTY_NAME",platform.platform()) except Exception: info["os"]=platform.platform() info.update({"hostname":socket.gethostname(),"kernel":platform.release(), "architecture":platform.machine(),"cpu_count":os.cpu_count() or 0}) return info def cpu_freq_mhz(): for q in ("/sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq","/sys/devices/system/cpu/cpu0/cpufreq/cpuinfo_cur_freq"): try:return int(Path(q).read_text().strip())/1000.0 except Exception:pass return 0.0 def root_readonly(): try: for line in Path("/proc/mounts").read_text().splitlines(): a=line.split() if len(a)>=4 and a[1]=="/":return "ro" in a[3].split(",") except Exception:pass return False def throttled_state(): raw=_run(["vcgencmd","get_throttled"]) m=re.search(r"0x([0-9a-fA-F]+)",raw);v=int(m.group(1),16) if m else 0 if not m: return {"raw":"UNKNOWN","error":"vcgencmd get_throttled unavailable or invalid"} return {"raw":f"0x{v:x}","under_voltage_now":bool(v&1),"freq_capped_now":bool(v&2),"throttled_now":bool(v&4),"soft_temp_limit_now":bool(v&8),"under_voltage_occurred":bool(v&(1<<16)),"freq_capped_occurred":bool(v&(1<<17)),"throttled_occurred":bool(v&(1<<18)),"soft_temp_limit_occurred":bool(v&(1<<19))} prev_cpu=None page=0; page_since=time.monotonic() cpu_hi=mem_hi=0 net_ok=link_ok=True net_iface=""; net_kind="NET"; net_metric="DOWN"; ip_label="IP4"; ip="NO IP" wifi_weak_since=None; wifi_weak_active=False last_net=last_watch=0 watch_state=[] def set_pi_led(name,off=True): base=Path("/sys/class/leds")/name try: if off: (base/"trigger").write_text("none") (base/"brightness").write_text("0") except Exception as e: print(f"board LED {name}: {e}",flush=True) def apply_dark_mode(): if DARK_PWR: set_pi_led("PWR",True) if DARK_ACT: set_pi_led("ACT",True) def cpu_pct(): global prev_cpu p=list(map(int,Path("/proc/stat").read_text().splitlines()[0].split()[1:9])) idle=p[3]+p[4]; total=sum(p) if prev_cpu is None: prev_cpu=(idle,total); return 0.0 di=idle-prev_cpu[0]; dt=total-prev_cpu[1]; prev_cpu=(idle,total) return 100.0*(dt-di)/dt if dt else 0.0 def temp_c(): return int(Path("/sys/class/thermal/thermal_zone0/temp").read_text())/1000.0 def mem_pct(): vals={} for l in Path("/proc/meminfo").read_text().splitlines(): if ":" in l: k,v=l.split(":",1); vals[k]=int(v.split()[0]) return 100*(vals["MemTotal"]-vals["MemAvailable"])/vals["MemTotal"] def disk_pct(): d=disk_usage("/"); return 100*d.used/d.total def _run(args,timeout=2): try: return subprocess.check_output(args,text=True,stderr=subprocess.DEVNULL,timeout=timeout).strip() except Exception: return "" def primary_network(): out=_run(["ip","-4","route","get","1.1.1.1"]) dev=re.search(r"\bdev\s+(\S+)",out); src=re.search(r"\bsrc\s+(\S+)",out); label="IP4" if not dev: out=_run(["ip","-6","route","get","2606:4700:4700::1111"]) dev=re.search(r"\bdev\s+(\S+)",out); src=re.search(r"\bsrc\s+(\S+)",out); label="IP6" if not dev: return "","NET","DOWN","IP4","NO IP" iface=dev.group(1); ipval=src.group(1) if src else "" if Path(f"/sys/class/net/{iface}/wireless").exists(): kind="WIF"; metric="--" try: for line in Path("/proc/net/wireless").read_text().splitlines(): if line.lstrip().startswith(iface+":"): metric=str(int(float(line.split()[3].rstrip(".")))); break except Exception: pass else: kind="ETH"; metric="UP" try: n=int(Path(f"/sys/class/net/{iface}/speed").read_text().strip()) metric=f"{n//1000}G" if n>=1000 and n%1000==0 else str(n) except Exception: pass return iface,kind,metric,label,(ipval or "NO IP") def get_ip(): return primary_network()[4] def internet(): try: s=socket.create_connection(("1.1.1.1",53),0.5);s.close();return True except: return False def load_cfg(): c=configparser.ConfigParser(); c.read(CFG) watches=[] if c.has_section("services"): for label,target in c["services"].items(): target=target.strip() paused=c.getboolean("service_monitor_disabled",label,fallback=False) if target and not target.startswith("#") and not paused: watches.append(("service",label.upper(),target)) if c.has_section("processes"): for label,target in c["processes"].items(): target=target.strip() if target and not target.startswith("#"): watches.append(("process",label.upper(),target)) return c,watches cfg,watches=load_cfg() def cf(section,key,default,cast=float): try: return cast(cfg.get(section,key)) except: return default DEVICE_ID=cf("device","identifier","auto",str).strip() if not DEVICE_ID or DEVICE_ID.lower()=="auto": DEVICE_ID=socket.gethostname() DEVICE_NAME=cf("device","name",DEVICE_ID,str).strip() or DEVICE_ID if DEVICE_NAME.lower()=="auto": DEVICE_NAME=DEVICE_ID DEVICE_INFO=device_details() SENSOR_INTERVAL=cf("timing","sensor_interval",1.0) NETWORK_INTERVAL=cf("timing","network_interval",5.0) SERVICE_INTERVAL=cf("timing","service_interval",5.0) HEALTH_INTERVAL=cf("timing","health_interval",1.0) CPU_HIGH=cf("alerts","cpu_high",90.0) CPU_RECOVER=cf("alerts","cpu_recover",75.0) CPU_HIGH_SEC=cf("alerts","cpu_high_seconds",30,int) MEM_HIGH=cf("alerts","memory_high",90.0) MEM_RECOVER=cf("alerts","memory_recover",85.0) MEM_HIGH_SEC=cf("alerts","memory_high_seconds",30,int) TEMP_HIGH=cf("alerts","temp_high",68.0) TEMP_CRIT=cf("alerts","temp_critical",75.0) DISK_HIGH=cf("alerts","disk_high",90.0) DISK_CRIT=cf("alerts","disk_critical",95.0) WIFI_WEAK=cf("alerts","wifi_weak",-75.0) WIFI_RECOVER=cf("alerts","wifi_recover",-68.0) WIFI_WEAK_SECONDS=cf("alerts","wifi_weak_seconds",30.0) DARK_PWR=cf("dark_mode","pi_pwr_led",True,lambda x: str(x).strip().lower() in ("off","0","false","no")) DARK_ACT=cf("dark_mode","pi_act_led",True,lambda x: str(x).strip().lower() in ("off","0","false","no")) ALERT_CATEGORIES={ "ROOT_READONLY":"STORAGE", "TEMP_CRITICAL":"CPU","TEMP_HIGH":"CPU","SOFT_TEMP_LIMIT":"CPU", "THROTTLED":"CPU","FREQ_CAPPED":"CPU","CPU_HIGH":"CPU", "POWER_UNDERVOLT":"POWER", "MEMORY_HIGH":"MEMORY", "DISK_CRITICAL":"STORAGE","DISK_HIGH":"STORAGE", "NETWORK_DOWN":"NETWORK","INTERNET_DOWN":"NETWORK","WIFI_WEAK":"NETWORK", "SERVICE_DOWN":"SERVICE","PROCESS_DOWN":"SERVICE", } def check_watch(w): typ,label,target=w args=["systemctl","is-active","--quiet",target] if typ=="service" else ["pgrep","-f","--",target] try: r=subprocess.run(args,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,timeout=2) return r.returncode==0 except (OSError,subprocess.TimeoutExpired): return False def diagnose(): print("Agent:") print(f"version={APP_VERSION} uptime_seconds={uptime_seconds()} config={CFG}") print(f"machine_id={DEVICE_ID} machine_name={DEVICE_NAME} hostname={socket.gethostname()}") print("loop_errors=UNAVAILABLE scope=separate_diagnostic_process") print("System:") print(f"load_average={load_average()} cpu_freq_mhz={cpu_freq_mhz():.0f} root_readonly={root_readonly()}") print("Power:") power=throttled_state() print("throttled="+power["raw"]) for key in ("under_voltage_now","freq_capped_now","throttled_now","soft_temp_limit_now", "under_voltage_occurred","freq_capped_occurred","throttled_occurred","soft_temp_limit_occurred"): print(f"{key}={power.get(key,'UNKNOWN')}") if "error" in power: print("error="+power["error"]) iface,kind,metric,label,address=primary_network() print("Network:") print(f"primary_interface={iface or 'NONE'} kind={kind} metric={metric} signal_percent={wifi_percent(metric) if kind=='WIF' else 'N/A'} ip_label={label} IP={address} internet={internet()}") print("Plugins (read-only API discovery; no hardware access):") for plugin in discover(cfg): print(f"plugin={plugin['id']} linked={plugin['linked']} online={plugin['online']}") print("Watches:") for w in watches: print(f"watch {w[1]}={'UP' if check_watch(w) else 'DOWN'} type={w[0]} target={w[2]}") return 0 ALERT_PRIORITIES={ "ROOT_READONLY":100, "TEMP_CRITICAL":90, "SOFT_TEMP_LIMIT":85, "POWER_UNDERVOLT":80, "THROTTLED":75, "FREQ_CAPPED":70, "NETWORK_DOWN":65, "INTERNET_DOWN":65, "SERVICE_DOWN":60, "PROCESS_DOWN":60, "DISK_CRITICAL":55, "TEMP_HIGH":50, "CPU_HIGH":45, "MEMORY_HIGH":45, "DISK_HIGH":40, "WIFI_WEAK":30, "NORMAL_HOME":0, } def alert_priority(request_id): return ALERT_PRIORITIES[request_id.split("|",1)[0]] def transition(key,active,on_event,off_event,**fields): old=event_state.get(key,False) if active!=old: event_state[key]=active log("WARN" if active else "INFO",on_event if active else off_event,**fields) def update_active_alerts(cpu,temp,mem,disk,sample_elapsed=0.0): global wifi_weak_since,wifi_weak_active global cpu_hi,mem_hi cpu_hi=cpu_hi+sample_elapsed if cpu>=CPU_HIGH else 0 if cpu=MEM_HIGH else 0 if mem=CPU_HIGH_SEC: alert_manager.request("CPU_HIGH",1,"CPU LOAD HIGH",f"LOAD {cpu:.1f}%",f"TEMP {temp:.1f}C","CHECK LOAD") else: alert_manager.clear("CPU_HIGH") if temp>=TEMP_CRIT: alert_manager.request("TEMP_CRITICAL",3,"TEMP CRITICAL",f"TEMP {temp:.1f}C","CHECK COOLING","CHECK COOLING") alert_manager.clear("TEMP_HIGH") elif temp>=TEMP_HIGH: alert_manager.request("TEMP_HIGH",1,"TEMP HIGH",f"TEMP {temp:.1f}C","CHECK COOLING","CHECK COOLING") alert_manager.clear("TEMP_CRITICAL") else: alert_manager.clear("TEMP_HIGH"); alert_manager.clear("TEMP_CRITICAL") if mem_hi>=MEM_HIGH_SEC: alert_manager.request("MEMORY_HIGH",1,"MEMORY HIGH",f"USED {mem:.1f}%","CHECK PROCESS","") else: alert_manager.clear("MEMORY_HIGH") if disk>=DISK_CRIT: alert_manager.request("DISK_CRITICAL",2,"DISK CRITICAL",f"USED {disk:.1f}%","FREE < 5%","CHECK STORAGE") alert_manager.clear("DISK_HIGH") elif disk>=DISK_HIGH: alert_manager.request("DISK_HIGH",1,"DISK SPACE LOW",f"USED {disk:.1f}%","FREE < 10%","CHECK STORAGE") alert_manager.clear("DISK_CRITICAL") else: alert_manager.clear("DISK_HIGH"); alert_manager.clear("DISK_CRITICAL") if not link_ok: alert_manager.request("NETWORK_DOWN",2,"NETWORK DOWN","NO LAN / WIFI","NO IP ADDRESS","") alert_manager.clear("INTERNET_DOWN") elif not net_ok: alert_manager.request("INTERNET_DOWN",2,"INTERNET DOWN","LOCAL LINK OK",ip,"") alert_manager.clear("NETWORK_DOWN") else: alert_manager.clear("NETWORK_DOWN"); alert_manager.clear("INTERNET_DOWN") for w,ok in zip(watches,watch_state): alert_id=("SERVICE_DOWN" if w[0]=="service" else "PROCESS_DOWN")+"|"+w[1] if not ok: alert_manager.request(alert_id,2,"SERVICE DOWN" if w[0]=="service" else "PROCESS DOWN",w[1],"CHECK / RESTART","CRITICAL",{"target":w[2]}) else: alert_manager.clear(alert_id) if net_kind=="WIF" and str(net_metric).lstrip("-").isdigit(): rssi=float(net_metric) if not wifi_weak_active: if rssi <= WIFI_WEAK: if wifi_weak_since is None: wifi_weak_since=time.monotonic() if time.monotonic()-wifi_weak_since >= WIFI_WEAK_SECONDS: wifi_weak_active=True else: wifi_weak_since=None elif rssi >= WIFI_RECOVER: wifi_weak_active=False; wifi_weak_since=None if wifi_weak_active: alert_manager.request("WIFI_WEAK",1,"WIFI SIGNAL WEAK",f"RSSI {rssi:.0f} dBm",f"IF {net_iface}","CHECK SIGNAL") else: alert_manager.clear("WIFI_WEAK") else: wifi_weak_since=None; wifi_weak_active=False; alert_manager.clear("WIFI_WEAK") if health.get("under_voltage_now"): alert_manager.request("POWER_UNDERVOLT",2,"POWER LOW","UNDERVOLTAGE","CHECK POWER","CRITICAL",{"raw":health.get("raw","")}) else: alert_manager.clear("POWER_UNDERVOLT") if health.get("throttled_now"): alert_manager.request("THROTTLED",2,"CPU THROTTLED",f"FREQ {cpu_freq_mhz():.0f}M",health.get("raw",""),"CHECK POWER") else: alert_manager.clear("THROTTLED") if health.get("freq_capped_now"): alert_manager.request("FREQ_CAPPED",2,"FREQ CAPPED",f"FREQ {cpu_freq_mhz():.0f}M",health.get("raw",""),"CHECK POWER") else: alert_manager.clear("FREQ_CAPPED") if health.get("soft_temp_limit_now"): alert_manager.request("SOFT_TEMP_LIMIT",2,"SOFT TEMP LIMIT","CPU TEMP LIMITED",health.get("raw",""),"CHECK COOLING") else: alert_manager.clear("SOFT_TEMP_LIMIT") if fs_readonly: alert_manager.request("ROOT_READONLY",3,"FILESYSTEM RO","ROOT READ ONLY","CHECK STORAGE","EMERGENCY") else: alert_manager.clear("ROOT_READONLY") class AlertManager: """Current monitoring facts. Hardware plugins own all display decisions.""" def __init__(self): self.requests={} def request(self,request_id,severity,title,l2="",l3="",l4="",metadata=None): now=time.monotonic();old=self.requests.get(request_id) value={"id":request_id,"priority":alert_priority(request_id),"severity":severity, "category":ALERT_CATEGORIES[request_id.split("|",1)[0]].lower(), "title":title[:21],"l2":l2[:21],"l3":l3[:21],"l4":l4[:21], "first_seen":old["first_seen"] if old else now,"metadata":metadata or {}} value["last_changed"]=old["last_changed"] if old and all(old[k]==v for k,v in value.items()) else now self.requests[request_id]=value if old is None:log("WARN","ALERT_ACTIVE",alert=request_id,severity=severity,detail=l2) def clear(self,request_id): if self.requests.pop(request_id,None) is not None:log("INFO","ALERT_RECOVERED",alert=request_id) def status_snapshot(self): values=sorted(self.requests.values(),key=lambda a:(-a["priority"],a["first_seen"],a["id"])) return {"owner":values[0]["id"] if values else "NORMAL_HOME","alerts":values} def cleanup(): publisher.close() log("INFO","STOP",loop_errors=agent["loop_errors"]) try:Path(STATUS_PATH).unlink(missing_ok=True) except Exception as e:log("ERROR","SHUTDOWN_CLEANUP",step="status_file",error=repr(e)) def write_status(cpu,temp,mem,disk): display_state=alert_manager.status_snapshot() memory_total,memory_used,memory_available,memory_calculated=memory_usage() disk_total,disk_used,disk_free=disk_usage("/") signal=wifi_percent(net_metric) if net_kind=="WIF" else None state={ "version":APP_VERSION,"timestamp":int(time.time()),"uptime_seconds":uptime_seconds(), "machine":{"id":DEVICE_ID,"name":DEVICE_NAME,"hostname":socket.gethostname()}, "device":DEVICE_INFO, "system":{"load_average":load_average(),"cpu_percent":round(cpu,1), "cpu_freq_mhz":round(cpu_freq_mhz()),"temperature_c":round(temp,1), "memory_percent":round(memory_calculated if memory_total else mem,1), "memory_total_bytes":memory_total,"memory_used_bytes":memory_used, "memory_available_bytes":memory_available, "disk_percent":round(disk,1),"disk_total_bytes":disk_total, "disk_used_bytes":disk_used,"disk_free_bytes":disk_free, "root_readonly":bool(fs_readonly)}, "power":health, "network":{"interface":net_iface,"kind":net_kind,"metric":net_metric, "signal_percent":signal, "ip_label":ip_label,"ip":ip,"link":bool(link_ok),"internet":bool(net_ok)}, "watches":[{"type":w[0],"name":w[1],"target":w[2],"up":bool(ok)} for w,ok in zip(watches,watch_state)], "display":display_state, "agent":{"loop_errors":agent["loop_errors"]}, "plugins":publisher.current_status(), } publisher.publish(state) path=Path(STATUS_PATH); path.parent.mkdir(mode=0o755,parents=True,exist_ok=True) temporary=path.with_suffix(".tmp") temporary.write_text(json.dumps(state,ensure_ascii=False,separators=(",",":"))) temporary.chmod(0o644); temporary.replace(path) # Dispatch only after all shared checks and configuration have been initialized. parser=argparse.ArgumentParser() parser.add_argument("--diagnose",action="store_true") args=parser.parse_args() if args.diagnose: raise SystemExit(diagnose()) watch_state=[] last_sensor=last_status=0.0 last_net=last_watch=last_health=0.0 health={}; fs_readonly=False agent={"last_sensor_ok":0.0,"last_network_ok":0.0,"last_watch_ok":0.0,"last_health_ok":0.0,"loop_errors":0} event_state={} history_logged=set() latest=(0.0,0.0,0.0,0.0) alert_manager=AlertManager() publisher=Publisher({k:v for k,v in registered(cfg).items() if not cfg.getboolean("plugin_paused",k,fallback=False)},DEVICE_ID,log) try: apply_dark_mode() log("INFO","START",version=APP_VERSION,uptime=uptime_seconds()) while running: try: now=time.monotonic() sample_elapsed=0.0 # Independent clocks: update data/state only. if now-last_sensor >= SENSOR_INTERVAL: sample_elapsed=now-last_sensor if last_sensor else 0.0 c=cpu_pct(); t=temp_c(); m=mem_pct(); d=disk_pct() latest=(c,t,m,d); last_sensor=now; agent["last_sensor_ok"]=now if now-last_net >= NETWORK_INTERVAL: old_id=(net_iface,ip) net_iface,net_kind,net_metric,ip_label,ip=primary_network(); link_ok=ip!="NO IP"; net_ok=internet() if link_ok else False new_id=(net_iface,ip) if new_id!=old_id: log("INFO","NET_SWITCH",old_if=old_id[0],old_ip=old_id[1],new_if=net_iface,new_ip=ip) transition("network_down",not link_ok,"NETWORK_DOWN","NETWORK_RECOVERED",iface=net_iface,ip=ip) transition("internet_down",link_ok and not net_ok,"INTERNET_DOWN","INTERNET_RECOVERED",iface=net_iface,ip=ip) last_net=now; agent["last_network_ok"]=now if now-last_watch >= SERVICE_INTERVAL: watch_state=[check_watch(w) for w in watches] for w,ok in zip(watches,watch_state): transition("watch:"+w[0]+":"+w[1],not ok,"WATCH_DOWN","WATCH_RECOVERED",name=w[1],target=w[2]) last_watch=now; agent["last_watch_ok"]=now if now-last_health >= HEALTH_INTERVAL: power_sample=throttled_state(); fs_readonly=root_readonly() transition("power_check", "error" in power_sample,"POWER_CHECK_ERROR","POWER_CHECK_RECOVERED") # An unavailable reading is not evidence that an existing fault recovered. if "error" not in power_sample: health=power_sample transition("undervoltage",health.get("under_voltage_now",False),"POWER_LOW","POWER_RECOVERED",raw=health.get("raw")) transition("throttled",health.get("throttled_now",False) or health.get("freq_capped_now",False),"THROTTLED","THROTTLE_RECOVERED",raw=health.get("raw"),freq_mhz=f"{cpu_freq_mhz():.0f}") transition("soft_temp",health.get("soft_temp_limit_now",False),"SOFT_TEMP_LIMIT","SOFT_TEMP_RECOVERED",raw=health.get("raw")) transition("root_ro",fs_readonly,"FILESYSTEM_RO","FILESYSTEM_RW") for k in ("under_voltage_occurred","freq_capped_occurred","throttled_occurred","soft_temp_limit_occurred"): if health.get(k) and k not in history_logged: history_logged.add(k); log("WARN","HISTORY_FLAG",flag=k,raw=health.get("raw")) agent["last_health_ok"]=now; last_health=now c,t,m,d=latest update_active_alerts(c,t,m,d,sample_elapsed) if now-last_status >= 1.0: write_status(c,t,m,d); last_status=now time.sleep(0.20) except Exception as e: agent["loop_errors"]+=1; log("ERROR","LOOP_ERROR",count=agent["loop_errors"],error=repr(e)); time.sleep(1) finally: cleanup()