feat: I2C-Lab v0.1.0 代码入库——RP2040 PIO 被动监控 probe 固件(pico-sdk C, 250MHz 双核) + Fluent 上位机(事务流/波形快照/CSV/演示模式)；release/ 交付物(exe/UF2/BIN/zip)不入库、本地保留待真机验收发布；根 README 工具表与 agent.md 登记；I2C-Inject README 标注后续并入 I2C-Lab；移除误登记的 OBI-1/ui-style-demo 相关内容

2026-10-13 15:41:29 +08:00
parent 74469cfe66
commit 42ff258b35
25 changed files with 3292 additions and 20 deletions
@@ -0,0 +1,333 @@
+#!/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'