Files
DCU_DesktopController/esp32_server.py
T
cbb d2c86c425a init: DCU ESP32 调试控制台前端与本地页面服务
纯前端上位机界面 + esp32_server.py 静态服务/桌面窗口外壳, 不含固件、
不直接访问硬件; 业务 API 由 ESP32_COMM 固件 80 端口跨域提供。

- tabs.html 单窗口多标签外壳, iframe 承载各功能页
- index.html OTA 中继(ESP32 WiFi / GD32 relay / BLE), js/ota + js/ble
- can.html / can_monitor.html CAN 通信与接收监控, js/can
- motor.html 电机 CAN FD 控制台, js/motor
- continuity_test.html LAN9252 通断测试, spi_test.html SPI 测速(临时, 自包含)
- 三入口: esp32_app.vbs 桌面窗口 / esp32_web.vbs 浏览器 / serve.bat 控制台调试
- 补 README(页面与目录约定、?v= 缓存版本等改动须知)、.gitignore、
  .gitattributes(* -text 保持 Windows 换行原样)
2026-09-03 21:30:32 +08:00

540 lines
20 KiB
Python

#!/usr/bin/env python3
"""esp32_server.py — ESP32 调试控制台的本地页面服务器 + 桌面窗口外壳。
本目录的 HTML/CSS/JS 是纯前端: 页面本体由本服务器提供(同源相对路径),
而所有业务 API 由 ESP32 固件的 80 端口跨域提供(js/common/base-url.js 里的
设备 IP, 存 localStorage 的 dcu_esp_ip)。因此本脚本不碰任何硬件, 只做静态服务。
GET / → index.html (OTA 控制台; 标签外壳为 tabs.html)
GET /<file> → 本目录静态文件(html/css/js)
POST /api/shutdown → 结束本服务进程(窗口模式父进程用它通知子进程退出)
启动模式:
--window pywebview 桌面窗口(esp32_app.vbs) —— GUI 进程 + 独立服务子进程
默认 浏览器标签页(esp32_web.vbs / serve.bat)
窗口模式为何要分两个进程: 实测 pythonw + pythonnet(pywebview 依赖) 环境下,
Python 后台线程会被静默杀死(py-spy 可见 serve 线程凭空消失, 页面随即失联)。
把 HTTP 服务放进独立子进程可彻底规避, 见 tools/ecat/ecat_server.py 同款结构。
pywebview 窗口没有浏览器刷新按钮/地址栏, 因此本服务器给 HTML 响应注入
_reload.html: 右下角 ⟳ 悬浮按钮 + F5/Ctrl+R 键盘拦截(WebView2 默认不响应)。
"""
import argparse
import collections
import ctypes
import os
import re
import socket
import subprocess
import sys
import threading
import time
import urllib.request
import webbrowser
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# 单实例互斥: 已有实例时, 新双击不创建第二个窗口(会隐形/端口漂移),
# 而是通过 /api/focus 让旧实例的窗口到前台。
g_mutex = None
# pywebview 窗口内的刷新能力: 悬浮按钮 + F5/Ctrl+R。浏览器 tab 打开时同样无害。
# __HOME__ 由模块加载时替换为工具主页路径(tabs.html 标签外壳)。
# 注入脚本对三种上下文自适应:
# 1) iframe 内的功能页(标签外壳承载): 移除悬浮按钮, 但仍覆写 window.open/_blank
# 让页面里的"新窗口打开"在 iframe 内跳转, 不弹系统浏览器;
# 2) 顶层标签外壳(tabs.html, 含 #view iframe): ⟳ 刷新当前 iframe, 不显示 ⌂;
# 3) 顶层独立页(浏览器直开某 html): ⟳ 刷新自身, 非主页时显示 ⌂ 回到标签外壳。
_RELOAD_SNIPPET = """
<style>#__rl,#__home{position:fixed;right:14px;width:34px;height:34px;z-index:99999;
border-radius:50%;border:1px solid rgba(255,255,255,.25);background:rgba(20,22,28,.72);
color:#9fb0c3;font-size:17px;line-height:32px;text-align:center;cursor:pointer;
user-select:none;opacity:.55;transition:opacity .15s}
#__rl{bottom:14px}#__home{bottom:56px;font-size:19px}
#__rl:hover,#__home:hover{opacity:1}#__rl:active{transform:rotate(90deg)}</style>
<div id="__rl" title="刷新当前页 (F5)">&#x21bb;</div>
<div id="__home" title="返回标签主页" style="display:none">&#x2302;</div>
<script>(function(){
var TOP=(window.top===window.self);
if(!TOP){var _r=document.getElementById('__rl'),_h=document.getElementById('__home');
if(_r)_r.remove(); if(_h)_h.remove();}
window.open=function(u){if(u)location.href=u;return null;};
document.addEventListener('click',function(e){
var a=e.target&&e.target.closest?e.target.closest('a[target=_blank]'):null;
if(a&&a.href){e.preventDefault();location.href=a.href;}
},true);
function reloadCur(){
if(TOP){var f=document.getElementById('view');
if(f){try{f.contentWindow.location.reload();}catch(_){f.src=f.src;} return;}}
location.reload();
}
window.addEventListener('keydown',function(e){
if(e.key==='F5'||(e.ctrlKey&&(e.key==='r'||e.key==='R'))){e.preventDefault();reloadCur();}
});
if(!TOP) return;
document.getElementById('__rl').addEventListener('click',reloadCur);
var home=document.getElementById('__home');
var p=location.pathname;
var isShell=!!document.getElementById('view');
if(!isShell&&!/\\/$/.test(p)&&!/\\/index\\.html$/.test(p)){
home.style.display='block';
home.addEventListener('click',function(){location.href='__HOME__';});
}
function ready(){try{window.pywebview&&window.pywebview.api&&
window.pywebview.api.page_loaded()}catch(e){}}
document.addEventListener('DOMContentLoaded',ready);
window.addEventListener('load',ready);
})();</script>
"""
_RELOAD_SNIPPET = _RELOAD_SNIPPET.replace("__HOME__", "/tabs.html")
_RELOAD_MARK = "</body>"
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
g_httpd = None
g_child = None
g_port = None
g_win = None
g_job = None # Job Object 句柄: 父进程一死内核连坐杀服务子进程
class _GuiApi:
"""暴露给窗口内 JS (window.pywebview.api) 的 GUI 控制接口。"""
def page_loaded(self):
"""页面每次加载完成后由注入脚本调用: 把窗口带到前台。"""
win = g_win
if win is not None:
try:
win.show()
except Exception:
pass
h = _native_handle(win) if win is not None else None
if h is not None:
_win32_foreground(h)
def reload_page(self):
win = g_win
if win is not None:
try:
win.load_url(win.url) # pywebview 无 reload, 用重载当前 URL 代替
except Exception:
pass
def _focus_existing(port):
"""让已有实例的窗口到前台。返回 True 表示确实存在旧实例。"""
try:
urllib.request.urlopen(urllib.request.Request(
f"http://127.0.0.1:{port}/api/focus", data=b"{}",
headers={"Content-Type": "application/json"}, method="POST"),
timeout=2)
return True
except Exception:
return False
class Handler(SimpleHTTPRequestHandler):
server_version = "Esp32Tools/1.0"
def __init__(self, *a, **kw):
super().__init__(*a, directory=BASE_DIR, **kw)
def log_message(self, fmt, *args):
pass # 静默访问日志
def end_headers(self):
# 本地调试: 禁缓存, 避免改了 css/js 还得手工 bump ?v= 才生效
self.send_header("Cache-Control", "no-store, must-revalidate")
super().end_headers()
def do_GET(self):
path = self.path.split("?")[0].lower()
if path.endswith(".html") or path == "/" or not os.path.splitext(path)[1]:
# HTML(含无扩展名的 SPA 路由): 注入刷新悬浮按钮/快捷键后返回
fp = self.translate_path(self.path)
if os.path.isdir(fp):
fp = os.path.join(fp, "index.html")
try:
with open(fp, "rb") as f:
html = f.read().decode("utf-8", "replace")
except OSError:
self.send_error(404, "File not found")
return
if _RELOAD_MARK in html:
html = html.replace(_RELOAD_MARK, _RELOAD_SNIPPET + _RELOAD_MARK, 1)
body = html.encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
else:
super().do_GET()
def do_POST(self):
path = self.path.split("?")[0]
if path == "/api/focus":
# 已有实例响应: 新实例应当退场
body = b'{"ok":true,"focused":true}'
self._post_json(body)
threading.Thread(target=lambda: print("FOCUS=1", flush=True), daemon=True).start()
return
if path == "/api/shutdown":
body = b'{"ok":true}'
self._post_json(body)
threading.Thread(target=_shutdown_self, daemon=True).start()
return
self.send_error(404)
def _post_json(self, body):
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _shutdown_self():
if g_httpd is not None:
try:
g_httpd.shutdown()
except Exception:
pass
def _win32_foreground(native_handle):
"""SetForegroundWindow: pywebview Window 没有 focus(), 要走 Win32。"""
try:
hwnd = int(native_handle)
ctypes.windll.user32.ShowWindow(hwnd, 9) # SW_RESTORE
ctypes.windll.user32.SetForegroundWindow(hwnd)
except Exception:
pass
def _native_handle(win):
try:
return win.native.Handle # WinForms Form
except Exception:
try:
return win.native
except Exception:
return None
def _port_taken(port):
"""Windows 的 SO_REUSEADDR 允许同端口重复 bind(实测会双监听、连接随机落),
故先用一次真实连接探测占用, 不能只靠 bind 是否抛错。"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(0.3)
return s.connect_ex(("127.0.0.1", port)) == 0
class QuietServer(ThreadingHTTPServer):
# 客户端中途断开(刷新/关窗)会触发 handle_error 打 traceback 到 stderr;
# 窗口模式下 stderr 是父进程的 PIPE, 4KB 缓冲一满子进程写日志即永久阻塞
# → 整个服务冻死(实测"窗口经常卡死"的根因)。此处静音, 父侧另有排空线程双保险。
def handle_error(self, request, client_address):
pass
def _bind_server(port):
for _ in range(6):
if _port_taken(port):
print(f"端口 {port} 已被占用, 尝试 {port + 1}")
port += 1
continue
try:
return QuietServer(("127.0.0.1", port), Handler), port
except OSError:
print(f"端口 {port} 绑定失败, 尝试 {port + 1}")
port += 1
return None, port
def _watch_parent(ppid):
"""兜底看门狗(Job Object 万一不可用时): 父 GUI 进程消失就自我了断,
避免窗口关了服务还在后台占端口。"""
def _loop():
k32 = ctypes.windll.kernel32
h = k32.OpenProcess(0x00100000, False, ppid) # SYNCHRONIZE
if not h:
return
k32.WaitForSingleObject(h, 0xFFFFFFFF) # 父进程退出即返回
os._exit(0)
threading.Thread(target=_loop, daemon=True).start()
def _run_browser(args):
"""浏览器模式: 服务跑在本进程主线程。"""
global g_httpd
if args.parent_pid:
_watch_parent(args.parent_pid)
httpd, port = _bind_server(args.port)
if httpd is None:
raise SystemExit("无法绑定端口")
g_httpd = httpd
url = f"http://127.0.0.1:{port}/tabs.html"
print(f"PORT={port}", flush=True) # 窗口模式父进程据此得知端口
print(f"ESP32 调试控制台: {url} (Ctrl+C 退出)")
print("页面里的设备 IP 决定 API 目标(默认 192.168.31.127), 本服务只提供页面")
if not args.no_browser:
threading.Timer(0.5, lambda: webbrowser.open(url)).start()
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
httpd.server_close()
print("\n已退出")
def _bind_child_lifetime(child):
"""把服务子进程放进 Job Object, 父进程句柄一关(正常退出/被任务管理器强杀/崩溃)
由内核连坐结束子进程, 杜绝"窗口关了端口还占着"的孤儿服务。
句柄故意不关闭: 生命周期就是靠父进程存活维系的。"""
k32 = ctypes.windll.kernel32
class _JOBLIM(ctypes.Structure):
_fields_ = [("PerProcessUserTimeLimit", ctypes.c_int64),
("PerJobUserTimeLimit", ctypes.c_int64),
("LimitFlags", ctypes.c_uint32),
("MinimumWorkingSetSize", ctypes.c_size_t),
("MaximumWorkingSetSize", ctypes.c_size_t),
("ActiveProcessLimit", ctypes.c_uint32),
("Affinity", ctypes.POINTER(ctypes.c_ulong)),
("PriorityClass", ctypes.c_uint32),
("SchedulingClass", ctypes.c_uint32)]
class _IOC(ctypes.Structure):
_fields_ = [("ReadOperationCount", ctypes.c_uint64),
("WriteOperationCount", ctypes.c_uint64),
("OtherOperationCount", ctypes.c_uint64),
("ReadTransferCount", ctypes.c_uint64),
("WriteTransferCount", ctypes.c_uint64),
("OtherTransferCount", ctypes.c_uint64)]
class _EXLIM(ctypes.Structure):
_fields_ = [("BasicLimitInformation", _JOBLIM),
("IoInfo", _IOC),
("ProcessMemoryLimit", ctypes.c_size_t),
("JobMemoryLimit", ctypes.c_size_t),
("PeakProcessMemoryUsed", ctypes.c_size_t),
("PeakJobMemoryUsed", ctypes.c_size_t)]
try:
job = k32.CreateJobObjectW(None, None)
if not job:
return None
info = _EXLIM()
info.BasicLimitInformation.LimitFlags = 0x2000 # KILL_ON_JOB_CLOSE
if not k32.SetInformationJobObject(job, 9, ctypes.byref(info),
ctypes.sizeof(info)): # ExtendedLimitInformation
return None
h = k32.OpenProcess(0x1F0FFF, False, child.pid) # PROCESS_ALL_ACCESS
if not h:
return None
k32.AssignProcessToJobObject(job, h)
k32.CloseHandle(h)
return job
except Exception:
return None
def _stop_child(child, port):
"""通知服务子进程优雅退出, 超时则强杀。"""
if child is None or child.poll() is not None:
return
try:
urllib.request.urlopen(urllib.request.Request(
f"http://127.0.0.1:{port}/api/shutdown", data=b"{}",
headers={"Content-Type": "application/json"}, method="POST"),
timeout=3)
except Exception:
pass
try:
child.wait(timeout=3)
except Exception:
pass
if child.poll() is None:
child.kill()
try:
child.wait(timeout=3)
except Exception:
pass
def _fail_box(title, msg):
"""GUI 进程启动失败时弹原生对话框, 避免无声无息。"""
try:
ctypes.windll.user32.MessageBoxW(None, msg, title, 0x10) # MB_ICONERROR
except Exception:
pass
def _acquire_single_instance():
"""命名 mutex 单实例守卫。返回 False=已有实例在跑。"""
global g_mutex
try:
g_mutex = ctypes.windll.kernel32.CreateMutexW(None, False, "DCU_Esp32Tools_SingleInstance")
return ctypes.windll.kernel32.GetLastError() != 183 # ERROR_ALREADY_EXISTS
except Exception:
return True # mutex 不可用时退化为允许多开
def _spawn_service_child(extra_args, fail_title):
"""启动服务子进程并持续排空其 stdout。
必须一直读: PIPE 缓冲 ~4KB, 读停则子进程任何 print/traceback 写满即永久阻塞。"""
script = os.path.abspath(__file__)
child = subprocess.Popen(
[sys.executable, script] + extra_args,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, cwd=BASE_DIR)
global g_job
g_job = _bind_child_lifetime(child) # 父进程消失 → 内核连坐杀子进程
holder = {"port": None, "focus": False}
tail = collections.deque(maxlen=40)
focus_evt = threading.Event()
def _drain():
for raw in child.stdout:
s = raw.decode("utf-8", "replace").rstrip()
if s:
tail.append(s)
if s.startswith("PORT=") and holder["port"] is None:
try:
holder["port"] = int(s[5:])
except ValueError:
pass
elif s.startswith("FOCUS="):
holder["focus"] = True
focus_evt.set()
threading.Thread(target=_drain, daemon=True).start()
deadline = time.time() + 30
while time.time() < deadline:
if holder["port"] is not None or child.poll() is not None:
break
time.sleep(0.1)
port = holder["port"]
if port is None:
child.terminate()
_fail_box(fail_title,
"\n".join(tail) or "服务子进程未在 30 秒内报告端口(无输出)")
os._exit(1)
return child, port, focus_evt
def _run_window(webview, args):
"""桌面窗口模式: 本进程只做 GUI, HTTP 服务跑在子进程。"""
global g_child, g_port, g_win
if not _acquire_single_instance():
# 已有实例: 尝试把它的窗口顶到前台后退出。旧实例是修过 focus 钩子的
# 新版本才能响应; 旧版无响应时提示用户用旧窗口。
probed = False
for p in (args.port, args.port + 1, args.port + 2):
if _focus_existing(p):
probed = True
break
if probed:
os._exit(0)
_fail_box("ESP32 调试控制台",
"已有一个实例在运行, 但无法唤醒其窗口。\n"
"请到任务管理器结束所有 pythonw.exe 后重新打开。")
os._exit(0)
child, port, focus_evt = _spawn_service_child(
["--no-browser", "--port", str(args.port),
"--parent-pid", str(os.getpid())], "ESP32 调试控制台启动失败")
g_child = child
g_port = port
url = f"http://127.0.0.1:{port}/tabs.html"
for _ in range(50): # 等 HTTP 就绪(最多 ~10s)
try:
urllib.request.urlopen(url, timeout=1)
break
except Exception:
if g_child.poll() is not None:
_fail_box("ESP32 调试控制台启动失败", "服务子进程意外退出")
os._exit(1)
time.sleep(0.2)
def _monitor():
while True:
if focus_evt.is_set():
_gui_bringToFront() # 新双击 → 旧窗口到前台
focus_evt.clear()
if g_child.poll() is not None:
os._exit(0) # 子进程没了 → GUI 一起退
time.sleep(0.5)
threading.Thread(target=_monitor, daemon=True).start()
g_win = webview.create_window("ESP32 调试控制台 — DCU", url,
width=1600, height=1000, min_size=(1100, 720),
js_api=_GuiApi(), hidden=True, text_select=True)
print(f"桌面窗口模式: {url} (关窗即退出)")
def _show_watchdog():
# 页面加载后由 JS 调 page_loaded → show; 若 20s 仍未显示则强制 show
deadline = time.time() + 20
while time.time() < deadline:
if g_win.events.loaded.is_set():
time.sleep(1)
break
time.sleep(0.5)
try:
g_win.show()
except Exception:
pass
h = _native_handle(g_win)
if h is not None:
_win32_foreground(h)
threading.Thread(target=_show_watchdog, daemon=True).start()
webview.start()
_stop_child(g_child, g_port)
os._exit(0)
def _gui_bringToFront():
"""把 GUI 窗口恢复/置顶(从子进程 monitor 线程调用)。"""
win = g_win
if win is None:
return
try:
win.show()
except Exception:
pass
h = _native_handle(win)
if h is not None:
_win32_foreground(h)
def main():
ap = argparse.ArgumentParser(description="ESP32 调试控制台页面服务器")
ap.add_argument("--port", type=int, default=8080)
ap.add_argument("--no-browser", action="store_true")
ap.add_argument("--window", action="store_true",
help="pywebview 桌面窗口模式")
ap.add_argument("--parent-pid", type=int, default=0,
help="内部用: GUI 父进程 pid, 父进程消失时本服务自动退出")
args = ap.parse_args()
if args.window:
try:
import webview
except ImportError:
print("pywebview 未安装: pip install pywebview —— 回退浏览器模式")
webview = None
if webview is not None:
_run_window(webview, args)
return
_run_browser(args)
if __name__ == "__main__":
main()