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

refactor: add single-owner display arbitration

Browse Source
This commit is contained in:
way
2026-09-26 22:57:10 +08:00
parent 34f334dd23
commit 6fc428ef18
4 changed files with 188 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/pigway_pi_control.py
+176 -159
View File
@@ -121,12 +121,16 @@ def rgb_off():
mcu(0x07,0x00) # seller closeRGB(): true OFF
_last_rgb=("off",)
def rgb(r,g,b):
def rgb_begin_mode():
global _last_rgb
mcu(0x07,0x00) # stop a previous mode once, never once per animation frame
_last_rgb=None
def rgb_frame(r,g,b):
global _last_rgb
scale=max(0,min(100,LED_BRIGHT))/100.0
state=(int(r*scale),int(g*scale),int(b*scale))
if _last_rgb==state: return
mcu(0x07,0x00) # stop built-in effect first
mcu(0x00,0xff) # seller protocol: >=3 => all three LEDs
mcu(0x01,state[0]); mcu(0x02,state[1]); mcu(0x03,state[2])
_last_rgb=state
@@ -313,35 +317,28 @@ def diagnose():
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],
"sev":sev,"color":color,"blink":blink}
DISPLAY_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,
}
class AlertManager:
def __init__(self):
self._active={}
def set_alert(self,alert_id,severity,title,l2="",l3="",l4="",metadata=None):
now=time.monotonic()
display={"id":alert_id,"title":title[:21],"l2":l2[:21],"l3":l3[:21],"l4":l4[:21],
"sev":severity,"metadata":metadata or {}}
current=self._active.get(alert_id)
if current is None:
display["active"]=True; display["first_seen"]=now; display["last_changed"]=now
self._active[alert_id]=display
log("WARN","ALERT_ACTIVE",alert=alert_id,severity=severity,detail=l2)
else:
changed=any(current.get(k)!=display.get(k) for k in ("sev","title","l2","l3","l4","metadata"))
current.update(display)
if changed: current["last_changed"]=now
def clear_alert(self,alert_id):
if alert_id in self._active:
self._active.pop(alert_id)
log("INFO","ALERT_RECOVERED",alert=alert_id)
def active_alerts(self):
return sorted(self._active.values(),key=lambda a:(-a["sev"],a["first_seen"],a["id"]))
def display_priority(request_id):
return DISPLAY_PRIORITIES[request_id.split("|",1)[0]]
def transition(key,active,on_event,off_event,**fields):
old=event_state.get(key,False)
@@ -354,39 +351,39 @@ def update_active_alerts(cpu,temp,mem,disk,sample_elapsed=0.0):
global cpu_hi,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: alerts.set_alert("CPU_HIGH",1,"CPU LOAD HIGH",f"LOAD {cpu:.1f}%",f"TEMP {temp:.1f}C","CHECK LOAD")
else: alerts.clear_alert("CPU_HIGH")
if cpu_hi>=CPU_HIGH_SEC: display.request("CPU_HIGH",1,"CPU LOAD HIGH",f"LOAD {cpu:.1f}%",f"TEMP {temp:.1f}C","CHECK LOAD")
else: display.clear("CPU_HIGH")
if temp>=TEMP_CRIT:
alerts.set_alert("TEMP_CRITICAL",3,"TEMP CRITICAL",f"TEMP {temp:.1f}C",f"FAN {FAN_NAME[fan_level]}","CHECK COOLING")
alerts.clear_alert("TEMP_HIGH")
display.request("TEMP_CRITICAL",3,"TEMP CRITICAL",f"TEMP {temp:.1f}C",f"FAN {FAN_NAME[fan_level]}","CHECK COOLING")
display.clear("TEMP_HIGH")
elif temp>=TEMP_HIGH:
alerts.set_alert("TEMP_HIGH",1,"TEMP HIGH",f"TEMP {temp:.1f}C",f"FAN {FAN_NAME[fan_level]}","CHECK COOLING")
alerts.clear_alert("TEMP_CRITICAL")
display.request("TEMP_HIGH",1,"TEMP HIGH",f"TEMP {temp:.1f}C",f"FAN {FAN_NAME[fan_level]}","CHECK COOLING")
display.clear("TEMP_CRITICAL")
else:
alerts.clear_alert("TEMP_HIGH"); alerts.clear_alert("TEMP_CRITICAL")
if mem_hi>=MEM_HIGH_SEC: alerts.set_alert("MEMORY_HIGH",1,"MEMORY HIGH",f"USED {mem:.1f}%","CHECK PROCESS","")
else: alerts.clear_alert("MEMORY_HIGH")
display.clear("TEMP_HIGH"); display.clear("TEMP_CRITICAL")
if mem_hi>=MEM_HIGH_SEC: display.request("MEMORY_HIGH",1,"MEMORY HIGH",f"USED {mem:.1f}%","CHECK PROCESS","")
else: display.clear("MEMORY_HIGH")
if disk>=DISK_CRIT:
alerts.set_alert("DISK_CRITICAL",2,"DISK CRITICAL",f"USED {disk:.1f}%","FREE < 5%","CHECK STORAGE")
alerts.clear_alert("DISK_HIGH")
display.request("DISK_CRITICAL",2,"DISK CRITICAL",f"USED {disk:.1f}%","FREE < 5%","CHECK STORAGE")
display.clear("DISK_HIGH")
elif disk>=DISK_HIGH:
alerts.set_alert("DISK_HIGH",1,"DISK SPACE LOW",f"USED {disk:.1f}%","FREE < 10%","CHECK STORAGE")
alerts.clear_alert("DISK_CRITICAL")
display.request("DISK_HIGH",1,"DISK SPACE LOW",f"USED {disk:.1f}%","FREE < 10%","CHECK STORAGE")
display.clear("DISK_CRITICAL")
else:
alerts.clear_alert("DISK_HIGH"); alerts.clear_alert("DISK_CRITICAL")
display.clear("DISK_HIGH"); display.clear("DISK_CRITICAL")
if not link_ok:
alerts.set_alert("NETWORK_DOWN",2,"NETWORK DOWN","NO LAN / WIFI","NO IP ADDRESS","")
alerts.clear_alert("INTERNET_DOWN")
display.request("NETWORK_DOWN",2,"NETWORK DOWN","NO LAN / WIFI","NO IP ADDRESS","")
display.clear("INTERNET_DOWN")
elif not net_ok:
alerts.set_alert("INTERNET_DOWN",2,"INTERNET DOWN","LOCAL LINK OK",ip,"")
alerts.clear_alert("NETWORK_DOWN")
display.request("INTERNET_DOWN",2,"INTERNET DOWN","LOCAL LINK OK",ip,"")
display.clear("NETWORK_DOWN")
else:
alerts.clear_alert("NETWORK_DOWN"); alerts.clear_alert("INTERNET_DOWN")
display.clear("NETWORK_DOWN"); display.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:
alerts.set_alert(alert_id,2,"SERVICE DOWN" if w[0]=="service" else "PROCESS DOWN",w[1],"CHECK / RESTART","CRITICAL",{"target":w[2]})
else: alerts.clear_alert(alert_id)
display.request(alert_id,2,"SERVICE DOWN" if w[0]=="service" else "PROCESS DOWN",w[1],"CHECK / RESTART","CRITICAL",{"target":w[2]})
else: display.clear(alert_id)
if net_kind=="WIF" and str(net_metric).lstrip("-").isdigit():
rssi=float(net_metric)
if not wifi_weak_active:
@@ -397,24 +394,24 @@ def update_active_alerts(cpu,temp,mem,disk,sample_elapsed=0.0):
elif rssi >= WIFI_RECOVER:
wifi_weak_active=False; wifi_weak_since=None
if wifi_weak_active:
alerts.set_alert("WIFI_WEAK",1,"WIFI SIGNAL WEAK",f"RSSI {rssi:.0f} dBm",f"IF {net_iface}","CHECK SIGNAL")
else: alerts.clear_alert("WIFI_WEAK")
display.request("WIFI_WEAK",1,"WIFI SIGNAL WEAK",f"RSSI {rssi:.0f} dBm",f"IF {net_iface}","CHECK SIGNAL")
else: display.clear("WIFI_WEAK")
else:
wifi_weak_since=None; wifi_weak_active=False; alerts.clear_alert("WIFI_WEAK")
wifi_weak_since=None; wifi_weak_active=False; display.clear("WIFI_WEAK")
if health.get("under_voltage_now"):
alerts.set_alert("POWER_UNDERVOLT",2,"POWER LOW","UNDERVOLTAGE","CHECK POWER","CRITICAL",{"raw":health.get("raw","")})
else: alerts.clear_alert("POWER_UNDERVOLT")
display.request("POWER_UNDERVOLT",2,"POWER LOW","UNDERVOLTAGE","CHECK POWER","CRITICAL",{"raw":health.get("raw","")})
else: display.clear("POWER_UNDERVOLT")
if health.get("throttled_now"):
alerts.set_alert("THROTTLED",2,"CPU THROTTLED",f"FREQ {cpu_freq_mhz():.0f}M",health.get("raw",""),"CHECK POWER")
else: alerts.clear_alert("THROTTLED")
display.request("THROTTLED",2,"CPU THROTTLED",f"FREQ {cpu_freq_mhz():.0f}M",health.get("raw",""),"CHECK POWER")
else: display.clear("THROTTLED")
if health.get("freq_capped_now"):
alerts.set_alert("FREQ_CAPPED",2,"FREQ CAPPED",f"FREQ {cpu_freq_mhz():.0f}M",health.get("raw",""),"CHECK POWER")
else: alerts.clear_alert("FREQ_CAPPED")
display.request("FREQ_CAPPED",2,"FREQ CAPPED",f"FREQ {cpu_freq_mhz():.0f}M",health.get("raw",""),"CHECK POWER")
else: display.clear("FREQ_CAPPED")
if health.get("soft_temp_limit_now"):
alerts.set_alert("SOFT_TEMP_LIMIT",2,"SOFT TEMP LIMIT","CPU TEMP LIMITED",health.get("raw",""),"CHECK COOLING")
else: alerts.clear_alert("SOFT_TEMP_LIMIT")
if fs_readonly: alerts.set_alert("ROOT_READONLY",3,"FILESYSTEM RO","ROOT READ ONLY","CHECK STORAGE","EMERGENCY")
else: alerts.clear_alert("ROOT_READONLY")
display.request("SOFT_TEMP_LIMIT",2,"SOFT TEMP LIMIT","CPU TEMP LIMITED",health.get("raw",""),"CHECK COOLING")
else: display.clear("SOFT_TEMP_LIMIT")
if fs_readonly: display.request("ROOT_READONLY",3,"FILESYSTEM RO","ROOT READ ONLY","CHECK STORAGE","EMERGENCY")
else: display.clear("ROOT_READONLY")
def rgb_mode_for(a):
if not a: return "OFF"
@@ -428,17 +425,15 @@ def breathe(level_seconds,elapsed):
return 0.05+0.95*(0.5-0.5*math.cos(2.0*math.pi*phase))
def rgb_engine(mode,elapsed):
if mode=="OFF":
rgb_off()
elif mode=="WARNING_BREATHE":
if mode=="WARNING_BREATHE":
level=breathe(WARNING_BREATHE_SECONDS,elapsed)
rgb(*(channel*level for channel in RGB_WARNING))
rgb_frame(*(channel*level for channel in RGB_WARNING))
elif mode=="CRITICAL_BREATHE":
level=breathe(CRITICAL_BREATHE_SECONDS,elapsed)
rgb(*(channel*level for channel in RGB_CRITICAL))
rgb_frame(*(channel*level for channel in RGB_CRITICAL))
else:
on=int(elapsed*1000/max(50,EMERGENCY_FLASH_MS))%2==0
rgb(RGB_EMERGENCY[0] if on else 0,0,0)
rgb_frame(RGB_EMERGENCY[0] if on else 0,0,0)
def render_home(cpu,temp,mem,disk):
@@ -475,52 +470,123 @@ def render_home(cpu,temp,mem,disk):
bus_write("write_i2c_block_data",OLED,0x40,data[i:i+16])
def render_alert(a,idx,total):
oled_show([(a["title"],f"{idx}/{total}"),
def render_alert(a):
oled_show([(a["title"],""),
(a["l2"],""),(a["l3"],""),(a["l4"],"")])
def alert_key(a):
return a["id"]
class DisplayController:
def __init__(self):
now=time.monotonic()
self.lock=threading.RLock()
self.requests={"NORMAL_HOME":{
"id":"NORMAL_HOME","priority":0,"severity":0,"rgb_mode":"OFF",
"title":"","l2":"","l3":"","l4":"","active":True,
"first_seen":now,"last_changed":now,"metadata":{"home":(0.0,0.0,0.0,0.0)},
}}
self.current_owner_id="NORMAL_HOME"
self.current_priority=0
self.current_rgb_mode="OFF"
self.owner_since=now
self.stop_event=threading.Event()
self.rgb_thread=threading.Thread(target=self._rgb_worker,name="rgb",daemon=True)
def find_live_alert(key, live):
for a in live:
if alert_key(a)==key: return a
return None
def start(self):
oled_init(); rgb_off(); self.refresh(); self.rgb_thread.start()
# RGB has its own clock and consumes only the highest-severity active alert.
# OLED rendering never reads or writes RGB state.
led_stop=threading.Event()
led_alert=None
def update_home(self,cpu,temp,mem,disk):
with self.lock:
request=self.requests["NORMAL_HOME"]
values=(cpu,temp,mem,disk)
if request["metadata"].get("home")!=values:
request["metadata"]={"home":values}; request["last_changed"]=time.monotonic()
def led_worker():
previous_error=None
current_mode="OFF"
animation_started=time.monotonic()
while not led_stop.is_set():
try:
new_mode=rgb_mode_for(led_alert)
if new_mode!=current_mode:
log("INFO","RGB_MODE",old=current_mode,new=new_mode)
current_mode=new_mode
# Keep the verified five-write MCU RGB transaction contiguous.
with bus_lock: rgb_engine(current_mode,time.monotonic()-animation_started)
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(max(0.02,min(0.05,RGB_UPDATE_MS/1000.0)))
def request(self,request_id,severity,title,l2="",l3="",l4="",metadata=None):
now=time.monotonic(); priority=display_priority(request_id)
request={"id":request_id,"priority":priority,"severity":severity,
"rgb_mode":rgb_mode_for({"sev":severity}),"title":title[:21],
"l2":l2[:21],"l3":l3[:21],"l4":l4[:21],"active":True,
"last_changed":now,"metadata":metadata or {}}
with self.lock:
current=self.requests.get(request_id)
if current is None:
request["first_seen"]=now; self.requests[request_id]=request
log("WARN","ALERT_ACTIVE",alert=request_id,severity=severity,detail=l2)
else:
request["first_seen"]=current["first_seen"]
changed=any(current.get(k)!=request.get(k) for k in
("priority","severity","rgb_mode","title","l2","l3","l4","metadata"))
if not changed: request["last_changed"]=current["last_changed"]
self.requests[request_id]=request
self._arbitrate_locked()
def clear(self,request_id):
if request_id=="NORMAL_HOME": return
with self.lock:
if self.requests.pop(request_id,None) is not None:
log("INFO","ALERT_RECOVERED",alert=request_id)
self._arbitrate_locked()
def _arbitrate_locked(self):
current=self.requests.get(self.current_owner_id)
ordered=sorted(self.requests.values(),key=lambda r:(-r["priority"],r["first_seen"],r["id"]))
candidate=ordered[0]
if current is not None and candidate["priority"]<=current["priority"]: return
self._change_owner_locked(candidate)
def _change_owner_locked(self,request):
old=self.current_owner_id
if request["id"]==old: return
self.current_owner_id=request["id"]
self.current_priority=request["priority"]
self.current_rgb_mode=request["rgb_mode"]
self.owner_since=time.monotonic()
log("INFO","DISPLAY_OWNER",old=old,new=request["id"],priority=request["priority"])
self._render_locked(request)
def _render_locked(self,request):
if request["id"]=="NORMAL_HOME": render_home(*request["metadata"]["home"])
else: render_alert(request)
def refresh(self):
with self.lock: self._render_locked(self.requests[self.current_owner_id])
def owner_snapshot(self):
with self.lock: return dict(self.requests[self.current_owner_id])
def _rgb_worker(self):
previous_error=None; applied_mode="OFF"; animation_started=time.monotonic()
while not self.stop_event.is_set():
try:
with self.lock: wanted_mode=self.current_rgb_mode
if wanted_mode!=applied_mode:
with bus_lock:
if wanted_mode=="OFF": rgb_off()
else: rgb_begin_mode()
log("INFO","RGB_MODE",old=applied_mode,new=wanted_mode)
applied_mode=wanted_mode
if applied_mode!="OFF":
with bus_lock: rgb_engine(applied_mode,time.monotonic()-animation_started)
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)
self.stop_event.wait(max(0.04,min(0.5,RGB_UPDATE_MS/1000.0)))
def shutdown(self):
self.stop_event.set()
if self.rgb_thread.is_alive(): self.rgb_thread.join()
for name,action in (("rgb",rgb_off),("oled_clear",lambda:oled_show([("","")]*4)),
("oled_off",lambda:cmd(0xAE))):
try: action()
except Exception as e: log("ERROR","SHUTDOWN_CLEANUP",step=name,error=repr(e))
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)):
display.shutdown()
for name,action in (("fan",lambda:mcu(0x08,0x00)),("bus_close",bus.close)):
try: action()
except Exception as e: log("ERROR","SHUTDOWN_CLEANUP",step=name,error=repr(e))
@@ -535,26 +601,15 @@ last_sensor=last_render=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,"i2c_errors":0}
alerts=AlertManager()
event_state={}
history_logged=set()
latest=(0.0,0.0,0.0,0.0)
active=[]
# Page state is a snapshot queue. Live sensor refreshes NEVER choose another page.
# Each cycle is HOME + every alert captured at the cycle boundary.
cycle_pages=["HOME"]
cycle_pos=0
page_started=time.monotonic()
current_alert_key=None
led_thread=threading.Thread(target=led_worker,name="rgb",daemon=True)
display=DisplayController()
try:
bus=SMBus(1)
FONT5=Path(FONT_PATH).read_bytes()
oled_init(); rgb_off(); apply_dark_mode()
display.start(); apply_dark_mode()
log("INFO","START",version=APP_VERSION,uptime=uptime_seconds())
led_thread.start()
while running:
try:
now=time.monotonic()
@@ -598,48 +653,10 @@ try:
agent["last_health_ok"]=now; last_health=now
c,t,m,d=latest
display.update_home(c,t,m,d)
update_active_alerts(c,t,m,d,sample_elapsed)
active=alerts.active_alerts()
led_alert=max(active,key=lambda a:a["sev"],default=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):
# 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
# 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_home(c,t,m,d)
display.refresh()
last_render=now
time.sleep(0.20)