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

feat(cp02): add PD decode tools and enhance build GUI

Browse Source 
- 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
This commit is contained in:
Ching L 2026-04-03 10:50:23 +08:00
parent 6ad3ab93a7
commit 8651878be8
5 changed files with 1147 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
CP02/README.md
View File

@ -9,6 +9,7 @@
- 将输入固件文件切分为前面部分和最后256字节 - 将输入固件文件切分为前面部分和最后256字节
- 调用合并脚本生成最终固件 - 调用合并脚本生成最终固件
- 自动删除最后1字节并检查文件大小限制 - 自动删除最后1字节并检查文件大小限制
- 合并完成后提示签名命令
### merge_2323_firmware.sh ### merge_2323_firmware.sh
- **功能**: 合并 PA768 updater、2323固件和IUM数据为单一固件文件 - **功能**: 合并 PA768 updater、2323固件和IUM数据为单一固件文件
@ -109,6 +110,27 @@
- `--check` 模式仅显示中文注释，不做修改 - `--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 ### build_gui.html
- **功能**: IonBridge 构建工具的可视化命令生成器 - **功能**: IonBridge 构建工具的可视化命令生成器
- **用法**: 在浏览器中打开 `build_gui.html` - **用法**: 在浏览器中打开 `build_gui.html`
@ -117,6 +139,7 @@
- 可配置 build 选项（production、coredump）和 flash 选项（littlefs、protected_data、nvs 等） - 可配置 build 选项（production、coredump）和 flash 选项（littlefs、protected_data、nvs 等）
- 一键生成 Build、Flash、Monitor、Clean 等命令并复制 - 一键生成 Build、Flash、Monitor、Clean 等命令并复制
- 自动保存配置和命令历史到 localStorage - 自动保存配置和命令历史到 localStorage
- 显示当前 variant 和模式对应的 sdkconfig 文件路径
### flash_guide.md ### flash_guide.md
- **功能**: IonBridge 烧录操作参考文档 - **功能**: IonBridge 烧录操作参考文档
@ -175,6 +198,15 @@ 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 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 文件 # 复制 .o 文件
./copy_o_files.sh /path/to/build /path/to/output ./copy_o_files.sh /path/to/build /path/to/output

32
CP02/build_gui.html
View File

@ -242,6 +242,15 @@ select:focus, input:focus { border-color: var(--accent); }
border-color: var(--accent); border-color: var(--accent);
color: var(--text); 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> </style>
</head> </head>
<body> <body>
@ -263,6 +272,7 @@ select:focus, input:focus { border-color: var(--accent); }
<option value="gpu_demo">GPU Demo</option> <option value="gpu_demo">GPU Demo</option>
<option value="bird">Bird</option> <option value="bird">Bird</option>
</select> </select>
<div id="sdkconfigInfo" class="sdkconfig-info"></div>
<label class="field-label">Serial Port</label> <label class="field-label">Serial Port</label>
<input type="text" id="port" value="/dev/tty.usbmodem2101"> <input type="text" id="port" value="/dev/tty.usbmodem2101">
</div> </div>
@ -390,6 +400,27 @@ document.addEventListener('change', (e) => {
if (e.target.matches('select')) saveState(); 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() { function buildCmd() {
let cmd = 'make build variant=' + val('variant'); let cmd = 'make build variant=' + val('variant');
if (checked('opt-production')) cmd += ' production=1'; if (checked('opt-production')) cmd += ' production=1';
@ -541,6 +572,7 @@ async function copyFromHistory(el, text) {
} }
loadState(); loadState();
updateSdkconfigInfo();
</script> </script>
</body> </body>
</html> </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()

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())

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" rm -f "$head_file" "$tail_file"
echo "⚠️⚠️⚠️ Remember signing output file! ⚠️⚠️⚠️" echo "⚠️⚠️⚠️ Remember signing output file! ⚠️⚠️⚠️"
echo "python signer.py --output_dir output --firmware /tmp/40_xiaomi_0403.bin"
else else
echo "合并失败，保留临时文件以便排查" echo "合并失败，保留临时文件以便排查"
fi fi