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

fix: use calibrated MCU RGB effects

Browse Source
This commit is contained in:
way
2026-09-27 09:25:19 +08:00
parent a84f4895ef
commit 249af5a3e3
4 changed files with 106 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/pigway_pi_control.py
+83 -50
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
import time,socket,subprocess,configparser,signal,re,errno,threading,argparse,math
import time,socket,subprocess,configparser,signal,re,errno,threading,argparse
from pathlib import Path
APP_VERSION="3.3.0"
from smbus2 import SMBus
@@ -121,19 +121,21 @@ def rgb_off():
mcu(0x07,0x00) # seller closeRGB(): true OFF
_last_rgb=("off",)
def rgb_begin_mode():
def rgb_mode(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(0x00,0xff) # seller protocol: >=3 => all three LEDs
mcu(0x01,state[0]); mcu(0x02,state[1]); mcu(0x03,state[2])
_last_rgb=state
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}")
if _last_rgb==(mode,): return
speed,color,effect=state
# 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)
_last_rgb=(mode,)
def set_pi_led(name,off=True):
base=Path("/sys/class/leds")/name
@@ -262,14 +264,12 @@ FAN_HYST=cf("fan","hysteresis",1.5)
ROW_Y=[cf("display","row1_y",0,int),cf("display","row2_y",8,int),cf("display","row3_y",16,int),cf("display","row4_y",24,int)]
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)
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",80,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))
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)))
def check_watch(w):
typ,label,target=w
@@ -419,22 +419,18 @@ def rgb_mode_for(a):
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.05+0.95*(0.5-0.5*math.cos(2.0*math.pi*phase))
def rgb_engine(mode,elapsed):
if mode=="WARNING_BREATHE":
level=breathe(WARNING_BREATHE_SECONDS,elapsed)
rgb_frame(*(channel*level for channel in RGB_WARNING))
elif mode=="CRITICAL_BREATHE":
level=breathe(CRITICAL_BREATHE_SECONDS,elapsed)
rgb_frame(*(channel*level for channel in RGB_CRITICAL))
def rgb_emergency(on):
global _last_rgb
state=("EMERGENCY_FLASH",bool(on))
if _last_rgb==state: return
if not on:
mcu(0x07,0x00)
else:
on=int(elapsed*1000/max(50,EMERGENCY_FLASH_MS))%2==0
rgb_frame(RGB_EMERGENCY[0] if on else 0,0,0)
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)
_last_rgb=state
def render_home(cpu,temp,mem,disk):
# 128x32 fixed columns. Both columns are LEFT-aligned.
@@ -470,8 +466,8 @@ def render_home(cpu,temp,mem,disk):
bus_write("write_i2c_block_data",OLED,0x40,data[i:i+16])
def render_alert(a):
oled_show([(a["title"],""),
def render_alert(a,index,total):
oled_show([(a["title"],f"{index}/{total}"),
(a["l2"],""),(a["l3"],""),(a["l4"],"")])
@@ -488,12 +484,14 @@ class DisplayController:
self.current_priority=0
self.current_rgb_mode="OFF"
self.owner_since=now
self.current_oled_page_id="NORMAL_HOME"
self.page_started=now
self.batch_depth=0
self.stop_event=threading.Event()
self.rgb_thread=threading.Thread(target=self._rgb_worker,name="rgb",daemon=True)
def start(self):
oled_init(); rgb_off(); self.refresh(); self.rgb_thread.start()
oled_init(); self.refresh(); rgb_off(); self.rgb_thread.start()
def update_home(self,cpu,temp,mem,disk):
with self.lock:
@@ -526,7 +524,10 @@ class DisplayController:
with self.lock:
if self.requests.pop(request_id,None) is not None:
log("INFO","ALERT_RECOVERED",alert=request_id)
if self.batch_depth==0: self._arbitrate_locked()
if self.batch_depth==0:
self._arbitrate_locked()
if self.current_oled_page_id not in self.requests:
self._show_page_locked(self.current_owner_id)
def begin_update(self):
with self.lock: self.batch_depth+=1
@@ -535,7 +536,10 @@ class DisplayController:
with self.lock:
if self.batch_depth<=0: raise RuntimeError("display update batch imbalance")
self.batch_depth-=1
if self.batch_depth==0: self._arbitrate_locked()
if self.batch_depth==0:
self._arbitrate_locked()
if self.current_oled_page_id not in self.requests:
self._show_page_locked(self.current_owner_id)
def _arbitrate_locked(self):
current=self.requests.get(self.current_owner_id)
@@ -552,14 +556,39 @@ class DisplayController:
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)
self._show_page_locked(request["id"])
def _render_locked(self,request):
if request["id"]=="NORMAL_HOME": render_home(*request["metadata"]["home"])
else: render_alert(request)
def _ordered_alerts_locked(self):
return sorted((r for r in self.requests.values() if r["id"]!="NORMAL_HOME"),
key=lambda r:(-r["priority"],r["first_seen"],r["id"]))
def _show_page_locked(self,page_id):
alerts=self._ordered_alerts_locked()
request=self.requests.get(page_id,self.requests["NORMAL_HOME"])
self.current_oled_page_id=request["id"]
self.page_started=time.monotonic()
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 refresh(self):
with self.lock: self._render_locked(self.requests[self.current_owner_id])
with self.lock:
alerts=self._ordered_alerts_locked()
pages=["NORMAL_HOME"]+[a["id"] for a in alerts]
if self.current_oled_page_id not in pages:
self._show_page_locked(self.current_owner_id)
return
if len(pages)>1 and time.monotonic()-self.page_started>=PAGE_INTERVAL:
pos=pages.index(self.current_oled_page_id)
self._show_page_locked(pages[(pos+1)%len(pages)])
return
request=self.requests[self.current_oled_page_id]
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 owner_snapshot(self):
with self.lock: return dict(self.requests[self.current_owner_id])
@@ -572,17 +601,21 @@ class DisplayController:
if wanted_mode!=applied_mode:
with bus_lock:
if wanted_mode=="OFF": rgb_off()
else: rgb_begin_mode()
elif wanted_mode=="EMERGENCY_FLASH": rgb_emergency(True)
else: rgb_mode(wanted_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)
applied_mode=wanted_mode; animation_started=time.monotonic()
if applied_mode=="EMERGENCY_FLASH":
elapsed=time.monotonic()-animation_started
on=int(elapsed*1000/max(50,EMERGENCY_FLASH_MS))%2==0
with bus_lock: rgb_emergency(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)
self.stop_event.wait(max(0.04,min(0.5,RGB_UPDATE_MS/1000.0)))
delay=0.04 if applied_mode=="EMERGENCY_FLASH" else 0.25
self.stop_event.wait(delay)
def shutdown(self):
self.stop_event.set()