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

feat: distinguish RGB alerts by fault type

Browse Source
This commit is contained in:
way
2026-09-27 09:47:03 +08:00
parent 249af5a3e3
commit ad22dcced7
4 changed files with 55 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/pigway_pi_control.py
+37 -17
View File
@@ -123,18 +123,15 @@ def rgb_off():
def rgb_mode(mode):
global _last_rgb
settings={
"WARNING_BREATHE":(RGB_WARNING_SPEED,RGB_WARNING_COLOR,0x01),
"CRITICAL_BREATHE":(RGB_CRITICAL_SPEED,RGB_CRITICAL_COLOR,0x01),
}
state=settings.get(mode)
if state is None: raise ValueError(f"unsupported RGB mode: {mode}")
match=re.fullmatch(r"(RED|GREEN|BLUE|YELLOW|PURPLE|CYAN|WHITE)_(SLOW|FAST)_BREATHE",mode)
if match is None: raise ValueError(f"unsupported RGB mode: {mode}")
if _last_rgb==(mode,): return
speed,color,effect=state
color=RGB_COLOR_VALUES[match.group(1)]
speed=RGB_WARNING_SPEED if match.group(2)=="SLOW" else RGB_CRITICAL_SPEED
# Hardware-calibrated MCU order. The final effect write commits the mode.
mcu(0x05,speed); time.sleep(RGB_WRITE_DELAY_MS/1000.0)
mcu(0x06,color); time.sleep(RGB_WRITE_DELAY_MS/1000.0)
mcu(0x04,effect)
mcu(0x04,0x01)
_last_rgb=(mode,)
def set_pi_led(name,off=True):
@@ -267,9 +264,17 @@ DARK_ACT=cf("dark_mode","pi_act_led",True,lambda x: str(x).strip().lower() in ("
EMERGENCY_FLASH_MS=cf("led","emergency_flash_ms",300,int)
RGB_WRITE_DELAY_MS=max(1,cf("led","write_delay_ms",10,int))
RGB_WARNING_SPEED=max(1,min(3,cf("led","warning_speed",1,int)))
RGB_WARNING_COLOR=max(0,min(6,cf("led","warning_color",3,int)))
RGB_CRITICAL_SPEED=max(1,min(3,cf("led","critical_speed",3,int)))
RGB_CRITICAL_COLOR=max(0,min(6,cf("led","critical_color",0,int)))
RGB_COLOR_VALUES={"RED":0,"GREEN":1,"BLUE":2,"YELLOW":3,"PURPLE":4,"CYAN":5,"WHITE":6}
RGB_OBJECT_COLORS={
"ROOT_READONLY":"RED",
"TEMP_CRITICAL":"RED","TEMP_HIGH":"RED","SOFT_TEMP_LIMIT":"RED",
"POWER_UNDERVOLT":"YELLOW","THROTTLED":"YELLOW","FREQ_CAPPED":"YELLOW","CPU_HIGH":"YELLOW",
"MEMORY_HIGH":"PURPLE",
"DISK_CRITICAL":"WHITE","DISK_HIGH":"WHITE",
"NETWORK_DOWN":"BLUE","INTERNET_DOWN":"BLUE","WIFI_WEAK":"BLUE",
"SERVICE_DOWN":"CYAN","PROCESS_DOWN":"CYAN",
}
def check_watch(w):
typ,label,target=w
@@ -413,11 +418,11 @@ def update_active_alerts(cpu,temp,mem,disk,sample_elapsed=0.0):
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"
if a["sev"]>=3: return "EMERGENCY_FLASH"
if a["sev"]>=2: return "CRITICAL_BREATHE"
return "WARNING_BREATHE"
def rgb_mode_for(request_id,severity):
if severity<=0: return "OFF"
color=RGB_OBJECT_COLORS[request_id.split("|",1)[0]]
if severity>=3: return "EMERGENCY_FLASH"
return f"{color}_{'FAST' if severity>=2 else 'SLOW'}_BREATHE"
def rgb_emergency(on):
global _last_rgb
@@ -503,7 +508,7 @@ class DisplayController:
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],
"rgb_mode":rgb_mode_for(request_id,severity),"title":title[:21],
"l2":l2[:21],"l3":l3[:21],"l4":l4[:21],"active":True,
"last_changed":now,"metadata":metadata or {}}
with self.lock:
@@ -517,7 +522,9 @@ class DisplayController:
("priority","severity","rgb_mode","title","l2","l3","l4","metadata"))
if not changed: request["last_changed"]=current["last_changed"]
self.requests[request_id]=request
if self.batch_depth==0: self._arbitrate_locked()
if self.batch_depth==0:
self._arbitrate_locked()
self._sync_rgb_locked()
def clear(self,request_id):
if request_id=="NORMAL_HOME": return
@@ -528,6 +535,7 @@ class DisplayController:
self._arbitrate_locked()
if self.current_oled_page_id not in self.requests:
self._show_page_locked(self.current_owner_id)
else: self._sync_rgb_locked()
def begin_update(self):
with self.lock: self.batch_depth+=1
@@ -540,6 +548,7 @@ class DisplayController:
self._arbitrate_locked()
if self.current_oled_page_id not in self.requests:
self._show_page_locked(self.current_owner_id)
else: self._sync_rgb_locked()
def _arbitrate_locked(self):
current=self.requests.get(self.current_owner_id)
@@ -567,12 +576,23 @@ class DisplayController:
request=self.requests.get(page_id,self.requests["NORMAL_HOME"])
self.current_oled_page_id=request["id"]
self.page_started=time.monotonic()
self._sync_rgb_locked()
if request["id"]=="NORMAL_HOME":
render_home(*request["metadata"]["home"])
else:
alert_ids=[a["id"] for a in alerts]
render_alert(request,alert_ids.index(request["id"])+1,len(alerts))
def _sync_rgb_locked(self):
# Alert pages show that alert's object color. HOME keeps the highest-
# priority active owner visible instead of briefly switching RGB off.
rgb_request=self.requests.get(self.current_oled_page_id)
if self.current_oled_page_id=="NORMAL_HOME":
rgb_request=self.requests.get(self.current_owner_id,self.requests["NORMAL_HOME"])
if rgb_request is None:
rgb_request=self.requests["NORMAL_HOME"]
self.current_rgb_mode=rgb_request["rgb_mode"]
def refresh(self):
with self.lock:
alerts=self._ordered_alerts_locked()