From 45aaba0c8a5c1441b65f005702e635a0e3b870d4 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Fri, 15 May 2026 15:11:59 +0800 Subject: [PATCH] fix: make led panel controls write to sysfs reliably --- README.md | 12 ++++++++ led_panel.py | 77 +++++++++++++++++++++++++++++++++-------------- test_led_panel.py | 52 ++++++++++++++++++++++++++++++++ 3 files changed, 118 insertions(+), 23 deletions(-) create mode 100644 test_led_panel.py diff --git a/README.md b/README.md index e8c75e0..daebef1 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,18 @@ blue:wan green:wlan mmc0:: red:power sim:en sim:en2 sim:sel sim:sel2 - “识别 3 秒”闪烁指定 LED - 无数据库、无 Node、无 pip 依赖 +## 更新已有安装 + +```bash +curl -fsSL https://gitea.chickliu.fun/Hermes/led-manager-panel/raw/branch/main/install.sh | sh +``` + +如果已经在浏览器打开旧页面，更新后请强制刷新一次页面，避免浏览器缓存旧 JS： + +```text +Ctrl+F5 / Cmd+Shift+R +``` + ## 一键安装 ```bash diff --git a/led_panel.py b/led_panel.py index 2e415e0..508495c 100755 --- a/led_panel.py +++ b/led_panel.py @@ -61,22 +61,24 @@ async function api(path, body) { return data; } function toast(msg, bad=false){ const t=document.getElementById('toast'); t.textContent=msg; t.style.display='block'; t.style.background=bad?'#450a0a':'#052e2b'; t.style.borderColor=bad?'#fb7185':'#14b8a6'; setTimeout(()=>t.style.display='none',2600); } -async function setBrightness(name, value){ try{ await api('/api/brightness',{name,value:Number(value)}); toast(name+' brightness='+value); setTimeout(load,300);}catch(e){toast(e.message,true);} } -async function setTrigger(name, trigger){ try{ await api('/api/trigger',{name,trigger}); toast(name+' trigger='+trigger); setTimeout(load,300);}catch(e){toast(e.message,true);} } -async function setDelay(name, field, value){ try{ await api('/api/delay',{name,field,value:Number(value)}); toast(name+' '+field+'='+value); }catch(e){toast(e.message,true);} } +async function setBrightness(name, value){ try{ const r=await api('/api/brightness',{name,value:Number(value)}); toast(name+' brightness='+(r.brightness ?? value)+(r.warning?'；'+r.warning:''), !!r.warning); setTimeout(load,300);}catch(e){toast(e.message,true);} } +async function setTrigger(name, trigger){ try{ const r=await api('/api/trigger',{name,trigger}); toast(name+' trigger='+(r.active_trigger || trigger)); setTimeout(load,300);}catch(e){toast(e.message,true);} } +async function setDelay(name, field, value){ try{ await api('/api/delay',{name,field,value:Number(value)}); toast(name+' '+field+'='+value); setTimeout(load,300);}catch(e){toast(e.message,true);} } async function identify(name){ try{ await api('/api/identify',{name}); toast('Blink '+name); setTimeout(load,3200);}catch(e){toast(e.message,true);} } function colorOf(n,b){ const s=n.toLowerCase(); if(Number(b)<=0) return '#475569'; if(s.includes('red')) return '#ef4444'; if(s.includes('green')) return '#22c55e'; if(s.includes('blue')) return '#3b82f6'; if(s.includes('wlan')) return '#22c55e'; if(s.includes('wan')) return '#3b82f6'; if(s.includes('power')) return '#ef4444'; if(s.includes('sim')) return '#f59e0b'; return '#5eead4'; } function esc(s){ return String(s).replace(/[&<>'"]/g, c=>({'&':'&','<':'<','>':'>',"'":''','"':'"'}[c])); } +function attr(s){ return esc(s); } function card(led){ const pct = led.max_brightness>0 ? Math.round(led.brightness*100/led.max_brightness) : 0; const triggers = led.triggers || []; - const triggerSelect = triggers.length ? `` : '无 trigger 文件'; - const delay = (led.delay_on !== null || led.delay_off !== null) ? `
timer:
` : ''; + const n = attr(led.name); + const triggerSelect = triggers.length ? `` : '无 trigger 文件'; + const delay = (led.delay_on !== null || led.delay_off !== null) ? `
timer:
` : ''; return `
${esc(led.name)} ${led.writable?'可写':'只读/权限不足'}
brightness: ${led.brightness} / ${led.max_brightness} (${pct}%)
trigger: ${esc(led.active_trigger || '-')}
path: ${esc(led.path)}
-
-
+
+
trigger${triggerSelect}
${delay}
`; } @@ -97,14 +99,14 @@ def safe_name(name: str) -> str: def led_path(name: str) -> Path: name = safe_name(name) + # /sys/class/leds entries are usually symlinks to /sys/devices/... + # Do NOT require the resolved target to stay under LED_ROOT, otherwise every real sysfs LED POST is rejected. + # safe_name() forbids slash and '..', so joining one basename under LED_ROOT is enough here. p = LED_ROOT / name - try: - root = LED_ROOT.resolve(strict=True) - resolved = p.resolve(strict=True) - except FileNotFoundError: + if not p.exists(): raise ValueError("led not found") - if root not in resolved.parents and resolved != root: - raise ValueError("invalid led path") + if not (p / "brightness").exists(): + raise ValueError("not a led sysfs entry") return p @@ -116,7 +118,33 @@ def read_text(path: Path) -> Optional[str]: def write_text(path: Path, value: str) -> None: - path.write_text(str(value)) + with path.open("w") as f: + f.write(str(value)) + f.flush() + + +def active_trigger_name(raw: Optional[str]) -> Optional[str]: + _, active = parse_triggers(raw) + return active + + +def set_trigger(p: Path, trigger: str) -> bool: + trigger_file = p / "trigger" + if not trigger_file.exists(): + return False + raw = read_text(trigger_file) + triggers, active = parse_triggers(raw) + names = {str(t["name"]) for t in triggers} + if names and trigger not in names: + raise ValueError(f"trigger not supported: {trigger}") + if active == trigger: + return True + write_text(trigger_file, trigger) + # Verify because LED sysfs can accept write but keep previous trigger if unsupported/raced. + new_active = active_trigger_name(read_text(trigger_file)) + if new_active and new_active != trigger: + raise ValueError(f"trigger write did not take effect: requested={trigger}, active={new_active}") + return True def parse_triggers(raw: Optional[str]) -> Tuple[List[Dict[str, object]], Optional[str]]: @@ -217,20 +245,23 @@ class Handler(BaseHTTPRequestHandler): max_brightness = int(read_text(p / "max_brightness") or 1) value = int(data.get("value", 0)) value = max(0, min(max_brightness, value)) - # manual brightness usually needs trigger none first, ignore if unsupported + # Manual brightness only takes effect when trigger is none. + # Some device LEDs do not support none; in that case still write brightness and report the actual state. + trigger_warning = None if (p / "trigger").exists(): try: - write_text(p / "trigger", "none") - except Exception: - pass + set_trigger(p, "none") + except Exception as e: + trigger_warning = str(e) write_text(p / "brightness", str(value)) - self.send_json({"ok": True}) + actual = read_text(p / "brightness") + self.send_json({"ok": True, "brightness": actual, "warning": trigger_warning}) elif parsed.path == "/api/trigger": trigger = str(data.get("trigger", "")) if not NAME_RE.match(trigger): raise ValueError("invalid trigger") - write_text(p / "trigger", trigger) - self.send_json({"ok": True}) + set_trigger(p, trigger) + self.send_json({"ok": True, "active_trigger": active_trigger_name(read_text(p / "trigger"))}) elif parsed.path == "/api/delay": field = str(data.get("field", "")) if field not in ("delay_on", "delay_off"): @@ -243,7 +274,7 @@ class Handler(BaseHTTPRequestHandler): old_brightness = read_text(p / "brightness") try: if (p / "trigger").exists(): - write_text(p / "trigger", "timer") + set_trigger(p, "timer") if (p / "delay_on").exists(): write_text(p / "delay_on", "200") if (p / "delay_off").exists(): write_text(p / "delay_off", "200") else: @@ -256,7 +287,7 @@ class Handler(BaseHTTPRequestHandler): if old_trigger and "[" in old_trigger: _, active = parse_triggers(old_trigger) if active: - try: write_text(p / "trigger", active) + try: set_trigger(p, active) except Exception: pass if old_brightness is not None: try: write_text(p / "brightness", old_brightness) diff --git a/test_led_panel.py b/test_led_panel.py new file mode 100644 index 0000000..58cf251 --- /dev/null +++ b/test_led_panel.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +import json +import tempfile +import threading +import urllib.request +from pathlib import Path +from led_panel import Handler, ThreadingHTTPServer +import led_panel + + +def req(port, path, body=None): + data = None + headers = {} + if body is not None: + data = json.dumps(body).encode() + headers['Content-Type'] = 'application/json' + r = urllib.request.Request(f'http://127.0.0.1:{port}{path}', data=data, headers=headers) + with urllib.request.urlopen(r, timeout=5) as resp: + return json.loads(resp.read().decode()) + + +def main(): + with tempfile.TemporaryDirectory() as td: + root = Path(td) + d = root / 'blue:wan' + d.mkdir() + (d / 'brightness').write_text('0') + (d / 'max_brightness').write_text('1') + (d / 'trigger').write_text('[none] timer') + (d / 'delay_on').write_text('500') + (d / 'delay_off').write_text('500') + led_panel.LED_ROOT = root + httpd = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + port = httpd.server_address[1] + t = threading.Thread(target=httpd.serve_forever, daemon=True) + t.start() + try: + leds = req(port, '/api/leds') + assert leds['ok'] and leds['leds'][0]['name'] == 'blue:wan' + b = req(port, '/api/brightness', {'name':'blue:wan','value':1}) + assert b['ok'] and (d / 'brightness').read_text() == '1' + tr = req(port, '/api/trigger', {'name':'blue:wan','trigger':'timer'}) + assert tr['ok'] + assert (d / 'trigger').read_text() == 'timer' + delay = req(port, '/api/delay', {'name':'blue:wan','field':'delay_on','value':123}) + assert delay['ok'] and (d / 'delay_on').read_text() == '123' + print('OK: API write regression passed') + finally: + httpd.shutdown() + +if __name__ == '__main__': + main()