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

fix: align RGB colors with fault domains

Browse Source
This commit is contained in:
way
2026-09-27 09:54:45 +08:00
parent ad22dcced7
commit ac9bc4befd
4 changed files with 37 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/pigway_pi_control.py
+29 -15
View File
@@ -262,14 +262,21 @@ ROW_Y=[cf("display","row1_y",0,int),cf("display","row2_y",8,int),cf("display","r
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"))
EMERGENCY_FLASH_MS=cf("led","emergency_flash_ms",300,int)
POWER_BLINK_MS=cf("led","power_blink_ms",750,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_CRITICAL_SPEED=max(1,min(3,cf("led","critical_speed",3,int)))
RGB_COLOR_VALUES={"RED":0,"GREEN":1,"BLUE":2,"YELLOW":3,"PURPLE":4,"CYAN":5,"WHITE":6}
RGB_STATIC_VALUES={
"RED":(0xff,0x00,0x00),"GREEN":(0x00,0xff,0x00),"BLUE":(0x00,0x00,0xff),
"YELLOW":(0xff,0xff,0x00),"PURPLE":(0xff,0x00,0xff),"CYAN":(0x00,0xff,0xff),
"WHITE":(0xff,0xff,0xff),"ORANGE":(0xff,0x28,0x00),
}
RGB_OBJECT_COLORS={
"ROOT_READONLY":"RED",
"ROOT_READONLY":"WHITE",
"TEMP_CRITICAL":"RED","TEMP_HIGH":"RED","SOFT_TEMP_LIMIT":"RED",
"POWER_UNDERVOLT":"YELLOW","THROTTLED":"YELLOW","FREQ_CAPPED":"YELLOW","CPU_HIGH":"YELLOW",
"THROTTLED":"RED","FREQ_CAPPED":"RED","CPU_HIGH":"RED",
"POWER_UNDERVOLT":"YELLOW",
"MEMORY_HIGH":"PURPLE",
"DISK_CRITICAL":"WHITE","DISK_HIGH":"WHITE",
"NETWORK_DOWN":"BLUE","INTERNET_DOWN":"BLUE","WIFI_WEAK":"BLUE",
@@ -420,21 +427,25 @@ def update_active_alerts(cpu,temp,mem,disk,sample_elapsed=0.0):
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"
alert_type=request_id.split("|",1)[0]
if alert_type=="POWER_UNDERVOLT": return "ORANGE_CRITICAL_BLINK"
color=RGB_OBJECT_COLORS[alert_type]
if severity>=3: return f"{color}_EMERGENCY_FLASH"
return f"{color}_{'FAST' if severity>=2 else 'SLOW'}_BREATHE"
def rgb_emergency(on):
def rgb_flash(mode,on):
global _last_rgb
state=("EMERGENCY_FLASH",bool(on))
color=mode.split("_",1)[0]
channels=RGB_STATIC_VALUES[color]
state=(mode,bool(on))
if _last_rgb==state: return
if not on:
mcu(0x07,0x00)
else:
mcu(0x00,0xff); time.sleep(RGB_WRITE_DELAY_MS/1000.0)
mcu(0x01,0xff); time.sleep(RGB_WRITE_DELAY_MS/1000.0)
mcu(0x02,0x00); time.sleep(RGB_WRITE_DELAY_MS/1000.0)
mcu(0x03,0x00)
mcu(0x01,channels[0]); time.sleep(RGB_WRITE_DELAY_MS/1000.0)
mcu(0x02,channels[1]); time.sleep(RGB_WRITE_DELAY_MS/1000.0)
mcu(0x03,channels[2])
_last_rgb=state
def render_home(cpu,temp,mem,disk):
@@ -614,27 +625,30 @@ class DisplayController:
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()
previous_error=None; applied_mode="OFF"; applied_flash=False; animation_started=time.monotonic()
while not self.stop_event.is_set():
try:
with self.lock: wanted_mode=self.current_rgb_mode
wanted_flash=wanted_mode.endswith(("_EMERGENCY_FLASH","_CRITICAL_BLINK"))
if wanted_mode!=applied_mode:
with bus_lock:
if wanted_mode=="OFF": rgb_off()
elif wanted_mode=="EMERGENCY_FLASH": rgb_emergency(True)
elif wanted_flash: rgb_flash(wanted_mode,True)
else: rgb_mode(wanted_mode)
log("INFO","RGB_MODE",old=applied_mode,new=wanted_mode)
applied_mode=wanted_mode; animation_started=time.monotonic()
if applied_mode=="EMERGENCY_FLASH":
applied_flash=applied_mode.endswith(("_EMERGENCY_FLASH","_CRITICAL_BLINK"))
if applied_flash:
elapsed=time.monotonic()-animation_started
on=int(elapsed*1000/max(50,EMERGENCY_FLASH_MS))%2==0
with bus_lock: rgb_emergency(on)
interval=POWER_BLINK_MS if applied_mode=="ORANGE_CRITICAL_BLINK" else EMERGENCY_FLASH_MS
on=int(elapsed*1000/max(50,interval))%2==0
with bus_lock: rgb_flash(applied_mode,on)
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)
delay=0.04 if applied_mode=="EMERGENCY_FLASH" else 0.25
delay=0.04 if applied_flash else 0.25
self.stop_event.wait(delay)
def shutdown(self):