'; }
}
window.addEventListener('load', ()=>{ load(); setInterval(load, 5000); });
"""
def safe_name(name: str) -> str:
name = urllib.parse.unquote(name)
if not NAME_RE.match(name) or "/" in name or ".." in name:
raise ValueError("invalid led name")
return name
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
if not p.exists():
raise ValueError("led not found")
if not (p / "brightness").exists():
raise ValueError("not a led sysfs entry")
return p
def read_text(path: Path) -> Optional[str]:
try:
return path.read_text().strip()
except Exception:
return None
def write_text(path: Path, value: str) -> None:
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]]:
if not raw:
return [], None
out = []
active = None
for part in raw.split():
is_active = part.startswith("[") and part.endswith("]")
name = part[1:-1] if is_active else part
if is_active:
active = name
out.append({"name": name, "active": is_active})
return out, active
def list_leds() -> List[Dict[str, object]]:
leds = []
if not LED_ROOT.exists():
return leds
for p in sorted(LED_ROOT.iterdir(), key=lambda x: x.name):
if not p.is_dir():
continue
name = p.name
b = read_text(p / "brightness")
mb = read_text(p / "max_brightness")
tr_raw = read_text(p / "trigger")
triggers, active = parse_triggers(tr_raw)
delay_on = read_text(p / "delay_on")
delay_off = read_text(p / "delay_off")
try:
brightness = int(b or 0)
except ValueError:
brightness = 0
try:
max_brightness = int(mb or 1)
except ValueError:
max_brightness = 1
writable = os.access(p / "brightness", os.W_OK)
leds.append({
"name": name,
"path": str(p),
"brightness": brightness,
"max_brightness": max_brightness,
"triggers": triggers,
"active_trigger": active,
"delay_on": int(delay_on) if delay_on and delay_on.isdigit() else None,
"delay_off": int(delay_off) if delay_off and delay_off.isdigit() else None,
"writable": writable,
})
return leds
class Handler(BaseHTTPRequestHandler):
server_version = "LedPanel/1.0"
def log_message(self, fmt: str, *args):
sys.stderr.write("%s - - [%s] %s\n" % (self.client_address[0], self.log_date_time_string(), fmt % args))
def send_json(self, obj: object, status: int = 200):
data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Cache-Control", "no-store")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def read_body(self) -> Dict[str, object]:
length = int(self.headers.get("Content-Length", "0") or 0)
if length > 4096:
raise ValueError("body too large")
raw = self.rfile.read(length).decode("utf-8") if length else "{}"
return json.loads(raw or "{}")
def do_GET(self):
parsed = urllib.parse.urlparse(self.path)
if parsed.path == "/":
body = f"""LED Manager