feat: add embedded app PC debugging skill

This commit is contained in:
zoukai committed 2026-09-11 17:17:09 +08:00
commit 23e3a4910a
12 files changed
+2818

No files matched your search

+66
View File
@@ -0,0 +1,66 @@
---
name: embedded-app-pc-debug
description: 用于 Windows 电脑端的嵌入式应用调试,包括固件远程打包、设备发现与身份核对、ROM 升级、串口实时监测、升级后验证和故障诊断。适用于 Eapil/CamTells 工作流,也允许覆盖主机、项目、设备、串口和输出路径;普通代码开发或仅查看已有日志时不要自动触发。
---
# Windows电脑端调试
完成从远程工程构建到设备升级验证的闭环,并保留足够证据判断失败发生在哪个阶段。
## 按需读取
- 涉及远程编译、工程定位、软链接、下载产物时,读取 [references/build-workflow.md](references/build-workflow.md)。
- 涉及广播搜索、基础信息、EPFP 文件传输或升级时,读取 [references/device-protocols.md](references/device-protocols.md)。
- 涉及串口监测、升级成功判据或故障定位时,读取 [references/validation-and-diagnostics.md](references/validation-and-diagnostics.md)。
- 涉及 Linux 打包任务与 Windows 设备任务之间交接时,读取 [references/cross-host-coordination.md](references/cross-host-coordination.md)。
只读取当前任务需要的参考文件。不要因为用户仅要求串口查看而加载远程打包流程。
## 默认环境
- SSH 配置主机:`emb.eapil.com`
- 远程项目根目录:`/CamTells`
- 默认 Public 来源:`~/work/develop/gitcode/EMX2.1`
- 本地归档根目录:`F:\tmp\embedded-app-pc-debug`
- 串口默认值:`115200`、`8N1`、无流控
这些都是可覆盖的默认值。不要硬编码项目、SN、IP、COM 口或 ROM 文件。
## 工作方式
1. 根据用户请求确定需要执行的阶段;不强迫每次都完成整条链路。
2. 自己调查文件、脚本参数、网卡和设备响应等可查事实,只向用户询问真正需要选择的事项。串口调试必须先取得本次用户确认的 COM 口;Telnet 调试必须先取得完整 SN 和型号。
3. 将每次完整任务归档到 `<根目录>\<项目>\<YYYYMMDD-HHmmss>`,分别保存 `firmware` 和 `logs`。
4. 搜索可以按型号、IP 或唯一响应列出候选,但这些条件都不能替代完整 SN。Telnet、升级及其他设备写操作必须以用户给出的完整 SN 精确匹配目标,并核对型号;多个候选或同一 SN 出现冲突时停止,不能默认第一台。
5. 先启动串口监测,再执行设备升级;串口输出同时实时显示并保存日志。需要在网页中查看日志或向设备输入命令时,使用单进程串口网页控制台,避免采集进程和发送进程争用同一个 COM 口。网页首次打开显示最新日志,所有发送内容都写入文本日志,空输入也必须发送真实 `CR`。刷新网页不能中断采集;多个页面共用服务时仅关闭最后一个页面才触发停止。最后页面关闭后,即使处于升级期间,也要在约 `5` 秒内结束采集、保留日志并释放 COM 口;之后必须重新启动控制台才能再次连接。
6. 在 ROM 传输前集中显示目标 SN、型号、IP、当前版本、ROM 绝对路径、大小、SHA-256、可解析的目标版本、串口状态和所有无法验证的项目。
7. 只有用户在看到上述信息后明确确认升级,才能传输 ROM、烧录或触发重启。之前对搜索、编译或诊断的授权不能替代这次确认。
8. 升级失败后先诊断并保留证据,不自动重试;再次升级需要新的明确确认。
用户如果只要求“显示串口”或“实时看日志”,仅打开只读串口监测;用户希望在页面持续查看时可以使用本地串口网页控制台,但不自动发送命令、搜索设备、登录、打包、传输文件或升级。
## Git 与构建边界
- 不执行编译前 `git status` 检查,不因未提交修改阻止打包。
- 不自动提交、推送、暂存、还原或清理任何仓库修改。
- 默认执行 `EmxGitInfo` 并保存输出作为版本凭证;失败时记录原因。临时验证包可继续,正式发布是否继续由用户决定。
- 不使用旧 ROM 冒充本次产物。构建脚本若明确宣告 ROM 目标成功并验证了本次 ROM,之后仅辅助全量镜像步骤失败时,可以交付“仅供局域网升级的 ROM”,但不能声称完整发布包成功。
## 工具脚本
- `scripts/device_upgrade.py`:V2 `30101` 身份预检与 `EPFP/4001` 升级。缺少 `--execute` 时绝不发送 ROM。
- `scripts/serial_monitor.ps1`:无第三方依赖的 Windows 串口实时输出、原始字节保存和带时间戳文本日志。
- `scripts/serial_web_console.ps1`:使用 Windows PowerShell 5.1 启动本地串口网页控制台;由单个进程独占 COM 口并同时负责接收、记录和发送,页面仅监听 `127.0.0.1`,写操作和关闭通知使用随机会话令牌。网页只提供用户主动输入和发送,不启动 U-Boot 持续按键或自动切换 A/B 分区。最后网页关闭后服务退出,不部署常驻管理进程或自动重连。必须用 `powershell.exe` 启动,不要改用 `pwsh.exe`。
- `scripts/v2_discover.py`:枚举适合广播的 Windows IPv4 网卡,从每个接口的本地 `8951` 向设备 `8941` 发送 V2 广播,并严格筛选完整 SN/型号。
运行脚本前先查看 `--help` 或 PowerShell 参数说明。不得绕过脚本的完整 SN、ROM 路径、日志目录和执行开关保护。
## 调试数据
这是设备调试 skill。用户已选择让终端和任务日志保留原始调试信息,不对 `30102`、串口或构建输出中的密码、密钥、Token 做自动脱敏;但不得把某台设备的真实凭据写入本 skill、脚本默认值或参考文档。
## 完成条件
不要把连接建立、ROM 发送结束、下载进度达到 100% 或 Flash 计数达到总量单独称为升级成功。至少需要明确的协议成功终态,并继续核对同一 SN 的重启、重发现、网络恢复和目标版本;无法取得某项证据时明确说明,不补造结论。
串口网页控制台发生变更时,还要验证服务代码能编译、Skill 结构有效、COM 口由单一进程占用、服务实际连接、接收字节或日志文件持续增长、普通发送与空回车都有 TX 记录,并确认页面首次打开定位到最新日志。还必须完成生命周期验收:刷新不中断、两个页面关闭一个仍运行、关闭最后页面后约 `5` 秒内进程退出且状态接口离线、日志具有结束标记、原日志与原始字节文件仍在原路径,并且下一次服务能重新打开同一 COM 口。仅看到服务进程或空白网页不算验证通过。
+6
View File
@@ -0,0 +1,6 @@
interface:
display_name: "Windows电脑端调试"
short_description: "完成 Windows 电脑端设备发现、固件升级、串口监测与验证"
default_prompt: "使用 $embedded-app-pc-debug 在 Windows 电脑端锁定目标设备,并完成固件升级、串口监测和结果验证。"
policy:
allow_implicit_invocation: true
+420
View File
@@ -0,0 +1,420 @@
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>串口实时控制台</title>
<style>
:root {
color-scheme: dark;
--bg: #0d1117;
--panel: #161b22;
--border: #30363d;
--text: #d7e0ea;
--muted: #8b949e;
--ok: #3fb950;
--warn: #d29922;
--danger: #f85149;
--button: #21262d;
--button-hover: #30363d;
--accent: #238636;
}
* { box-sizing: border-box; }
html, body { height: 100%; margin: 0; }
body {
display: flex;
flex-direction: column;
min-height: 100dvh;
overflow: hidden;
background: var(--bg);
color: var(--text);
font-family: Consolas, "Cascadia Mono", "Microsoft YaHei UI", monospace;
}
header {
display: flex;
align-items: center;
gap: 16px;
min-height: 43px;
padding: 8px 12px;
background: var(--panel);
border-bottom: 1px solid var(--border);
font-size: 13px;
white-space: nowrap;
}
#connection { color: var(--warn); font-weight: 700; }
#meta { color: var(--muted); overflow: hidden; text-overflow: ellipsis; }
.follow { margin-left: auto; color: var(--muted); }
main { flex: 1; min-height: 0; }
#log {
height: 100%;
margin: 0;
padding: 10px 12px 24px;
overflow: auto;
white-space: pre-wrap;
overflow-wrap: anywhere;
tab-size: 4;
font: 13px/1.42 Consolas, "Cascadia Mono", "Microsoft YaHei UI", monospace;
}
.console {
flex: none;
padding: 10px 12px 11px;
background: var(--panel);
border-top: 1px solid var(--border);
box-shadow: 0 -8px 24px rgba(0, 0, 0, .18);
}
.quick {
display: flex;
align-items: center;
flex-wrap: wrap;
gap: 6px;
}
.quick { margin-bottom: 8px; }
.quick-label { color: var(--muted); margin-right: 2px; }
.command-row {
display: grid;
grid-template-columns: auto minmax(0, 1fr) auto;
align-items: end;
gap: 7px;
}
.command-prompt {
display: flex;
align-items: center;
height: 38px;
padding: 0 9px;
color: #79c0ff;
background: #0d1117;
border: 1px solid var(--border);
border-radius: 6px;
font-weight: 700;
user-select: none;
}
button, textarea {
color: var(--text);
background: var(--button);
border: 1px solid var(--border);
border-radius: 6px;
font: 13px Consolas, "Cascadia Mono", "Microsoft YaHei UI", monospace;
}
button { min-height: 30px; padding: 4px 9px; cursor: pointer; }
button:hover:not(:disabled) { background: var(--button-hover); }
button:disabled { cursor: not-allowed; opacity: .45; }
button.primary { background: var(--accent); border-color: #2ea043; }
button.danger { color: #ffb3ad; }
textarea {
display: block;
width: 100%;
height: 38px;
min-height: 38px;
max-height: 116px;
margin: 0;
padding: 8px 10px;
resize: none;
overflow-y: auto;
line-height: 1.4;
outline: none;
}
textarea:focus { border-color: #58a6ff; box-shadow: 0 0 0 1px #58a6ff; }
textarea.masked { -webkit-text-security: disc; }
#send { min-width: 92px; height: 38px; }
@media (max-width: 720px) {
header { flex-wrap: wrap; }
.follow { margin-left: 0; }
.command-row { grid-template-columns: auto minmax(0, 1fr); }
#send { grid-column: 1 / -1; width: 100%; }
}
</style>
</head>
<body data-token="__SESSION_TOKEN__" data-port="__SERIAL_PORT__">
<header>
<span id="connection">正在连接…</span>
<span id="meta">串口状态读取中</span>
<label class="follow"><input id="follow" type="checkbox" checked> 自动滚动</label>
</header>
<main><pre id="log">正在读取串口日志…</pre></main>
<section class="console" aria-label="串口命令区">
<div class="quick">
<span class="quick-label">快捷填入:</span>
<button class="fill" data-command="printenv systemSelect">查询分区</button>
<button class="fill" data-command="setenv systemSelect A">填入分区 A</button>
<button class="fill" data-command="setenv systemSelect B">填入分区 B</button>
<button class="fill" data-command="saveenv">保存环境</button>
<button class="fill" data-command="reset">重启设备</button>
<button class="fill" data-command="ifconfig">ifconfig</button>
<button class="fill" data-command="dmesg">dmesg</button>
</div>
<div class="command-row">
<span id="commandPrompt" class="command-prompt">COM &gt;</span>
<textarea id="command" rows="1" spellcheck="false" autocomplete="off" placeholder="输入命令;留空按 Enter 发送回车"></textarea>
<button id="send" class="primary" disabled>发送 Enter</button>
</div>
</section>
<script>
const token = document.body.dataset.token;
const configuredPort = document.body.dataset.port;
const clientId = window.crypto && typeof window.crypto.randomUUID === 'function'
? window.crypto.randomUUID()
: `${Date.now()}-${Math.random().toString(16).slice(2)}`;
const logView = document.querySelector('#log');
const connectionView = document.querySelector('#connection');
const metaView = document.querySelector('#meta');
const followView = document.querySelector('#follow');
const commandView = document.querySelector('#command');
const commandPromptView = document.querySelector('#commandPrompt');
const sendButton = document.querySelector('#send');
const decoder = new TextDecoder('utf-8');
const commandHistory = [];
let historyIndex = 0;
let historyDraft = '';
let logPolling = false;
let statusPolling = false;
let hasRenderedLog = false;
let serialOpen = false;
let sendBusy = false;
let promptMode = 'unknown';
let pageClosing = false;
commandPromptView.textContent = `${configuredPort} >`;
function clientUrl(path) {
const separator = path.includes('?') ? '&' : '?';
return `${path}${separator}client=${encodeURIComponent(clientId)}`;
}
// 仅清理终端显示控制字符,不修改设备输出中的普通调试内容。
function removeTerminalControls(text) {
text = text.replace(/\x1b\][^\x07]*(?:\x07|\x1b\\)/g, '');
text = text.replace(/\x1b\[[0-?]*[ -\/]*[@-~]/g, '');
const output = [];
for (const char of text) {
if (char === '\b') {
if (output.length && output[output.length - 1] !== '\n') output.pop();
} else {
output.push(char);
}
}
return output.join('').replace(/[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]/g, '');
}
function updateSessionState(text) {
const lastLogin = text.lastIndexOf('login:');
const lastPassword = text.lastIndexOf('Password:');
const lastFailure = text.lastIndexOf('Login incorrect');
const lastShell = Math.max(text.lastIndexOf(']~#'), text.lastIndexOf(']~$'));
const latestPrompt = Math.max(lastLogin, lastPassword, lastFailure, lastShell);
if (latestPrompt === lastShell && lastShell >= 0) {
promptMode = 'shell';
} else if (latestPrompt === lastPassword && lastPassword >= 0) {
promptMode = 'password';
} else if (latestPrompt === lastLogin && lastLogin >= 0) {
promptMode = 'login';
} else if (latestPrompt === lastFailure && lastFailure >= 0) {
promptMode = 'failed';
} else {
promptMode = 'unknown';
}
commandView.classList.toggle('masked', promptMode === 'password');
}
function resizeCommandInput() {
commandView.style.height = '38px';
commandView.style.height = `${Math.min(116, Math.max(38, commandView.scrollHeight))}px`;
}
function setCommandValue(value) {
commandView.value = value;
resizeCommandInput();
commandView.focus();
commandView.setSelectionRange(value.length, value.length);
}
function rememberCommand(command) {
if (!command.length) return;
const duplicateIndex = commandHistory.indexOf(command);
if (duplicateIndex >= 0) commandHistory.splice(duplicateIndex, 1);
commandHistory.push(command);
if (commandHistory.length > 50) commandHistory.shift();
historyIndex = commandHistory.length;
historyDraft = '';
}
function updateInputControls() {
sendButton.disabled = !serialOpen || sendBusy;
}
function applyConnectionState(isOpen) {
serialOpen = isOpen;
updateInputControls();
}
async function refreshLog() {
if (logPolling) return;
logPolling = true;
const nearBottom = logView.scrollHeight - logView.scrollTop - logView.clientHeight < 100;
const shouldFollowLatest = followView.checked && (!hasRenderedLog || nearBottom);
try {
const response = await fetch(`${clientUrl('/api/log')}&t=${Date.now()}`, { cache: 'no-store' });
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const bytes = new Uint8Array(await response.arrayBuffer());
const fullText = removeTerminalControls(decoder.decode(bytes));
updateSessionState(fullText);
const visibleText = fullText.length > 500000
? `[前面的日志已在页面中省略,完整数据仍保存在文件中]\n${fullText.slice(-500000)}`
: fullText;
logView.textContent = visibleText || `日志文件当前为空,正在等待 ${configuredPort} 数据…`;
// 首次打开必须显示最新日志;等待浏览器完成布局后再滚动到底部,避免大日志仍停在开头。
if (shouldFollowLatest) {
requestAnimationFrame(() => {
logView.scrollTop = logView.scrollHeight;
});
}
hasRenderedLog = true;
} catch (error) {
logView.textContent = `${logView.textContent}\n[网页] 日志读取失败:${error.message}`;
} finally {
logPolling = false;
}
}
async function refreshStatus() {
if (statusPolling) return;
statusPolling = true;
try {
const response = await fetch(`${clientUrl('/api/status')}&t=${Date.now()}`, { cache: 'no-store' });
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const status = await response.json();
applyConnectionState(Boolean(status.isOpen));
connectionView.textContent = status.isOpen ? `● ${status.port} 已连接` : `● ${status.port} 已断开`;
connectionView.style.color = status.isOpen ? 'var(--ok)' : 'var(--danger)';
const lastRx = status.lastReceiveUtc ? new Date(status.lastReceiveUtc).toLocaleTimeString() : '暂无数据';
metaView.textContent = `115200 8N1 · 本次接收 ${Number(status.receivedBytes).toLocaleString()} 字节 · 最后数据 ${lastRx}`;
updateInputControls();
} catch (error) {
applyConnectionState(false);
connectionView.textContent = `● ${configuredPort} 服务离线`;
connectionView.style.color = 'var(--danger)';
metaView.textContent = `无法连接本地串口服务:${error.message}`;
} finally {
statusPolling = false;
}
}
async function postCommand(path, body = '') {
const response = await fetch(path, {
method: 'POST',
headers: {
'Content-Type': 'text/plain; charset=utf-8',
'X-Serial-Token': token
},
body
});
let result = {};
try { result = await response.json(); } catch { }
if (!response.ok) throw new Error(result.error || `HTTP ${response.status}`);
return result;
}
async function sendCommand(commandOverride) {
const fromInput = typeof commandOverride !== 'string';
const command = fromInput ? commandView.value : commandOverride;
const wasSecret = fromInput && promptMode === 'password';
if (!serialOpen || sendBusy) return;
sendBusy = true;
updateInputControls();
try {
await postCommand('/api/send?ending=CR', command);
if (!wasSecret) rememberCommand(command);
if (fromInput && commandView.value === command) setCommandValue('');
if (wasSecret) commandView.classList.remove('masked');
setTimeout(refreshLog, 50);
} catch (error) {
connectionView.textContent = `● ${configuredPort} 发送失败:${error.message}`;
connectionView.style.color = 'var(--danger)';
} finally {
sendBusy = false;
updateInputControls();
commandView.focus();
}
}
function disconnectClient() {
if (pageClosing) return;
pageClosing = true;
const disconnectUrl = `${clientUrl('/api/client/disconnect')}&token=${encodeURIComponent(token)}`;
// 浏览器关闭标签页时普通异步请求可能被直接取消;sendBeacon 专门用于离页通知。
// 某些内置浏览器未实现 sendBeacon,因此同时保留 keepalive fetch 作为兼容兜底。
if (navigator.sendBeacon) {
try {
navigator.sendBeacon(disconnectUrl, new Blob([], { type: 'text/plain;charset=UTF-8' }));
} catch (_) { }
}
fetch(disconnectUrl, {
method: 'POST',
headers: {
'Content-Type': 'text/plain; charset=utf-8',
'X-Serial-Token': token
},
body: '',
keepalive: true
}).catch(() => { });
}
document.querySelectorAll('.fill').forEach((button) => {
button.addEventListener('click', () => {
setCommandValue(button.dataset.command || '');
});
});
sendButton.addEventListener('click', sendCommand);
followView.addEventListener('change', () => {
if (followView.checked) {
requestAnimationFrame(() => {
logView.scrollTop = logView.scrollHeight;
});
}
});
commandView.addEventListener('input', resizeCommandInput);
commandView.addEventListener('keydown', (event) => {
// 中文输入法确认候选时的 Enter 不应被当成串口发送。
if (event.isComposing || event.keyCode === 229) return;
if (event.key === 'Enter' && !event.shiftKey) {
event.preventDefault();
if (event.repeat && commandView.value.length === 0) return;
sendCommand();
return;
}
if (event.key === 'Escape') {
event.preventDefault();
setCommandValue('');
return;
}
const atStart = commandView.selectionStart === 0 && commandView.selectionEnd === 0;
const atEnd = commandView.selectionStart === commandView.value.length && commandView.selectionEnd === commandView.value.length;
if (event.key === 'ArrowUp' && commandHistory.length && (atStart || !commandView.value.includes('\n'))) {
event.preventDefault();
if (historyIndex === commandHistory.length) historyDraft = commandView.value;
historyIndex = Math.max(0, historyIndex - 1);
setCommandValue(commandHistory[historyIndex]);
} else if (event.key === 'ArrowDown' && commandHistory.length && (atEnd || !commandView.value.includes('\n'))) {
event.preventDefault();
historyIndex = Math.min(commandHistory.length, historyIndex + 1);
setCommandValue(historyIndex === commandHistory.length ? historyDraft : commandHistory[historyIndex]);
}
});
window.addEventListener('pagehide', disconnectClient);
window.addEventListener('beforeunload', disconnectClient);
window.addEventListener('pageshow', () => {
pageClosing = false;
refreshStatus();
});
refreshLog();
refreshStatus();
setInterval(refreshLog, 800);
setInterval(refreshStatus, 500);
</script>
</body>
</html>
+64
View File
@@ -0,0 +1,64 @@
# 远程打包与产物归档
仅在任务涉及编译新固件或下载远程产物时读取本文件。
## 默认值与覆盖
| 项目 | 默认值 |
|---|---|
| SSH Host | `emb.eapil.com` |
| 工程根目录 | `/CamTells` |
| Public 来源 | `~/work/develop/gitcode/EMX2.1` |
| 本地归档根目录 | `F:\tmp\embedded-app-pc-debug` |
用户提供的值优先。凭据只使用现有 SSH 配置、密钥代理或交互式认证,不把密码、私钥和 Token 保存到 skill 或任务日志。
## 工程定位
1. 先用现有 SSH/VS Code 配置解析 Host,不猜测用户名、端口或私钥路径。
2. 将用户给出的项目名视为 `/CamTells` 下的相对路径并精确匹配。
3. 如果有多个候选,列出远程绝对路径等待选择;不要凭目录修改时间挑选。
4. 确认工程目录包含待用脚本,再确定真实命令行参数。
## 构建顺序
1. 在当前工程目录执行 `EmxGitInfo`,将标准输出、标准错误和退出码保存为 `logs/EmxGitInfo.txt`。不先运行 Git 工作区检查。
2. 查看 `scripts/product.py`、`release.py` 的帮助、源码入口或项目说明,确认参数和要求的工作目录。不同工程的参数及启动目录可能不同,不能照搬其他项目命令;脚本位于 `scripts` 不代表应从工程根目录启动,也不代表一定应从 `scripts` 启动。
3. 只有实际缺少 `Public`、第三方库映射或脚本明确报资源缺失时,才检查并运行 `softLink.py`。创建前核对目标与默认来源 `~/work/develop/gitcode/EMX2.1`,避免覆盖已有有效链接。
4. 先运行 `product.py` 构建环境包,成功后运行 `release.py` 编译项目。
5. 每一步记录:工作目录、完整命令、开始/结束时间、退出码和关键输出。
6. 尊重构建脚本内部的校验与重建循环。脚本因随机命中、签名或封装检查而自动重建时,让脚本自行完成;不要在外层重复启动、重复改版本或把一次内部重建误报为多个发布版本。
7. 构建失败时不下载旧产物。若本次 ROM 已由脚本明确生成、校验通过,而随后仅全量烧录镜像或其他辅助目标失败,应分开报告:允许交付已验证 ROM 用于局域网升级,但完整发布包仍判为失败或不完整。
## 打包任务交付清单
Linux 打包任务应一次性交付以下事实,供 Windows 设备任务直接使用:
- 工程绝对路径、实际运行目录和构建命令;
- ROM 与 `devInfo.json` 的远程绝对路径;
- ROM 大小、MD5、SHA-256、目标内外层版本;
- 构建记录、警告以及哪些发布目标成功或失败;
- 产物是否已下载或排队到 Windows 任务。
Windows 任务已取得内容完整且自洽的交付清单后,不重复 SSH 到 Linux 重新计算同一组构建事实。只有交付缺字段、相互矛盾、文件校验不一致或用户明确要求复核时,才回到远端检查。
## 下载与归档
创建:
```text
F:\tmp\embedded-app-pc-debug\<项目名>\<YYYYMMDD-HHmmss>\
├─ firmware\
└─ logs\
```
下载前解析远程 ROM 的真实绝对路径。下载后核对交付清单中的远端大小与本地大小,并计算本地 SHA-256;已有远端 SHA-256 时必须比较。多个 ROM 候选时列出文件名、大小、时间和可解析版本;只有唯一候选才可自动选择。
建议日志:
- `build.log`
- `EmxGitInfo.txt`
- `download.log`
- `firmware.sha256`
不要自动提交、推送、清理或还原远程仓库。
+38
View File
@@ -0,0 +1,38 @@
# Linux 打包任务与 Windows 设备任务协作
仅在固件由 Linux 任务构建、由 Windows 任务下载并升级设备时读取本文件。
## 职责边界
- Linux 任务负责工程定位、构建、产物验证、版本与哈希、构建日志和警告。
- Windows 任务负责接收产物、本地完整性比较、广播发现、完整 SN/型号核对、串口监测、EPFP 升级和升级后验证。
- 已由上游任务验证并完整交付的事实,下游不重复计算。发现字段缺失、矛盾、哈希不一致或用户要求复核时再回到上游。
## 建议交接流程
1. Windows 任务通过可用的 Codex 任务工具联系或等待 Linux 任务;等待超时只表示尚未得到新状态,不要重复发送同一请求。
2. Linux 任务完成后用短消息交付路径、大小、哈希、版本和结论;ROM、`devInfo.json` 与大日志使用 SCP 或共享文件传递。
3. Windows 任务比较本地文件与交付清单,然后开始设备侧预检。校验一致时不再登录 Linux 重算哈希或重跑构建命令。
4. 升级出现异常时,Windows 任务上传本次 skill 生成的串口日志快照,并把精确时间线与诊断问题发回 Linux 任务。
5. Linux 任务回复源码/构建侧分析;Windows 任务负责将分析与设备现场证据合并成最终验证报告。
## 工具与降级原则
- 优先使用任务列表、读取、消息和等待能力;不同主机或版本的 Codex 工具能力可能不同,先确认工具实际返回结果。
- 不得把“文件页已打开/已排队”说成“消息已送达”,也不得在读取结果为空时编造对方结论。
- 读取不到任务正文时,先核对任务 ID 和主机 ID,再查看最新任务状态。只有任务工具确实不能返回内容时,才在已授权主机上读取对应会话记录或请用户转交摘要。
- 任务 ID、主机 ID、项目路径和远端临时目录都属于本次运行参数,不写成通用固定值。
## 最小交接消息
```text
项目:<绝对路径>
ROM:<绝对路径>
devInfo:<绝对路径>
大小:<bytes>
MD5:<md5>
SHA-256:<sha256>
目标版本:inner=<值>,outer=<值>
构建结论:<ROM成功/完整发布成功/仅ROM可用/失败>
警告:<无或摘要>
```
+108
View File
@@ -0,0 +1,108 @@
# 设备发现、基础信息与升级协议
仅在任务涉及设备网络发现、版本查询或 ROM 传输时读取本文件。优先使用用户提供的最新协议、现有已验证实现和抓包;文档歧义不得伪装成确定事实。
## 本机参考资料
存在时优先核对:
- `C:\Users\ZK\Documents\ChatGPT\日常开发\调试资料\设备端协议V2-工厂整合版本.md`
- `C:\Users\ZK\Documents\ChatGPT\日常开发\调试资料\设备端统一配置协议V2.md`
- `C:\Users\ZK\Downloads\设备端TCP文件传输类协议.md`
- `C:\Users\ZK\Downloads\工厂协议V1.md`
- `E:\CodeSpace\git-pro\eapil_qt_player_v2\EpProtocolDemo`
## V2 协议矩阵
| 功能 | 传输与端口 | 帧格式 |
|---|---|---|
| 设备发现请求 | UDP,客户端从本地 `8951` 向设备 `8941` 广播 | `EPBP + uint32长度 + UTF-8 JSON` |
| 设备发现响应 | UDP,客户端 `8951` 接收 | 同上,完整帧在一个数据报内 |
| 统一配置 | TCP,设备 `8942` | `EPTP + uint32长度 + UTF-8 JSON` |
| 固件升级 | TCP,设备 `8943` | `EPFP + uint32命令 + uint32长度 + 二进制数据` |
多字节整数使用网络字节序。配置 TCP 长度端序在整合文档中没有单独声明,但现有设备和客户端验证采用网络序。
## 发现与身份
V2 发现请求 JSON:
```json
{"cmd":60000,"session":12345}
```
响应必须回显相同 `cmd/session`。身份字段包括 `sn`、顶层烧录 `mac`、`inner`、`outer`、`hardware` 和 `net`。顶层 `mac` 与 `net.mac` 可能不同。型号判断同时展示:
- `inner.type`
- `outer.type`
- `hardware.type`
非关键网络字段允许缺失。完整 SN 是升级身份核对的首要依据。V1 的端口与帧格式必须从 V1 文档或已验证的 `EpProtocolDemo/V1` 获取,不能套用 V2。
执行发现时使用 `scripts/v2_discover.py`,并遵守以下规则:
1. 默认枚举已启用、已连接且具有有效 IPv4 的物理网卡;回环、断开、APIPA、VPN、VMware、Npcap 等接口要列出并注明排除原因。用户明确要求所有活动接口时才加入虚拟接口。
2. 每个接口分别绑定其本地 IPv4 和客户端端口 `8951`,同时发送有限广播 `255.255.255.255` 和按前缀计算的定向广播。不能把 `/22` 的广播地址按 `/24` 猜测。
3. 如果 UDP `8951` 被 EapilCameraConfigurer 或其他进程占用,报告占用者,不结束进程;请用户释放端口,或使用已有工具明确支持的搜索结果。
4. 从临时源端口发包只可作为诊断。设备可能只向 `8951` 回复,因此临时端口无响应不能证明设备离线。
5. 打印所有合法候选并按完整 SN 精确匹配;不得选择第一台。响应需校验 `EPBP`、长度、`cmd/session`。只有 SN、源 IP、MAC 和型号都一致的重复响应可以合并;同一 SN 对应不同身份信息时报告冲突并禁止自动选择。
6. 未提供完整 `--expected-sn` 时脚本只是列表模式,即使退出码为 0 也不代表身份核验成功。Telnet、升级及其他设备写操作必须提供完整 SN,并根据任务要求同时传入型号。
## 基础信息
通过 TCP `8942` 发送 `cmd=30101`。响应必须满足:同步头正确、长度合理、`cmd/session` 匹配且 `code=0`。读取:
- `sn`、`mac`
- `inner.type/version`
- `outer.type/version`
- `hardware.type/version`
`hardware.version` 不是固件版本。升级后应根据 ROM 类型比较 `inner.version` 或 `outer.version`。如果无法从 ROM 或构建记录确定应变化的字段,必须明确说明并让用户确认预期。
`30102` 可能返回密码和密钥,只在调试任务确实需要时调用。用户选择保留原始调试数据,因此响应可原样写入本次任务日志,但不得写入 skill 本体。
## 固件升级 4001
请求:
```text
EPFP
+ uint32 4001
+ uint32 ROM字节数
+ ROM原始字节
```
ROM 前没有 MD5 或目标路径业务前缀。不要把普通文件写入 `2001` 的 MD5 机制声称为升级 `4001` 自带校验。
状态业务数据为 21 字节:
```text
uint8 status
+ uint32 sessionId
+ uint32 downloadTotal
+ uint32 downloaded
+ uint32 flashTotal
+ uint32 flashed
```
状态:`0` 空闲、`1` 开始、`2` 完成、`3` 失败。`sessionId` 不等同于配置 JSON session,不要求匹配。解析器要兼容完整 `EPFP/cmd/length` 状态帧与设备实测出现的裸 21 字节状态,但完整帧必须校验响应命令为 `4001`。
只有 `status=2` 是协议成功终态;下载或 Flash 计数达到总量不能替代它。`status=3` 立即失败。连续 60 秒没有下载或 Flash 进度变化视为停滞,值允许覆盖。没有依据时不自动续传或重试。
设备报告的 `downloadTotal`、`flashTotal` 可能与 ROM 文件字节数不同,不能因为这种封装头、元数据或分区差异单独判失败。`status=2` 是协议成功终态;如果此时 `downloaded/flashed` 仍小于设备各自报告的非零总量,记录协议计数异常警告并纳入验证报告,但不自行推翻设备的成功终态。
## 升级前保护
- ROM 必须是普通文件,大小大于 0 且不超过无符号 32 位长度。
- 计算并记录 SHA-256;协议本身没有 ROM 签名或来源认证,不能声称 SHA-256 提供设备侧认证。
- 精确核对设备 SN。
- 同时展示当前型号和 ROM 目标型号。ROM 元数据无法解析时显示“无法验证型号”,并重新取得用户确认。
- 多个 ROM 或设备候选时等待选择。搜索端口被占用或没有完成完整 SN 核对时不得升级。
- 确认后调用 `scripts/device_upgrade.py --execute`,不绕过其保护参数。
## 已知协议歧义
- 状态章节没有明确说明每次状态是否带 EPFP 外层头,因此客户端兼容完整帧和裸状态。
- 没有规定分块大小、连接关闭方、断线恢复或统一总超时;使用可配置值并保存最后进度。
- 没有定义 ROM 与型号的自动匹配机制。
- TCP 是字节流,必须累积并按长度解析,不能把单次 `recv` 当作完整帧。
+100
View File
@@ -0,0 +1,100 @@
# 串口监测、验证与故障诊断
仅在任务涉及串口、设备重启或升级结果判断时读取本文件。
## 串口准备
1. 每次串口任务先请用户确认当前 COM 口;可以枚举 Windows 串口及描述帮助确认,但不要沿用上一次的端口或自行挑选。
2. 默认 `115200`、8 数据位、1 停止位、无校验、无流控,允许覆盖。
3. 打开失败、打开后没有输出、乱码或访问拒绝时立即停手,请用户核对 COM 口、波特率/校验/数据位/停止位/流控、端口是否已释放,必要时重新插拔或更换转接器。不要强制结束 SecureCRT、串口工具或其他进程。
4. 只读采集使用 `scripts/serial_monitor.ps1`;需要网页实时显示或网页命令输入时,使用 `scripts/serial_web_console.ps1`。确认 COM 口已打开并开始记录后,才传输 ROM。
5. 串口网页控制台必须由同一个进程独占 COM 口并完成收、发和日志写入,不能再启动第二个采集或发送进程。服务只绑定 `127.0.0.1`,同时写本任务自己的文本日志和原始字节文件。调试信息不脱敏。
6. 不主动读取、复制或上传 SecureCRT 的历史日志。只有本 skill 在本次任务生成的文本和原始字节日志可以默认使用;其他工具日志需用户明确授权。
只读串口脚本只负责监测。网页控制台允许用户直接输入命令,页面按 Enter 后固定追加 `CR` 并立即发送,不做危险命令二次确认。命令框为空时也发送 `CR`,表现为真实回车。服务接口仍支持无结尾、`CR`、`LF`、`CRLF`,供明确的自动化流程调用。AI 仍只能在用户明确要求或确认具体命令后代为发送,不得自行发送登录名、密码、Shell 命令、break、DTR 或 RTS 脉冲。所有写入数据都在文本日志中记录为 `[TX:来源]`,不可见字符显示为 `<CR>`、`<LF>` 等;RX 文本日志和原始字节日志保持不变。
网页输入区保持在日志下方,只保留快捷填入、命令框和发送按钮,不显示行尾选择、独立空回车、历史下拉、清空、密码开关、U-Boot 连续发送按钮及交互状态栏。页面首次打开时默认定位到最新日志;启用自动滚动后持续跟随最新输出,用户手动向上查看旧日志时不抢回位置,再次勾选自动滚动立即回到底部。Enter 发送、Shift+Enter 输入多行、上下方向键浏览当前页面的隐式历史、Esc 清空输入;处理中文输入法组合状态,确认候选时不能误发串口命令,并阻止长按 Enter 连续发送空回车。发送请求未完成时阻止重复提交,成功后立即刷新 TX 日志并保持输入焦点。页面识别 `Password:` 阶段并自动遮挡输入,而且密码不加入页面命令历史。按用户要求,实际发送内容仍会写入 TX 文本日志。
每个网页标签维护独立客户端心跳。最后一个页面正常发送关闭通知后等待约 `2` 秒,若没有刷新后的新页面接管,则服务自动结束日志采集、写入结束标记并释放 COM 口;多个标签同时打开时,关闭其中一个不能停止其他页面。页面异常退出或内置浏览器不触发离页通知时,心跳超过 `3` 秒后清理失联页面,再进入 `2` 秒宽限期,因此最后页面关闭后最迟约 `5` 秒退出。此规则同样适用于升级期间:需要全程监测时必须保持至少一个页面打开。服务启动后尚未打开过网页时不要自动退出。
关闭后的文本日志和原始字节文件保留在原路径,不自动移动、删除或截断。服务退出后原网址不可用,不部署常驻管理服务,也不在再次访问时自动打开串口;需要继续查看时重新运行启动脚本。关闭通知只接受回环地址请求,并校验本次随机会话令牌。
用户只要求实时显示串口时,到此为止:打开终端面板并持续监测,不附带设备搜索、协议登录、文件上传或升级。
## U-Boot 与 A/B 分区手工操作
1. 串口网页控制台不提供 U-Boot 持续按键、定时发送、提示符自动识别停止或一键切换 A/B 分区;不要通过隐藏接口或后台线程补回这些行为。
2. 需要在启动阶段输入 `eapil_uboot_key_str` 时,由用户在普通命令框中自行输入和发送;设备无法由电脑控制重启时,可以在控制台已经连接后请用户手动重启。
3. 进入 U-Boot 后仍由用户按需填入并主动发送命令。常见核对顺序为:`printenv systemSelect`、`setenv systemSelect A` 或 `setenv systemSelect B`、`saveenv`、再次 `printenv systemSelect`、`reset`。快捷按钮只能填入文本,不能自动发送或串联执行。
4. 分区恢复成功需要看到内核命令行中的 `systemSelect=...`、系统明确报告运行于目标 A/B 分区、核心应用恢复,并且没有 panic、看门狗或循环重启。网络恢复属于后续独立诊断,不作为这一步成功的必要条件。
## Telnet 与交互命令
1. 先让用户选择串口或 Telnet。串口需要本次确认的 COM 口;Telnet 需要设备完整 SN 和型号,然后广播发现并精确核对 IP,不能沿用历史 IP。
2. Telnet 登录名和密码从当前 Linux 工程的准确配置读取,优先检查 `product.json` 中的用户配置。不要从截图 OCR、相似机型、旧固件或历史记忆猜测,也不要把凭据写入 skill。
3. 当前配置认证失败或设备版本与配置不一致时再调查历史固件配置。限制失败登录次数;首次明确失败后先停手核实,不连续试猜密码。
4. 串口与 Telnet 是否共用密码必须由工程配置或用户确认,不能推断。
5. 用户明确要求 `ifconfig` 等命令时,若已有已验证且仍连接的会话,就只发送该命令并实时展示结果;登录验证只执行最少的只读命令,完成后退出。
## 成功判据
完整成功需要组合证据:
1. EPFP 返回明确 `status=2`。
2. 串口出现正常校验、烧录、启动过程,没有明确失败终态;如果存在设备已知成功标记,例如 `update successful`,记录其时间。
3. 设备完成重启。
4. 广播重新发现同一个完整 SN。
5. 网络恢复,目标 IP 可达;IP 变化时以同一 SN 为准更新目标。
6. `30101` 查询成功。
7. 应变化的 `inner.version` 或 `outer.version` 与目标构建一致。
缺少任一证据时不要补造“全部成功”。可以报告“协议升级完成,但尚未验证版本”或“设备已重启,但未收到协议终态”等部分结论。
## 失败分类
| 阶段 | 典型证据 | 后续检查 |
|---|---|---|
| 串口准备 | 端口不存在、拒绝访问、被占用 | 核对 COM 口和占用程序 |
| 设备发现 | 无响应、候选冲突 | 网卡、V1/V2、广播出口、防火墙、二层网络 |
| 身份预检 | SN 或型号不匹配 | 停止,不传 ROM,重新选择目标/固件 |
| TCP 连接 | 8942/8943 拒绝或超时 | IP、服务状态、设备网络、端口监听 |
| ROM 发送 | 连接断开、发送异常 | 最后发送量、串口错误、设备重启时点 |
| 下载阶段 | downloaded 不增长 | 网络、设备缓存、服务负载 |
| Flash 阶段 | flashed 不增长、status=3 | ROM 兼容性、分区、校验、Flash 错误 |
| 重启阶段 | 协议完成但串口无启动 | 启动加载器、内核、供电、看门狗 |
| 网络恢复 | 串口正常但未重发现 | DHCP/Wi-Fi、网线、IP 变化、服务启动 |
| 版本回查 | 设备在线但版本未变 | 比较字段选错、升级分区未切换、回滚 |
日志中的 `error`、`failed`、`timeout` 字样不是自动失败判据。结合发生时间、是否重复、协议终态、设备功能、重启次数以及是否出现 panic、segfault、core dump 或看门狗循环判断。升级停止和启动阶段的一次性模块告警可以记录为警告;若同一 SN 已正常重启、网络恢复、版本正确并持续心跳,不应仅凭孤立字样推翻成功结论。
## 日志快照与远端协查
持续监测期间不要为了上传而关闭串口。复制本任务生成的活动文本日志和原始字节文件形成稳定快照,记录时间范围并计算 SHA-256,再上传到约定的远端临时目录,例如 `/tmp/<项目>-windows-logs/<时间戳>`。
请求 Linux 任务协查时附带:完整 SN、型号、IP、COM/波特率、升级前后版本、问题时间窗口、症状和复现步骤、远端日志路径及 SHA-256、希望对方回答的问题。不要在任务消息里粘贴整份大日志。
## 超时和重试
- 有进度就继续等待,连续 60 秒无下载或烧写变化才判定疑似停滞。
- `status=3` 立即失败。
- TCP 断开但未收到 `status=2` 时不能直接成功;结合串口和重启证据报告“协议终态缺失”。
- 保存最后状态、进度、串口末尾片段和错误时间线。
- 不自动再次刷机。分析后重新展示预检信息,并等待新的明确确认。
## 验证报告
`verification.txt` 按以下顺序记录:项目与 ROM、目标设备身份、升级前后版本、协议结果、串口关键事件、网络与身份复查、未完成检查、最终结论。包含 SHA-256、升级前后 IP 和日志路径,不用把全部串口内容复制进摘要。
## 串口网页控制台变更验收
更新串口服务或页面后至少完成以下检查:
1. 使用 `serial_web_console.ps1 -ValidateOnly` 验证 C# 服务能够编译且网页资源存在,再运行 Skill 的 `quick_validate.py`。
2. 启动前确认目标 COM 口没有被第二个串口工具占用;启动后核对状态接口报告端口已打开,不能只根据 PowerShell 进程存在作结论。
3. 设备有输出时核对接收字节数和文本日志文件大小确实增长;设备静默时明确报告“尚未收到数据”,不能把空白页面说成实时采集正常。
4. 发送普通文本后,核对设备响应及对应 `[TX:命令]`;命令框为空时发送 Enter,日志必须出现 `[TX:命令] <CR>`。
5. 首次打开或刷新页面时应显示最新日志;勾选自动滚动后立即回到底部并持续跟随,用户向上查看历史时页面不能强制抢回滚动位置。
6. 核对 Enter 发送、Shift+Enter 换行、上下键隐式历史、Esc 清空、输入法候选确认不误发、长按 Enter 不连续空发,以及请求进行中不重复提交。
7. 出现 `Password:` 后输入框应自动遮挡,密码不加入页面历史;按当前调试策略,TX 文本日志仍保存实际发送内容。
8. 打开两个页面时关闭其中一个,服务应继续运行;关闭最后一个页面后,正常离页约 `2` 秒、未触发离页事件时最迟约 `5` 秒,状态接口应离线、宿主进程应退出,日志应写入结束标记且 COM 口应释放。刷新页面不能误停服务。
9. 自动停止后核对文本日志和原始字节文件仍位于原路径且未被截断;再次启动服务并确认能够重新打开同一 COM 口。升级期间关闭最后页面也按相同规则停止,不能宣称仍在后台监测。
+631
View File
@@ -0,0 +1,631 @@
using System;
using System.Collections.Generic;
using System.Globalization;
using System.IO;
using System.IO.Ports;
using System.Net;
using System.Text;
using System.Threading;
namespace EmbeddedAppPcDebug
{
/// <summary>
/// 在单一进程内独占串口,同时提供只允许本机访问的日志与命令接口。
/// </summary>
public sealed class SerialWebConsoleHost : IDisposable
{
private const int WebClientTimeoutSeconds = 3;
private const int LastClientGraceMilliseconds = 2000;
private readonly SerialPort _serialPort;
private readonly Encoding _encoding;
private readonly Decoder _decoder;
private readonly string _logPath;
private readonly string _rawPath;
private readonly string _htmlTemplate;
private readonly string _listenPrefix;
private readonly string _sessionToken;
private readonly int _readBufferSize;
private readonly object _writeLock = new object();
private readonly object _logLock = new object();
private readonly object _stateLock = new object();
private readonly Dictionary<string, DateTime> _webClients = new Dictionary<string, DateTime>(StringComparer.Ordinal);
private HttpListener _listener;
private FileStream _rawStream;
private StreamWriter _textWriter;
private Thread _receiveThread;
private Thread _httpThread;
private Thread _clientMonitorThread;
private volatile bool _stopping;
private volatile bool _started;
private bool _hadWebClient;
private DateTime _noClientSinceUtc = DateTime.MinValue;
private string _lastError = string.Empty;
private string _pendingLine = string.Empty;
private long _receivedBytes;
private DateTime _lastReceiveUtc = DateTime.MinValue;
public SerialWebConsoleHost(
string portName,
int baudRate,
int dataBits,
string parity,
string stopBits,
string handshake,
string encodingName,
int readTimeoutMilliseconds,
int readBufferSize,
bool dtrEnable,
bool rtsEnable,
string logPath,
string rawPath,
int webPort,
string htmlPath)
{
if (string.IsNullOrWhiteSpace(portName)) throw new ArgumentException("串口名称不能为空。", "portName");
if (string.IsNullOrWhiteSpace(logPath)) throw new ArgumentException("文本日志路径不能为空。", "logPath");
if (string.IsNullOrWhiteSpace(rawPath)) throw new ArgumentException("原始日志路径不能为空。", "rawPath");
if (string.IsNullOrWhiteSpace(htmlPath) || !File.Exists(htmlPath)) throw new FileNotFoundException("找不到串口控制台页面。", htmlPath);
_encoding = Encoding.GetEncoding(encodingName);
_decoder = _encoding.GetDecoder();
_logPath = Path.GetFullPath(logPath);
_rawPath = Path.GetFullPath(rawPath);
_htmlTemplate = File.ReadAllText(htmlPath, new UTF8Encoding(false));
_listenPrefix = "http://127.0.0.1:" + webPort.ToString(CultureInfo.InvariantCulture) + "/";
_sessionToken = Guid.NewGuid().ToString("N");
_readBufferSize = readBufferSize;
_serialPort = new SerialPort();
_serialPort.PortName = portName;
_serialPort.BaudRate = baudRate;
_serialPort.DataBits = dataBits;
_serialPort.Parity = (Parity)Enum.Parse(typeof(Parity), parity, true);
_serialPort.StopBits = (StopBits)Enum.Parse(typeof(StopBits), stopBits, true);
_serialPort.Handshake = (Handshake)Enum.Parse(typeof(Handshake), handshake, true);
_serialPort.DtrEnable = dtrEnable;
_serialPort.RtsEnable = rtsEnable;
_serialPort.ReadTimeout = readTimeoutMilliseconds;
_serialPort.ReadBufferSize = Math.Max(4096, readBufferSize);
}
public bool IsStopping { get { return _stopping; } }
public string Url { get { return _listenPrefix; } }
public void Start()
{
if (_started) throw new InvalidOperationException("串口网页服务已经启动。");
string logDirectory = Path.GetDirectoryName(_logPath);
string rawDirectory = Path.GetDirectoryName(_rawPath);
if (!string.IsNullOrEmpty(logDirectory)) Directory.CreateDirectory(logDirectory);
if (!string.IsNullOrEmpty(rawDirectory)) Directory.CreateDirectory(rawDirectory);
try
{
FileStream textStream = new FileStream(_logPath, FileMode.Append, FileAccess.Write, FileShare.ReadWrite);
_textWriter = new StreamWriter(textStream, new UTF8Encoding(false));
_textWriter.AutoFlush = true;
_rawStream = new FileStream(_rawPath, FileMode.Append, FileAccess.Write, FileShare.ReadWrite);
_serialPort.Open();
WriteSystemLog(string.Format(
CultureInfo.InvariantCulture,
"已打开 {0},{1} {2}{3}{4},流控 {5};网页控制台:{6};原始数据:{7}",
_serialPort.PortName,
_serialPort.BaudRate,
_serialPort.DataBits,
_serialPort.Parity.ToString().Substring(0, 1),
_serialPort.StopBits == StopBits.One ? "1" : (_serialPort.StopBits == StopBits.Two ? "2" : "1.5"),
_serialPort.Handshake,
_listenPrefix,
_rawPath));
_listener = new HttpListener();
_listener.Prefixes.Add(_listenPrefix);
_listener.Start();
_stopping = false;
_started = true;
_receiveThread = StartBackgroundThread(ReceiveLoop, "SerialWebConsole-RX");
_httpThread = StartBackgroundThread(HttpLoop, "SerialWebConsole-HTTP");
_clientMonitorThread = StartBackgroundThread(WebClientMonitorLoop, "SerialWebConsole-WebClient");
}
catch
{
Stop();
throw;
}
}
private static Thread StartBackgroundThread(ThreadStart action, string name)
{
Thread thread = new Thread(action);
thread.IsBackground = true;
thread.Name = name;
thread.Start();
return thread;
}
private void ReceiveLoop()
{
byte[] byteBuffer = new byte[_readBufferSize];
char[] charBuffer = new char[_encoding.GetMaxCharCount(_readBufferSize)];
while (!_stopping)
{
try
{
int count = _serialPort.Read(byteBuffer, 0, byteBuffer.Length);
if (count <= 0) continue;
lock (_logLock)
{
_rawStream.Write(byteBuffer, 0, count);
_rawStream.Flush();
}
int charCount = _decoder.GetChars(byteBuffer, 0, count, charBuffer, 0, false);
if (charCount > 0)
{
string text = new string(charBuffer, 0, charCount);
Console.Write(text);
WriteReceivedText(text);
}
Interlocked.Add(ref _receivedBytes, count);
lock (_stateLock)
{
_lastReceiveUtc = DateTime.UtcNow;
}
}
catch (TimeoutException)
{
// 读取超时用于让线程定期检查退出状态,不属于串口错误。
}
catch (Exception ex)
{
if (!_stopping)
{
SetFatalError("串口读取失败:" + ex.Message);
}
}
}
}
private void WriteReceivedText(string text)
{
lock (_logLock)
{
_pendingLine += text;
while (true)
{
int newlineIndex = _pendingLine.IndexOf('\n');
if (newlineIndex < 0) break;
string line = _pendingLine.Substring(0, newlineIndex).TrimEnd('\r');
_pendingLine = _pendingLine.Substring(newlineIndex + 1);
_textWriter.WriteLine("[" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture) + "] " + line);
}
}
}
private void WriteCommand(string text, string ending)
{
string terminator;
switch ((ending ?? string.Empty).ToUpperInvariant())
{
case "NONE": terminator = string.Empty; break;
case "CR": terminator = "\r"; break;
case "LF": terminator = "\n"; break;
case "CRLF": terminator = "\r\n"; break;
default: throw new ArgumentException("换行方式必须是 None、CR、LF 或 CRLF。", "ending");
}
string payload = (text ?? string.Empty) + terminator;
WriteBytes(_encoding.GetBytes(payload));
WriteTransmitLog(payload, "命令");
}
private void WriteTransmitLog(string text, string source)
{
// 将不可见的行尾转成明确标记,空命令回车会显示为 <CR>。
StringBuilder visible = new StringBuilder();
foreach (char c in text ?? string.Empty)
{
switch (c)
{
case '\r': visible.Append("<CR>"); break;
case '\n': visible.Append("<LF>"); break;
case '\t': visible.Append("<TAB>"); break;
default:
if (c < 32) visible.Append("<0x" + ((int)c).ToString("X2", CultureInfo.InvariantCulture) + ">");
else visible.Append(c);
break;
}
}
if (visible.Length == 0) visible.Append("<EMPTY>");
string line = "[" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture) + "] [TX:" + source + "] " + visible;
Console.WriteLine(line);
lock (_logLock)
{
if (_textWriter != null) _textWriter.WriteLine(line);
}
}
private void WriteBytes(byte[] bytes)
{
if (bytes == null || bytes.Length == 0) return;
lock (_writeLock)
{
if (_stopping || !_serialPort.IsOpen) throw new IOException("串口当前未打开。");
_serialPort.BaseStream.Write(bytes, 0, bytes.Length);
_serialPort.BaseStream.Flush();
}
}
private void TouchWebClient(string clientId)
{
if (string.IsNullOrWhiteSpace(clientId) || clientId.Length > 128) return;
lock (_stateLock)
{
_webClients[clientId] = DateTime.UtcNow;
_hadWebClient = true;
_noClientSinceUtc = DateTime.MinValue;
}
}
private void RemoveWebClient(string clientId)
{
if (string.IsNullOrWhiteSpace(clientId)) return;
lock (_stateLock)
{
_webClients.Remove(clientId);
if (_hadWebClient && _webClients.Count == 0 && _noClientSinceUtc == DateTime.MinValue)
_noClientSinceUtc = DateTime.UtcNow;
}
}
private void WebClientMonitorLoop()
{
while (!_stopping)
{
bool shouldStop = false;
lock (_stateLock)
{
DateTime now = DateTime.UtcNow;
List<string> expiredClients = new List<string>();
foreach (KeyValuePair<string, DateTime> item in _webClients)
{
if ((now - item.Value).TotalSeconds >= WebClientTimeoutSeconds)
expiredClients.Add(item.Key);
}
foreach (string clientId in expiredClients) _webClients.Remove(clientId);
if (_webClients.Count > 0)
{
_noClientSinceUtc = DateTime.MinValue;
}
else if (_hadWebClient)
{
if (_noClientSinceUtc == DateTime.MinValue) _noClientSinceUtc = now;
shouldStop = (now - _noClientSinceUtc).TotalMilliseconds >= LastClientGraceMilliseconds;
}
}
if (shouldStop)
{
WriteSystemLog("最后一个串口网页已关闭,正在停止日志采集并释放串口。");
_stopping = true;
try { if (_listener != null) _listener.Stop(); } catch { }
break;
}
Thread.Sleep(250);
}
}
private void HttpLoop()
{
while (!_stopping)
{
try
{
HttpListenerContext context = _listener.GetContext();
ThreadPool.QueueUserWorkItem(delegate { HandleHttpRequest(context); });
}
catch (HttpListenerException)
{
if (!_stopping) SetFatalError("本地网页服务异常停止。");
}
catch (ObjectDisposedException)
{
// 服务关闭时 GetContext 会退出。
}
catch (Exception ex)
{
if (!_stopping) SetFatalError("本地网页服务失败:" + ex.Message);
}
}
}
private void HandleHttpRequest(HttpListenerContext context)
{
try
{
if (context.Request.RemoteEndPoint == null || !IPAddress.IsLoopback(context.Request.RemoteEndPoint.Address))
{
SendTextResponse(context, 403, "text/plain; charset=utf-8", "只允许本机访问。");
return;
}
string path = context.Request.Url == null ? "/" : context.Request.Url.AbsolutePath;
string method = context.Request.HttpMethod.ToUpperInvariant();
if (method == "GET" && (path == "/" || path == "/index.html"))
{
string html = _htmlTemplate
.Replace("__SESSION_TOKEN__", _sessionToken)
.Replace("__SERIAL_PORT__", WebUtility.HtmlEncode(_serialPort.PortName));
SendTextResponse(context, 200, "text/html; charset=utf-8", html);
return;
}
if (method == "GET" && path == "/api/status")
{
TouchWebClient(context.Request.QueryString["client"]);
SendTextResponse(context, 200, "application/json; charset=utf-8", BuildStatusJson());
return;
}
if (method == "GET" && path == "/api/log")
{
TouchWebClient(context.Request.QueryString["client"]);
SendLogTail(context, 2 * 1024 * 1024);
return;
}
if (method == "GET" && path == "/favicon.ico")
{
context.Response.StatusCode = 204;
context.Response.Close();
return;
}
if (method == "POST")
{
if (path == "/api/send")
{
ValidateCommandRequest(context);
string ending = context.Request.QueryString["ending"] ?? "CR";
string command = ReadRequestBody(context.Request, 16384);
WriteCommand(command, ending);
SendTextResponse(context, 200, "application/json; charset=utf-8", "{\"status\":\"written\"}");
return;
}
if (path == "/api/client/disconnect")
{
ValidateDisconnectRequest(context);
RemoveWebClient(context.Request.QueryString["client"]);
SendTextResponse(context, 200, "application/json; charset=utf-8", "{\"status\":\"disconnected\"}");
return;
}
}
SendTextResponse(context, 404, "text/plain; charset=utf-8", "未找到接口。");
}
catch (ArgumentException ex)
{
SendTextResponse(context, 400, "application/json; charset=utf-8", "{\"error\":\"" + JsonEscape(ex.Message) + "\"}");
}
catch (UnauthorizedAccessException ex)
{
SendTextResponse(context, 403, "application/json; charset=utf-8", "{\"error\":\"" + JsonEscape(ex.Message) + "\"}");
}
catch (Exception ex)
{
lock (_stateLock) _lastError = ex.Message;
SendTextResponse(context, 503, "application/json; charset=utf-8", "{\"error\":\"" + JsonEscape(ex.Message) + "\"}");
}
}
private void ValidateCommandRequest(HttpListenerContext context)
{
string token = context.Request.Headers["X-Serial-Token"];
if (!string.Equals(token, _sessionToken, StringComparison.Ordinal))
throw new UnauthorizedAccessException("本地会话令牌无效。");
string origin = context.Request.Headers["Origin"];
if (!string.IsNullOrWhiteSpace(origin) &&
!string.Equals(origin.TrimEnd('/') + "/", _listenPrefix, StringComparison.OrdinalIgnoreCase))
throw new UnauthorizedAccessException("请求来源不是当前串口控制台。");
}
private void ValidateDisconnectRequest(HttpListenerContext context)
{
// sendBeacon 无法自定义请求头,因此仅离页通知接口允许在查询参数中携带会话令牌。
string token = context.Request.Headers["X-Serial-Token"];
if (string.IsNullOrWhiteSpace(token)) token = context.Request.QueryString["token"];
if (!string.Equals(token, _sessionToken, StringComparison.Ordinal))
throw new UnauthorizedAccessException("本地会话令牌无效。");
string origin = context.Request.Headers["Origin"];
if (!string.IsNullOrWhiteSpace(origin) &&
!string.Equals(origin.TrimEnd('/') + "/", _listenPrefix, StringComparison.OrdinalIgnoreCase))
throw new UnauthorizedAccessException("请求来源不是当前串口控制台。");
}
private static string ReadRequestBody(HttpListenerRequest request, int maxCharacters)
{
Encoding bodyEncoding = request.ContentEncoding ?? Encoding.UTF8;
using (StreamReader reader = new StreamReader(request.InputStream, bodyEncoding, true, 1024, false))
{
char[] buffer = new char[maxCharacters + 1];
int total = 0;
while (total < buffer.Length)
{
int count = reader.Read(buffer, total, buffer.Length - total);
if (count <= 0) break;
total += count;
}
if (total > maxCharacters) throw new ArgumentException("单次命令不能超过 16384 个字符。");
return new string(buffer, 0, total);
}
}
private void SendLogTail(HttpListenerContext context, int maxBytes)
{
byte[] data;
using (FileStream stream = new FileStream(_logPath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
{
long start = Math.Max(0, stream.Length - maxBytes);
stream.Position = start;
data = new byte[stream.Length - start];
int offset = 0;
while (offset < data.Length)
{
int count = stream.Read(data, offset, data.Length - offset);
if (count <= 0) break;
offset += count;
}
if (offset != data.Length) Array.Resize(ref data, offset);
}
context.Response.StatusCode = 200;
context.Response.ContentType = "text/plain; charset=utf-8";
context.Response.Headers["Cache-Control"] = "no-store";
context.Response.ContentLength64 = data.Length;
context.Response.OutputStream.Write(data, 0, data.Length);
context.Response.Close();
}
private string BuildStatusJson()
{
string lastError;
DateTime lastReceive;
int activeWebClients;
lock (_stateLock)
{
lastError = _lastError;
lastReceive = _lastReceiveUtc;
activeWebClients = _webClients.Count;
}
return "{" +
"\"port\":\"" + JsonEscape(_serialPort.PortName) + "\"," +
"\"isOpen\":" + (_serialPort.IsOpen && !_stopping ? "true" : "false") + "," +
"\"receivedBytes\":" + Interlocked.Read(ref _receivedBytes).ToString(CultureInfo.InvariantCulture) + "," +
"\"lastReceiveUtc\":\"" + (lastReceive == DateTime.MinValue ? string.Empty : lastReceive.ToString("o", CultureInfo.InvariantCulture)) + "\"," +
"\"activeWebClients\":" + activeWebClients.ToString(CultureInfo.InvariantCulture) + "," +
"\"lastError\":\"" + JsonEscape(lastError) + "\"" +
"}";
}
private static string JsonEscape(string value)
{
if (string.IsNullOrEmpty(value)) return string.Empty;
StringBuilder builder = new StringBuilder(value.Length + 16);
foreach (char c in value)
{
switch (c)
{
case '\\': builder.Append("\\\\"); break;
case '"': builder.Append("\\\""); break;
case '\r': builder.Append("\\r"); break;
case '\n': builder.Append("\\n"); break;
case '\t': builder.Append("\\t"); break;
default:
if (c < 32) builder.Append("\\u" + ((int)c).ToString("x4", CultureInfo.InvariantCulture));
else builder.Append(c);
break;
}
}
return builder.ToString();
}
private static void SendTextResponse(HttpListenerContext context, int statusCode, string contentType, string text)
{
try
{
byte[] data = new UTF8Encoding(false).GetBytes(text ?? string.Empty);
context.Response.StatusCode = statusCode;
context.Response.ContentType = contentType;
context.Response.Headers["Cache-Control"] = "no-store";
context.Response.ContentLength64 = data.Length;
context.Response.OutputStream.Write(data, 0, data.Length);
context.Response.Close();
}
catch
{
try { context.Response.Abort(); } catch { }
}
}
private void SetFatalError(string message)
{
lock (_stateLock) _lastError = message;
WriteSystemLog(message);
_stopping = true;
try { if (_listener != null) _listener.Stop(); } catch { }
}
private void WriteSystemLog(string message)
{
string line = "[" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture) + "] " + message;
Console.WriteLine(line);
lock (_logLock)
{
if (_textWriter != null) _textWriter.WriteLine(line);
}
}
public void Stop()
{
_stopping = true;
try { if (_listener != null) _listener.Stop(); } catch { }
try { if (_serialPort.IsOpen) _serialPort.Close(); } catch { }
JoinThread(_httpThread);
JoinThread(_clientMonitorThread);
JoinThread(_receiveThread);
lock (_logLock)
{
if (_textWriter != null && !string.IsNullOrEmpty(_pendingLine))
{
_textWriter.WriteLine("[" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture) + "] " + _pendingLine.TrimEnd('\r'));
_pendingLine = string.Empty;
}
if (_textWriter != null)
{
_textWriter.WriteLine("[" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture) + "] 串口网页服务结束");
_textWriter.Dispose();
_textWriter = null;
}
if (_rawStream != null)
{
_rawStream.Dispose();
_rawStream = null;
}
}
if (_listener != null)
{
try { _listener.Close(); } catch { }
_listener = null;
}
_serialPort.Dispose();
_started = false;
}
private static void JoinThread(Thread thread)
{
if (thread == null || thread == Thread.CurrentThread) return;
try { thread.Join(1500); } catch { }
}
public void Dispose()
{
Stop();
}
}
}
+478
View File
@@ -0,0 +1,478 @@
#!/usr/bin/env python3
"""设备 V2 基础信息预检与 EPFP/4001 固件升级工具。"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import select
import socket
import stat
import struct
import sys
import time
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
INFO_PORT = 8942
UPGRADE_PORT = 8943
INFO_COMMAND = 30101
UPGRADE_COMMAND = 4001
MAX_JSON_SIZE = 1024 * 1024
MAX_STATUS_SIZE = 1024 * 1024
MAX_ROM_SIZE = 0xFFFFFFFF
STATUS_SIZE = 21
def configure_redirected_output() -> None:
"""输出被管道捕获时统一使用 UTF-8,避免中文日志乱码。"""
for stream in (sys.stdout, sys.stderr):
if not stream.isatty() and hasattr(stream, "reconfigure"):
stream.reconfigure(encoding="utf-8", errors="backslashreplace")
configure_redirected_output()
class UpgradeError(RuntimeError):
"""表示预检、协议或升级过程失败。"""
class EventLogger:
"""同时向终端和 JSONL 文件写入带时间戳的事件。"""
def __init__(self, log_dir: Path) -> None:
log_dir.mkdir(parents=True, exist_ok=True)
stamp = time.strftime("%Y%m%d-%H%M%S")
self.path = log_dir / ("device-upgrade-" + stamp + ".jsonl")
self._stream = self.path.open("a", encoding="utf-8", buffering=1)
def close(self) -> None:
self._stream.close()
def emit(self, event: str, **fields: Any) -> None:
record: Dict[str, Any] = {
"time": time.strftime("%Y-%m-%d %H:%M:%S"),
"event": event,
}
record.update(fields)
line = json.dumps(record, ensure_ascii=False, separators=(",", ":"))
print(line, flush=True)
self._stream.write(line + "\n")
def recv_exact(sock: socket.socket, size: int) -> bytes:
"""读取指定字节数;对端提前关闭时立即失败。"""
data = bytearray()
while len(data) < size:
chunk = sock.recv(size - len(data))
if not chunk:
raise UpgradeError("设备在响应完整之前关闭了连接")
data.extend(chunk)
return bytes(data)
def query_device_info(host: str, timeout: float, logger: EventLogger) -> Dict[str, Any]:
"""通过 EPTP/30101 查询并严格校验设备基础信息。"""
session = int(time.time() * 1000) & 0xFFFFFFFF
request_object = {"cmd": INFO_COMMAND, "session": session}
request = json.dumps(request_object, separators=(",", ":")).encode("utf-8")
logger.emit("info_request", host=host, port=INFO_PORT, request=request_object)
with socket.create_connection((host, INFO_PORT), timeout=timeout) as sock:
sock.settimeout(timeout)
sock.sendall(b"EPTP" + struct.pack("!I", len(request)) + request)
header = recv_exact(sock, 8)
if header[:4] != b"EPTP":
raise UpgradeError("基础信息响应同步头不是 EPTP:%r" % (header[:4],))
payload_size = struct.unpack("!I", header[4:])[0]
if payload_size == 0 or payload_size > MAX_JSON_SIZE:
raise UpgradeError("基础信息响应长度不合理:%d" % payload_size)
payload = recv_exact(sock, payload_size)
try:
response = json.loads(payload.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise UpgradeError("基础信息响应不是有效 UTF-8 JSON:%s" % exc) from exc
if not isinstance(response, dict):
raise UpgradeError("基础信息响应 JSON 顶层不是对象")
if response.get("cmd") != INFO_COMMAND:
raise UpgradeError("基础信息响应 cmd 不匹配:%r" % response.get("cmd"))
if response.get("session") != session:
raise UpgradeError(
"基础信息响应 session 不匹配:期望 %d,实际 %r"
% (session, response.get("session"))
)
if response.get("code") != 0:
raise UpgradeError("基础信息响应 code 非 0:%r" % response.get("code"))
logger.emit("device_info", response=response)
return response
def response_data(response: Dict[str, Any]) -> Dict[str, Any]:
"""取得响应 data 对象,格式错误时拒绝继续。"""
data = response.get("data")
if not isinstance(data, dict):
raise UpgradeError("基础信息响应缺少 data 对象")
return data
def collect_models(data: Dict[str, Any]) -> List[str]:
"""收集 inner、outer 和 hardware 中可用的型号。"""
models: List[str] = []
for section_name in ("inner", "outer", "hardware"):
section = data.get(section_name)
if isinstance(section, dict):
value = section.get("type")
if isinstance(value, str) and value and value not in models:
models.append(value)
return models
def collect_versions(data: Dict[str, Any]) -> Dict[str, Any]:
"""收集各固件分区和硬件版本,便于升级前后比对。"""
versions: Dict[str, Any] = {}
for section_name in ("inner", "outer", "hardware"):
section = data.get(section_name)
if isinstance(section, dict):
versions[section_name] = section.get("version")
return versions
def validate_identity(
response: Dict[str, Any], expected_sn: str, expected_model: Optional[str]
) -> Tuple[Dict[str, Any], List[str]]:
"""对完整 SN 做强校验,并按需校验型号。"""
data = response_data(response)
actual_sn = data.get("sn")
if actual_sn != expected_sn:
raise UpgradeError("设备 SN 不匹配:期望 %s,实际 %r" % (expected_sn, actual_sn))
models = collect_models(data)
if expected_model and expected_model not in models:
raise UpgradeError(
"设备型号不匹配:期望 %s,设备报告 %s"
% (expected_model, models if models else "无可用型号")
)
return data, models
def infer_rom_version(path: Path) -> Optional[str]:
"""仅从路径中识别明确的四段数字版本号,不猜测其他命名。"""
candidates: List[str] = []
pattern = re.compile(r"(?<!\d)(\d+\.\d+\.\d+\.\d+)(?!\d)")
for part in path.parts:
candidates.extend(pattern.findall(part))
return candidates[-1] if candidates else None
def validate_rom(path: Path) -> Dict[str, Any]:
"""确认 ROM 为普通非空文件,并计算 SHA-256。"""
absolute = path.expanduser().resolve(strict=True)
metadata = os.stat(str(absolute))
if not stat.S_ISREG(metadata.st_mode):
raise UpgradeError("ROM 不是普通文件:%s" % absolute)
if metadata.st_size <= 0:
raise UpgradeError("ROM 文件为空:%s" % absolute)
if metadata.st_size > MAX_ROM_SIZE:
raise UpgradeError("ROM 大于协议 uint32 长度上限:%d" % metadata.st_size)
digest = hashlib.sha256()
with absolute.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return {
"path": str(absolute),
"size": metadata.st_size,
"modified": time.strftime(
"%Y-%m-%d %H:%M:%S", time.localtime(metadata.st_mtime)
),
"sha256": digest.hexdigest(),
"target_version_from_path": infer_rom_version(absolute),
}
def parse_status(payload: bytes) -> Dict[str, int]:
"""解析 21 字节升级状态业务数据。"""
if len(payload) < STATUS_SIZE:
raise UpgradeError("升级状态不足 21 字节:%d" % len(payload))
status, session, download_total, downloaded, flash_total, flashed = struct.unpack(
"!BIIIII", payload[:STATUS_SIZE]
)
if status not in (0, 1, 2, 3):
raise UpgradeError("未知升级状态值:%d" % status)
return {
"status": status,
"session": session,
"download_total": download_total,
"downloaded": downloaded,
"flash_total": flash_total,
"flashed": flashed,
}
def extract_status_payloads(buffer: bytearray) -> Iterable[Tuple[str, bytes]]:
"""从 TCP 字节流中提取完整 EPFP 帧或兼容设备实测的裸状态。"""
while buffer:
# 同步头被拆包时等待后续数据,避免把半个 EPFP 当裸状态。
if len(buffer) < 4 and b"EPFP".startswith(bytes(buffer)):
return
if buffer[:4] == b"EPFP":
if len(buffer) < 12:
return
command, payload_size = struct.unpack("!II", buffer[4:12])
if command != UPGRADE_COMMAND:
raise UpgradeError("升级状态帧 cmd 不匹配:%d" % command)
if payload_size < STATUS_SIZE or payload_size > MAX_STATUS_SIZE:
raise UpgradeError("升级状态帧长度不合理:%d" % payload_size)
if len(buffer) < 12 + payload_size:
return
payload = bytes(buffer[12 : 12 + payload_size])
del buffer[: 12 + payload_size]
yield "framed", payload
continue
if buffer[0] in (0, 1, 2, 3):
# 合法裸状态内部也可能包含字节串 EPFP,不能在其业务数据中误重同步。
if len(buffer) < STATUS_SIZE:
return
payload = bytes(buffer[:STATUS_SIZE])
del buffer[:STATUS_SIZE]
yield "raw", payload
continue
sync_position = buffer.find(b"EPFP")
if sync_position > 0:
# 首字节不是合法 status,且后方存在同步头时,丢弃无效前缀。
del buffer[:sync_position]
continue
if len(buffer) < STATUS_SIZE:
return
# 保留这 21 字节交给状态解析器报告明确的非法 status。
payload = bytes(buffer[:STATUS_SIZE])
del buffer[:STATUS_SIZE]
yield "raw", payload
def transfer_rom(
host: str,
rom_path: Path,
rom_size: int,
connect_timeout: float,
stall_timeout: float,
total_timeout: float,
chunk_size: int,
logger: EventLogger,
) -> Dict[str, Any]:
"""发送 ROM,并在同一循环中处理回包、停滞和终态。"""
result: Dict[str, Any] = {"status": None, "bytes_sent": 0}
receive_buffer = bytearray()
pending = bytearray(b"EPFP" + struct.pack("!II", UPGRADE_COMMAND, rom_size))
started = time.monotonic()
last_progress = started
previous_progress: Optional[Tuple[int, int]] = None
upload_complete = False
logger.emit(
"upgrade_connect", host=host, port=UPGRADE_PORT, rom_size=rom_size
)
with socket.create_connection((host, UPGRADE_PORT), timeout=connect_timeout) as sock:
sock.setblocking(False)
with rom_path.open("rb") as rom:
while True:
now = time.monotonic()
if now - started >= total_timeout:
raise UpgradeError("升级超过总超时 %.1f 秒" % total_timeout)
if now - last_progress >= stall_timeout:
raise UpgradeError("设备下载或烧录进度连续 %.1f 秒无变化" % stall_timeout)
readable, writable, exceptional = select.select(
[sock], [] if upload_complete else [sock], [sock], 0.5
)
if exceptional:
raise UpgradeError("升级 TCP 连接出现异常状态")
if readable:
chunk = sock.recv(4096)
if not chunk:
raise UpgradeError("设备在 status=2 之前关闭升级连接")
logger.emit("upgrade_rx", size=len(chunk), hex=chunk.hex())
receive_buffer.extend(chunk)
for frame_type, payload in extract_status_payloads(receive_buffer):
status = parse_status(payload)
logger.emit("device_status", frame=frame_type, **status)
progress = (status["downloaded"], status["flashed"])
if previous_progress is None or progress != previous_progress:
previous_progress = progress
last_progress = time.monotonic()
result.update(status)
if status["status"] == 3:
raise UpgradeError("设备返回升级失败状态 status=3")
if status["status"] == 2:
incomplete_counters = {}
if (
status["download_total"] > 0
and status["downloaded"] < status["download_total"]
):
incomplete_counters["download"] = {
"total": status["download_total"],
"current": status["downloaded"],
}
if (
status["flash_total"] > 0
and status["flashed"] < status["flash_total"]
):
incomplete_counters["flash"] = {
"total": status["flash_total"],
"current": status["flashed"],
}
if incomplete_counters:
logger.emit(
"protocol_warning",
message="status=2,但设备进度计数尚未达到其自身总量",
counters=incomplete_counters,
)
logger.emit("protocol_success", status=2)
return result
if writable:
if not pending:
next_chunk = rom.read(chunk_size)
if next_chunk:
pending.extend(next_chunk)
else:
upload_complete = True
logger.emit(
"client_upload_complete",
bytes_sent=result["bytes_sent"],
rom_size=rom_size,
)
if pending:
sent = sock.send(pending)
if sent <= 0:
raise UpgradeError("升级 TCP 连接无法继续发送")
# 前 12 字节是 EPFP 头,不计入 ROM 上传进度。
previous_wire_bytes = result.get("wire_bytes_sent", 0)
wire_bytes = previous_wire_bytes + sent
result["wire_bytes_sent"] = wire_bytes
result["bytes_sent"] = max(0, min(rom_size, wire_bytes - 12))
del pending[:sent]
def build_parser() -> argparse.ArgumentParser:
"""创建命令行参数解析器。"""
parser = argparse.ArgumentParser(
description="V2 设备 30101 身份预检与 EPFP/4001 ROM 升级;默认仅预检。"
)
parser.add_argument("--host", required=True, help="目标设备 IPv4/主机名")
parser.add_argument("--expected-sn", required=True, help="必须精确匹配的完整设备 SN")
parser.add_argument("--expected-model", help="可选,必须出现在设备型号字段中的型号")
parser.add_argument("--rom", required=True, type=Path, help="ROM 文件绝对或相对路径")
parser.add_argument("--log-dir", required=True, type=Path, help="本次任务日志目录")
parser.add_argument(
"--execute",
action="store_true",
help="实际传输 ROM;未指定时只查询身份、校验 ROM 和输出预检信息",
)
parser.add_argument("--connect-timeout", type=float, default=10.0)
parser.add_argument("--stall-timeout", type=float, default=60.0)
parser.add_argument("--total-timeout", type=float, default=600.0)
parser.add_argument("--chunk-size", type=int, default=256 * 1024)
return parser
def validate_numeric_options(args: argparse.Namespace) -> None:
"""拒绝会让超时或分块保护失效的参数。"""
if args.connect_timeout <= 0:
raise UpgradeError("--connect-timeout 必须大于 0")
if args.stall_timeout <= 0:
raise UpgradeError("--stall-timeout 必须大于 0")
if args.total_timeout <= 0:
raise UpgradeError("--total-timeout 必须大于 0")
if args.chunk_size <= 0 or args.chunk_size > 16 * 1024 * 1024:
raise UpgradeError("--chunk-size 必须在 1 到 16777216 之间")
def main(argv: Optional[List[str]] = None) -> int:
"""执行预检;仅在明确给出 --execute 时进入升级传输。"""
args = build_parser().parse_args(argv)
logger: Optional[EventLogger] = None
try:
validate_numeric_options(args)
logger = EventLogger(args.log_dir.expanduser().resolve())
logger.emit("mode", execute=bool(args.execute), log_file=str(logger.path))
rom = validate_rom(args.rom)
logger.emit("rom", **rom)
response = query_device_info(args.host, args.connect_timeout, logger)
data, models = validate_identity(response, args.expected_sn, args.expected_model)
summary = {
"host": args.host,
"sn": data.get("sn"),
"mac": data.get("mac"),
"models": models,
"versions": collect_versions(data),
"expected_model": args.expected_model,
"rom": rom,
"rom_model_verification": (
"由调用方提供并已核对设备型号"
if args.expected_model
else "无法从协议或 ROM 自动验证目标型号"
),
}
logger.emit("preflight", **summary)
if not args.execute:
logger.emit("preflight_only", message="未指定 --execute,未连接 8943,未发送 ROM")
return 0
result = transfer_rom(
host=args.host,
rom_path=Path(rom["path"]),
rom_size=int(rom["size"]),
connect_timeout=args.connect_timeout,
stall_timeout=args.stall_timeout,
total_timeout=args.total_timeout,
chunk_size=args.chunk_size,
logger=logger,
)
if result.get("status") != 2:
raise UpgradeError("升级没有取得 status=2 成功终态")
logger.emit(
"upgrade_result",
protocol_success=True,
postcheck_required=True,
result=result,
)
return 0
except (OSError, ValueError, UpgradeError) as exc:
if logger is not None:
logger.emit("error", type=type(exc).__name__, message=str(exc))
else:
print("错误:%s" % exc, file=sys.stderr, flush=True)
return 1
finally:
if logger is not None:
logger.close()
if __name__ == "__main__":
raise SystemExit(main())
+222
View File
@@ -0,0 +1,222 @@
<#
.SYNOPSIS
实时显示并保存设备串口数据。
.DESCRIPTION
使用 .NET SerialPort 直接独占打开串口。收到的数据会原样写入二进制文件,
同时解码后实时显示到当前 PowerShell,并按行写入带时间戳的文本日志。
默认只读取数据,不向设备发送指令。DtrEnable 和 RtsEnable 会改变硬件控制线,
只有用户明确要求时才能启用。
.EXAMPLE
.\serial_monitor.ps1 -Port COMx -LogPath F:\tmp\embedded-app-pc-debug\task\logs\serial.log
.EXAMPLE
.\serial_monitor.ps1 -Port COMx -LogPath .\serial.log -DurationSeconds 60
.PARAMETER DtrEnable
改变 DTR 硬件控制线;仅在用户明确要求时使用。
.PARAMETER RtsEnable
改变 RTS 硬件控制线;仅在用户明确要求时使用。
#>
[CmdletBinding()]
param(
[Parameter(Mandatory = $true)]
[ValidateNotNullOrEmpty()]
[string]$Port,
[Parameter(Mandatory = $true)]
[ValidateNotNullOrEmpty()]
[string]$LogPath,
[ValidateRange(1, 4000000)]
[int]$BaudRate = 115200,
[ValidateRange(5, 8)]
[int]$DataBits = 8,
[ValidateSet('None', 'Odd', 'Even', 'Mark', 'Space')]
[string]$Parity = 'None',
[ValidateSet('One', 'OnePointFive', 'Two')]
[string]$StopBits = 'One',
[ValidateSet('None', 'XOnXOff', 'RequestToSend', 'RequestToSendXOnXOff')]
[string]$Handshake = 'None',
[string]$EncodingName = 'utf-8',
[ValidateRange(0, 2147483647)]
[int]$DurationSeconds = 0,
[ValidateRange(1, 5000)]
[int]$PollMilliseconds = 100,
[ValidateRange(64, 1048576)]
[int]$BufferSize = 4096,
[string]$RawPath,
[switch]$DtrEnable,
[switch]$RtsEnable
)
$ErrorActionPreference = 'Stop'
$serialPort = $null
$textWriter = $null
$rawStream = $null
$pendingLine = ''
function Write-TimestampedLines {
param(
[Parameter(Mandatory = $true)]
[string]$Text,
[Parameter(Mandatory = $true)]
[System.IO.StreamWriter]$Writer
)
$script:pendingLine += $Text
while ($true) {
$newlineIndex = $script:pendingLine.IndexOf("`n", [System.StringComparison]::Ordinal)
if ($newlineIndex -lt 0) {
break
}
$line = $script:pendingLine.Substring(0, $newlineIndex).TrimEnd("`r")
$script:pendingLine = $script:pendingLine.Substring($newlineIndex + 1)
$Writer.WriteLine(('[{0}] {1}' -f (Get-Date -Format 'yyyy-MM-dd HH:mm:ss.fff'), $line))
}
}
try {
$resolvedLogPath = [System.IO.Path]::GetFullPath($LogPath)
if ([string]::IsNullOrWhiteSpace($RawPath)) {
$resolvedRawPath = $resolvedLogPath + '.raw.bin'
}
else {
$resolvedRawPath = [System.IO.Path]::GetFullPath($RawPath)
}
$logDirectory = [System.IO.Path]::GetDirectoryName($resolvedLogPath)
$rawDirectory = [System.IO.Path]::GetDirectoryName($resolvedRawPath)
if (-not [string]::IsNullOrWhiteSpace($logDirectory)) {
[System.IO.Directory]::CreateDirectory($logDirectory) | Out-Null
}
if (-not [string]::IsNullOrWhiteSpace($rawDirectory)) {
[System.IO.Directory]::CreateDirectory($rawDirectory) | Out-Null
}
$utf8NoBom = New-Object System.Text.UTF8Encoding($false)
$textFileStream = New-Object System.IO.FileStream(
$resolvedLogPath,
[System.IO.FileMode]::Append,
[System.IO.FileAccess]::Write,
[System.IO.FileShare]::Read
)
$textWriter = New-Object System.IO.StreamWriter($textFileStream, $utf8NoBom)
$textWriter.AutoFlush = $true
$rawStream = New-Object System.IO.FileStream(
$resolvedRawPath,
[System.IO.FileMode]::Append,
[System.IO.FileAccess]::Write,
[System.IO.FileShare]::Read
)
try {
$encoding = [System.Text.Encoding]::GetEncoding($EncodingName)
}
catch {
throw "无法识别编码 '$EncodingName':$($_.Exception.Message)"
}
$decoder = $encoding.GetDecoder()
$serialPort = New-Object System.IO.Ports.SerialPort
$serialPort.PortName = $Port
$serialPort.BaudRate = $BaudRate
$serialPort.DataBits = $DataBits
$serialPort.Parity = [System.Enum]::Parse([System.IO.Ports.Parity], $Parity)
$serialPort.StopBits = [System.Enum]::Parse([System.IO.Ports.StopBits], $StopBits)
$serialPort.Handshake = [System.Enum]::Parse([System.IO.Ports.Handshake], $Handshake)
$serialPort.DtrEnable = $DtrEnable.IsPresent
$serialPort.RtsEnable = $RtsEnable.IsPresent
$serialPort.ReadTimeout = $PollMilliseconds
$serialPort.ReadBufferSize = [Math]::Max(4096, $BufferSize)
try {
$serialPort.Open()
}
catch {
throw "无法打开串口 $Port。串口可能不存在或正被其他程序占用:$($_.Exception.Message)"
}
$startTime = Get-Date
$startMessage = '[{0}] 已打开 {1},{2} {3}{4}{5},流控 {6};原始数据:{7}' -f `
(Get-Date -Format 'yyyy-MM-dd HH:mm:ss.fff'),
$Port,
$BaudRate,
$DataBits,
$Parity.Substring(0, 1),
$(if ($StopBits -eq 'One') { '1' } elseif ($StopBits -eq 'Two') { '2' } else { '1.5' }),
$Handshake,
$resolvedRawPath
[Console]::WriteLine($startMessage)
$textWriter.WriteLine($startMessage)
$byteBuffer = New-Object byte[] $BufferSize
$charBuffer = New-Object char[] ($encoding.GetMaxCharCount($BufferSize))
while ($true) {
if ($DurationSeconds -gt 0 -and ((Get-Date) - $startTime).TotalSeconds -ge $DurationSeconds) {
break
}
try {
$count = $serialPort.Read($byteBuffer, 0, $byteBuffer.Length)
}
catch [System.TimeoutException] {
continue
}
if ($count -le 0) {
continue
}
$rawStream.Write($byteBuffer, 0, $count)
$rawStream.Flush()
$charCount = $decoder.GetChars($byteBuffer, 0, $count, $charBuffer, 0, $false)
if ($charCount -gt 0) {
$text = New-Object string($charBuffer, 0, $charCount)
[Console]::Write($text)
Write-TimestampedLines -Text $text -Writer $textWriter
}
}
}
catch {
# 使用字符串连接保留底层异常文本,避免异常内容再次参与格式化。
$errorMessage = '[' + (Get-Date -Format 'yyyy-MM-dd HH:mm:ss.fff') + '] 串口监测失败:' + [string]$_.Exception.Message
[Console]::Error.WriteLine($errorMessage)
if ($null -ne $textWriter) {
$textWriter.WriteLine($errorMessage)
}
exit 1
}
finally {
if ($null -ne $textWriter -and -not [string]::IsNullOrEmpty($script:pendingLine)) {
$textWriter.WriteLine(('[{0}] {1}' -f (Get-Date -Format 'yyyy-MM-dd HH:mm:ss.fff'), $script:pendingLine.TrimEnd("`r")))
}
if ($null -ne $serialPort) {
if ($serialPort.IsOpen) {
$serialPort.Close()
}
$serialPort.Dispose()
}
if ($null -ne $rawStream) {
$rawStream.Dispose()
}
if ($null -ne $textWriter) {
$textWriter.WriteLine('[{0}] 串口监测结束' -f (Get-Date -Format 'yyyy-MM-dd HH:mm:ss.fff'))
$textWriter.Dispose()
}
}
+146
View File
@@ -0,0 +1,146 @@
<#
.SYNOPSIS
启动支持网页命令输入的串口实时控制台。
.DESCRIPTION
使用单一进程独占串口,同时提供串口读取、原始日志、时间戳文本日志和本机网页命令接口。
网页仅监听 127.0.0.1;普通命令支持 None、CR、LF、CRLF 四种结尾。
.EXAMPLE
.\serial_web_console.ps1 -Port COM56 -LogPath F:\tmp\serial.log
.EXAMPLE
.\serial_web_console.ps1 -Port COM56 -LogPath .\serial.log -WebPort 8766
.PARAMETER ValidateOnly
只编译并验证程序和页面资源,不打开串口,也不监听网页端口。
#>
[CmdletBinding()]
param(
[Parameter(Mandatory = $true)]
[ValidateNotNullOrEmpty()]
[string]$Port,
[Parameter(Mandatory = $true)]
[ValidateNotNullOrEmpty()]
[string]$LogPath,
[ValidateRange(1, 4000000)]
[int]$BaudRate = 115200,
[ValidateRange(5, 8)]
[int]$DataBits = 8,
[ValidateSet('None', 'Odd', 'Even', 'Mark', 'Space')]
[string]$Parity = 'None',
[ValidateSet('One', 'OnePointFive', 'Two')]
[string]$StopBits = 'One',
[ValidateSet('None', 'XOnXOff', 'RequestToSend', 'RequestToSendXOnXOff')]
[string]$Handshake = 'None',
[string]$EncodingName = 'utf-8',
[ValidateRange(1, 5000)]
[int]$PollMilliseconds = 100,
[ValidateRange(64, 1048576)]
[int]$BufferSize = 4096,
[string]$RawPath,
[ValidateRange(1024, 65535)]
[int]$WebPort = 8766,
[string]$HtmlPath,
[ValidateRange(0, 2147483647)]
[int]$DurationSeconds = 0,
[switch]$DtrEnable,
[switch]$RtsEnable,
[switch]$ValidateOnly
)
$ErrorActionPreference = 'Stop'
$broker = $null
try {
if ($PSVersionTable.PSEdition -ne 'Desktop') {
throw '该脚本需要由 Windows PowerShell 5.1 运行。请使用 powershell.exe,而不是 pwsh.exe。'
}
$sourcePath = Join-Path $PSScriptRoot 'SerialWebConsoleHost.cs'
if (-not (Test-Path -LiteralPath $sourcePath -PathType Leaf)) {
throw "找不到串口服务源文件:$sourcePath"
}
if ([string]::IsNullOrWhiteSpace($HtmlPath)) {
$HtmlPath = Join-Path (Split-Path $PSScriptRoot -Parent) 'assets\serial-console.html'
}
$resolvedHtmlPath = [System.IO.Path]::GetFullPath($HtmlPath)
if (-not (Test-Path -LiteralPath $resolvedHtmlPath -PathType Leaf)) {
throw "找不到串口控制台页面:$resolvedHtmlPath"
}
# Windows PowerShell 5.1 自带完整 .NET Framework,可直接编译 SerialPort 和 HttpListener。
Add-Type -Path $sourcePath -ErrorAction Stop
if ($ValidateOnly.IsPresent) {
Write-Output "验证通过:串口服务代码已编译,网页资源存在。"
Write-Output "源文件:$sourcePath"
Write-Output "页面:$resolvedHtmlPath"
return
}
$resolvedLogPath = [System.IO.Path]::GetFullPath($LogPath)
if ([string]::IsNullOrWhiteSpace($RawPath)) {
$resolvedRawPath = $resolvedLogPath + '.raw.bin'
}
else {
$resolvedRawPath = [System.IO.Path]::GetFullPath($RawPath)
}
$broker = [EmbeddedAppPcDebug.SerialWebConsoleHost]::new(
$Port,
$BaudRate,
$DataBits,
$Parity,
$StopBits,
$Handshake,
$EncodingName,
$PollMilliseconds,
$BufferSize,
$DtrEnable.IsPresent,
$RtsEnable.IsPresent,
$resolvedLogPath,
$resolvedRawPath,
$WebPort,
$resolvedHtmlPath
)
$broker.Start()
$startTime = Get-Date
Write-Output "串口网页控制台已启动:$($broker.Url)"
Write-Output "按 Ctrl+C 可停止串口采集和网页服务。"
while (-not $broker.IsStopping) {
if ($DurationSeconds -gt 0 -and ((Get-Date) - $startTime).TotalSeconds -ge $DurationSeconds) {
break
}
Start-Sleep -Milliseconds 200
}
}
catch {
$message = '[' + (Get-Date -Format 'yyyy-MM-dd HH:mm:ss.fff') + '] 串口网页控制台失败:' + [string]$_.Exception.Message
[Console]::Error.WriteLine($message)
exit 1
}
finally {
if ($null -ne $broker) {
$broker.Dispose()
}
}
+539
View File
@@ -0,0 +1,539 @@
#!/usr/bin/env python3
"""通过多个 Windows IPv4 接口执行 V2 EPBP 设备发现。"""
from __future__ import annotations
import argparse
import errno
import ipaddress
import json
import secrets
import select
import socket
import struct
import subprocess
import sys
import time
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
DEVICE_PORT = 8941
CLIENT_PORT = 8951
DISCOVERY_COMMAND = 60000
MAX_FRAME_SIZE = 1024 * 1024
class DiscoveryError(RuntimeError):
"""表示参数、网卡枚举或协议校验失败。"""
@dataclass(frozen=True)
class Interface:
"""保存一个 IPv4 接口及其筛选信息。"""
alias: str
description: str
address: ipaddress.IPv4Address
prefix_length: int
hardware: bool
@property
def network(self) -> ipaddress.IPv4Network:
"""返回接口对应的 IPv4 网络。"""
return ipaddress.ip_network(
"%s/%d" % (self.address, self.prefix_length), strict=False
)
def configure_output() -> None:
"""输出被重定向时统一使用 UTF-8。"""
for stream in (sys.stdout, sys.stderr):
if not stream.isatty() and hasattr(stream, "reconfigure"):
stream.reconfigure(encoding="utf-8", errors="backslashreplace")
def emit(event: str, **fields: Any) -> None:
"""输出单行 JSON,便于终端查看和日志保存。"""
record: Dict[str, Any] = {
"time": time.strftime("%Y-%m-%d %H:%M:%S"),
"event": event,
}
record.update(fields)
print(json.dumps(record, ensure_ascii=False, separators=(",", ":")), flush=True)
def parse_bool(value: Any) -> bool:
"""兼容 PowerShell JSON 中的布尔值和字符串。"""
if isinstance(value, bool):
return value
return str(value).strip().lower() == "true"
def powershell_interfaces() -> List[Interface]:
"""通过 Windows PowerShell 枚举已启用且具有 IPv4 的接口。"""
command = r"""
[Console]::OutputEncoding = New-Object System.Text.UTF8Encoding($false)
$OutputEncoding = [Console]::OutputEncoding
$items = Get-NetIPConfiguration | Where-Object {
$_.NetAdapter.Status -eq 'Up' -and $null -ne $_.IPv4Address
} | ForEach-Object {
$cfg = $_
foreach ($ipv4 in @($cfg.IPv4Address)) {
[PSCustomObject]@{
Alias = [string]$cfg.InterfaceAlias
Description = [string]$cfg.InterfaceDescription
Address = [string]$ipv4.IPAddress
PrefixLength = [int]$ipv4.PrefixLength
Hardware = [bool]$cfg.NetAdapter.HardwareInterface
}
}
}
@($items) | ConvertTo-Json -Compress
"""
try:
result = subprocess.run(
["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", command],
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
timeout=20,
)
except (OSError, subprocess.TimeoutExpired) as exc:
raise DiscoveryError("无法调用 PowerShell 枚举网卡:%s" % exc) from exc
if result.returncode != 0:
raise DiscoveryError("PowerShell 网卡枚举失败:%s" % result.stderr.strip())
try:
decoded = json.loads(result.stdout or "[]")
except json.JSONDecodeError as exc:
raise DiscoveryError("PowerShell 网卡结果不是有效 JSON:%s" % exc) from exc
if isinstance(decoded, dict):
decoded = [decoded]
interfaces: List[Interface] = []
for item in decoded:
if not isinstance(item, dict):
continue
try:
address = ipaddress.IPv4Address(str(item.get("Address", "")))
prefix_length = int(item.get("PrefixLength"))
ipaddress.ip_network("%s/%d" % (address, prefix_length), strict=False)
except (ipaddress.AddressValueError, ValueError, TypeError):
continue
interfaces.append(
Interface(
alias=str(item.get("Alias") or "未知接口"),
description=str(item.get("Description") or ""),
address=address,
prefix_length=prefix_length,
hardware=parse_bool(item.get("Hardware")),
)
)
return interfaces
def explicit_interfaces(values: Sequence[str]) -> List[Interface]:
"""解析重复传入的 IP/前缀接口参数。"""
interfaces: List[Interface] = []
for index, value in enumerate(values, 1):
try:
parsed = ipaddress.ip_interface(value)
except ValueError as exc:
raise DiscoveryError("无效的 --interface %r:%s" % (value, exc)) from exc
if not isinstance(parsed, ipaddress.IPv4Interface):
raise DiscoveryError("--interface 只支持 IPv4:%s" % value)
interfaces.append(
Interface(
alias="手工接口%d" % index,
description="由 --interface 指定",
address=parsed.ip,
prefix_length=parsed.network.prefixlen,
hardware=True,
)
)
return interfaces
def exclusion_reason(interface: Interface, include_virtual: bool) -> Optional[str]:
"""返回接口的排除原因;返回空值表示可用于搜索。"""
if interface.address.is_loopback:
return "回环地址"
if interface.address.is_link_local:
return "APIPA/链路本地地址"
if interface.address.is_unspecified or interface.address.is_multicast:
return "不是可用单播地址"
if not interface.hardware and not include_virtual:
return "虚拟或 VPN 接口(使用 --include-virtual 可加入)"
return None
def make_frame(session: int) -> bytes:
"""生成 V2 发现请求帧。"""
payload = json.dumps(
{"cmd": DISCOVERY_COMMAND, "session": session}, separators=(",", ":")
).encode("utf-8")
return b"EPBP" + struct.pack("!I", len(payload)) + payload
def parse_frame(packet: bytes, expected_session: int) -> Dict[str, Any]:
"""严格解析并校验一个完整 EPBP 数据报。"""
if len(packet) < 8 or packet[:4] != b"EPBP":
raise DiscoveryError("响应同步头不是 EPBP")
payload_size = struct.unpack("!I", packet[4:8])[0]
if payload_size <= 0 or payload_size > MAX_FRAME_SIZE:
raise DiscoveryError("响应长度不合理:%d" % payload_size)
if len(packet) != 8 + payload_size:
raise DiscoveryError(
"响应数据报长度不匹配:声明 %d,实际 %d" % (payload_size, len(packet) - 8)
)
try:
response = json.loads(packet[8:].decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise DiscoveryError("响应不是有效 UTF-8 JSON:%s" % exc) from exc
if not isinstance(response, dict):
raise DiscoveryError("响应 JSON 顶层不是对象")
if response.get("cmd") != DISCOVERY_COMMAND:
raise DiscoveryError("响应 cmd 不匹配:%r" % response.get("cmd"))
if response.get("session") != expected_session:
raise DiscoveryError("响应 session 不匹配:%r" % response.get("session"))
return response
def response_data(response: Dict[str, Any]) -> Dict[str, Any]:
"""兼容身份位于顶层或 data 对象的设备响应。"""
data = response.get("data")
return data if isinstance(data, dict) else response
def identity(response: Dict[str, Any]) -> Tuple[Optional[str], List[str], Optional[str]]:
"""提取完整 SN、型号集合和 MAC。"""
data = response_data(response)
sn = data.get("sn") if isinstance(data.get("sn"), str) else None
mac = data.get("mac") if isinstance(data.get("mac"), str) else None
models: List[str] = []
for section_name in ("inner", "outer", "hardware"):
section = data.get(section_name)
if isinstance(section, dict):
model = section.get("type")
if isinstance(model, str) and model and model not in models:
models.append(model)
return sn, models, mac
def port_owner(port: int) -> Optional[Dict[str, Any]]:
"""在 Windows 上查询 UDP 端口占用者,不结束任何进程。"""
if sys.platform != "win32":
return None
command = (
"$e=Get-NetUDPEndpoint -LocalPort %d -ErrorAction SilentlyContinue | "
"Select-Object -First 1; if($null -ne $e){$p=Get-Process -Id $e.OwningProcess "
"-ErrorAction SilentlyContinue;[PSCustomObject]@{PID=$e.OwningProcess;"
"Process=$p.ProcessName;Address=$e.LocalAddress}|ConvertTo-Json -Compress}" % port
)
try:
result = subprocess.run(
["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", command],
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
encoding="utf-8",
errors="replace",
timeout=10,
)
return json.loads(result.stdout) if result.stdout.strip() else None
except (OSError, subprocess.TimeoutExpired, json.JSONDecodeError):
return None
def open_sockets(
interfaces: Iterable[Interface], client_port: int
) -> List[Tuple[Interface, socket.socket]]:
"""为每个 IPv4 接口创建独立的广播收发套接字。"""
opened: List[Tuple[Interface, socket.socket]] = []
for interface in interfaces:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
try:
sock.bind((str(interface.address), client_port))
except OSError as exc:
sock.close()
owner = port_owner(client_port) if exc.errno in (errno.EACCES, errno.EADDRINUSE, 10013, 10048) else None
emit(
"bind_error",
interface=interface.alias,
address=str(interface.address),
port=client_port,
error=str(exc),
owner=owner,
action="请释放端口;本脚本不会结束占用进程",
)
continue
sock.setblocking(False)
opened.append((interface, sock))
emit(
"socket_ready",
interface=interface.alias,
address=str(interface.address),
port=client_port,
)
return opened
def discover(
opened: Sequence[Tuple[Interface, socket.socket]],
device_port: int,
timeout: float,
session: int,
) -> List[Dict[str, Any]]:
"""向每个接口发送两类广播并收集合法响应。"""
frame = make_frame(session)
socket_map = {sock: interface for interface, sock in opened}
for interface, sock in opened:
destinations = ["255.255.255.255", str(interface.network.broadcast_address)]
for destination in dict.fromkeys(destinations):
try:
sock.sendto(frame, (destination, device_port))
emit(
"broadcast_sent",
interface=interface.alias,
source=str(interface.address),
destination=destination,
device_port=device_port,
session=session,
)
except OSError as exc:
emit(
"send_error",
interface=interface.alias,
destination=destination,
error=str(exc),
)
deadline = time.monotonic() + timeout
found: List[Dict[str, Any]] = []
while socket_map:
remaining = deadline - time.monotonic()
if remaining <= 0:
break
readable, _, _ = select.select(list(socket_map), [], [], remaining)
if not readable:
break
for sock in readable:
interface = socket_map[sock]
try:
packet, peer = sock.recvfrom(MAX_FRAME_SIZE + 9)
response = parse_frame(packet, session)
except (OSError, DiscoveryError) as exc:
emit(
"invalid_response",
interface=interface.alias,
error=str(exc),
)
continue
sn, models, mac = identity(response)
candidate = {
"source_ip": peer[0],
"source_port": peer[1],
"received_on": str(interface.address),
"interface": interface.alias,
"sn": sn,
"models": models,
"mac": mac,
"response": response,
}
found.append(candidate)
emit("candidate", **candidate)
return found
def deduplicate(candidates: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""只合并 SN、源地址、MAC 和型号完全一致的重复响应。"""
unique: Dict[Tuple[str, ...], Dict[str, Any]] = {}
for candidate in candidates:
sn = candidate.get("sn")
key = (
"identity",
str(sn or ""),
str(candidate.get("source_ip") or ""),
str(candidate.get("mac") or ""),
json.dumps(candidate.get("models", []), ensure_ascii=False, separators=(",", ":")),
)
unique.setdefault(key, candidate)
return list(unique.values())
def find_identity_conflicts(candidates: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""找出同一完整 SN 对应多个网络或型号身份的冲突。"""
grouped: Dict[str, List[Dict[str, Any]]] = {}
for candidate in candidates:
sn = candidate.get("sn")
if isinstance(sn, str) and sn:
grouped.setdefault(sn, []).append(candidate)
return [
{"sn": sn, "candidates": items}
for sn, items in grouped.items()
if len(items) > 1
]
def select_expected(
candidates: Sequence[Dict[str, Any]], expected_sn: Optional[str], expected_model: Optional[str]
) -> List[Dict[str, Any]]:
"""按完整 SN 和型号筛选候选。"""
selected = list(candidates)
if expected_sn:
selected = [item for item in selected if item.get("sn") == expected_sn]
if expected_model:
selected = [item for item in selected if expected_model in item.get("models", [])]
return selected
def build_parser() -> argparse.ArgumentParser:
"""创建命令行参数解析器。"""
parser = argparse.ArgumentParser(
description="从每个活动 IPv4 接口的本地 8951 端口执行 V2 EPBP 广播搜索。"
)
parser.add_argument(
"--interface",
action="append",
default=[],
metavar="IP/PREFIX",
help="手工指定接口,可重复;给出后不自动枚举,格式为本机IPv4/前缀长度",
)
parser.add_argument(
"--include-virtual",
action="store_true",
help="自动枚举时同时使用活动的虚拟、VPN 等非物理接口",
)
parser.add_argument("--expected-sn", help="期望的完整设备 SN,按完全相等匹配")
parser.add_argument("--expected-model", help="期望型号,匹配 inner/outer/hardware.type")
parser.add_argument("--timeout", type=float, default=3.0, help="接收等待秒数")
parser.add_argument("--client-port", type=int, default=CLIENT_PORT)
parser.add_argument("--device-port", type=int, default=DEVICE_PORT)
return parser
def validate_options(args: argparse.Namespace) -> None:
"""拒绝无效端口和等待时间。"""
if args.timeout <= 0 or args.timeout > 300:
raise DiscoveryError("--timeout 必须在 0 到 300 秒之间")
for name in ("client_port", "device_port"):
value = getattr(args, name)
if value <= 0 or value > 65535:
raise DiscoveryError("--%s 必须在 1 到 65535 之间" % name.replace("_", "-"))
def main(argv: Optional[List[str]] = None) -> int:
"""执行网卡筛选、广播搜索和唯一身份匹配。"""
opened: List[Tuple[Interface, socket.socket]] = []
try:
args = build_parser().parse_args(argv)
validate_options(args)
if args.client_port != CLIENT_PORT:
emit(
"diagnostic_port",
client_port=args.client_port,
warning="设备可能只回复到 8951;非标准源端口无响应不能证明设备离线",
)
interfaces = (
explicit_interfaces(args.interface) if args.interface else powershell_interfaces()
)
selected_interfaces: List[Interface] = []
for interface in interfaces:
reason = exclusion_reason(interface, args.include_virtual)
emit(
"interface",
alias=interface.alias,
description=interface.description,
address=str(interface.address),
prefix_length=interface.prefix_length,
directed_broadcast=str(interface.network.broadcast_address),
hardware=interface.hardware,
selected=reason is None,
exclusion_reason=reason,
)
if reason is None:
selected_interfaces.append(interface)
if not selected_interfaces:
raise DiscoveryError("没有可用于广播的 IPv4 接口")
opened = open_sockets(selected_interfaces, args.client_port)
if not opened:
raise DiscoveryError(
"所有接口都无法绑定本地端口 %d;请检查端口占用" % args.client_port
)
session = secrets.randbelow(0xFFFFFFFF) + 1
candidates = deduplicate(
discover(opened, args.device_port, args.timeout, session)
)
conflicts = find_identity_conflicts(candidates)
if conflicts:
emit(
"identity_conflict",
message="同一完整 SN 对应不同 IP、MAC 或型号,禁止自动选择设备",
conflicts=conflicts,
)
return 2
matches = select_expected(candidates, args.expected_sn, args.expected_model)
identity_verified = bool(args.expected_sn and len(matches) == 1)
emit(
"summary",
mode="identity_check" if args.expected_sn else "list_only",
identity_verified=identity_verified,
candidate_count=len(candidates),
candidates=candidates,
expected_sn=args.expected_sn,
expected_model=args.expected_model,
match_count=len(matches),
matches=matches,
)
if not candidates:
emit(
"not_found",
message=(
"未收到合法响应;若使用的不是客户端端口 8951,此结果不能证明设备离线"
),
)
return 2
if args.expected_sn:
if len(matches) != 1:
emit("identity_error", message="期望身份没有唯一匹配,禁止自动选择设备")
return 2
return 0
except (OSError, ValueError, DiscoveryError) as exc:
emit("error", type=type(exc).__name__, message=str(exc))
return 1
finally:
for _, sock in opened:
sock.close()
configure_output()
if __name__ == "__main__":
raise SystemExit(main())