cbb
2d1743ca4f
- 抽象电机传输层，WiFi 与 EtherCAT 实现分离，解码、状态面板、波形和命令序列共用 - 新增 EtherCAT 后端反向代理与自动拉起，解决跨端口 CORS 与预检失败问题 - 刷新速度按链路独立存储，支持 EtherCAT 低延迟读取与 WiFi 稳定轮询 - 更新电机页链路选择、不支持页签置灰样式、README 说明及相关测试脚本
778 lines
32 KiB
Python
778 lines
32 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)。因此本脚本不碰任何硬件, 只做静态服务
|
|
—— 唯一例外是 /ecat/api/* 反向代理(见下)。
|
|
|
|
GET / → 302 到 /tabs.html (标签外壳, 工具主页)
|
|
GET /<file> → 本目录静态文件(tabs.html / pages/ / lab/ / css/ / js/)
|
|
POST /api/shutdown → 结束本服务进程(窗口模式父进程用它通知子进程退出)
|
|
POST /ecat/api/<路径> → 反向代理到 tools/ecat 的 ecat_server.py 的 /api/<路径>
|
|
|
|
为什么要代理 EtherCAT: 电机页(pages/motor.html)支持 WiFi 与 EtherCAT 两种传输,
|
|
后者的后端 tools/ecat/src/ecat_server.py 只绑 127.0.0.1 且完全没有 CORS
|
|
(既无 Access-Control-Allow-Origin, 也没有 do_OPTIONS —— 预检直接 501)。
|
|
我们的 API 是 POST + application/json, 属非简单请求, 浏览器必发预检,
|
|
因此页面无法直接跨端口调它。改那边要动另一个 git 仓库, 把页面搬过去又会让
|
|
WiFi/ECAT 分裂成两份页面, 所以在本服务做同源转发, 两个仓库都不用改。
|
|
|
|
目录约定: tabs.html 在根(工具入口), 功能页在 pages/, 自包含实验页在 lab/。
|
|
|
|
启动模式:
|
|
--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 json
|
|
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)">↻</div>
|
|
<div id="__home" title="返回标签主页" style="display:none">⌂</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');
|
|
// 顶层且非标签外壳(浏览器直开 pages/xxx.html) → 给个 ⌂ 回到外壳。
|
|
// 判定基准是外壳自身(有 #view / 路径为 tabs.html 或 /), 不能再看 index.html:
|
|
// 重排后 pages/index.html 只是普通功能页, 直开它同样该显示 ⌂。
|
|
if(!isShell&&!/\\/$/.test(p)&&!/\\/tabs\\.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 句柄: 父进程一死内核连坐杀服务子进程
|
|
|
|
# ── EtherCAT 后端反向代理 ──────────────────────────────────────────────
|
|
# 电机页的 EtherCAT 传输要打 tools/ecat/src/ecat_server.py 的 /api/motor/*,
|
|
# 但那个服务**只绑 127.0.0.1 且完全没有 CORS**(无 Access-Control-Allow-Origin,
|
|
# 也没有 do_OPTIONS —— 预检直接 501)。而我们的 API 是 POST + JSON,
|
|
# 属非简单请求, 浏览器必发预检 → 从本页面(:8080)直接 fetch 它(:8090)必然失败。
|
|
#
|
|
# 三种解法里选反向代理:
|
|
# 1) 改 ecat_server.py 加 CORS —— 它是**独立 git 仓库**, 不该为本页面改;
|
|
# 2) 把电机页搬进 ecat_server 的 _STATIC_MAP —— 那是硬白名单, 且会让
|
|
# WiFi 版和 ECAT 版分裂成两份页面, 正好违背"所有组件复用"的要求;
|
|
# 3) 本服务代理 —— 页面永远同源(:8080), 两个仓库都不用动。
|
|
# 端口自动探测: ecat_server 的 _bind_server 会在 8090..8095 间漂移。
|
|
ECAT_PROXY_PREFIX = "/ecat/api/"
|
|
ECAT_PORTS = (8090, 8091, 8092, 8093, 8094, 8095)
|
|
g_ecat_port = None # 记住上次探到的端口, 避免每次请求都从 8090 试起
|
|
|
|
# ── EtherCAT 服务自动启动 ─────────────────────────────────────────────
|
|
# 电机页切到 EtherCAT 链路时, 不该要求用户先去双击 ecat_app.vbs。
|
|
# 本服务在第一次代理请求时按需拉起 tools/ecat 的服务进程。
|
|
#
|
|
# 三条必须小心的地方(都在 ecat_server.py 里读出来的, 不是猜的):
|
|
# 1) **必须先探测再拉起**。ecat_server 的 _run_browser 在没有 --parent-pid 时
|
|
# 会调 _kill_stale_instances(), 那会把用户已经开着的 EtherCAT 窗口
|
|
# 优雅关掉(还会把从站带回 INIT)。所以已有实例时只复用, 绝不重复拉起。
|
|
# 2) 拉起时**必须带 --parent-pid**: 一是跳过上面那个清场逻辑, 二是让它在
|
|
# 本服务退出后自我了断, 不留孤儿进程占着 Npcap 适配器和端口。
|
|
# 3) 用 pythonw.exe + venv: pysoem/pywebview 都装在 tools/ecat\venv 里,
|
|
# 用系统 python 会 ImportError; 用 python.exe 会弹黑框。
|
|
ECAT_DIR = os.path.normpath(os.path.join(BASE_DIR, "..", "ecat"))
|
|
ECAT_ENTRY = os.path.join(ECAT_DIR, "src", "ecat_server.py")
|
|
ECAT_PYW = os.path.join(ECAT_DIR, "venv", "Scripts", "pythonw.exe")
|
|
g_ecat_child = None # 由本服务拉起的 ecat 服务子进程(复用已有实例时为 None)
|
|
g_ecat_lock = threading.Lock() # 串行化自动启动, 避免并发请求拉起多个进程
|
|
|
|
|
|
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
|
|
|
|
|
|
def _ecat_find_port():
|
|
"""返回正在监听的 ecat 服务端口, 没有则 None。"""
|
|
ports = ([g_ecat_port] if g_ecat_port else []) + \
|
|
[p for p in ECAT_PORTS if p != g_ecat_port]
|
|
for p in ports:
|
|
if _port_taken(p):
|
|
return p
|
|
return None
|
|
|
|
|
|
def _ecat_autostart():
|
|
"""按需拉起 tools/ecat 的服务进程。返回 (port, log_lines)。
|
|
|
|
port 为 None 表示拉不起来(缺 venv / 缺文件 / 起了但没监听)。
|
|
已有实例时**只复用不重启** —— 详见文件头 ECAT_DIR 附近的注释:
|
|
ecat_server 独立启动会 _kill_stale_instances(), 把用户开着的窗口关掉。
|
|
"""
|
|
global g_ecat_child, g_ecat_port
|
|
with g_ecat_lock:
|
|
port = _ecat_find_port()
|
|
if port: # 已经有人在跑(用户手开的, 或我们上次拉起的)
|
|
g_ecat_port = port
|
|
return port, []
|
|
|
|
logs = []
|
|
if not os.path.isfile(ECAT_ENTRY):
|
|
return None, [f"找不到 EtherCAT 服务入口: {ECAT_ENTRY}"]
|
|
if not os.path.isfile(ECAT_PYW):
|
|
return None, [
|
|
f"找不到 EtherCAT 服务的 venv: {ECAT_PYW}",
|
|
"请在 tools/ecat 下重建: py -3.12 -m venv venv 后 "
|
|
"venv\\Scripts\\python.exe -m pip install -r requirements.txt",
|
|
]
|
|
|
|
# 复用已被本进程拉起但已退出的槽位
|
|
if g_ecat_child is not None and g_ecat_child.poll() is not None:
|
|
g_ecat_child = None
|
|
if g_ecat_child is None:
|
|
try:
|
|
# --parent-pid: 跳过 ecat 侧的 _kill_stale_instances 清场,
|
|
# 并让它在本服务退出后自我了断(不留孤儿占 Npcap/端口)。
|
|
# --no-browser: 我们只要后台服务, 不要它再弹一个浏览器标签。
|
|
# cwd=src: ecat_server.py 与同目录 py 之间靠相对 import 解析。
|
|
g_ecat_child = subprocess.Popen(
|
|
[ECAT_PYW, ECAT_ENTRY, "--no-browser",
|
|
"--parent-pid", str(os.getpid())],
|
|
cwd=os.path.join(ECAT_DIR, "src"),
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
stdin=subprocess.DEVNULL,
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
env=dict(os.environ, PYTHONUTF8="1"),
|
|
)
|
|
logs.append(f"已自动启动 EtherCAT 服务(pid {g_ecat_child.pid})")
|
|
except OSError as e:
|
|
return None, [f"启动 EtherCAT 服务失败: {type(e).__name__}: {e}"]
|
|
|
|
# 等它把端口监听起来。pysoem 首次导入 + find_adapters 要一两秒,
|
|
# 给到 20s: 慢机器上 Npcap 枚举适配器偶发很久, 提前放弃会让用户
|
|
# 看到"启动失败"而其实几秒后就好了。
|
|
deadline = time.time() + 20
|
|
while time.time() < deadline:
|
|
if g_ecat_child.poll() is not None:
|
|
return None, logs + [
|
|
f"EtherCAT 服务启动后立刻退出(exit={g_ecat_child.returncode}) —— "
|
|
"多为 venv 缺 pysoem 或未安装 Npcap; "
|
|
"可双击 tools/ecat/ecat_web.bat 看具体报错"]
|
|
port = _ecat_find_port()
|
|
if port:
|
|
g_ecat_port = port
|
|
logs.append(f"EtherCAT 服务已就绪(端口 {port})")
|
|
return port, logs
|
|
time.sleep(0.3)
|
|
return None, logs + ["EtherCAT 服务启动超时(20s 内未监听端口)"]
|
|
|
|
|
|
def _ecat_post(port, path, body, timeout):
|
|
"""向 ecat_server 转发一次 POST, 返回 (status, body_bytes)。
|
|
|
|
**不要改回 urllib/http.client 并给它传 timeout。** 那样每次请求会多花约 10ms:
|
|
给 socket 设 timeout 会让 Python 把它切成非阻塞模式, 而 Windows 上非阻塞
|
|
connect 要靠 select 轮询完成, 于是撞上约 10ms 的系统调度粒度。实测
|
|
(本机回环, /api/motor/state):
|
|
urllib.urlopen(req, timeout=5) p50 11.0 ms
|
|
http.client + timeout p50 10.7 ms
|
|
裸 socket 连接阶段不设 timeout p50 0.58 ms
|
|
ecat_server 自己只用 0.58ms, 那 10ms 全是这一层白交的过路费 ——
|
|
电机页 EtherCAT 链路的刷新上限直接被它按在 ~76Hz。
|
|
|
|
所以这里手写最小 HTTP/1.0 客户端: 连接阶段**不设** timeout(本机回环连接
|
|
要么立刻成功要么立刻 ECONNREFUSED, 不会挂住), 连上之后才设读写超时。
|
|
"""
|
|
payload = (
|
|
f"POST {path} HTTP/1.0\r\n"
|
|
f"Host: 127.0.0.1:{port}\r\n"
|
|
"Content-Type: application/json\r\n"
|
|
f"Content-Length: {len(body or b'')}\r\n"
|
|
"Connection: close\r\n"
|
|
"\r\n"
|
|
).encode("ascii") + (body or b"")
|
|
|
|
sock = socket.create_connection(("127.0.0.1", port)) # 注意: 不传 timeout
|
|
try:
|
|
sock.settimeout(timeout) # 连上之后再设: 只作用于收发, 不影响 connect
|
|
sock.sendall(payload)
|
|
chunks = []
|
|
while True:
|
|
chunk = sock.recv(65536)
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
finally:
|
|
sock.close()
|
|
|
|
raw = b"".join(chunks)
|
|
head, _, data = raw.partition(b"\r\n\r\n")
|
|
first = head.split(b"\r\n", 1)[0].split(b" ", 2)
|
|
if len(first) < 2 or not first[1].isdigit():
|
|
raise OSError(f"ecat_server 回了无法解析的响应头: {head[:120]!r}")
|
|
# 404 等业务错误要原样透传: ecat_server 的 404 体里带
|
|
# "服务在改动之后没重启" 的排查提示, 吞掉它会让人白排查半天。
|
|
return int(first[1]), data
|
|
|
|
|
|
def _ecat_proxy(path, body, timeout=6.0):
|
|
"""把 /ecat/api/xxx 转发到 ecat_server 的 /api/xxx。
|
|
|
|
返回 (status, body_bytes)。端口在 ECAT_PORTS 里探测并缓存到 g_ecat_port:
|
|
ecat_server 的 _bind_server 会在 8090..8095 之间漂移(端口被占就挪),
|
|
写死 8090 会在它挪了之后表现为"页面点了没反应"。
|
|
|
|
连不上时返回 503 + 与 ecat_server 同构的 {"ok":false,"log":[...]} 信封,
|
|
这样前端只有一条错误处理路径 —— 它不需要知道中间还隔着一层代理。
|
|
"""
|
|
global g_ecat_port
|
|
# /ecat/api/motor/state → /api/motor/state (只剥掉 "/ecat" 前缀, 保留 /api)
|
|
target = "/api/" + path[len(ECAT_PROXY_PREFIX):]
|
|
ports = ([g_ecat_port] if g_ecat_port else []) + \
|
|
[p for p in ECAT_PORTS if p != g_ecat_port]
|
|
last = None
|
|
for port in ports:
|
|
try:
|
|
status, data = _ecat_post(port, target, body, timeout)
|
|
g_ecat_port = port
|
|
return status, data
|
|
except Exception as e: # 连不上/超时 → 试下一个端口
|
|
last = e
|
|
if g_ecat_port == port:
|
|
g_ecat_port = None # 缓存端口失效, 下轮重新全量探测
|
|
|
|
# 一个端口都不通 → 服务没在跑, 按需自动拉起后重试一次。
|
|
# 放在这里而不是每次请求前探测: 正常情况(服务在跑)一次都不会走到,
|
|
# 不给高频的 /motor/state 轮询增加任何开销。
|
|
port, start_logs = _ecat_autostart()
|
|
if port:
|
|
try:
|
|
status, data = _ecat_post(port, target, body, timeout)
|
|
g_ecat_port = port
|
|
return status, data
|
|
except Exception as e:
|
|
last = e
|
|
|
|
msg = (f"EtherCAT 服务不可用: {type(last).__name__ if last else 'unknown'}")
|
|
body = json.dumps({"ok": False, "log": start_logs + [msg]},
|
|
ensure_ascii=False).encode("utf-8")
|
|
return 503, body
|
|
|
|
|
|
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()
|
|
# 根路径不再有 index.html(已移入 pages/), 直接送去标签外壳
|
|
if path in ("/", "/index.html"):
|
|
self.send_response(302)
|
|
self.send_header("Location", "/tabs.html")
|
|
self.send_header("Content-Length", "0")
|
|
self.end_headers()
|
|
return
|
|
if path.endswith(".html") 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
|
|
if path.startswith(ECAT_PROXY_PREFIX):
|
|
# 电机页的 EtherCAT 传输: 同源转发到 ecat_server(见 _ecat_proxy 注释)。
|
|
# 超时给足: /api/op/start 要走 PREOP→SAFEOP→OP, 内部还会自动重试一次,
|
|
# 实测最坏 ~24s, 用默认 6s 会在正常进 OP 的过程中假失败。
|
|
try:
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
except ValueError:
|
|
length = 0
|
|
raw = self.rfile.read(length) if length > 0 else b"{}"
|
|
timeout = 30.0 if path.endswith(("/op/start", "/motor/start",
|
|
"/flash", "/eeprom/write",
|
|
"/eeprom/read", "/eeprom/verify")) else 6.0
|
|
status, body = _ecat_proxy(path, raw, timeout)
|
|
self._post_json(body, status)
|
|
return
|
|
self.send_error(404)
|
|
|
|
def _post_json(self, body, status=200):
|
|
self.send_response(status)
|
|
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()
|