${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()