chronaxis.fast.can_io

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""修复版 ChrgInfo 文件循环：先选 ID，再解码一次并更新该帧的全部公式。"""23from dataclasses import dataclass, field4from itertools import islice5from pathlib import Path6from types import SimpleNamespace78import can9import cantools1011from ..legacy.binding import dbc_candidates, select_message12from ..legacy.decode_policy import check_decode_length13from ..timebase.conversion import convert_time, checked_add_ns14from .compute import prepare_metric, _prepare_program_metric151617@dataclass18class _Group:19    """同一实际报文类型共享一次解码。只有已请求的公式进入该组。"""2021    message: object22    entries: list = field(default_factory=list)232425_UNSUPPORTED_FD = object()262728def _prepare(spec, db, *, _programs_checked=False):29    """预建 ID 索引及每文件状态；内部已编译路径仅绑定指令，低层默认检查文本。3031    参数为声明、DBC及内部准备状态。返回组索引和累计器；字段、歧义或公式错误32    在开日志前抛出。准备状态由调用链传入，不从用户配置或 spec 标志中读取。33    """34    groups = {}35    metrics = {}36    for item in spec['metrics']:37        name = item['name']38        if name in metrics:39            raise ValueError(f'重复的计算项 {name}')40        identifier = item['frame_id']41        if type(identifier) is not int or not 0 <= identifier <= 0x1fffffff:42            raise ValueError(f'{name} 的 frame_id 超出 CAN 范围')43        required = {signal['signal'] for signal in item['signals'].values()}44        if item.get('direction'):45            required.add(item['direction']['signal'])46        ide, fd = item.get('is_extended'), item.get('is_fd')47        if any(flag is not None and type(flag) is not bool for flag in (ide,fd)):48            raise ValueError(f'{name} 的帧型选择需要布尔值')49        choices = dbc_candidates(db,identifier,required,ide,fd)50        factory = _prepare_program_metric if _programs_checked else prepare_metric51        metric = metrics[name] = factory(item)52        # 未指定帧型时，只允许实际数据出现一种匹配身份。53        observed = set() if None in (ide,fd) else None54        by_kind = groups.setdefault(identifier,{})55        for extended in ((False,True) if ide is None else (ide,)):56            for actual_fd in ((False,True) if fd is None else (fd,)):57                kind = extended,actual_fd58                message = select_message(choices,SimpleNamespace(is_extended_id=extended,is_fd=actual_fd))59                if message is None:60                    if fd is None and actual_fd and any(m.is_extended_frame == extended for m in choices):61                        by_kind.setdefault(kind,_UNSUPPORTED_FD)62                    continue63                if message.is_container or message.is_multiplexed():64                    raise ValueError(f'{name} 的目标报文使用 container 或 multiplex')65                if by_kind.get(kind) is _UNSUPPORTED_FD:66                    raise ValueError(f'同 ID {identifier:#x} 的计算项对 FD 定义有冲突')67                group = by_kind.setdefault(kind,_Group(message))68                if group.message is not message:69                    raise ValueError(f'ID {identifier:#x} 在同一实际帧型中对应多个 DBC 定义')70                group.entries.append((name,metric,observed))71    if not metrics:72        raise ValueError('至少需要一个逐消息计算项')73    return groups,metrics747576def run_one(spec, database=None, *, cancel_event=None):77    """流式运行单文件低层入口，保留原调用的行及提示返回结构。7879    参数: spec编译声明，database可选DBC，cancel_event可选取消事件。80    返回: (汇总行list, 提示list)。异常: 读取/解码/数值/取消错误。81    """82    from .session import run_session83    channel = spec['channel']84    if type(channel) is not int or channel < 1:85        raise ValueError('配置通道从1开始')86    domain = spec.get('domain_ns')87    if domain is not None and (len(domain) != 2 or any(type(t) is not int for t in domain) or domain[0] >= domain[1]):88        raise ValueError('domain_ns需要递增的两个整数纳秒')89    return run_session(spec, database, cancel_event=cancel_event)90