fix: redesign RGB alert engine

app/pigway_pi_control.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-import time,socket,subprocess,configparser,signal,re,errno,threading,argparse
+import time,socket,subprocess,configparser,signal,re,errno,threading,argparse,math
 from pathlib import Path
 APP_VERSION="3.3.0"
 from smbus2 import SMBus
@@ -236,7 +236,7 @@ OLED_REFRESH=cf("timing","oled_refresh_interval",1.0)
 PAGE_INTERVAL=cf("timing","page_interval",10.0)
 NETWORK_INTERVAL=cf("timing","network_interval",5.0)
 SERVICE_INTERVAL=cf("timing","service_interval",5.0)
-HEALTH_INTERVAL=cf("timing","health_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)
@@ -259,9 +259,13 @@ 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"))
 LED_BRIGHT=cf("led","brightness",10,int)
-LED_WARN_ON=cf("led","warning_on_ms",300,int); LED_WARN_OFF=cf("led","warning_off_ms",2700,int)
-LED_CRIT_ON=cf("led","critical_on_ms",400,int); LED_CRIT_OFF=cf("led","critical_off_ms",1100,int)
-LED_EMERG_ON=cf("led","emergency_on_ms",400,int); LED_EMERG_OFF=cf("led","emergency_off_ms",400,int)
+WARNING_BREATHE_SECONDS=cf("led","warning_breathe_seconds",3.0)
+CRITICAL_BREATHE_SECONDS=cf("led","critical_breathe_seconds",1.5)
+EMERGENCY_FLASH_MS=cf("led","emergency_flash_ms",300,int)
+RGB_UPDATE_MS=cf("led","rgb_update_ms",40,int)
+RGB_WARNING=(cf("led","warning_r",255,int),cf("led","warning_g",255,int),cf("led","warning_b",0,int))
+RGB_CRITICAL=(cf("led","critical_r",255,int),cf("led","critical_g",72,int),cf("led","critical_b",0,int))
+RGB_EMERGENCY=(cf("led","emergency_r",255,int),cf("led","emergency_g",0,int),cf("led","emergency_b",0,int))
 
 def check_watch(w):
     typ,label,target=w
@@ -357,16 +361,29 @@ def make_alerts(cpu,temp,mem,disk,sample_elapsed=0.0):
     a.sort(key=lambda x:x["sev"],reverse=True)
     return a
 
-def led_for(a):
-    # Color = fault object. Blink frequency = severity. All alerts blink.
-    if not a:
-        rgb_off(); return
-    if a["sev"]>=3: on_ms,off_ms=LED_EMERG_ON,LED_EMERG_OFF
-    elif a["sev"]>=2: on_ms,off_ms=LED_CRIT_ON,LED_CRIT_OFF
-    else: on_ms,off_ms=LED_WARN_ON,LED_WARN_OFF
-    cycle=max(1,on_ms+off_ms)
-    phase=int(time.monotonic()*1000)%cycle
-    rgb(*a["color"]) if phase<on_ms else rgb_off()
+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 breathe(level_seconds,elapsed):
+    # A full sine-derived ease in/out cycle gives continuous light transitions.
+    phase=(elapsed/max(0.1,level_seconds))%1.0
+    return 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":
+        level=breathe(WARNING_BREATHE_SECONDS,elapsed)
+        rgb(*(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))
+    else:
+        on=int(elapsed*1000/max(50,EMERGENCY_FLASH_MS))%2==0
+        rgb(RGB_EMERGENCY[0] if on else 0,0,0)
 
 
 def render_home(cpu,temp,mem,disk):
@@ -417,21 +434,29 @@ def find_live_alert(key, live):
         if alert_key(a)==key: return a
     return None
 
-# RGB has its own clock so slow network/watch checks cannot stretch blink pulses.
+# 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 led_worker():
     previous_error=None
+    current_mode="OFF"
+    mode_started=time.monotonic()
     while not led_stop.is_set():
         try:
-            with bus_lock: led_for(led_alert)
+            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; mode_started=time.monotonic()
+            # Keep the verified five-write MCU RGB transaction contiguous.
+            with bus_lock: rgb_engine(current_mode,time.monotonic()-mode_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(0.02)
+        led_stop.wait(max(0.02,min(0.05,RGB_UPDATE_MS/1000.0)))
 
 def cleanup():
     log("INFO","STOP",loop_errors=agent["loop_errors"],i2c_errors=agent["i2c_errors"])