334 lines
13 KiB
Python
334 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
i2cdec.py — I2C 领域模型/解码(纯标准库)
|
|
==================================================
|
|
本模块是 I2C-Lab 三方共用的"协议语法"权威:
|
|
- 固件监控流 T 行(事务)与 W 行(边沿快照)的解析(与 firmware/main.c 语法一致);
|
|
- 边沿序列 -> 事务解码 decode_edges()(算法自 I2C-Inject i2c_sniff_decode.py,
|
|
在真机总线上验证过, 引脚编码沿用: val bit0=SDA bit1=SCL);
|
|
- 事务 -> 边沿序列的参考编码器 encode_edges()(自测向量生成, 也规定了对齐时序);
|
|
- 事务显示行/CSV 导出。
|
|
|
|
时间单位一律 us(int)。地址为 7bit。
|
|
"""
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 领域模型
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class Transaction:
|
|
"""一次 I2C 事务: START .. (STOP | repeated START | 流被截断)
|
|
|
|
字段:
|
|
t0_us / t1_us START 边沿时刻 / 收尾边沿时刻(STOP/rep-START/最后边沿)
|
|
addr7 7bit 地址
|
|
read True=读(R), False=写(W)
|
|
acks 每字节一个应答, True=ACK(SDA=0), [0] 是地址字节应答
|
|
data 地址字节之后的数据字节(bytes)
|
|
ended 收尾方式: 'stop' / 'rstart' / 'cut'(流截断) / 'partial'(字节中途断)
|
|
truncated 固件长事务截断(data/acks 只含前段, 尾部丢弃)
|
|
"""
|
|
|
|
__slots__ = ('t0_us', 't1_us', 'addr7', 'read', 'acks', 'data', 'ended',
|
|
'truncated')
|
|
|
|
def __init__(self, t0_us=0, addr7=0, read=False, acks=(), data=b'',
|
|
ended='stop', t1_us=None, truncated=False):
|
|
self.t0_us = int(t0_us)
|
|
self.t1_us = int(t0_us if t1_us is None else t1_us)
|
|
self.addr7 = int(addr7)
|
|
self.read = bool(read)
|
|
self.acks = tuple(bool(a) for a in acks)
|
|
self.data = bytes(data)
|
|
self.ended = ended
|
|
self.truncated = bool(truncated)
|
|
|
|
# --- 便捷属性 ---
|
|
|
|
@property
|
|
def addr_ack(self):
|
|
return self.acks[0] if self.acks else None
|
|
|
|
@property
|
|
def data_acks(self):
|
|
return self.acks[1:]
|
|
|
|
@property
|
|
def complete(self):
|
|
return self.ended == 'stop'
|
|
|
|
@property
|
|
def dur_us(self):
|
|
return self.t1_us - self.t0_us
|
|
|
|
def key(self):
|
|
"""解码自测用等值比较(不含时间戳的不变内容)"""
|
|
return (self.addr7, self.read, self.acks, self.data, self.ended)
|
|
|
|
# --- 显示 ---
|
|
|
|
def addr_hex(self):
|
|
return f'0x{self.addr7:02X}'
|
|
|
|
def rw_char(self):
|
|
return 'R' if self.read else 'W'
|
|
|
|
def ack_str(self):
|
|
return ''.join('A' if a else 'N' for a in self.acks) or '-'
|
|
|
|
def flag_str(self):
|
|
return {'stop': '', 'rstart': '+', 'cut': ' <未收尾>',
|
|
'partial': ' <半字节断>'}.get(self.ended, '')
|
|
|
|
def data_hex(self, limit=0):
|
|
"""数据 hex; limit>0 时截断并加省略号"""
|
|
h = ' '.join(f'{b:02X}' for b in self.data)
|
|
if limit and len(self.data) > limit:
|
|
h = ' '.join(f'{b:02X}' for b in self.data[:limit]) + ' …'
|
|
return h
|
|
|
|
def format(self, idx=None, with_dur=True):
|
|
"""单行显示: [idx] t=<t0>us <dur>us 0x32 W: FF 60 ACK:AA"""
|
|
head = f'[{idx:3d}] ' if idx is not None else ''
|
|
dur = f' {self.dur_us:>6d}us' if with_dur else ''
|
|
body = self.data_hex(12) if self.data else '(无数据)'
|
|
return (f'{head}t={self.t0_us:9d}us{dur} {self.addr_hex()} '
|
|
f'{self.rw_char()}: {body:<30s} ACK:{self.ack_str()}'
|
|
f'{self.flag_str()}')
|
|
|
|
def __repr__(self):
|
|
return (f'Transaction(t0={self.t0_us}, addr=0x{self.addr7:02X}, '
|
|
f'{self.rw_char()}, data={self.data_hex()}, acks={self.ack_str()}, '
|
|
f'ended={self.ended})')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 边沿序列 -> 事务(算法自 I2C-Inject i2c_sniff_decode.py, 真机验证过)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def decode_edges(edges):
|
|
"""边沿序列 -> [Transaction]。
|
|
|
|
edges: 可迭代的 (t_us, pins), pins bit0=SDA bit1=SCL, 值为新电平(变化沿)。
|
|
首个边沿只用于建立基线电平。规则:
|
|
SCL 高电平期间 SDA 下降 = START/repeated START(收掉当前事务, 新开一个);
|
|
SCL 高电平期间 SDA 上升 = STOP;
|
|
SCL 上升沿采样 SDA 为数据位, 第 9 位为应答(0=ACK)。
|
|
事务由 START..STOP 组成; repeated START 结束上一段(ended='rstart');
|
|
输入流耗尽仍未收尾的按 'cut'(若正好断在半字节则 'partial')。
|
|
"""
|
|
return _decode_impl(edges)
|
|
|
|
|
|
def _decode_impl(edges):
|
|
"""decode_edges 的真实实现(按位状态机, 与旧脚本语义一致)"""
|
|
trans = []
|
|
cur = None # {'t0', 'bytes'[], 'acks'[], }
|
|
bits = []
|
|
psda = pscl = None
|
|
t_last = 0
|
|
|
|
def close(t, ended):
|
|
nonlocal cur, bits
|
|
if cur is not None:
|
|
if bits and ended == 'cut':
|
|
ended = 'partial'
|
|
d = cur['bytes']
|
|
trans.append(Transaction(
|
|
t0_us=cur['t0'], t1_us=t, addr7=(d[0] >> 1) if d else 0,
|
|
read=bool(d[0] & 1) if d else False,
|
|
acks=cur['acks'], data=bytes(d[1:]), ended=ended))
|
|
cur = None
|
|
bits = []
|
|
|
|
for t, v in edges:
|
|
t_last = t
|
|
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
|
|
close(t, 'rstart')
|
|
cur = {'t0': t, 'bytes': [], 'acks': []}
|
|
elif psda == 0 and sda == 1: # STOP
|
|
close(t, 'stop')
|
|
elif scl == 1 and pscl == 0: # SCL 上升沿采样
|
|
if cur is not None:
|
|
bits.append(sda)
|
|
if len(bits) == 8:
|
|
cur['bytes'].append(
|
|
int(''.join(map(str, bits)), 2))
|
|
elif len(bits) == 9: # 第 9 位 = 应答(0=ACK)
|
|
cur['acks'].append(sda == 0)
|
|
bits = [] # 8 数据位不清, 凑满 9 位才清
|
|
psda, pscl = sda, scl
|
|
close(t_last, 'cut')
|
|
return trans
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 事务 -> 边沿序列(参考编码器: 自测向量生成 + 对齐时序规定)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class EdgeEncoder:
|
|
"""把事务序列编码成 (t_us, pins) 边沿流。pins bit0=SDA bit1=SCL。
|
|
|
|
定时(khz=100 → half=5us): SDA 在 SCL 变低后 1us 换位, SCL 高/低各
|
|
half us。时间戳与 decode_edges 的采样规则严格对齐:
|
|
START 的 t0 = SDA 下降沿时刻, STOP 的 t1 = SDA 上升沿时刻。
|
|
"""
|
|
|
|
def __init__(self, khz=100.0, t0_us=1000):
|
|
self.half = max(1, round(500.0 / khz)) # 半周期 us
|
|
self.t = int(t0_us)
|
|
self.sda = 1
|
|
self.scl = 1
|
|
self.edges = [(self.t, self.pins())] # 初始基线电平
|
|
|
|
def pins(self):
|
|
return (self.scl << 1) | self.sda
|
|
|
|
def _set(self, sda=None, scl=None, dt=1):
|
|
"""电平目标: 只有真发生变化的沿才推进时间并记录(边沿流语义)"""
|
|
if sda is not None and sda != self.sda:
|
|
self.t += dt
|
|
self.sda = sda
|
|
self.edges.append((self.t, self.pins()))
|
|
if scl is not None and scl != self.scl:
|
|
self.t += dt
|
|
self.scl = scl
|
|
self.edges.append((self.t, self.pins()))
|
|
|
|
def _bit(self, level):
|
|
"""发一位: SCL 低电平窗口内置 SDA=level, SCL 走一拍(高 half 低 half)"""
|
|
self._set(sda=level, scl=0)
|
|
self._set(scl=1, dt=self.half)
|
|
self._set(scl=0, dt=self.half)
|
|
|
|
def start(self):
|
|
"""(repeated) START: SDA 高->低(SCL 高, t0 基准沿), 随后 SCL 拉低"""
|
|
self._set(sda=1)
|
|
self._set(scl=1)
|
|
self._set(sda=0, dt=self.half) # t0 基准沿
|
|
t0 = self.t
|
|
self._set(scl=0, dt=self.half)
|
|
return t0
|
|
|
|
def stop(self):
|
|
"""STOP: SCL 低时 SDA 置低, SCL 升高, SDA 升高(t1 基准沿)"""
|
|
self._set(sda=0, scl=0)
|
|
self._set(scl=1, dt=self.half)
|
|
self._set(sda=1, dt=self.half) # t1 基准沿
|
|
return self.t
|
|
|
|
def byte(self, value, ack):
|
|
"""发 8 数据位 + 1 应答位(ack=True 为 ACK, 即 SDA=0)"""
|
|
for i in range(7, -1, -1):
|
|
self._bit((value >> i) & 1)
|
|
self._bit(0 if ack else 1)
|
|
return self.t
|
|
|
|
|
|
def encode_edges(transactions, khz=100.0, t0_us=1000, idle_us=50):
|
|
"""事务列表 -> (edges, refs)。事务间 SDA/SCL 保持高(总线空闲 idle_us)。
|
|
|
|
ended 字段决定收尾: 'stop' 发 STOP; 'rstart' 直接进入下一个事务的
|
|
repeated START; 'cut'/'partial' 发完最后字节悬停(仅自测用)。
|
|
refs[i] = (t0, t1): t0 恒有; t1 仅 'stop' 时有(STOP 沿时刻, 与解码器
|
|
对 t1 的定义一致), 其余为 None。
|
|
"""
|
|
enc = EdgeEncoder(khz=khz, t0_us=t0_us)
|
|
refs = []
|
|
for k, tr in enumerate(transactions):
|
|
t0 = enc.start()
|
|
enc.byte(tr.addr7 << 1 | (1 if tr.read else 0), tr.acks[0])
|
|
for i, b in enumerate(tr.data):
|
|
ack = tr.acks[i + 1] if i + 1 < len(tr.acks) else True
|
|
enc.byte(b, ack)
|
|
t1 = enc.stop() if tr.ended == 'stop' else None
|
|
if tr.ended == 'cut':
|
|
enc._set(scl=0)
|
|
refs.append((t0, t1))
|
|
if k + 1 < len(transactions):
|
|
enc.t += idle_us # 空闲间隙(不出边沿)
|
|
return enc.edges, refs
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 固件监控流行解析(T/W 行, 语法见 firmware/main.c)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def to_tline(tr, idx=None):
|
|
"""Transaction -> 固件 T 行语法(本函数是语法权威之一, 与固件对齐):
|
|
|
|
T <t0_us> <dur_us> <addr7:02X> <R|W> <flags> <acks> <n> <b1..bn>
|
|
flags: 结束码 + 可选 'L'(长事务截断): '-' 'r' 'c' 'p' ( + 'L' )
|
|
acks : n+1 个 A/N, 首个是地址应答; n=0 时地址应答仍保留
|
|
"""
|
|
flags = {'stop': '-', 'rstart': 'r', 'cut': 'c', 'partial': 'p'}[tr.ended]
|
|
if tr.truncated:
|
|
flags += 'L'
|
|
n = len(tr.data)
|
|
body = ' '.join(f'{b:02X}' for b in tr.data)
|
|
return (f'T {tr.t0_us} {tr.dur_us} {tr.addr7:02X} {tr.rw_char()} '
|
|
f'{flags} {tr.ack_str()} {n} {body}'.rstrip())
|
|
|
|
|
|
def parse_tline(line):
|
|
"""固件 T 行 -> Transaction; 非法行抛 ValueError。"""
|
|
parts = line.split()
|
|
if len(parts) < 7 or parts[0] != 'T':
|
|
raise ValueError(f'非 T 行: {line!r}')
|
|
try:
|
|
t0, dur = int(parts[1]), int(parts[2])
|
|
addr = int(parts[3], 16)
|
|
rw = parts[4]
|
|
flags = parts[5]
|
|
acks_s = parts[6]
|
|
n = int(parts[7])
|
|
data = bytes(int(x, 16) for x in parts[8:8 + n])
|
|
except (IndexError, ValueError) as e:
|
|
raise ValueError(f'T 行字段非法: {line!r} ({e})') from e
|
|
if len(data) != n:
|
|
raise ValueError(f'T 行数据不足: 声明 {n} 实得 {len(data)}')
|
|
if rw not in ('R', 'W'):
|
|
raise ValueError(f'T 行方向非法: {rw!r}')
|
|
if not flags or len(flags) > 2 or flags[0] not in '-rcp':
|
|
raise ValueError(f'T 行 flags 非法: {flags!r}')
|
|
truncated = 'L' in flags[1:]
|
|
ended = {'-': 'stop', 'r': 'rstart', 'c': 'cut', 'p': 'partial'}[flags[0]]
|
|
if len(acks_s) != n + 1 or set(acks_s) - {'A', 'N'}:
|
|
raise ValueError(f'T 行 acks 长度/字符非法: {acks_s!r}')
|
|
return Transaction(t0_us=t0, t1_us=t0 + dur, addr7=addr, read=(rw == 'R'),
|
|
acks=tuple(c == 'A' for c in acks_s), data=data,
|
|
ended=ended, truncated=truncated)
|
|
|
|
|
|
def parse_wline(line):
|
|
"""固件 W 行(边沿快照) -> (t_us, pins); 'W.' 结尾行返回 None。"""
|
|
if line == 'W.':
|
|
return None
|
|
parts = line.split()
|
|
if len(parts) != 3 or parts[0] != 'W':
|
|
raise ValueError(f'非 W 行: {line!r}')
|
|
t, pins = int(parts[1]), int(parts[2])
|
|
if not 0 <= pins <= 3:
|
|
raise ValueError(f'W 行 pins 越界: {pins}')
|
|
return (t, pins)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 导出
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def transactions_to_csv(transactions):
|
|
"""事务列表 -> CSV 文本(含表头; 数据列 hex 串, 应答列 A/N 串)"""
|
|
rows = ['idx,t0_us,dur_us,addr7,rw,addr_ack,data_hex,data_acks,ended']
|
|
for i, tr in enumerate(transactions):
|
|
rows.append(','.join((
|
|
str(i), str(tr.t0_us), str(tr.dur_us), f'{tr.addr7:02X}',
|
|
tr.rw_char(), 'A' if tr.addr_ack else 'N',
|
|
tr.data_hex(), ''.join('A' if a else 'N' for a in tr.data_acks),
|
|
tr.ended)))
|
|
return '\n'.join(rows) + '\n'
|