way/pigway-device-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

fix: harden P0 health monitoring RC2

Browse Source
This commit is contained in:
way
2026-09-26 17:10:53 +08:00
parent 83ec3f4379
commit 3a302b9c78
4 changed files with 245 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/pigway_pi_control.py
+201 -133
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
import os,time,socket,subprocess,configparser,signal,re,sys,json
import time,socket,subprocess,configparser,signal,re,errno,threading,argparse
from pathlib import Path
APP_VERSION="3.3.0"
from smbus2 import SMBus
@@ -45,19 +45,22 @@ def root_readonly():
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
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_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_occurred":bool(v&(1<<19))}
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))}
def diagnose():
print(f"Pigway Pi Control {APP_VERSION} diagnostic")
print("uptime_seconds=",uptime_seconds());print("load_average=",load_average());print("cpu_freq_mhz=",round(cpu_freq_mhz()))
print("root_readonly=",root_readonly());print("throttled=",json.dumps(throttled_state()))
print("network=",primary_network());print("internet=",internet())
out=_run(["i2cdetect","-y","1"],4) if False else ""
return 0
bus=SMBus(1)
bus=None
bus_lock=threading.RLock()
FONT_PATH="/usr/local/share/pigway-pi-control/oled_font_5x7.bin"
FONT5=Path(FONT_PATH).read_bytes()
FONT5=b""
agent={"loop_errors":0,"i2c_errors":0}
def bus_write(method,*args):
with bus_lock:
try: return getattr(bus,method)(*args)
except OSError:
agent["i2c_errors"]+=1
raise
fan_level=0
prev_cpu=None
page=0; page_since=time.monotonic()
@@ -68,7 +71,7 @@ wifi_weak_since=None; wifi_weak_active=False
last_net=last_watch=0
watch_state=[]
def cmd(c): bus.write_byte_data(OLED,0x00,c)
def cmd(c): bus_write("write_byte_data",OLED,0x00,c)
def oled_init():
# Seller-native physical 128x32 configuration.
for c in [0xAE,0xD5,0x80,0xA8,0x1F,0xD3,0x00,0x40,0x8D,0x14,
@@ -107,10 +110,10 @@ def oled_show(lines_lr, title_zh=False):
data.append(v)
cmd(0x21);cmd(0);cmd(127);cmd(0x22);cmd(0);cmd(3)
for i in range(0,len(data),16):
bus.write_i2c_block_data(OLED,0x40,data[i:i+16])
bus_write("write_i2c_block_data",OLED,0x40,data[i:i+16])
def mcu(reg,val): bus.write_byte_data(MCU,reg,val)
def mcu(reg,val): bus_write("write_byte_data",MCU,reg,val)
_last_rgb=None
def rgb_off():
global _last_rgb
@@ -200,7 +203,6 @@ def primary_network():
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
if label=="IP6" and len(ipval)>18: ipval=ipval[:7]+".."+ipval[-7:]
return iface,kind,metric,label,(ipval or "NO IP")
def get_ip():
@@ -263,11 +265,49 @@ LED_EMERG_ON=cf("led","emergency_on_ms",400,int); LED_EMERG_OFF=cf("led","emerge
def check_watch(w):
typ,label,target=w
if typ=="service":
r=subprocess.run(["systemctl","is-active","--quiet",target])
else:
r=subprocess.run(["pgrep","-f",target],stdout=subprocess.DEVNULL)
return r.returncode==0
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("loop_errors=UNAVAILABLE i2c_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} ip_label={label} IP={address} internet={internet()}")
print("I2C:")
# SMBus receive-byte performs a read only; no register selection or display commands.
try:
with SMBus(1) as probe:
for address,name in ((MCU,"MCU"),(OLED,"OLED")):
try:
probe.read_byte(address)
print(f"0x{address:02X} {name}=PRESENT")
except OSError as e:
status="MISSING" if e.errno in (errno.ENXIO,getattr(errno,"EREMOTEIO",121)) else "ERROR"
print(f"0x{address:02X} {name}={status} error={e}")
except OSError as e:
for address,name in ((MCU,"MCU"),(OLED,"OLED")):
print(f"0x{address:02X} {name}=ERROR error={e}")
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
def alert(title,l2="",l3="",l4="",sev=1,color=(20,16,0),blink=0):
return {"title":title[:21],"l2":l2[:21],"l3":l3[:21],"l4":l4[:21],
@@ -279,12 +319,12 @@ def transition(key,active,on_event,off_event,**fields):
event_state[key]=active
log("WARN" if active else "INFO",on_event if active else off_event,**fields)
def make_alerts(cpu,temp,mem,disk):
def make_alerts(cpu,temp,mem,disk,sample_elapsed=0.0):
global wifi_weak_since,wifi_weak_active
global cpu_hi,mem_hi
a=[]
cpu_hi=cpu_hi+1 if cpu>=CPU_HIGH else 0 if cpu<CPU_RECOVER else cpu_hi
mem_hi=mem_hi+1 if mem>=MEM_HIGH else 0 if mem<MEM_RECOVER else mem_hi
cpu_hi=cpu_hi+sample_elapsed if cpu>=CPU_HIGH else 0 if cpu<CPU_RECOVER else cpu_hi
mem_hi=mem_hi+sample_elapsed if mem>=MEM_HIGH else 0 if mem<MEM_RECOVER else mem_hi
if cpu_hi>=CPU_HIGH_SEC:a.append(alert("CPU LOAD HIGH",f"LOAD {cpu:.1f}%",f"TEMP {temp:.1f}C","FOR >=30 SEC",1,(24,18,0)))
if temp>=TEMP_CRIT:a.append(alert("TEMP CRITICAL",f"TEMP {temp:.1f}C",f"FAN {FAN_NAME[fan_level]}","CHECK COOLING",3,(32,0,0),2))
elif temp>=TEMP_HIGH:a.append(alert("TEMP HIGH",f"TEMP {temp:.1f}C",f"FAN {FAN_NAME[fan_level]}","CHECK COOLING",1,(24,0,0)))
@@ -333,11 +373,12 @@ def render_home(cpu,temp,mem,disk):
# 128x32 fixed columns. Both columns are LEFT-aligned.
# Seller 5x7 font has 6px character pitch; right column starts at x=72.
im=Image.new("1",(W,H),0); pix=im.load()
display_ip=ip[:7]+".."+ip[-7:] if ip_label=="IP6" and len(ip)>18 else ip
left=[
f"CPU {cpu:.1f}%",
f"TMP {temp:.1f}C",
f"DSK {disk:.1f}%",
ip_label+" "+ip,
ip_label+" "+display_ip,
]
right=[
f"MEM {mem:.1f}%",
@@ -359,7 +400,7 @@ def render_home(cpu,temp,mem,disk):
data.append(v)
cmd(0x21);cmd(0);cmd(127);cmd(0x22);cmd(0);cmd(3)
for i in range(0,len(data),16):
bus.write_i2c_block_data(OLED,0x40,data[i:i+16])
bus_write("write_i2c_block_data",OLED,0x40,data[i:i+16])
def render_alert(a,idx,total):
@@ -367,18 +408,55 @@ def render_alert(a,idx,total):
(a["l2"],""),(a["l3"],""),(a["l4"],"")])
if "--diagnose" in sys.argv:
raise SystemExit(diagnose())
def alert_key(a):
# Sensor values change without changing the identity of a fault.
return a["title"]+("|"+a["l2"] if a["title"] in ("SERVICE DOWN","PROCESS DOWN") else "")
oled_init(); rgb_off(); apply_dark_mode(); log("INFO","START",version=APP_VERSION,uptime=uptime_seconds())
watch_state=[check_watch(w) for w in watches]
def find_live_alert(key, live):
for a in live:
if alert_key(a)==key: return a
return None
# RGB has its own clock so slow network/watch checks cannot stretch blink pulses.
led_stop=threading.Event()
led_alert=None
def led_worker():
previous_error=None
while not led_stop.is_set():
try:
with bus_lock: led_for(led_alert)
if previous_error is not None: log("INFO","RGB_RECOVERED")
previous_error=None
except Exception as e:
if repr(e)!=previous_error: log("ERROR","RGB_ERROR",error=repr(e))
previous_error=repr(e)
led_stop.wait(0.02)
def cleanup():
log("INFO","STOP",loop_errors=agent["loop_errors"],i2c_errors=agent["i2c_errors"])
led_stop.set()
if led_thread.is_alive(): led_thread.join()
if bus is None: return
# Attempt every step even when a previous device write fails.
for name,action in (("rgb",rgb_off),("fan",lambda:mcu(0x08,0x00)),
("oled_clear",lambda:oled_show([("","")]*4)),
("oled_off",lambda:cmd(0xAE)),("bus_close",bus.close)):
try: action()
except Exception as e: log("ERROR","SHUTDOWN_CLEANUP",step=name,error=repr(e))
# 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_render=0.0
last_net=last_watch=last_health=time.monotonic()
health=throttled_state(); fs_readonly=root_readonly()
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,"i2c_errors":0}
event_state={}
prev_net_identity=None
prev_watch_map={w[1]:ok for w,ok in zip(watches,watch_state)}
history_logged=set()
prev_alert_keys=set()
latest=(0.0,0.0,0.0,0.0)
@@ -391,119 +469,109 @@ cycle_pos=0
page_started=time.monotonic()
current_alert_key=None
def alert_key(a):
# Stable identity: title + monitored object/primary detail.
return f"{a['title']}|{a['l2']}"
def find_live_alert(key, live):
for a in live:
if alert_key(a)==key: return a
return None
while running:
try:
now=time.monotonic()
led_thread=threading.Thread(target=led_worker,name="rgb",daemon=True)
try:
bus=SMBus(1)
FONT5=Path(FONT_PATH).read_bytes()
oled_init(); rgb_off(); apply_dark_mode()
log("INFO","START",version=APP_VERSION,uptime=uptime_seconds())
led_thread.start()
while running:
try:
now=time.monotonic()
sample_elapsed=0.0
# Independent clocks: update data/state only.
if now-last_sensor >= SENSOR_INTERVAL:
c=cpu_pct(); t=temp_c(); m=mem_pct(); d=disk_pct()
latest=(c,t,m,d); update_fan(t); last_sensor=now; agent["last_sensor_ok"]=now
# 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); update_fan(t); 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 old_id!=("","NO IP") and 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_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[1],not ok,"WATCH_DOWN","WATCH_RECOVERED",name=w[1],target=w[2])
last_watch=now; agent["last_watch_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:
health=throttled_state(); fs_readonly=root_readonly()
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("root_ro",fs_readonly,"FILESYSTEM_RO","FILESYSTEM_RW")
for k in ("under_voltage_occurred","freq_capped_occurred","throttled_occurred","soft_temp_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"))
la=load_average()
agent["last_health_ok"]=now; last_health=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
active=make_alerts(c,t,m,d)
new_alert_keys={alert_key(a) for a in active}
for a in active:
k=alert_key(a)
if k not in prev_alert_keys:
log("WARN","ALERT_ACTIVE",alert=a["title"],detail=a["l2"],severity=a["sev"])
for k in prev_alert_keys-new_alert_keys:
log("INFO","ALERT_RECOVERED",alert=k)
prev_alert_keys=new_alert_keys
c,t,m,d=latest
active=make_alerts(c,t,m,d,sample_elapsed)
led_alert=max(active,key=lambda a:a["sev"],default=None)
new_alert_keys={alert_key(a) for a in active}
for a in active:
k=alert_key(a)
if k not in prev_alert_keys:
log("WARN","ALERT_ACTIVE",alert=a["title"],detail=a["l2"],severity=a["sev"])
for k in prev_alert_keys-new_alert_keys:
log("INFO","ALERT_RECOVERED",alert=k)
prev_alert_keys=new_alert_keys
# If the currently displayed alert truly disappeared, skip it immediately.
if cycle_pages[cycle_pos]!="HOME":
key=cycle_pages[cycle_pos]
if find_live_alert(key,active) is None:
# If the currently displayed alert truly disappeared, skip it immediately.
if cycle_pages[cycle_pos]!="HOME":
key=cycle_pages[cycle_pos]
if find_live_alert(key,active) is None:
cycle_pos += 1
if cycle_pos>=len(cycle_pages):
# End of cycle: capture a fresh snapshot.
cycle_pages=["HOME"]+[alert_key(a) for a in active]
cycle_pos=0
page_started=now
# ONLY this timer performs normal page changes.
if now-page_started >= PAGE_INTERVAL:
cycle_pos += 1
if cycle_pos>=len(cycle_pages):
# End of cycle: capture a fresh snapshot.
# A complete cycle ended. Snapshot all currently active alerts.
cycle_pages=["HOME"]+[alert_key(a) for a in active]
cycle_pos=0
page_started=now
# ONLY this timer performs normal page changes.
if now-page_started >= PAGE_INTERVAL:
cycle_pos += 1
if cycle_pos>=len(cycle_pages):
# A complete cycle ended. Snapshot all currently active alerts.
# Special case: while idle on HOME with no alert pages queued, begin a new
# alert cycle immediately when the first alert appears. HOME keeps its
# current 10-second slot; alerts follow it and cannot be kicked out by refresh.
if cycle_pages==["HOME"] and active:
cycle_pages=["HOME"]+[alert_key(a) for a in active]
cycle_pos=0
page_started=now
# Special case: while idle on HOME with no alert pages queued, begin a new
# alert cycle immediately when the first alert appears. HOME keeps its
# current 10-second slot; alerts follow it and cannot be kicked out by refresh.
if cycle_pages==["HOME"] and active:
cycle_pages=["HOME"]+[alert_key(a) for a in active]
# Render CURRENT page only. 1-second refresh does not affect cycle_pos.
if now-last_render >= OLED_REFRESH:
current=cycle_pages[cycle_pos]
if current=="HOME":
render_home(c,t,m,d)
else:
a=find_live_alert(current,active)
if a:
# Number refers to this cycle's alert slot, all using same 6x8 font.
render_alert(a,cycle_pos,len(cycle_pages)-1)
else:
# Render CURRENT page only. 1-second refresh does not affect cycle_pos.
if now-last_render >= OLED_REFRESH:
current=cycle_pages[cycle_pos]
if current=="HOME":
render_home(c,t,m,d)
last_render=now
else:
a=find_live_alert(current,active)
if a:
# Number refers to this cycle's alert slot, all using same 6x8 font.
render_alert(a,cycle_pos,len(cycle_pages)-1)
else:
render_home(c,t,m,d)
last_render=now
# LED follows OLED page; HOME reflects highest currently active risk.
current=cycle_pages[cycle_pos]
if current=="HOME":
led_for(active[0] if active else None)
else:
a=find_live_alert(current,active)
led_for(a if a else (active[0] if active else None))
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)
log("INFO","STOP",reason="signal")
try:
rgb_off(); mcu(0x08,0x00)
oled_show([("",""),("",""),("",""),("","")]); cmd(0xAE)
except Exception as e: log("ERROR","SHUTDOWN_CLEANUP",error=repr(e))
try: bus.close()
except Exception: pass
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()