520 lines
29 KiB
Python
Executable File
520 lines
29 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Linux LED Manager Panel
|
||
A tiny dependency-free web panel for /sys/class/leds on Debian/OpenWrt/Linux.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import html
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
import urllib.parse
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
LED_ROOT = Path(os.environ.get("LED_ROOT", "/sys/class/leds"))
|
||
CONFIG_PATH = Path(os.environ.get("LED_CONFIG", "/etc/led-manager/config.json"))
|
||
# BOSS's device exposes more LED class entries than the physically controllable lights.
|
||
# Keep the panel focused on the real RGB lights by default; override with --colors if needed.
|
||
CONTROLLABLE_COLORS = tuple(c.strip().lower() for c in os.environ.get("LED_COLORS", "red,green,blue").split(",") if c.strip())
|
||
NAME_RE = re.compile(r"^[A-Za-z0-9_.:+@=-]+$")
|
||
|
||
CSS = r"""
|
||
:root { color-scheme: dark; --bg:#070b16; --card:#101827; --muted:#8ea0c0; --text:#eef5ff; --line:#25324a; --accent:#5eead4; --danger:#fb7185; --warn:#fbbf24; }
|
||
* { box-sizing: border-box; }
|
||
body { margin:0; min-height:100vh; font-family: -apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,Arial,sans-serif; background: radial-gradient(circle at top left,rgba(59,130,246,.26),transparent 28rem), radial-gradient(circle at top right,rgba(20,184,166,.18),transparent 24rem), linear-gradient(135deg,#070b16 0%,#0b1020 58%,#111827 100%); color:var(--text); }
|
||
.container { max-width: 1180px; margin: 0 auto; padding: 22px; }
|
||
.header { display:grid; grid-template-columns: minmax(0,1fr) auto; gap:16px; align-items:start; margin-bottom:18px; }
|
||
.hero { border:1px solid rgba(148,163,184,.18); background:rgba(15,23,42,.62); backdrop-filter: blur(16px); border-radius:24px; padding:18px 20px; box-shadow:0 18px 48px rgba(0,0,0,.28); }
|
||
h1 { margin:0 0 8px; font-size:30px; letter-spacing:.2px; }
|
||
.sub { color:var(--muted); font-size:14px; line-height:1.6; }
|
||
.toolbar { display:flex; gap:10px; flex-wrap:wrap; justify-content:flex-end; }
|
||
.summary { display:flex; gap:10px; flex-wrap:wrap; margin-top:14px; }
|
||
.pill { border:1px solid rgba(148,163,184,.18); background:rgba(15,23,42,.72); color:var(--muted); padding:7px 10px; border-radius:999px; font-size:13px; }
|
||
.pill b { color:var(--text); }
|
||
.grid { display:grid; grid-template-columns: repeat(3,minmax(0,1fr)); gap:16px; }
|
||
.card { position:relative; overflow:hidden; background: linear-gradient(180deg,rgba(18,26,47,.96),rgba(10,15,28,.96)); border:1px solid var(--line); border-radius:24px; padding:18px; box-shadow: 0 14px 42px rgba(0,0,0,.28); }
|
||
.card::before { content:''; position:absolute; inset:-40% -30% auto auto; width:180px; height:180px; border-radius:50%; background:radial-gradient(circle,var(--led-color),transparent 68%); opacity:.16; pointer-events:none; }
|
||
.card.on { border-color: color-mix(in srgb, var(--led-color) 45%, var(--line)); }
|
||
.card.off { opacity:.82; }
|
||
.card-head { display:flex; align-items:center; justify-content:space-between; gap:12px; }
|
||
.name { font-weight:800; font-size:19px; display:flex; align-items:center; gap:10px; word-break:break-all; }
|
||
.dot { width:14px; height:14px; border-radius:50%; background:var(--led-color); color:var(--led-color); box-shadow:0 0 18px currentColor; }
|
||
.bulb-wrap { display:grid; place-items:center; margin:18px 0 12px; }
|
||
.bulb { width:118px; height:118px; border-radius:999px; display:grid; place-items:center; color:#020617; font-size:28px; font-weight:900; background:radial-gradient(circle at 34% 30%,#fff 0 4%,var(--led-color) 10%,rgba(71,85,105,.24) 72%); box-shadow:0 0 calc(12px + var(--level)*42px) color-mix(in srgb, var(--led-color) 72%, transparent), inset 0 0 28px rgba(255,255,255,.12); filter:saturate(calc(.55 + var(--level))); }
|
||
.card.off .bulb { background:#1e293b; color:#64748b; box-shadow:inset 0 0 24px rgba(0,0,0,.35); }
|
||
.meta { margin-top:8px; color:var(--muted); font-size:13px; line-height:1.55; }
|
||
.row { display:flex; gap:8px; flex-wrap:wrap; align-items:center; margin-top:12px; }
|
||
.stack { display:grid; gap:10px; margin-top:14px; }
|
||
button, select, input[type=number], input[type=text] { border:1px solid var(--line); background:#0f172a; color:var(--text); border-radius:12px; padding:10px 12px; font-size:14px; }
|
||
button { cursor:pointer; transition:.15s ease; }
|
||
button:hover { border-color:var(--accent); transform: translateY(-1px); }
|
||
button.primary { background:#0f766e; border-color:#14b8a6; }
|
||
button.off { background:#3f1d2a; border-color:#7f1d1d; }
|
||
button.warn { background:#422006; border-color:#92400e; }
|
||
.actions { display:grid; grid-template-columns:1fr 1fr; gap:8px; }
|
||
.actions button { width:100%; }
|
||
.sub-actions button { padding:8px 10px; font-size:13px; }
|
||
input[type=range] { width:100%; accent-color: var(--led-color); }
|
||
.range-line { display:grid; grid-template-columns: 1fr 66px; gap:10px; align-items:center; margin-top:12px; }
|
||
.badge { color:#07111f; background:var(--accent); padding:3px 8px; border-radius:999px; font-size:12px; font-weight:800; }
|
||
.badge.saved { background:#fbbf24; }
|
||
.err { background:#450a0a; border:1px solid #7f1d1d; color:#fecaca; padding:12px; border-radius:14px; margin-bottom:16px; }
|
||
.toast { position:fixed; right:16px; bottom:16px; max-width:520px; padding:12px 14px; border-radius:14px; background:#052e2b; border:1px solid #14b8a6; box-shadow:0 18px 40px rgba(0,0,0,.35); display:none; z-index:20; }
|
||
.small { font-size:12px; color:var(--muted); }
|
||
.path { word-break:break-all; }
|
||
a { color:var(--accent); }
|
||
@media (max-width: 900px) { .grid{grid-template-columns:1fr;} .header{grid-template-columns:1fr;} .toolbar{justify-content:flex-start;} }
|
||
@media (max-width: 640px) { .container{padding:14px;} h1{font-size:25px;} .hero{padding:16px;} .bulb{width:96px;height:96px;} .actions{grid-template-columns:1fr;} }
|
||
"""
|
||
|
||
JS = r"""
|
||
async function api(path, body) {
|
||
const opt = body ? {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify(body)} : {};
|
||
const r = await fetch(path, opt);
|
||
const data = await r.json();
|
||
if (!r.ok || data.ok === false) throw new Error(data.error || ('HTTP '+r.status));
|
||
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{ const r=await api('/api/brightness',{name,value:Number(value)}); toast(name+' 亮度='+(r.brightness ?? value)+(r.warning?'；'+r.warning:''), !!r.warning); setTimeout(load,260);}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,260);}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,260);}catch(e){toast(e.message,true);} }
|
||
async function identify(name){ try{ await api('/api/identify',{name}); toast('正在识别 '+name); setTimeout(load,3200);}catch(e){toast(e.message,true);} }
|
||
async function saveLed(name){ try{ await api('/api/save',{name}); toast('已保存 '+name+'，重启后自动恢复'); setTimeout(load,260);}catch(e){toast(e.message,true);} }
|
||
async function deleteSaved(name){ try{ await api('/api/delete-saved',{name}); toast('已取消持久化 '+name); setTimeout(load,260);}catch(e){toast(e.message,true);} }
|
||
async function applySaved(){ try{ const r=await api('/api/apply-saved',{}); toast('已应用持久化配置：'+(r.applied||0)+' 个'); setTimeout(load,260);}catch(e){toast(e.message,true);} }
|
||
async function turnOffAll(){ try{ const r=await api('/api/all-off',{}); const failed=(r.results||[]).filter(x=>!x.ok); toast(failed.length ? ('已关闭 '+(r.changed||0)+' 个，失败 '+failed.length+' 个') : ('已关闭全部 LED：'+(r.changed||0)+' 个'), failed.length>0); setTimeout(load,260);}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 labelOf(n){ const s=n.toLowerCase(); if(s.includes('red')) return '红灯'; if(s.includes('green') || s.includes('wlan')) return '绿灯'; if(s.includes('blue') || s.includes('wan')) return '蓝灯'; return n; }
|
||
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 level = Math.max(0, Math.min(1, led.max_brightness>0 ? led.brightness/led.max_brightness : 0));
|
||
const on = Number(led.brightness)>0;
|
||
const triggers = led.triggers || [];
|
||
const n = attr(led.name);
|
||
const color = colorOf(led.name, led.brightness);
|
||
const triggerSelect = triggers.length ? `<select data-led="${n}" onchange="setTrigger(this.dataset.led, this.value)">${triggers.map(t=>`<option value="${attr(t.name)}" ${t.active?'selected':''}>${esc(t.name)}</option>`).join('')}</select>` : '<span class="small">无 trigger 文件</span>';
|
||
const delay = (led.delay_on !== null || led.delay_off !== null) ? `<div class="row"><span class="small">闪烁间隔</span><input data-led="${n}" type="number" min="0" value="${led.delay_on ?? 500}" onchange="setDelay(this.dataset.led,'delay_on',this.value)" title="亮起毫秒"><input data-led="${n}" type="number" min="0" value="${led.delay_off ?? 500}" onchange="setDelay(this.dataset.led,'delay_off',this.value)" title="熄灭毫秒"></div>` : '';
|
||
return `<div class="card ${on?'on':'off'}" style="--led-color:${color};--level:${level}">
|
||
<div class="card-head"><div class="name"><span class="dot"></span><span>${esc(labelOf(led.name))}</span></div><div>${led.saved?'<span class="badge saved">已保存</span>':(led.writable?'<span class="badge">可写</span>':'<span class="small">只读</span>')}</div></div>
|
||
<div class="bulb-wrap"><div class="bulb">${pct}%</div></div>
|
||
<div class="range-line"><input data-led="${n}" type="range" min="0" max="${led.max_brightness || 1}" value="${led.brightness}" oninput="this.nextElementSibling.value=this.value" onchange="setBrightness(this.dataset.led, this.value)"><input data-led="${n}" type="number" min="0" max="${led.max_brightness || 1}" value="${led.brightness}" onchange="setBrightness(this.dataset.led, this.value)"></div>
|
||
<div class="actions"><button class="primary" data-led="${n}" data-max="${led.max_brightness || 1}" onclick="setBrightness(this.dataset.led, this.dataset.max)">打开</button><button class="off" data-led="${n}" onclick="setBrightness(this.dataset.led, 0)">关闭</button><button class="warn" data-led="${n}" onclick="setTrigger(this.dataset.led, 'timer')">闪烁</button><button data-led="${n}" onclick="identify(this.dataset.led)">识别 3 秒</button></div>
|
||
<div class="stack"><div class="row"><span class="small">模式</span>${triggerSelect}</div>${delay}<div class="row sub-actions"><button data-led="${n}" onclick="saveLed(this.dataset.led)">保存开机状态</button><button data-led="${n}" onclick="deleteSaved(this.dataset.led)">取消保存</button></div></div>
|
||
<div class="meta">名称：<b>${esc(led.name)}</b><br>trigger：<b>${esc(led.active_trigger || '-')}</b><br><span class="path">${esc(led.path)}</span></div>
|
||
</div>`;
|
||
}
|
||
async function load(){
|
||
try { const data = await api('/api/leds'); const leds=data.leds||[]; document.getElementById('root').innerHTML = leds.map(card).join('') || '<div class="err">没有发现可控制的红绿蓝 LED</div>'; document.getElementById('count').textContent=leds.length; document.getElementById('oncount').textContent=leds.filter(x=>Number(x.brightness)>0).length; document.getElementById('savedcount').textContent=leds.filter(x=>x.saved).length; }
|
||
catch(e){ document.getElementById('root').innerHTML='<div class="err">'+esc(e.message)+'</div>'; }
|
||
}
|
||
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"""<!doctype html><html><head><meta charset='utf-8'><meta name='viewport' content='width=device-width,initial-scale=1'><title>LED Manager</title><style>{CSS}</style></head><body><div class='container'><div class='header'><div class='hero'><h1>RGB LED 控制台</h1><div class='sub'>只显示实际可控的红 / 绿 / 蓝灯；配置保存到 <code>{html.escape(str(CONFIG_PATH))}</code>，页面每 5 秒自动刷新。</div><div class='summary'><span class='pill'>可控灯 <b id='count'>-</b></span><span class='pill'>已亮 <b id='oncount'>-</b></span><span class='pill'>已保存 <b id='savedcount'>-</b></span><span class='pill'>目录 <b>{html.escape(str(LED_ROOT))}</b></span></div></div><div class='toolbar'><button class='off' onclick='turnOffAll()'>关闭全部</button><button onclick='applySaved()'>应用保存</button><button onclick='load()'>刷新</button></div></div><div id='root' class='grid'></div></div><div id='toast' class='toast'></div><script>{JS}</script></body></html>"""
|
||
data = body.encode("utf-8")
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
elif parsed.path == "/api/leds":
|
||
self.send_json({"ok": True, "leds": list_leds()})
|
||
else:
|
||
self.send_json({"ok": False, "error": "not found"}, 404)
|
||
|
||
def do_POST(self):
|
||
parsed = urllib.parse.urlparse(self.path)
|
||
try:
|
||
data = self.read_body()
|
||
name = None
|
||
p = None
|
||
if parsed.path not in ("/api/apply-saved", "/api/all-off"):
|
||
name = safe_name(str(data.get("name", "")))
|
||
p = led_path(name)
|
||
if parsed.path == "/api/brightness":
|
||
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 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:
|
||
set_trigger(p, "none")
|
||
except Exception as e:
|
||
trigger_warning = str(e)
|
||
write_text(p / "brightness", str(value))
|
||
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")
|
||
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"):
|
||
raise ValueError("invalid delay field")
|
||
value = max(0, int(data.get("value", 0)))
|
||
write_text(p / field, str(value))
|
||
self.send_json({"ok": True})
|
||
elif parsed.path == "/api/save":
|
||
config = load_config()
|
||
config[name] = current_led_config(name)
|
||
save_config(config)
|
||
self.send_json({"ok": True, "config": config[name], "path": str(CONFIG_PATH)})
|
||
elif parsed.path == "/api/delete-saved":
|
||
config = load_config()
|
||
existed = name in config
|
||
config.pop(name, None)
|
||
save_config(config)
|
||
self.send_json({"ok": True, "deleted": existed, "path": str(CONFIG_PATH)})
|
||
elif parsed.path == "/api/apply-saved":
|
||
results = apply_saved_config()
|
||
self.send_json({"ok": True, "applied": sum(1 for r in results if r.get("ok")), "results": results})
|
||
elif parsed.path == "/api/all-off":
|
||
results = turn_off_all_leds()
|
||
self.send_json({"ok": all(r.get("ok") for r in results), "changed": sum(1 for r in results if r.get("brightness") == 0), "results": results})
|
||
elif parsed.path == "/api/identify":
|
||
old_trigger = read_text(p / "trigger")
|
||
old_brightness = read_text(p / "brightness")
|
||
try:
|
||
if (p / "trigger").exists():
|
||
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:
|
||
mb = int(read_text(p / "max_brightness") or 1)
|
||
for _ in range(6):
|
||
write_text(p / "brightness", str(mb)); time.sleep(0.2)
|
||
write_text(p / "brightness", "0"); time.sleep(0.2)
|
||
time.sleep(3)
|
||
finally:
|
||
if old_trigger and "[" in old_trigger:
|
||
_, active = parse_triggers(old_trigger)
|
||
if active:
|
||
try: set_trigger(p, active)
|
||
except Exception: pass
|
||
if old_brightness is not None:
|
||
try: write_text(p / "brightness", old_brightness)
|
||
except Exception: pass
|
||
self.send_json({"ok": True})
|
||
else:
|
||
self.send_json({"ok": False, "error": "not found"}, 404)
|
||
except PermissionError:
|
||
self.send_json({"ok": False, "error": "权限不足：请用 root 运行，或给 /sys/class/leds 写权限"}, 403)
|
||
except Exception as e:
|
||
self.send_json({"ok": False, "error": str(e)}, 400)
|
||
|
||
|
||
def main():
|
||
global LED_ROOT, CONFIG_PATH, CONTROLLABLE_COLORS
|
||
ap = argparse.ArgumentParser(description="Tiny web panel for /sys/class/leds")
|
||
ap.add_argument("--host", default="0.0.0.0", help="listen host, default 0.0.0.0")
|
||
ap.add_argument("--port", type=int, default=8088, help="listen port, default 8088")
|
||
ap.add_argument("--led-root", default=str(LED_ROOT), help="LED sysfs root, default /sys/class/leds")
|
||
ap.add_argument("--config", default=str(CONFIG_PATH), help="persistent config path, default /etc/led-manager/config.json")
|
||
ap.add_argument("--colors", default=",".join(CONTROLLABLE_COLORS), help="comma-separated LED color prefixes to expose, default red,green,blue; empty means expose all")
|
||
ap.add_argument("--no-apply", action="store_true", help="do not apply saved config on startup")
|
||
args = ap.parse_args()
|
||
LED_ROOT = Path(args.led_root)
|
||
CONFIG_PATH = Path(args.config)
|
||
CONTROLLABLE_COLORS = tuple(c.strip().lower() for c in args.colors.split(",") if c.strip())
|
||
if not args.no_apply:
|
||
results = apply_saved_config()
|
||
if results:
|
||
print(f"Applied saved LED config: {results}", flush=True)
|
||
httpd = ThreadingHTTPServer((args.host, args.port), Handler)
|
||
print(f"LED Manager Panel listening on http://{args.host}:{args.port} ; LED_ROOT={LED_ROOT}; CONFIG_PATH={CONFIG_PATH}; COLORS={','.join(CONTROLLABLE_COLORS) or 'all'}", flush=True)
|
||
httpd.serve_forever()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|