118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
# RP2040 注入器嗅探数据 -> I2C 协议解码
|
|
# 用法: python i2c_sniff_decode.py [COM口] [采集ms, 默认700]
|
|
# 帧格式: 固件 'c' 命令回吐 "C <t_us> <val>" 行, val bit0=SDA bit1=SCL
|
|
import serial, sys, time
|
|
|
|
PORT = sys.argv[1] if len(sys.argv) > 1 else 'COM9'
|
|
MS = int(sys.argv[2]) if len(sys.argv) > 2 else 700
|
|
|
|
KNOWN_REGS = {0x44: 'SYS_STATUS', 0x6E: 'TTL_8B_CFG', 0xFF: 'BANK'}
|
|
|
|
s = serial.Serial(PORT, timeout=1)
|
|
time.sleep(0.3)
|
|
s.reset_input_buffer()
|
|
s.write(b's\n')
|
|
time.sleep(0.3)
|
|
print(s.read(200).decode('utf-8', 'replace').strip())
|
|
s.write(('c %d\n' % MS).encode())
|
|
|
|
edges = []
|
|
t_deadline = time.time() + MS / 1000 + 60
|
|
while time.time() < t_deadline:
|
|
ln = s.readline().decode('ascii', 'replace').strip()
|
|
if ln == 'C.':
|
|
break
|
|
if ln.startswith('C '):
|
|
try:
|
|
a, b = ln[2:].split()
|
|
edges.append((int(a), int(b)))
|
|
except ValueError:
|
|
pass
|
|
elif ln.startswith('='):
|
|
print(ln)
|
|
s.close()
|
|
print('edges:', len(edges))
|
|
|
|
# ---------------- I2C 解码 ----------------
|
|
trans, cur, bits = [], None, []
|
|
psda = pscl = None
|
|
|
|
def flush_byte():
|
|
global bits
|
|
if cur is not None and len(bits) == 8:
|
|
cur['bytes'].append(int(''.join(str(x) for x in bits), 2))
|
|
bits = []
|
|
|
|
for t, v in edges:
|
|
sda, scl = v & 1, (v >> 1) & 1
|
|
if pscl is not None:
|
|
if scl == 1 and pscl == 1: # SCL 高电平期间 SDA 变化
|
|
if psda == 1 and sda == 0: # START / repeated START
|
|
if cur is not None:
|
|
cur['t1'] = t
|
|
cur['inc'] = bool(bits)
|
|
trans.append(cur)
|
|
cur = {'t0': t, 'bytes': [], 'acks': [], 'inc': False}
|
|
bits = []
|
|
elif psda == 0 and sda == 1: # STOP
|
|
if cur is not None:
|
|
cur['t1'] = t
|
|
cur['inc'] = bool(bits)
|
|
trans.append(cur)
|
|
cur = None
|
|
bits = []
|
|
elif scl == 1 and pscl == 0: # SCL 上升沿采样
|
|
if cur is not None:
|
|
bits.append(sda)
|
|
if len(bits) == 8:
|
|
cur['bytes'].append(int(''.join(str(x) for x in bits), 2))
|
|
bits = []
|
|
elif len(bits) == 9:
|
|
cur['acks'].append(sda) # 0=ACK
|
|
bits = []
|
|
psda, pscl = sda, scl
|
|
if cur is not None:
|
|
cur['t1'] = edges[-1][0] if edges else 0
|
|
cur['inc'] = bool(bits)
|
|
trans.append(cur)
|
|
|
|
# ---------------- 打印 ----------------
|
|
bank = None
|
|
print(f'\n解码出 {len(trans)} 个事务:')
|
|
for k, tr in enumerate(trans):
|
|
d = tr['bytes']
|
|
if not d:
|
|
print(' [%2d] t=%dus <空/片段>' % (k, tr['t0']))
|
|
continue
|
|
addr7, rw = d[0] >> 1, 'R' if d[0] & 1 else 'W'
|
|
acks = tr['acks']
|
|
body = d[1:]
|
|
ann = []
|
|
if addr7 == 0x32 and rw == 'W' and body:
|
|
if body[0] == 0xFF and len(body) >= 2:
|
|
bank = body[1]
|
|
ann.append('bank=%02X' % bank)
|
|
body = body[2:]
|
|
if body:
|
|
r = body[0]
|
|
note = KNOWN_REGS.get(r, '')
|
|
if r == 0xFF:
|
|
bank = body[1] if len(body) > 1 else bank
|
|
note = 'bank=%02X' % bank if len(body) > 1 else note
|
|
ann.append(('[%02X%02X] %s' % (bank or 0, r, note)) if note else '[%02X?%02X]' % (bank or 0, r))
|
|
ackstr = ''.join('A' if a == 0 else 'N' for a in acks[1:]) if len(acks) > 1 else ''
|
|
flag = ' <未收尾>' if tr.get('inc') else ''
|
|
print(' [%2d] t=%7dus 0x%02X %s: %-30s ACK:%s%s'
|
|
% (k, tr['t0'], addr7, rw,
|
|
' '.join('%02X' % x for x in d[1:12]) + (' …' if len(d) > 12 else ''),
|
|
ackstr[:16], flag))
|
|
|
|
# 轮询周期统计(相邻事务起始间隔的主值)
|
|
if len(trans) > 3:
|
|
gaps = [b['t0'] - a['t0'] for a, b in zip(trans, trans[1:])]
|
|
gaps = [g for g in gaps if 1000 < g < 5000000]
|
|
if gaps:
|
|
gaps.sort()
|
|
print('\n相邻事务间隔(us): 中位=%d 最小=%d 最大=%d'
|
|
% (gaps[len(gaps) // 2], gaps[0], gaps[-1]))
|