chronaxis.execution.signal_validity

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""从本次已经加载的DBC元数据补全必要信号的有效期。23仅处理尚未确定valid_for的信号，在读取首帧前按所属报文周期乘三。4不加载DBC、不扫描报文、不改变实际时间；返回的新定义交给计算状态使用，5静态配置仍保留缺省声明。无需有效期的无关信号不会额外请求周期。6"""78from dataclasses import replace9from ..legacy.records import field_to_dict10from ..sources.can_spec import can_field_from_key11from ..planning.signal_validity import triple_period121314def resolve_signal_validity(definitions, source_cursors):15    """在来源初始化后补全待DBC的有效期，显式有效期原样复用。1617    参数: definitions为本次实际需要的SignalDefinition元组；source_cursors为18        来源名到已初始化游标的映射，metadata.fields含本次请求的locator/cycle_ns。19    返回: SignalDefinition元组，全部有效期为正int64纳秒。20    异常: ValueError为缺失或不唯一的周期；PlanValidationError为周期三倍溢出。21    副作用: 无，不修改原声明或游标，不读取文件。22    """23    resolved = []24    for definition in definitions:25        # 已显式填写或由信号period求出的值无需再查询DBC。26        if definition.valid_for_ns is not None:27            resolved.append(definition)28            continue29        # 用完整规范定位找本次事实，不能按同名信号借用另一通道或报文的周期。30        cursor = source_cursors.get(definition.source)31        metadata = None if cursor is None else cursor.metadata32        locator = field_to_dict(can_field_from_key(definition.field))33        entries = (() if metadata is None else34                   tuple(item for item in metadata.get('fields', ()) if item.get('locator') == locator))35        if len(entries) != 1 or 'cycle_ns' not in entries[0]:36            raise ValueError(f'{definition.name}无法取得唯一DBC周期；请填写本信号period或valid_for')37        # 只替换运行所需定义；静态声明保持原样，不把尚未读取的周期写成既定事实。38        validity = triple_period(entries[0]['cycle_ns'], f'信号{definition.name}.valid_for')39        resolved.append(replace(definition, valid_for_ns=validity))40    return tuple(resolved)41