chronaxis.planning.signal_groups

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""把同一报文的集中声明展开为既有信号绑定，不增加运行时分组或重复读取。"""2from .diagnostic_collection import usable_node, collect_node_error3from .checks import array_at, object_at, choice_at, text_at, model_at, PlanValidationError456SHARED = ('period', 'valid_for', 'same_time_policy')7LOCATOR = ('frame_id', 'channel', 'is_extended', 'is_fd')8910def signal_entries(value):11    """按原顺序提供信号副本和声明路径。参数: signals数组。返回: 迭代器。异常: 结构错误。1213    集中项只允许声明一套报文定位；子项可覆盖周期、有效期和同刻策略，14    不能把另一个报文塞进本组。没有额外组名、引用或运行时对象。15    """16    for index, raw in enumerate(array_at(value, '$.signals')):17        path = f'$.signals[{index}]'18        if type(raw) is not dict or 'signals' not in raw:19            yield raw, path20            continue21        group = object_at(raw, path, ('frame_id', 'signals'),22                          ('source', 'channel', 'is_extended', 'is_fd') + SHARED)23        children = array_at(group['signals'], path + '.signals')24        if not children:25            raise PlanValidationError(path + '.signals', '报文组至少需要一个信号')26        # 即使所有子项都覆盖共享参数，也不能把填写错误当成未提供。27        from .signal_validity import declared_valid_for28        declared_valid_for(group, path, dbc_available=True)29        if 'same_time_policy' in group:30            choice_at(group['same_time_policy'], ('error', 'last_by_sequence'), path + '.same_time_policy')31        common = {key: group[key] for key in SHARED if key in group}32        if 'source' in group:33            common['source'] = group['source']34        locator = {key: group[key] for key in LOCATOR if key in group}35        from ..legacy.records import field_from_dict36        # 预检使用转换后的副本；展开给子项的locator仍保留用户1基编号。37        from .channels import channel_index38        checked = dict(locator, signal='_group_check')39        checked['channel'] = channel_index(locator['channel'], path + '.channel') if 'channel' in locator else 040        model_at(field_from_dict, path, checked)41        for child_index, child in enumerate(children):42            child_path = f'{path}.signals[{child_index}]'43            if not usable_node(child_path, child):44                continue45            try:46                child_path = f'{path}.signals[{child_index}]'47                if type(child) is not dict:48                    raise PlanValidationError(child_path, '信号必须为对象')49                forbidden = set(child) & ({'field', 'source', 'signals'} | set(LOCATOR))50                if forbidden:51                    raise PlanValidationError(child_path, '组内信号不能重复声明报文定位：' + ', '.join(sorted(forbidden)))52                if 'signal' not in child:53                    raise PlanValidationError(child_path + '.signal', '需要DBC信号名称')54                text_at(child['signal'], child_path + '.signal')55                expanded = dict(common, **child)56                expanded['field'] = dict(locator, signal=expanded.pop('signal'))57                yield expanded, child_path58            except PlanValidationError as error:59                collect_node_error(error, child_path, child)606162def declared_signals(value):63    """遍历原声明的叶信号供默认展开使用。参数: 已校验数组。返回: 原字典迭代器。异常: 无。"""64    for item in value:65        if 'signals' in item:66            yield from item['signals']67        else:68            yield item69