'; }
}
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_color(name: str) -> Optional[str]:
low = name.lower()
for color in CONTROLLABLE_COLORS:
if low == color or low.startswith(color + ":") or low.startswith(color + "-") or low.startswith(color + "_"):
return color
return None
def is_controllable_led_name(name: str) -> bool:
return not CONTROLLABLE_COLORS or led_color(name) is not None
def led_path(name: str) -> Path:
name = safe_name(name)
if not is_controllable_led_name(name):
raise ValueError("led hidden by color filter")
# /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 load_config() -> Dict[str, Dict[str, object]]:
try:
if not CONFIG_PATH.exists():
return {}
data = json.loads(CONFIG_PATH.read_text())
if not isinstance(data, dict):
return {}
leds = data.get("leds", data)
if not isinstance(leds, dict):
return {}
out = {}
for k, v in leds.items():
if not isinstance(v, dict):
continue
name = safe_name(str(k))
if is_controllable_led_name(name):
out[name] = v
return out
except Exception:
return {}
def save_config(config: Dict[str, Dict[str, object]]) -> None:
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
tmp = CONFIG_PATH.with_suffix(CONFIG_PATH.suffix + ".tmp")
filtered = {safe_name(str(k)): v for k, v in config.items() if isinstance(v, dict) and is_controllable_led_name(safe_name(str(k)))}
payload = {"version": 1, "colors": list(CONTROLLABLE_COLORS), "leds": filtered}
tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n")
os.replace(tmp, CONFIG_PATH)
try:
os.chmod(CONFIG_PATH, 0o644)
except Exception:
pass
def current_led_config(name: str) -> Dict[str, object]:
p = led_path(name)
cfg: Dict[str, object] = {}
active = active_trigger_name(read_text(p / "trigger"))
if active:
cfg["trigger"] = active
b = read_text(p / "brightness")
if b is not None:
try: cfg["brightness"] = int(b)
except ValueError: pass
for field in ("delay_on", "delay_off"):
v = read_text(p / field)
if v is not None:
try: cfg[field] = int(v)
except ValueError: pass
return cfg
def apply_led_config(name: str, cfg: Dict[str, object]) -> Dict[str, object]:
p = led_path(name)
result: Dict[str, object] = {"name": name, "ok": True, "errors": []}
# Apply trigger first so timer/netdev files appear before setting delay fields.
if "trigger" in cfg:
try:
set_trigger(p, str(cfg["trigger"]))
except Exception as e:
result["ok"] = False
result["errors"].append(f"trigger: {e}")
for field in ("delay_on", "delay_off"):
if field in cfg and (p / field).exists():
try:
write_text(p / field, str(int(cfg[field])))
except Exception as e:
result["ok"] = False
result["errors"].append(f"{field}: {e}")
if "brightness" in cfg:
try:
max_brightness = int(read_text(p / "max_brightness") or 1)
value = max(0, min(max_brightness, int(cfg["brightness"])))
# If trigger is not explicitly set, try none for manual brightness.
if "trigger" not in cfg and (p / "trigger").exists():
try: set_trigger(p, "none")
except Exception: pass
write_text(p / "brightness", str(value))
except Exception as e:
result["ok"] = False
result["errors"].append(f"brightness: {e}")
return result
def apply_saved_config() -> List[Dict[str, object]]:
config = load_config()
results = []
for name, cfg in config.items():
try:
results.append(apply_led_config(name, cfg))
except Exception as e:
results.append({"name": name, "ok": False, "errors": [str(e)]})
return results
def turn_off_all_leds() -> List[Dict[str, object]]:
results = []
for led in list_leds():
name = str(led["name"])
result: Dict[str, object] = {"name": name, "ok": True, "errors": []}
try:
p = led_path(name)
if (p / "trigger").exists():
try:
set_trigger(p, "none")
except Exception as e:
result["ok"] = False
result["errors"].append(f"trigger: {e}")
write_text(p / "brightness", "0")
try:
result["brightness"] = int(read_text(p / "brightness") or 0)
except ValueError:
result["brightness"] = read_text(p / "brightness")
except Exception as e:
result["ok"] = False
result["errors"].append(str(e))
results.append(result)
return results
def list_leds() -> List[Dict[str, object]]:
leds = []
saved_config = load_config()
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
if not is_controllable_led_name(name):
continue
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,
"saved": name in saved_config,
})
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