# -*- coding: utf-8 -*- # LPC 0140650 web UI，运行于 Ubuntu 192.168.58.7:8090 # LPC 自治架构：Modbus 轮询+闭环在 LPC Lua（触发器 #14/#15），本程序只做监控/记录/HTTP 转发 import json import re import threading import time import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse, parse_qs LPC = "http://192.168.58.109" PORT = 8090 HIST_FILE = "/home/tonycao/rs485-lpc/history.jsonl" SAMPLE_INTERVAL = 30 # 秒 HIST_KEEP = 48 * 3600 # 48 小时 TRIG = {"BRT": 2, "LUX": 8, "LXT": 9, "LX2": 10, "FX": 11, "SC": 12, "FAD": 18, "PER": 19, "DBD": 20} TRIG_LINK_ON = 16 TRIG_LINK_OFF = 17 _hist = [] # [[t_ms, brt, out, lux], ...] lux 可为 None _hist_lock = threading.Lock() _samples_since_compact = 0 def lpc_json(path): with urllib.request.urlopen(LPC + path, timeout=5) as r: return json.loads(r.read().decode("utf-8", "replace")) def lpc_lua(names): d = lpc_json("/api/lua?variables=" + ",".join(names)) return d def lua_num(v): try: return int(round(float(v))) except Exception: return None def read_lux(): try: return lua_num(lpc_lua(["lux_value"]).get("lux_value")) except Exception: return None def link_on(): try: return bool(lpc_lua(["lux_link_on"]).get("lux_link_on")) except Exception: return False def fire_trigger(num, var=None): url = LPC + "/api/trigger?num=%d" % num if var is not None: url += "&var=%d" % var req = urllib.request.Request(url, data=b"", method="POST") with urllib.request.urlopen(req, timeout=5) as r: r.read() def sample_once(): t = int(time.time() * 1000) brt = None out = None try: g = lpc_json("/api/group") brt = g["groups"][0]["level"] except Exception: pass try: o = lpc_json("/api/output?universe=dmx:1") vals = [] for c in o["channels"][:32]: m = re.search(r"\d+", str(c)) vals.append(int(m.group(0)) if m else 0) out = round(sum(vals) * 100.0 / (255 * len(vals)), 1) except Exception: pass lux = read_lux() row = [t, brt, out, lux] with _hist_lock: _hist.append(row) cutoff = t - HIST_KEEP * 1000 while _hist and _hist[0][0] < cutoff: _hist.pop(0) try: with open(HIST_FILE, "a") as f: f.write(json.dumps(row, separators=(",", ":")) + "\n") except Exception: pass global _samples_since_compact _samples_since_compact += 1 if _samples_since_compact >= 120: # 每小时压实一次文件 _samples_since_compact = 0 compact_history() def compact_history(): with _hist_lock: data = list(_hist) try: with open(HIST_FILE, "w") as f: for r in data: f.write(json.dumps(r, separators=(",", ":")) + "\n") except Exception: pass def load_history(): try: cutoff = int(time.time() * 1000) - HIST_KEEP * 1000 with open(HIST_FILE) as f: for line in f: line = line.strip() if not line: continue try: r = json.loads(line) except Exception: continue if isinstance(r, list) and len(r) == 4 and r[0] >= cutoff: _hist.append(r) except Exception: pass def sampler_loop(): while True: try: sample_once() except Exception: pass time.sleep(SAMPLE_INTERVAL) PAGE = r""" LPC 灯光控制

LPC 0140650 实时状态

照度联动

照度：- Lux

目标照度： Lux

轮询周期： s 渐变时间： s 死区： Lux

闭环在 LPC 内运行（触发器 #14 每秒轮询传感器，#15 Lua 自动调光）；轮询周期=闭环调节间隔，照度采集与显示仍每秒刷新；FAD 单位为 0.1s

控制

效果：

场景：

总亮度： 100%

历史曲线（48 小时，30s 采样）

输出%（左轴 0-100）总调光 BRT%（左轴）照度 Lux（右轴，对数 1-200k）

Timeline

DMX 输出（通道 1~32）

""" class H(BaseHTTPRequestHandler): def _send(self, code, body, ctype="application/json; charset=utf-8"): b = body if isinstance(body, bytes) else body.encode("utf-8") self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) def log_message(self, *a): pass def _reqlog(self): if self.command != "POST": return try: with open("/home/tonycao/rs485-lpc/lpc_web.req.log", "a") as f: f.write("%s %s %s %s\n" % (time.strftime("%F %T"), self.client_address[0], self.command, self.path)) except Exception: pass def do_GET(self): u = urlparse(self.path) if u.path in ("/", "/index.html"): self._send(200, PAGE, "text/html; charset=utf-8") elif u.path.startswith("/api/"): try: with urllib.request.urlopen(LPC + u.path + (("?" + u.query) if u.query else ""), timeout=5) as r: self._send(200, r.read(), "application/json") except Exception as e: self._send(502, json.dumps({"error": str(e)})) elif u.path == "/luxlink": cfg = {} try: d = lpc_lua(["lux_target", "lux_fade", "lux_poll_s", "lux_deadband"]) cfg = {"target": lua_num(d.get("lux_target")), "fade": d.get("lux_fade"), "period": lua_num(d.get("lux_poll_s")), "deadband": lua_num(d.get("lux_deadband"))} except Exception: pass self._send(200, json.dumps({"on": link_on(), "lux": read_lux(), **cfg})) elif u.path == "/history": with _hist_lock: data = list(_hist) self._send(200, json.dumps({"points": data}, separators=(",", ":"))) else: self._send(404, "{}") def do_POST(self): self._reqlog() u = urlparse(self.path) q = parse_qs(u.query) if u.path == "/send": cmd = q.get("cmd", [""])[0] m = re.fullmatch(r"(BRT|FX|SC|LUX|LXT|LX2|FAD|PER|DBD)=(\d+)", cmd) if not m: self._send(400, json.dumps({"error": "bad cmd"})) return try: fire_trigger(TRIG[m.group(1)], int(m.group(2))) self._send(200, json.dumps({"ok": True})) except Exception as e: self._send(502, json.dumps({"error": str(e)})) elif u.path == "/luxlink": try: fire_trigger(TRIG_LINK_ON if q.get("on", ["0"])[0] == "1" else TRIG_LINK_OFF) self._send(200, json.dumps({"ok": True})) except Exception as e: self._send(502, json.dumps({"error": str(e)})) else: self._send(404, "{}") if __name__ == "__main__": load_history() threading.Thread(target=sampler_loop, daemon=True).start() print("LPC web UI: http://0.0.0.0:%d (history %d pts)" % (PORT, len(_hist))) ThreadingHTTPServer(("0.0.0.0", PORT), H).serve_forever()