ching/work_script
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: ching:e5a9369440bad44c0e5a054d1718a405a01531e5
Branches Tags
ching:main
..
compare: ching:8651878be87a24529ed96da629a2b56f2fede205
Branches Tags
ching:main

2 Commits
e5a9369440 ... 8651878be8

Author SHA1 Message Date
Ching L
8651878be8 feat(cp02): add PD decode tools and enhance build GUI 
- Add charging_viewer.py for reading PowerZ .sqlite/.db charging data
  - Add powerz_pd_decode.py for decoding USB PD protocol messages
  - Add sdkconfig file path display to build_gui.html
  - Add signing reminder to split_and_merge.sh
 2026-04-03 10:50:23 +08:00
Ching L
6ad3ab93a7 feat(cp02): add build GUI, flash guide, and comment removal tool 
Add build_gui.html for visual command generation, flash_guide.md for
flashing reference, and remove_chinese_comments.py for cleaning source
files. Update extract_firmware.sh to use output directory instead of
prefix, and update README with all new tools.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
 2026-02-26 13:57:52 +08:00
8 changed files with 1931 additions and 24 deletions
Show Stats Expand all files Collapse all files

70
CP02/README.md
View File

@ -9,6 +9,7 @@
- 将输入固件文件切分为前面部分和最后256字节
- 调用合并脚本生成最终固件
- 自动删除最后1字节并检查文件大小限制
- 合并完成后提示签名命令
### merge_2323_firmware.sh
- **功能**: 合并 PA768 updater、2323固件和IUM数据为单一固件文件
@ -21,12 +22,13 @@
### extract_firmware.sh
- **功能**: 从合并的 bin 文件中提取 firmware、IUM 和 metadata
- **用法**: `./extract_firmware.sh <merged_bin> [output_prefix]`
- **用法**: `./extract_firmware.sh <merged_bin> [output_dir]`
- **描述**:
- 从固定偏移位置提取 firmware (0x04000)
- 提取 IUM 数据 (0x1dc00, 256 bytes)
- 解析并生成 metadata.json (blocks 和 last_block_size)
- 如未指定前缀，自动使用原文件名
- 如未指定输出目录，默认输出到当前目录
- 自动创建不存在的输出目录
### get_last_bytes.sh
- **功能**: 获取 IUM 文件的最后两个字节并以十六进制格式返回
@ -99,6 +101,53 @@
- 自动处理重名文件（添加 _1, _2 等后缀）
- 显示复制进度和统计信息
### remove_chinese_comments.py
- **功能**: 从源代码文件中移除中文注释
- **用法**: `python3 remove_chinese_comments.py [--check] <file> [file2 ...]`
- **描述**:
- 支持 C/C++（`//` 注释）和 Python（`#` 注释）文件
- 整行中文注释直接删除，行尾中文注释只移除注释部分
- `--check` 模式仅显示中文注释，不做修改
- 交互式确认后才执行删除
### charging_viewer.py
- **功能**: 充电数据查看器，读取 PowerZ 等工具导出的 .sqlite / .db 充电测试数据
- **用法**: `python3 charging_viewer.py [--no-page] [目录或文件路径 ...]`
- **描述**:
- 自动检测 pd_chart 和 table_1 表格式
- 显示功率、电压、电流、温度、充电量等统计信息
- 解析 PD 消息头、PDO（Fixed/Battery/Variable/PPS）、RDO
- 从 VBUS/CC 波形分析 PD 协商事件（电压跳变检测）
- 支持翻页输出（自动使用 less）
- 无参数时扫描当前目录下所有 .sqlite/.db 文件
### powerz_pd_decode.py
- **功能**: 解码 PowerZ 抓取的 USB PD 协议消息
- **用法**: `python3 powerz_pd_decode.py <sqlite文件> [--window START END]`
- **描述**:
- 从 PowerZ sqlite 文件的 pd_table 读取原始 PD 数据
- 解码 Source_Capabilities（Fixed/PPS/AVS PDO）
- 解码 Request/EPR_Request（Fixed RDO 和 PPS RDO）
- 关联 Request 与对应的 Source_Cap 显示选中的 PDO
- `--window` 参数可查看指定时间窗口内的所有消息
### build_gui.html
- **功能**: IonBridge 构建工具的可视化命令生成器
- **用法**: 在浏览器中打开 `build_gui.html`
- **描述**:
- 支持选择 variant（CP02、CP02S、3536、GPU Demo、Bird）
- 可配置 build 选项（production、coredump）和 flash 选项（littlefs、protected_data、nvs 等）
- 一键生成 Build、Flash、Monitor、Clean 等命令并复制
- 自动保存配置和命令历史到 localStorage
- 显示当前 variant 和模式对应的 sdkconfig 文件路径
### flash_guide.md
- **功能**: IonBridge 烧录操作参考文档
- **描述**:
- 非 Secure 设备（开发板）的 build/flash/monitor 命令示例
- Secure 设备（量产板）的签名烧录流程
- sign_key 和 psn 参数说明
### merge_encrypted_firmware.py
- **功能**: 合成已加密的固件和 IUM 为最终 programmer 固件
- **用法**: `python3 merge_encrypted_firmware.py <base.bin> <encrypted.bin> <ium.bin> -o <output.bin>`
@ -123,8 +172,8 @@
./merge_2323_firmware.sh -u PA768_Updater.bin -f 2323_firmware.bin -i 2323_ium.bin -o merged.bin
# 提取固件组件
./extract_firmware.sh 40_ufcs2.bin firmware # 指定输出前缀
./extract_firmware.sh merged.bin # 使用原文件名作为前缀
./extract_firmware.sh 40_ufcs2.bin /tmp/output # 输出到指定目录
./extract_firmware.sh merged.bin # 输出到当前目录
# 获取 IUM 文件的最后两个字节
./get_last_bytes.sh ium.bin # 使用默认文件
@ -149,6 +198,19 @@ python3 txt_to_bin_converter.py ium11.txt ium11.bin # 文本转二
# 合成加密固件
python3 merge_encrypted_firmware.py base.bin encrypted.bin ium.bin -o programmer.bin
# 查看充电测试数据
python3 charging_viewer.py . # 扫描当前目录
python3 charging_viewer.py data.sqlite # 指定文件
python3 charging_viewer.py --no-page /path/to/captures/ # 不翻页输出
# 解码 PowerZ PD 消息
python3 powerz_pd_decode.py capture.sqlite # 显示摘要
python3 powerz_pd_decode.py capture.sqlite --window 1.0 5.0 # 查看时间窗口
# 复制 .o 文件
./copy_o_files.sh /path/to/build /path/to/output
# 移除中文注释
python3 remove_chinese_comments.py --check main.c # 仅检查
python3 remove_chinese_comments.py main.c utils.py # 交互式移除
```

578
CP02/build_gui.html Normal file
View File

@ -0,0 +1,578 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>IonBridge Build Tool</title>
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
:root {
--bg: #1a1b26;
--surface: #24283b;
--surface-hover: #2f3349;
--border: #3b4261;
--text: #c0caf5;
--text-dim: #565f89;
--accent: #7aa2f7;
--accent-hover: #89b4fa;
--green: #9ece6a;
--red: #f7768e;
--orange: #ff9e64;
--cyan: #7dcfff;
}
body {
font-family: -apple-system, BlinkMacSystemFont, 'SF Pro Text', system-ui, sans-serif;
background: var(--bg);
color: var(--text);
height: 100vh;
display: flex;
flex-direction: column;
overflow: hidden;
}
.header {
padding: 14px 24px;
background: var(--surface);
border-bottom: 1px solid var(--border);
flex-shrink: 0;
}
.header h1 { font-size: 17px; font-weight: 600; color: var(--accent); }
.main { display: flex; flex: 1; overflow: hidden; }
.sidebar {
width: 300px;
background: var(--surface);
border-right: 1px solid var(--border);
padding: 16px;
overflow-y: auto;
flex-shrink: 0;
}
.section { margin-bottom: 20px; }
.section-title {
font-size: 11px;
font-weight: 600;
text-transform: uppercase;
letter-spacing: 0.8px;
color: var(--text-dim);
margin-bottom: 10px;
}
label.field-label {
display: block;
font-size: 12px;
color: var(--text-dim);
margin-bottom: 4px;
}
select, input[type="text"] {
width: 100%;
padding: 7px 10px;
background: var(--bg);
border: 1px solid var(--border);
border-radius: 6px;
color: var(--text);
font-size: 13px;
outline: none;
margin-bottom: 10px;
}
select:focus, input:focus { border-color: var(--accent); }
.checkbox-group { display: flex; flex-wrap: wrap; gap: 6px; }
.checkbox-item {
display: flex;
align-items: center;
gap: 5px;
padding: 5px 9px;
background: var(--bg);
border: 1px solid var(--border);
border-radius: 6px;
cursor: pointer;
font-size: 12px;
transition: all 0.15s;
user-select: none;
}
.checkbox-item:hover { border-color: var(--accent); }
.checkbox-item input[type="checkbox"] { accent-color: var(--accent); width: 14px; height: 14px; }
.checkbox-item.checked {
border-color: var(--accent);
background: rgba(122, 162, 247, 0.08);
}
.actions { display: flex; flex-direction: column; gap: 6px; }
.btn {
padding: 9px 14px;
border: none;
border-radius: 7px;
font-size: 13px;
font-weight: 500;
cursor: pointer;
transition: all 0.15s;
display: flex;
align-items: center;
justify-content: center;
gap: 6px;
}
.btn-primary { background: var(--accent); color: var(--bg); }
.btn-primary:hover { background: var(--accent-hover); }
.btn-green { background: var(--green); color: var(--bg); }
.btn-green:hover { opacity: 0.9; }
.btn-orange { background: var(--orange); color: var(--bg); }
.btn-orange:hover { opacity: 0.9; }
.btn-ghost {
background: var(--surface-hover);
color: var(--text);
border: 1px solid var(--border);
}
.btn-ghost:hover { background: var(--border); }
.btn-row { display: flex; gap: 6px; }
.btn-row .btn { flex: 1; }
.output-area {
flex: 1;
display: flex;
flex-direction: column;
padding: 24px;
overflow-y: auto;
}
.output-title {
font-size: 11px;
font-weight: 600;
text-transform: uppercase;
letter-spacing: 0.8px;
color: var(--text-dim);
margin-bottom: 12px;
}
.cmd-block {
background: var(--surface);
border: 1px solid var(--border);
border-radius: 8px;
margin-bottom: 12px;
overflow: hidden;
transition: border-color 0.3s;
}
.cmd-block.just-copied { border-color: var(--green); }
.cmd-header {
display: flex;
align-items: center;
justify-content: space-between;
padding: 8px 14px;
background: var(--surface-hover);
border-bottom: 1px solid var(--border);
}
.cmd-label {
font-size: 12px;
font-weight: 600;
color: var(--text-dim);
}
.cmd-step {
font-size: 11px;
color: var(--text-dim);
background: var(--bg);
padding: 2px 8px;
border-radius: 10px;
}
.copy-btn {
padding: 4px 12px;
font-size: 11px;
background: var(--accent);
color: var(--bg);
border: none;
border-radius: 4px;
cursor: pointer;
font-weight: 500;
transition: all 0.15s;
}
.copy-btn:hover { background: var(--accent-hover); }
.copy-btn.copied { background: var(--green); }
.cmd-body {
padding: 12px 14px;
font-family: 'SF Mono', 'Fira Code', 'Cascadia Code', 'Menlo', monospace;
font-size: 13px;
line-height: 1.8;
color: var(--cyan);
user-select: all;
white-space: pre-wrap;
word-break: break-all;
}
.copy-all-bar {
display: flex;
align-items: center;
justify-content: space-between;
margin-bottom: 16px;
}
.copy-all-btn {
padding: 8px 20px;
font-size: 13px;
font-weight: 500;
background: var(--accent);
color: var(--bg);
border: none;
border-radius: 7px;
cursor: pointer;
transition: all 0.15s;
}
.copy-all-btn:hover { background: var(--accent-hover); }
.copy-all-btn.copied { background: var(--green); }
.empty-state {
display: flex;
flex-direction: column;
align-items: center;
justify-content: center;
flex: 1;
color: var(--text-dim);
gap: 8px;
}
.empty-state .icon { font-size: 40px; opacity: 0.3; }
.empty-state .text { font-size: 14px; }
.history-item {
padding: 8px 12px;
background: var(--bg);
border: 1px solid var(--border);
border-radius: 6px;
margin-bottom: 6px;
font-family: 'SF Mono', monospace;
font-size: 11px;
color: var(--text-dim);
cursor: pointer;
transition: all 0.15s;
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
}
.history-item:hover {
border-color: var(--accent);
color: var(--text);
}
.sdkconfig-info {
font-size: 11px;
color: var(--text-dim);
background: var(--bg);
padding: 5px 8px;
border-radius: 4px;
margin-bottom: 10px;
font-family: 'SF Mono', monospace;
}
</style>
</head>
<body>
<div class="header">
<h1>IonBridge Build Tool</h1>
</div>
<div class="main">
<div class="sidebar">
<div class="section">
<div class="section-title">Target</div>
<label class="field-label">Variant</label>
<select id="variant">
<option value="cp02" selected>CP02 (FPGA)</option>
<option value="cp02s">CP02S (GPU)</option>
<option value="3536">3536 / 68W (I2C)</option>
<option value="gpu_demo">GPU Demo</option>
<option value="bird">Bird</option>
</select>
<div id="sdkconfigInfo" class="sdkconfig-info"></div>
<label class="field-label">Serial Port</label>
<input type="text" id="port" value="/dev/tty.usbmodem2101">
</div>
<div class="section">
<div class="section-title">Build Options</div>
<div class="checkbox-group">
<label class="checkbox-item" onclick="toggleCb(this)">
<input type="checkbox" id="opt-production"> production
</label>
<label class="checkbox-item checked" onclick="toggleCb(this)">
<input type="checkbox" id="opt-coredump" checked> coredump
</label>
</div>
</div>
<div class="section">
<div class="section-title">Flash Options</div>
<div class="checkbox-group">
<label class="checkbox-item" onclick="toggleCb(this)">
<input type="checkbox" id="opt-littlefs"> littlefs
</label>
<label class="checkbox-item" onclick="toggleCb(this)">
<input type="checkbox" id="opt-protected_data"> protected_data
</label>
<label class="checkbox-item" onclick="toggleCb(this)">
<input type="checkbox" id="opt-nvs"> nvs
</label>
<label class="checkbox-item" onclick="toggleCb(this)">
<input type="checkbox" id="opt-phy_init"> phy_init
</label>
<label class="checkbox-item" onclick="toggleCb(this)">
<input type="checkbox" id="opt-unlock"> unlock
</label>
</div>
<label class="field-label" style="margin-top:10px">PSN</label>
<input type="text" id="psn" placeholder="e.g. 1331040606001272">
<label class="field-label">Sign Key (overrides PSN)</label>
<input type="text" id="sign_key" placeholder="e.g. /Users/ching/Downloads/xxx_sign_key.pem">
</div>
<div class="section">
<div class="section-title">Generate Command</div>
<div class="actions">
<button class="btn btn-primary" onclick="gen('build')">Build</button>
<div class="btn-row">
<button class="btn btn-green" onclick="gen('flash')">Flash</button>
<button class="btn btn-orange" onclick="gen('monitor')">Monitor</button>
</div>
<button class="btn btn-ghost" onclick="gen('build_flash')">Build + Flash</button>
<button class="btn btn-ghost" onclick="gen('update_version')">Update Version</button>
<div class="btn-row">
<button class="btn btn-ghost" onclick="gen('clean')">Clean</button>
<button class="btn btn-ghost" onclick="gen('fullclean')">Full Clean</button>
</div>
</div>
</div>
<div class="section" id="historySection" style="display:none">
<div class="section-title">History</div>
<div id="history"></div>
</div>
</div>
<div class="output-area">
<div id="emptyState" class="empty-state">
<div class="icon">&#8592;</div>
<div class="text">Select an action to generate commands</div>
</div>
<div id="output" style="display:none"></div>
</div>
</div>
<script>
const STORAGE_KEY = 'ionbridge_build_gui';
const FIELDS = ['variant', 'port', 'psn', 'sign_key'];
const CHECKBOXES = ['opt-production', 'opt-coredump', 'opt-littlefs', 'opt-protected_data', 'opt-nvs', 'opt-phy_init', 'opt-unlock'];
let historyList = [];
function val(id) { return document.getElementById(id).value.trim(); }
function checked(id) { return document.getElementById(id).checked; }
function saveState() {
const state = {};
FIELDS.forEach(id => state[id] = document.getElementById(id).value);
CHECKBOXES.forEach(id => state[id] = document.getElementById(id).checked);
state.history = historyList;
localStorage.setItem(STORAGE_KEY, JSON.stringify(state));
}
function loadState() {
const raw = localStorage.getItem(STORAGE_KEY);
if (!raw) return;
try {
const state = JSON.parse(raw);
FIELDS.forEach(id => { if (state[id] !== undefined) document.getElementById(id).value = state[id]; });
CHECKBOXES.forEach(id => {
if (state[id] !== undefined) {
const cb = document.getElementById(id);
cb.checked = state[id];
cb.closest('.checkbox-item').classList.toggle('checked', state[id]);
}
});
if (state.history) {
historyList = state.history;
if (historyList.length > 0) renderHistory();
}
} catch(e) {}
}
function toggleCb(label) {
setTimeout(() => {
const cb = label.querySelector('input[type="checkbox"]');
label.classList.toggle('checked', cb.checked);
saveState();
}, 0);
}
// Auto-save on any input change
document.addEventListener('input', (e) => {
if (e.target.matches('select, input[type="text"]')) saveState();
});
document.addEventListener('change', (e) => {
if (e.target.matches('select')) saveState();
});
// Develop: sdkconfig.{variant_config}, Production: sdkconfig-{variant}.production (fallback: sdkconfig.production)
const SDKCONFIG_MAP = {
cp02: { develop: 'sdkconfig.develop', production: 'sdkconfig.production' },
cp02s: { develop: 'sdkconfig.cp02s.develop', production: 'sdkconfig-cp02s.production' },
'3536': { develop: 'sdkconfig.3536.develop', production: 'sdkconfig.production' },
gpu_demo: { develop: 'sdkconfig.develop', production: 'sdkconfig.production' },
bird: { develop: 'sdkconfig.bird', production: 'sdkconfig.production' },
};
function updateSdkconfigInfo() {
const v = val('variant');
const mode = checked('opt-production') ? 'production' : 'develop';
const cfg = SDKCONFIG_MAP[v];
const file = cfg ? cfg[mode] : '?';
document.getElementById('sdkconfigInfo').textContent = 'sdkconfig \u2190 configs/' + file;
}
// Update on variant or production change
document.getElementById('variant').addEventListener('change', updateSdkconfigInfo);
document.getElementById('opt-production').addEventListener('change', updateSdkconfigInfo);
function buildCmd() {
let cmd = 'make build variant=' + val('variant');
if (checked('opt-production')) cmd += ' production=1';
if (!checked('opt-coredump')) cmd += ' coredump=0';
return cmd;
}
function flashCmd() {
let cmd = 'make flash';
if (checked('opt-littlefs')) cmd += ' littlefs=1';
if (checked('opt-protected_data')) cmd += ' protected_data=1';
if (checked('opt-nvs')) cmd += ' nvs=1';
if (checked('opt-phy_init')) cmd += ' phy_init=1';
if (checked('opt-unlock')) cmd += ' unlock=1';
const signKey = val('sign_key');
const psn = val('psn');
if (signKey) cmd += ' sign_key=' + signKey;
else if (psn) cmd += ' psn=' + psn;
const port = val('port');
if (port) cmd += ' PORT=' + port;
return cmd;
}
function monitorCmd() {
let cmd = 'make monitor';
const port = val('port');
if (port) cmd += ' PORT=' + port;
return cmd;
}
function gen(action) {
let commands = [];
switch (action) {
case 'build':
commands = [{label: 'Build', cmd: buildCmd()}];
break;
case 'flash':
commands = [{label: 'Flash', cmd: flashCmd()}];
break;
case 'monitor':
commands = [{label: 'Monitor', cmd: monitorCmd()}];
break;
case 'build_flash':
commands = [
{label: 'Build', cmd: buildCmd()},
{label: 'Flash', cmd: flashCmd()},
];
break;
case 'update_version':
commands = [{label: 'Update Version', cmd: 'make update_version variant=' + val('variant')}];
break;
case 'clean':
commands = [{label: 'Clean', cmd: 'make clean'}];
break;
case 'fullclean':
commands = [{label: 'Full Clean', cmd: 'make fullclean'}];
break;
}
renderOutput(commands);
addHistory(commands);
}
function renderOutput(commands) {
document.getElementById('emptyState').style.display = 'none';
const out = document.getElementById('output');
out.style.display = 'block';
const allCmds = commands.map(c => c.cmd).join('\n');
const multi = commands.length > 1;
let html = '<div class="copy-all-bar">';
html += '<div class="output-title">Generated Commands</div>';
html += '<button class="copy-all-btn" onclick="copyAll(this, ' + JSON.stringify(allCmds).replace(/"/g, '&quot;') + ')">Copy All</button>';
html += '</div>';
commands.forEach((c, i) => {
html += '<div class="cmd-block">';
html += '<div class="cmd-header">';
html += '<div style="display:flex;align-items:center;gap:8px">';
if (multi) html += '<span class="cmd-step">Step ' + (i + 1) + '</span>';
html += '<span class="cmd-label">' + c.label + '</span>';
html += '</div>';
html += '<button class="copy-btn" onclick="copyOne(this, ' + JSON.stringify(c.cmd).replace(/"/g, '&quot;') + ')">Copy</button>';
html += '</div>';
html += '<div class="cmd-body">' + escapeHtml(c.cmd) + '</div>';
html += '</div>';
});
out.innerHTML = html;
}
function escapeHtml(text) {
const div = document.createElement('div');
div.textContent = text;
return div.innerHTML;
}
async function copyOne(btn, text) {
await navigator.clipboard.writeText(text);
btn.textContent = 'Copied!';
btn.classList.add('copied');
btn.closest('.cmd-block').classList.add('just-copied');
setTimeout(() => {
btn.textContent = 'Copy';
btn.classList.remove('copied');
btn.closest('.cmd-block').classList.remove('just-copied');
}, 1500);
}
async function copyAll(btn, text) {
await navigator.clipboard.writeText(text);
btn.textContent = 'Copied!';
btn.classList.add('copied');
setTimeout(() => {
btn.textContent = 'Copy All';
btn.classList.remove('copied');
}, 1500);
}
function addHistory(commands) {
const text = commands.map(c => c.cmd).join(' && ');
// Avoid duplicates at top
if (historyList.length > 0 && historyList[0] === text) return;
historyList.unshift(text);
if (historyList.length > 10) historyList.pop();
renderHistory();
saveState();
}
function renderHistory() {
const section = document.getElementById('historySection');
const container = document.getElementById('history');
section.style.display = 'block';
container.innerHTML = historyList.map(cmd =>
'<div class="history-item" onclick="copyFromHistory(this, ' + JSON.stringify(cmd).replace(/"/g, '&quot;') + ')" title="Click to copy">' + escapeHtml(cmd) + '</div>'
).join('');
}
async function copyFromHistory(el, text) {
await navigator.clipboard.writeText(text);
el.style.borderColor = 'var(--green)';
el.style.color = 'var(--green)';
setTimeout(() => {
el.style.borderColor = '';
el.style.color = '';
}, 1000);
}
loadState();
updateSdkconfigInfo();
</script>
</body>
</html>

775
CP02/charging_viewer.py Normal file
View File

@ -0,0 +1,775 @@
#!/usr/bin/env python3
"""充电数据查看器 - 读取 .sqlite / .db 充电测试数据并显示统计信息"""
import sqlite3
import sys
import os
import glob
import io
import subprocess
import shutil
# ── USB PD 消息类型 ──────────────────────────────────────────
PD_CTRL_MSG = {
0x01: 'GoodCRC', 0x02: 'GotoMin', 0x03: 'Accept', 0x04: 'Reject',
0x05: 'Ping', 0x06: 'PS_RDY', 0x07: 'Get_Source_Cap',
0x08: 'Get_Sink_Cap', 0x09: 'DR_Swap', 0x0A: 'PR_Swap',
0x0B: 'VCONN_Swap', 0x0C: 'Wait', 0x0D: 'Soft_Reset',
0x0E: 'Data_Reset', 0x0F: 'Data_Reset_Cpl',
0x10: 'Not_Supported', 0x11: 'Get_Src_Cap_Ext',
0x12: 'Get_Status', 0x13: 'FR_Swap', 0x14: 'Get_PPS_Status',
0x15: 'Get_Country_Codes', 0x16: 'Get_Snk_Cap_Ext',
0x17: 'Get_Source_Info', 0x18: 'Get_Revision',
}
PD_DATA_MSG = {
0x01: 'Source_Cap', 0x02: 'Request', 0x03: 'BIST',
0x04: 'Sink_Cap', 0x05: 'Battery_Status', 0x06: 'Alert',
0x07: 'Get_Country_Info', 0x08: 'Enter_USB', 0x09: 'EPR_Request',
0x0A: 'EPR_Mode', 0x0B: 'Source_Info', 0x0C: 'Revision',
0x0F: 'Vendor_Defined',
}
PD_SPEC_REV = {0: '1.0', 1: '2.0', 2: '3.0', 3: '3.1'}
# ── PD 解析 ──────────────────────────────────────────────────
def _to_int(val):
"""尝试将值转为整数"""
if isinstance(val, int):
return val
if isinstance(val, float):
return int(val)
if isinstance(val, str):
s = val.strip().replace(' ', '')
if s.startswith(('0x', '0X')):
try:
return int(s, 16)
except ValueError:
return None
try:
return int(s)
except ValueError:
try:
return int(s, 16)
except ValueError:
return None
if isinstance(val, (bytes, bytearray)):
return int.from_bytes(val, 'little')
return None
def parse_pd_header(val):
"""解析 PD 消息头 (16-bit)"""
h = _to_int(val)
if h is None or not (0 <= h <= 0xFFFF):
return None
num_do = (h >> 12) & 7
msg_type = h & 0x1F
if num_do == 0:
name = PD_CTRL_MSG.get(msg_type, f'Ctrl_0x{msg_type:02X}')
else:
name = PD_DATA_MSG.get(msg_type, f'Data_0x{msg_type:02X}')
return {
'raw': h, 'name': name, 'num_do': num_do,
'msg_id': (h >> 9) & 7,
'spec': PD_SPEC_REV.get((h >> 6) & 3, '?'),
'prole': 'SRC' if (h >> 8) & 1 else 'SNK',
'drole': 'DFP' if (h >> 5) & 1 else 'UFP',
}
def parse_pdo(val):
"""解析 Power Data Object"""
pdo = _to_int(val)
if pdo is None:
return str(val)
t = (pdo >> 30) & 3
if t == 0: # Fixed
v = ((pdo >> 10) & 0x3FF) * 0.05
i_max = (pdo & 0x3FF) * 0.01
flags = []
if (pdo >> 29) & 1: flags.append('DRP')
if (pdo >> 27) & 1: flags.append('UP')
if (pdo >> 26) & 1: flags.append('USB')
if (pdo >> 25) & 1: flags.append('DRD')
f = f' [{",".join(flags)}]' if flags else ''
return f'Fixed {v:.1f}V/{i_max:.2f}A({v*i_max:.0f}W){f}'
elif t == 1: # Battery
max_v = ((pdo >> 20) & 0x3FF) * 0.05
min_v = ((pdo >> 10) & 0x3FF) * 0.05
max_p = (pdo & 0x3FF) * 0.25
return f'Batt {min_v:.1f}-{max_v:.1f}V/{max_p:.1f}W'
elif t == 2: # Variable
max_v = ((pdo >> 20) & 0x3FF) * 0.05
min_v = ((pdo >> 10) & 0x3FF) * 0.05
max_i = (pdo & 0x3FF) * 0.01
return f'Var {min_v:.1f}-{max_v:.1f}V/{max_i:.2f}A'
else: # Augmented
sub = (pdo >> 28) & 3
if sub == 0: # SPR PPS
max_v = ((pdo >> 17) & 0xFF) * 0.1
min_v = ((pdo >> 8) & 0xFF) * 0.1
max_i = (pdo & 0x7F) * 0.05
return f'PPS {min_v:.1f}-{max_v:.1f}V/{max_i:.2f}A'
return f'APDO 0x{pdo:08X}'
def parse_rdo(val):
"""解析 Request Data Object"""
rdo = _to_int(val)
if rdo is None:
return str(val)
pos = (rdo >> 28) & 0x7
op_i = ((rdo >> 10) & 0x3FF) * 0.01
max_i = (rdo & 0x3FF) * 0.01
flags = []
if (rdo >> 27) & 1: flags.append('GiveBack')
if (rdo >> 26) & 1: flags.append('Mismatch')
f = f' [{",".join(flags)}]' if flags else ''
return f'Pos{pos} Op:{op_i:.2f}A Max:{max_i:.2f}A{f}'
def _parse_data_blob(val):
"""尝试将 data/payload 拆分为多个 32-bit 数据对象"""
if val is None:
return []
if isinstance(val, (bytes, bytearray)):
return [int.from_bytes(val[i:i+4], 'little')
for i in range(0, len(val) - 3, 4)]
if isinstance(val, str):
s = val.strip().replace(' ', '')
if s.startswith(('0x', '0X')):
s = s[2:]
if len(s) >= 8 and all(c in '0123456789abcdefABCDEF' for c in s):
return [int(s[i:i+8], 16) for i in range(0, len(s) - 7, 8)]
return []
# ── PD 标准电压 ──────────────────────────────────────────────
PD_STD_VOLTAGES = {5: '5V', 9: '9V', 12: '12V', 15: '15V', 20: '20V'}
def _nearest_pd_voltage(v):
"""找到最近的标准 PD 电压"""
nearest = min(PD_STD_VOLTAGES, key=lambda x: abs(x - v))
return nearest if abs(nearest - v) < 1.5 else None
# ── 数据读取 ─────────────────────────────────────────────────
def read_pd_chart(db_path):
"""读取 pd_chart 表"""
conn = sqlite3.connect(db_path)
cur = conn.cursor()
tables = [r[0] for r in cur.execute(
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()]
if 'pd_chart' not in tables:
conn.close()
return None
rows = cur.execute(
"SELECT Time, VBUS, IBUS, CC1, CC2 FROM pd_chart ORDER BY Time"
).fetchall()
conn.close()
return {
'type': 'pd_chart',
'columns': ['Time(s)', 'VBUS(V)', 'IBUS(A)', 'CC1(V)', 'CC2(V)'],
'rows': rows,
} if rows else None
def read_table_1(db_path):
"""读取 table_1 表"""
conn = sqlite3.connect(db_path)
cur = conn.cursor()
tables = [r[0] for r in cur.execute(
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()]
if 'table_1' not in tables:
conn.close()
return None
rows = cur.execute(
"SELECT ElapsedTime, Unix, VBUS, IBUS, DP, DM, CC1, CC2, TEMP, CHARGE, ENERGY "
"FROM table_1 ORDER BY Unix"
).fetchall()
params = None
if 'table_1_param' in tables:
p = cur.execute("SELECT * FROM table_1_param").fetchone()
if p:
params = p
conn.close()
if not rows:
return None
return {
'type': 'table_1',
'columns': ['时间', 'Unix(s)', 'VBUS(V)', 'IBUS(A)', 'DP(V)', 'DM(V)',
'CC1(V)', 'CC2(V)', 'TEMP(°C)', 'CHARGE(Ah)', 'ENERGY(Wh)'],
'rows': rows,
'params': params,
}
def read_extra_tables(db_path):
"""读取已知表以外的所有表 (PD 消息等)"""
known = {'pd_chart', 'table_1', 'table_1_param'}
conn = sqlite3.connect(db_path)
cur = conn.cursor()
tables = [r[0] for r in cur.execute(
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()]
results = []
for table in tables:
if table in known or table.startswith('sqlite_'):
continue
cols = [r[1] for r in cur.execute(
f"PRAGMA table_info([{table}])").fetchall()]
rows = cur.execute(f"SELECT * FROM [{table}]").fetchall()
if rows:
results.append({'table': table, 'columns': cols, 'rows': rows})
conn.close()
return results or None
def load_file(path):
"""自动检测文件类型并加载数据"""
data = read_pd_chart(path)
if data:
return data
data = read_table_1(path)
if data:
return data
return None
# ── 显示函数 ─────────────────────────────────────────────────
def fmt_duration(seconds):
m, s = divmod(int(seconds), 60)
h, m = divmod(m, 60)
if h > 0:
return f"{h}h{m:02d}m{s:02d}s"
if m > 0:
return f"{m}m{s:02d}s"
return f"{s}s"
def show_stats(data, filename):
"""显示数据统计信息"""
rows = data['rows']
if not rows:
print(" (无数据)")
return
print(f"\n{'='*70}")
print(f" 文件: {os.path.basename(filename)}")
print(f" 类型: {data['type']} | 数据点: {len(rows):,}")
print(f"{'='*70}")
if data['type'] == 'pd_chart':
t_idx, v_idx, i_idx = 0, 1, 2
else:
t_idx, v_idx, i_idx = 1, 2, 3
times = [r[t_idx] for r in rows]
vbus = [r[v_idx] for r in rows]
ibus = [r[i_idx] for r in rows]
power = [v * i for v, i in zip(vbus, ibus)]
duration = max(times) - min(times)
max_p = max(power)
max_p_idx = power.index(max_p)
avg_p = sum(power) / len(power)
max_v = max(vbus)
avg_v = sum(vbus) / len(vbus)
max_i = max(ibus)
avg_i = sum(ibus) / len(ibus)
print(f"\n ── 功率 ──────────────────────────────────")
print(f" 最大功率: {max_p:8.2f} W (@ {times[max_p_idx]:.1f}s)")
print(f" 平均功率: {avg_p:8.2f} W")
print(f"\n ── 电压 ──────────────────────────────────")
print(f" 最大电压: {max_v:8.3f} V")
print(f" 最小电压: {min(vbus):8.3f} V")
print(f" 平均电压: {avg_v:8.3f} V")
print(f"\n ── 电流 ──────────────────────────────────")
print(f" 最大电流: {max_i:8.3f} A")
print(f" 平均电流: {avg_i:8.3f} A")
print(f"\n ── 时间 ──────────────────────────────────")
print(f" 测试时长: {fmt_duration(duration)} ({duration:.1f}s)")
if data['type'] == 'table_1':
temps = [r[8] for r in rows]
charge = rows[-1][9]
energy = rows[-1][10]
print(f"\n ── 温度 ──────────────────────────────────")
print(f" 最高温度: {max(temps):8.1f} °C")
print(f" 最低温度: {min(temps):8.1f} °C")
print(f" 平均温度: {sum(temps)/len(temps):8.1f} °C")
print(f"\n ── 充电量 ────────────────────────────────")
print(f" 总电荷量: {charge*1000:8.2f} mAh")
print(f" 总能量: {energy*1000:8.2f} mWh")
if data.get('params'):
p = data['params']
print(f"\n ── 参数 ──────────────────────────────────")
print(f" 名称: {p[0]}")
print(f" 采样率: {p[3]}")
# 功率分段概览
print(f"\n ── 功率变化概览 ──────────────────────────")
n_segments = min(10, len(rows))
seg_size = len(rows) // n_segments
print(f" {'时间段':>10s} {'平均功率':>8s} {'最大功率':>8s} {'柱状图'}")
max_seg_p = 0
segments = []
for i in range(n_segments):
start = i * seg_size
end = (i + 1) * seg_size if i < n_segments - 1 else len(rows)
seg_power = power[start:end]
seg_avg = sum(seg_power) / len(seg_power)
seg_max = max(seg_power)
t_start = times[start]
t_end = times[end - 1]
segments.append((t_start, t_end, seg_avg, seg_max))
if seg_max > max_seg_p:
max_seg_p = seg_max
bar_width = 30
for t_start, t_end, seg_avg, seg_max in segments:
bar_len = int(seg_avg / max(max_seg_p, 0.01) * bar_width) if max_seg_p > 0 else 0
bar = '█' * bar_len + '░' * (bar_width - bar_len)
label = f"{t_start:6.0f}-{t_end:.0f}s"
print(f" {label:>12s} {seg_avg:7.1f}W {seg_max:7.1f}W {bar}")
print()
def show_data_preview(data, n=10):
"""显示前 N 条数据预览"""
rows = data['rows']
cols = data['columns'] + ['Power(W)']
if data['type'] == 'pd_chart':
v_idx, i_idx = 1, 2
else:
v_idx, i_idx = 2, 3
widths = [max(len(c), 10) for c in cols]
header = ' '.join(c.rjust(w) for c, w in zip(cols, widths))
print(f" {header}")
print(f" {'─' * len(header)}")
display = rows[:n]
for row in display:
power = row[v_idx] * row[i_idx]
vals = []
for v in row:
if isinstance(v, float):
vals.append(f"{v:.4f}")
else:
vals.append(str(v))
vals.append(f"{power:.2f}")
line = ' '.join(v.rjust(w) for v, w in zip(vals, widths))
print(f" {line}")
if len(rows) > n:
print(f" ... (共 {len(rows):,} 条)")
# ── PD 消息显示 ──────────────────────────────────────────────
def _find_col(cols_lower, *keywords):
"""查找包含关键词的列索引"""
for kw in keywords:
for i, c in enumerate(cols_lower):
if kw in c:
return i
return None
def show_extra_tables(extra_tables):
"""显示额外的表 (PD 消息等)"""
for mt in extra_tables:
table = mt['table']
cols = mt['columns']
rows = mt['rows']
cl = [c.lower() for c in cols]
print(f"\n ── {table} ({len(rows)} 行) ── 列: {', '.join(cols)} ──")
hdr_idx = _find_col(cl, 'header')
time_idx = _find_col(cl, 'time', 'timestamp')
sop_idx = _find_col(cl, 'sop')
type_idx = _find_col(cl, 'type', 'msgtype', 'msg_type')
if hdr_idx is not None:
_show_pd_msg_table(rows, cols, cl, time_idx, hdr_idx, sop_idx)
elif type_idx is not None:
_show_typed_msg_table(rows, cols, cl, time_idx, type_idx)
else:
_show_generic_table(rows, cols)
def _show_pd_msg_table(rows, cols, cl, time_idx, hdr_idx, sop_idx):
"""显示含 PD header 的消息表"""
skip = {time_idx, hdr_idx, sop_idx} - {None}
ignore_names = {'id', 'index', 'rowid', '_id', 'no', 'cc', 'dir', 'direction'}
do_indices = []
data_idx = None
for i, c in enumerate(cl):
if i in skip or c in ignore_names:
continue
if c in ('data', 'payload'):
data_idx = i
continue
do_indices.append(i)
sop_names = {0: 'SOP', 1: "SOP'", 2: 'SOP"', 3: "SOP'D", 4: 'SOP"D'}
type_counts = {}
for idx, row in enumerate(rows):
hdr = parse_pd_header(row[hdr_idx])
t_str = ''
if time_idx is not None:
tv = row[time_idx]
t_str = f'{tv:8.3f}s' if isinstance(tv, (int, float)) else str(tv)[:12]
sop_str = ''
if sop_idx is not None:
sv = _to_int(row[sop_idx])
if sv is not None:
sop_str = sop_names.get(sv, str(row[sop_idx])[:6])
else:
sop_str = str(row[sop_idx])[:6]
if hdr:
type_counts[hdr['name']] = type_counts.get(hdr['name'], 0) + 1
hdr_hex = f'0x{hdr["raw"]:04X}'
meta = f'PD{hdr["spec"]} {hdr["prole"]}/{hdr["drole"]}'
print(f' #{idx+1:<4d} {t_str:>10s} {sop_str:<6s}'
f' [{hdr["name"]:<20s}] {hdr_hex} {meta}')
if hdr['num_do'] > 0:
data_vals = []
if data_idx is not None and row[data_idx]:
data_vals = _parse_data_blob(row[data_idx])
if not data_vals and do_indices:
data_vals = [row[i] for i in do_indices
if row[i] is not None and row[i] != 0 and row[i] != '']
_print_data_objects(hdr['name'], data_vals)
else:
print(f' #{idx+1:<4d} {t_str:>10s} {sop_str:<6s}'
f' Header={row[hdr_idx]}')
_print_type_summary(type_counts)
def _show_typed_msg_table(rows, cols, cl, time_idx, type_idx):
"""显示含 type 列的消息表"""
skip = {time_idx, type_idx} - {None}
data_indices = [i for i in range(len(cols)) if i not in skip
and cl[i] not in ('id', 'index', 'rowid', '_id', 'no')]
type_counts = {}
for idx, row in enumerate(rows):
t_str = ''
if time_idx is not None:
tv = row[time_idx]
t_str = f'{tv:8.3f}s' if isinstance(tv, (int, float)) else str(tv)[:12]
type_name = str(row[type_idx])
type_counts[type_name] = type_counts.get(type_name, 0) + 1
extra = ''
if data_indices:
parts = []
for i in data_indices:
v = row[i]
if v is not None and v != '' and v != 0:
iv = _to_int(v)
if iv is not None and isinstance(iv, int) and iv > 0xFFFF:
parts.append(f'{cols[i]}=0x{iv:08X}')
else:
parts.append(f'{cols[i]}={v}')
if parts:
extra = ' ' + ', '.join(parts)
print(f' #{idx+1:<4d} {t_str:>10s} [{type_name:<20s}]{extra}')
_print_type_summary(type_counts)
def _print_data_objects(msg_name, data_vals):
"""打印 PD 数据对象"""
for j, dv in enumerate(data_vals):
iv = _to_int(dv)
hs = f'0x{iv:08X}' if iv is not None else str(dv)
if msg_name in ('Source_Cap', 'Sink_Cap'):
print(f' PDO{j+1}: {hs} -> {parse_pdo(dv)}')
elif msg_name == 'Request' and j == 0:
print(f' RDO: {hs} -> {parse_rdo(dv)}')
else:
print(f' DO{j+1}: {hs}')
def _print_type_summary(type_counts):
"""打印消息类型统计"""
if type_counts:
print(f'\n ── 消息类型统计 ──────────────────────────')
for name, cnt in sorted(type_counts.items(), key=lambda x: -x[1]):
print(f' {name:<24s} {cnt:>5d} 条')
def _show_generic_table(rows, cols):
"""通用表格显示"""
widths = [len(c) for c in cols]
for row in rows[:100]:
for i, v in enumerate(row):
s = f'{v:.4f}' if isinstance(v, float) else str(v)
widths[i] = max(widths[i], min(len(s), 24))
header = ' '.join(c.ljust(w) for c, w in zip(cols, widths))
print(f' {header}')
print(f' {"─" * len(header)}')
for row in rows:
vals = []
for v in row:
if isinstance(v, float):
vals.append(f'{v:.4f}')
elif isinstance(v, int) and abs(v) > 0xFFFF:
vals.append(f'0x{v:X}')
else:
vals.append(str(v) if v is not None else '')
line = ' '.join(s.ljust(w)[:w] for s, w in zip(vals, widths))
print(f' {line}')
# ── 数据库结构 & PD 事件分析 ──────────────────────────────────
def show_db_schema(db_path):
"""显示数据库所有表和列信息"""
conn = sqlite3.connect(db_path)
cur = conn.cursor()
tables = cur.execute(
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()
print(f"\n ── 数据库结构 ────────────────────────────")
for (table,) in tables:
if table.startswith('sqlite_'):
continue
cols = cur.execute(f"PRAGMA table_info([{table}])").fetchall()
col_desc = ', '.join(f'{c[1]}({c[2]})' for c in cols)
count = cur.execute(f"SELECT COUNT(*) FROM [{table}]").fetchone()[0]
print(f' {table}: {count} 行 [{col_desc}]')
conn.close()
def analyze_pd_events(data):
"""从 VBUS/IBUS/CC 波形数据分析 PD 协商事件"""
rows = data['rows']
if not rows or len(rows) < 10:
return
if data['type'] == 'pd_chart':
t_idx, v_idx, i_idx, cc1_idx, cc2_idx = 0, 1, 2, 3, 4
else:
t_idx, v_idx, i_idx = 1, 2, 3
cc1_idx, cc2_idx = 6, 7
# ── 活跃 CC 通道检测
cc1_avg = sum(r[cc1_idx] for r in rows) / len(rows)
cc2_avg = sum(r[cc2_idx] for r in rows) / len(rows)
active_cc = 'CC1' if cc1_avg > cc2_avg else 'CC2'
active_cc_v = max(cc1_avg, cc2_avg)
# ── 检测 VBUS 电压跳变 (PD 协商事件)
# 用滑动窗口平滑噪声
win = min(5, max(1, len(rows) // 20))
def avg_v(center, radius=win):
lo = max(0, center - radius)
hi = min(len(rows), center + radius + 1)
vals = [rows[j][v_idx] for j in range(lo, hi)]
return sum(vals) / len(vals)
events = []
cooldown = 0
for idx in range(win, len(rows) - win):
if cooldown > 0:
cooldown -= 1
continue
v_before = avg_v(idx - win)
v_after = avg_v(idx + win)
dv = v_after - v_before
if abs(dv) < 1.0:
continue
t = rows[idx][t_idx]
i_val = rows[idx][i_idx]
from_pdv = _nearest_pd_voltage(v_before)
to_pdv = _nearest_pd_voltage(v_after)
from_s = f'{from_pdv}V' if from_pdv else f'{v_before:.1f}V'
to_s = f'{to_pdv}V' if to_pdv else f'{v_after:.1f}V'
if v_after < 0.5:
etype = '断开'
msgs = 'Hard_Reset / 拔出'
elif dv > 0:
if from_pdv == 5:
etype = 'PD 协商成功'
msgs = 'Source_Cap → Request → Accept → PS_RDY'
else:
etype = '电压切换'
msgs = 'Request → Accept → PS_RDY'
else:
if to_pdv == 5 and (from_pdv and from_pdv > 5):
etype = '重置/重新协商'
msgs = 'Soft_Reset → Source_Cap → ...'
elif v_after < 0.5:
etype = '断开'
msgs = 'Hard_Reset'
else:
etype = '电压下降'
msgs = 'Request → Accept → PS_RDY'
# 计算跳变后稳定功率
stable_start = min(idx + win * 2, len(rows) - 1)
stable_end = min(stable_start + win * 4, len(rows))
if stable_end > stable_start:
stable_v = sum(rows[j][v_idx] for j in range(stable_start, stable_end)) / (stable_end - stable_start)
stable_i = sum(rows[j][i_idx] for j in range(stable_start, stable_end)) / (stable_end - stable_start)
else:
stable_v, stable_i = v_after, i_val
events.append({
'time': t, 'type': etype,
'from': from_s, 'to': to_s,
'voltage': stable_v, 'current': stable_i,
'power': stable_v * stable_i, 'msgs': msgs,
})
cooldown = win * 4
# ── 输出
print(f"\n ── PD 协商事件 ──────────────────────────")
print(f" 活跃通道: {active_cc} (平均 {active_cc_v:.2f}V)")
if not events:
avg_vbus = sum(r[v_idx] for r in rows) / len(rows)
pdv = _nearest_pd_voltage(avg_vbus)
v_str = f'{pdv}V' if pdv else f'{avg_vbus:.1f}V'
avg_ibus = sum(r[i_idx] for r in rows) / len(rows)
print(f' VBUS 全程稳定在 {v_str} ({avg_ibus:.2f}A, {avg_vbus*avg_ibus:.1f}W)')
print(f' 未检测到 PD 电压协商变化')
return
for e in events:
print(f'\n {e["time"]:8.2f}s [{e["type"]}]')
print(f' {e["from"]} → {e["to"]} '
f'({e["voltage"]:.1f}V {e["current"]:.2f}A {e["power"]:.1f}W)')
print(f' 推测: {e["msgs"]}')
# 最终稳定状态
last_50 = rows[-min(50, len(rows)):]
final_v = sum(r[v_idx] for r in last_50) / len(last_50)
final_i = sum(r[i_idx] for r in last_50) / len(last_50)
final_pdv = _nearest_pd_voltage(final_v)
final_v_str = f'{final_pdv}V' if final_pdv else f'{final_v:.1f}V'
print(f'\n 最终状态: {final_v_str} {final_i:.2f}A ({final_v*final_i:.1f}W)')
# ── 翻页输出 ─────────────────────────────────────────────────
def page_output(text):
"""内容超过终端高度时使用 less 翻页"""
try:
term_h = os.get_terminal_size().lines
except OSError:
term_h = 40
if text.count('\n') <= term_h - 2:
sys.stdout.write(text)
return
pager = shutil.which('less')
if pager:
try:
proc = subprocess.Popen([pager, '-R'], stdin=subprocess.PIPE)
proc.communicate(input=text.encode('utf-8'))
except (BrokenPipeError, KeyboardInterrupt):
pass
else:
sys.stdout.write(text)
# ── 主程序 ───────────────────────────────────────────────────
def main():
no_page = '--no-page' in sys.argv
args = [a for a in sys.argv[1:] if a != '--no-page']
targets = args if args else ['.']
files = []
for target in targets:
if os.path.isdir(target):
files.extend(glob.glob(os.path.join(target, '*.sqlite')))
files.extend(glob.glob(os.path.join(target, '*.db')))
elif os.path.isfile(target):
files.append(target)
else:
print(f" 跳过: {target} (不存在)")
if not files:
print("未找到 .sqlite 或 .db 文件")
print(f"用法: {sys.argv[0]} [--no-page] [目录或文件路径 ...]")
sys.exit(1)
files.sort()
buf = io.StringIO()
old_stdout = sys.stdout
sys.stdout = buf
try:
print(f"\n找到 {len(files)} 个数据文件")
for f in files:
data = load_file(f)
extra = read_extra_tables(f)
if data:
show_stats(data, f)
analyze_pd_events(data)
show_data_preview(data)
elif not extra:
print(f"\n 跳过: {os.path.basename(f)} (无法识别的表结构)")
if extra:
if not data:
print(f"\n{'='*70}")
print(f" 文件: {os.path.basename(f)}")
print(f"{'='*70}")
show_extra_tables(extra)
show_db_schema(f)
finally:
sys.stdout = old_stdout
text = buf.getvalue()
if no_page or not sys.stdout.isatty():
sys.stdout.write(text)
else:
page_output(text)
if __name__ == '__main__':
main()

34
CP02/extract_firmware.sh
View File

@ -1,5 +1,5 @@
#!/bin/bash
# 用法: ./extract_firmware.sh <merged_bin> [output_prefix]
# 用法: ./extract_firmware.sh <merged_bin> [output_dir]
# 从合并的 bin 文件中提取 head file (firmware), tail file (IUM), 以及生成 metadata.json
#
# 文件布局:
@ -8,26 +8,17 @@
# 0x1dd00: metadata (blocks: uint16, last_block_size: uint8)
INPUT="$1"
# 如果没有传 PREFIX，使用原文件名（去掉扩展名）
if [ -n "$2" ]; then
PREFIX="$2"
else
# 获取文件名（不含路径），然后去掉扩展名
BASENAME=$(basename "$INPUT")
PREFIX="${BASENAME%.*}"
fi
OUTPUT_DIR="${2:-.}"
if [ -z "$INPUT" ]; then
echo "用法: $0 <merged_bin> [output_prefix]"
echo "示例: $0 40_ufcs2.bin firmware"
echo " $0 40_ufcs2.bin # 使用原文件名作为前缀"
echo "用法: $0 <merged_bin> [output_dir]"
echo "示例: $0 40_ufcs2.bin /tmp/output"
echo " $0 40_ufcs2.bin # 输出到当前目录"
echo ""
echo "输出文件:"
echo " {prefix}.firmware - firmware 部分"
echo " {prefix}.ium - IUM 部分 (256 bytes)"
echo " {prefix}_metadata.json - 元数据"
echo ""
echo "注: 如果不指定 output_prefix，将使用原文件名（去掉扩展名）"
echo " firmware.bin - firmware 部分"
echo " ium.bin - IUM 部分 (256 bytes)"
echo " metadata.json - 元数据"
exit 1
fi
@ -36,6 +27,9 @@ if [ ! -f "$INPUT" ]; then
exit 1
fi
# 创建输出目录（如果不存在）
mkdir -p "$OUTPUT_DIR"
# 常量定义
FIRMWARE_OFFSET=$((0x4000)) # 16384
IUM_OFFSET=$((0x1dc00)) # 121856
@ -79,7 +73,7 @@ echo " 计算出 firmware 大小: $FIRMWARE_SIZE bytes"
echo ""
# 提取 head file (firmware: 从 0x4000 开始)
HEAD_FILE="firmware.bin"
HEAD_FILE="$OUTPUT_DIR/firmware.bin"
echo "提取 head file (firmware)..."
dd if="$INPUT" of="$HEAD_FILE" bs=1 skip=$FIRMWARE_OFFSET count=$FIRMWARE_SIZE status=none
@ -92,7 +86,7 @@ else
fi
# 提取 tail file (IUM: 从 0x1dc00 开始, 256 bytes)
TAIL_FILE="ium.bin"
TAIL_FILE="$OUTPUT_DIR/ium.bin"
echo "提取 tail file (IUM)..."
dd if="$INPUT" of="$TAIL_FILE" bs=1 skip=$IUM_OFFSET count=$IUM_SIZE status=none
@ -105,7 +99,7 @@ else
fi
# 生成 metadata.json
METADATA_FILE="metadata.json"
METADATA_FILE="$OUTPUT_DIR/metadata.json"
echo "生成 metadata.json..."
cat > "$METADATA_FILE" << EOF

38
CP02/flash_guide.md Normal file
View File

@ -0,0 +1,38 @@
# IonBridge 烧录指南
所有命令在 IonBridge 项目根目录下执行（`cd /Users/ching/develop/IonBridge`）。
## 非 Secure 设备（开发板）
```bash
# Build + Flash firmware + littlefs
make build variant=cp02
make flash littlefs=1 PORT=/dev/tty.usbmodem2101
# 只烧 firmware
make build variant=cp02
make flash PORT=/dev/tty.usbmodem2101
# 查看日志
make monitor PORT=/dev/tty.usbmodem2101
```
## Secure 设备（量产板，Secure Boot + Flash Encryption）
```bash
# Build + Flash firmware + littlefs
make build variant=cp02 production=1
make flash littlefs=1 sign_key=/Users/ching/Downloads/1331040606001272_sign_key.pem PORT=/dev/tty.usbmodem2101
# 只烧 littlefs（不更新 firmware，不需要重新 build）
make flash_littlefs_partition secure=1 PORT=/dev/tty.usbmodem2101
# 查看日志
make monitor PORT=/dev/tty.usbmodem2101
```
## 备注
- `sign_key` 可用 `psn` 代替：`psn=1331040606001272`（自动查找 `sign_keys/` 下的 key）
- Secure 设备用 `production=1` 构建后，`secure=1` 会自动检测
- `PORT` 根据实际设备端口修改

307
CP02/powerz_pd_decode.py Normal file
View File

@ -0,0 +1,307 @@
#!/usr/bin/env python3
import argparse
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Any
CTRL_MESSAGES = {
1: "GoodCRC",
2: "GotoMin",
3: "Accept",
4: "Reject",
5: "Ping",
6: "PS_RDY",
7: "Get_Source_Cap",
8: "Get_Sink_Cap",
9: "DR_Swap",
10: "PR_Swap",
11: "VCONN_Swap",
12: "Wait",
13: "Soft_Reset",
16: "Not_Supported",
17: "Get_Source_Cap_Ext",
18: "Get_Status",
19: "FR_Swap",
20: "Get_PPS_Status",
21: "Get_Country_Codes",
22: "Get_Sink_Cap_Ext",
23: "Get_Source_Info",
24: "Get_Revision",
}
DATA_MESSAGES = {
1: "Source_Capabilities",
2: "Request",
3: "BIST",
4: "Sink_Capabilities",
5: "Battery_Status",
6: "Alert",
7: "Get_Country_Info",
8: "Enter_USB",
9: "EPR_Request",
10: "EPR_Mode",
11: "Source_Info",
12: "Revision",
15: "Vendor_Defined",
}
EXT_MESSAGES = {
1: "Source_Cap_Ext",
2: "Status",
3: "Get_Battery_Cap",
4: "Get_Battery_Status",
5: "Battery_Capabilities",
6: "Get_Mfr_Info",
7: "Mfr_Info",
8: "Security_Request",
9: "Security_Response",
10: "FW_Update_Request",
11: "FW_Update_Response",
12: "PPS_Status",
13: "Country_Info",
14: "Country_Codes",
15: "Sink_Cap_Ext",
16: "Extended_Control",
17: "EPR_Source_Cap",
18: "EPR_Sink_Cap",
30: "Vendor_Defined_Ext",
}
@dataclass
class Message:
time: float
vbus: float
ibus: float
raw: bytes
pd: bytes
header: dict[str, int]
cls: str
name: str
decoded: Any = None
def parse_header(halfword: int) -> dict[str, int]:
return {
"type": halfword & 0x1F,
"rev": (halfword >> 6) & 0x3,
"role": (halfword >> 8) & 0x1,
"msgid": (halfword >> 9) & 0x7,
"nobj": (halfword >> 12) & 0x7,
"ext": (halfword >> 15) & 0x1,
}
def classify_message(header: dict[str, int]) -> tuple[str, str]:
if header["ext"]:
return "ext", EXT_MESSAGES.get(header["type"], f"EXT_{header['type']}")
if header["nobj"] == 0:
return "ctrl", CTRL_MESSAGES.get(header["type"], f"CTRL_{header['type']}")
return "data", DATA_MESSAGES.get(header["type"], f"DATA_{header['type']}")
def parse_pdo(word: int) -> dict[str, Any]:
supply_type = (word >> 30) & 0x3
if supply_type == 0:
return {
"kind": "fixed",
"mv": ((word >> 10) & 0x3FF) * 50,
"ma": (word & 0x3FF) * 10,
}
if supply_type == 3:
subtype = (word >> 28) & 0xF
if subtype == 0xC:
return {
"kind": "pps",
"min_mv": ((word >> 8) & 0xFF) * 100,
"max_mv": ((word >> 17) & 0xFF) * 100,
"ma": (word & 0x7F) * 50,
}
if subtype in (0xD, 0xE):
return {
"kind": "avs",
"min_mv": ((word >> 8) & 0xFF) * 100,
"max_mv": ((word >> 17) & 0x1FF) * 100,
"pdp_w": word & 0x7F,
}
return {"kind": "other", "raw": f"0x{word:08X}"}
def parse_fixed_rdo(word: int) -> dict[str, Any]:
return {
"kind": "fixed",
"obj_pos": (word >> 28) & 0xF,
"op_ma": (word & 0x3FF) * 10,
"max_ma": ((word >> 10) & 0x3FF) * 10,
"mismatch": (word >> 26) & 0x1,
"epr": (word >> 22) & 0x1,
}
def parse_pps_rdo(word: int) -> dict[str, Any]:
return {
"kind": "pps",
"obj_pos": (word >> 28) & 0xF,
"op_ma": (word & 0x7F) * 50,
"out_mv": ((word >> 9) & 0xFFF) * 20,
"epr": (word >> 22) & 0x1,
}
def decode_message(time: float, vbus: float, ibus: float, raw: bytes) -> Message | None:
if len(raw) < 8:
return None
# PowerZ 的 Raw 前 6 字节是记录头，后面是标准 PD message header + payload。
pd = raw[6:]
if len(pd) < 2:
return None
header = parse_header(int.from_bytes(pd[:2], "little"))
cls, name = classify_message(header)
return Message(time, vbus, ibus, raw, pd, header, cls, name)
def load_messages(db_path: Path) -> list[Message]:
conn = sqlite3.connect(str(db_path))
rows = conn.execute(
"select Time, Vbus, Ibus, Raw from pd_table "
"where Raw is not null and length(Raw) > 0 order by Time"
).fetchall()
conn.close()
messages: list[Message] = []
for time, vbus, ibus, raw in rows:
msg = decode_message(time, vbus, ibus, raw if isinstance(raw, bytes) else bytes(raw))
if msg is not None:
messages.append(msg)
return messages
def decode_source_caps(messages: list[Message]) -> list[Message]:
caps: list[Message] = []
for msg in messages:
if msg.name != "Source_Capabilities":
continue
objs = []
for idx in range(msg.header["nobj"]):
offset = 2 + idx * 4
if offset + 4 > len(msg.pd):
break
objs.append(parse_pdo(int.from_bytes(msg.pd[offset : offset + 4], "little")))
msg.decoded = objs
caps.append(msg)
return caps
def decode_requests(messages: list[Message], source_caps: list[Message]) -> list[Message]:
requests: list[Message] = []
for msg in messages:
if msg.name not in ("Request", "EPR_Request") or len(msg.pd) < 6:
continue
word = int.from_bytes(msg.pd[2:6], "little")
obj_pos = (word >> 28) & 0xF
selected = None
for cap_msg in reversed(source_caps):
if cap_msg.time <= msg.time and 0 < obj_pos <= len(cap_msg.decoded):
selected = cap_msg.decoded[obj_pos - 1]
break
if selected and selected.get("kind") in ("pps", "avs"):
decoded = parse_pps_rdo(word)
else:
decoded = parse_fixed_rdo(word)
decoded["selected"] = selected
msg.decoded = decoded
requests.append(msg)
return requests
def format_pdo(pdo: dict[str, Any]) -> str:
if pdo["kind"] == "fixed":
watts = pdo["mv"] * pdo["ma"] / 1_000_000
return f"fixed {pdo['mv'] / 1000:.1f}V {pdo['ma'] / 1000:.3f}A ({watts:.1f}W)"
if pdo["kind"] == "pps":
return (
f"PPS {pdo['min_mv'] / 1000:.1f}-{pdo['max_mv'] / 1000:.1f}V "
f"{pdo['ma'] / 1000:.3f}A"
)
if pdo["kind"] == "avs":
return (
f"AVS {pdo['min_mv'] / 1000:.1f}-{pdo['max_mv'] / 1000:.1f}V "
f"{pdo['pdp_w']}W"
)
return str(pdo)
def format_request(req: dict[str, Any]) -> str:
selected = req.get("selected")
suffix = f", src={format_pdo(selected)}" if selected else ""
if req["kind"] == "pps":
return (
f"obj={req['obj_pos']} req={req['out_mv'] / 1000:.3f}V "
f"{req['op_ma'] / 1000:.3f}A{suffix}"
)
return (
f"obj={req['obj_pos']} op={req['op_ma'] / 1000:.3f}A "
f"max={req['max_ma'] / 1000:.3f}A{suffix}"
)
def print_summary(messages: list[Message], source_caps: list[Message], requests: list[Message]) -> None:
print(f"Decoded PD messages: {len(messages)}")
print()
print("Source Capabilities:")
for msg in source_caps:
print(f"{msg.time:8.3f}s VBUS={msg.vbus:6.3f}V IBUS={msg.ibus:5.3f}A")
for idx, pdo in enumerate(msg.decoded, start=1):
print(f" PDO{idx}: {format_pdo(pdo)}")
print()
print("Requests:")
for msg in requests:
print(
f"{msg.time:8.3f}s VBUS={msg.vbus:6.3f}V IBUS={msg.ibus:5.3f}A "
f"{msg.name}: {format_request(msg.decoded)}"
)
def print_window(messages: list[Message], start: float, end: float) -> None:
print()
print(f"Messages in {start:.3f}s - {end:.3f}s:")
for msg in messages:
if start <= msg.time <= end and msg.name != "GoodCRC":
print(
f"{msg.time:8.3f}s {msg.name:18} "
f"VBUS={msg.vbus:6.3f}V IBUS={msg.ibus:5.3f}A "
f"nobj={msg.header['nobj']} pd={msg.pd.hex()}"
)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Decode PD messages from a PowerZ sqlite capture."
)
parser.add_argument("sqlite", type=Path, help="Path to PowerZ .sqlite file")
parser.add_argument(
"--window",
nargs=2,
type=float,
metavar=("START", "END"),
help="Print decoded messages in a time window",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
messages = load_messages(args.sqlite)
source_caps = decode_source_caps(messages)
requests = decode_requests(messages, source_caps)
print_summary(messages, source_caps, requests)
if args.window:
print_window(messages, args.window[0], args.window[1])
return 0
if __name__ == "__main__":
raise SystemExit(main())

152
CP02/remove_chinese_comments.py Executable file
View File

@ -0,0 +1,152 @@
#!/usr/bin/env python3
"""Remove Chinese comments (// and # style) from source files."""
from __future__ import annotations
import re
import sys
from pathlib import Path
# File extensions and their comment patterns
COMMENT_PATTERNS = {
# C/C++ style
'.c': r'//.*',
'.cpp': r'//.*',
'.h': r'//.*',
'.hpp': r'//.*',
# Python style
'.py': r'#.*',
}
def get_comment_pattern(filepath: str) -> str:
"""Get comment pattern based on file extension."""
ext = Path(filepath).suffix.lower()
return COMMENT_PATTERNS.get(ext, r'//.*')
def get_comment_prefix(filepath: str) -> str:
"""Get comment prefix based on file extension."""
ext = Path(filepath).suffix.lower()
if ext == '.py':
return '#'
return '//'
def contains_chinese(text: str) -> bool:
"""Check if text contains Chinese characters."""
return bool(re.search(r'[\u4e00-\u9fff]', text))
def find_chinese_comments(filepath: str):
"""Find all lines with Chinese comments.
Returns list of (line_number, line_content) tuples.
"""
results = []
path = Path(filepath)
if not path.exists():
print(f"Error: File not found: {filepath}")
sys.exit(1)
pattern = get_comment_pattern(filepath)
with open(path, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
match = re.search(pattern, line)
if match and contains_chinese(match.group()):
results.append((line_num, line.rstrip()))
return results
def remove_chinese_comments(filepath: str) -> str:
"""Remove Chinese comments from file content.
Returns the modified content.
"""
path = Path(filepath)
pattern = get_comment_pattern(filepath)
prefix = get_comment_prefix(filepath)
with open(path, 'r', encoding='utf-8') as f:
lines = f.readlines()
new_lines = []
for line in lines:
match = re.search(pattern, line)
if match and contains_chinese(match.group()):
stripped = line.lstrip()
if stripped.startswith(prefix):
# Entire line is a comment - skip it
continue
else:
# Comment is at end of code - remove just the comment
comment_start = match.start()
new_line = line[:comment_start].rstrip() + '\n'
new_lines.append(new_line)
else:
new_lines.append(line)
return ''.join(new_lines)
def main():
if len(sys.argv) < 2:
print("Usage: python remove_chinese_comments.py <file> [file2 ...]")
print(" python remove_chinese_comments.py --check <file> [file2 ...]")
print("\nOptions:")
print(" --check Only show Chinese comments without removing")
sys.exit(1)
check_only = False
files = sys.argv[1:]
if files[0] == '--check':
check_only = True
files = files[1:]
if not files:
print("Error: No files specified")
sys.exit(1)
total_found = 0
for filepath in files:
comments = find_chinese_comments(filepath)
if not comments:
print(f"\n{filepath}: No Chinese comments found")
continue
total_found += len(comments)
print(f"\n{'='*60}")
print(f"File: {filepath}")
print(f"Found {len(comments)} Chinese comment(s):")
print('-'*60)
for line_num, line in comments:
print(f" L{line_num}: {line}")
if check_only:
continue
# Ask for confirmation
print('-'*60)
response = input(f"Remove these {len(comments)} comment(s)? [y/N]: ").strip().lower()
if response == 'y':
new_content = remove_chinese_comments(filepath)
with open(filepath, 'w', encoding='utf-8') as f:
f.write(new_content)
print(f"✓ Removed {len(comments)} Chinese comment(s) from {filepath}")
else:
print(f"✗ Skipped {filepath}")
print(f"\n{'='*60}")
print(f"Total: {total_found} Chinese comment(s) found")
if __name__ == '__main__':
main()

1
CP02/split_and_merge.sh
View File

@ -79,6 +79,7 @@ if ./merge_2323_firmware.sh -u "$updater" -f "$head_file" -i "$tail_file" -o "$o
# 清理临时文件
rm -f "$head_file" "$tail_file"
echo "⚠️⚠️⚠️ Remember signing output file! ⚠️⚠️⚠️"
echo "python signer.py --output_dir output --firmware /tmp/40_xiaomi_0403.bin"
else
echo "合并失败，保留临时文件以便排查"
fi