chronaxis.configuration.imports
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""展开分析配置明确列出的一层本地公共定义。23主配置按imports顺序引入sources/signals/derived，最后追加本地声明；重复名称4直接拒绝。导入文件不能再导入文件，也不能声明运行、规则或输出字段。5仅打开配置文件，数据/DBC路径按各自声明文件转为词法绝对路径；不读取业务输入。6位置随本次加载保留，成功计划不持有原文或第二份配置，不执行环境替换或模板。7"""8from copy import deepcopy9from pathlib import Path10from types import MappingProxyType1112from . import load_configuration13from .diagnostics import locate_error14from .models import ConfigurationDocument, ConfigurationOrigin, field_path_text1516SECTIONS=('sources','signals','derived') # 唯一允许共享的定义，不包含执行节点。171819def _reject(document, path, message, code='invalid_import', hint=None):20 """在实际声明处拒绝导入。参数: 文档、tuple路径、原因/代码/建议。返回: 不返回。异常: PlanValidationError。"""21 from ..planning.checks import PlanValidationError22 raise locate_error(PlanValidationError(field_path_text(path),message,code,hint=hint),document)232425def _source_paths(source, base, path, document):26 """固定一项来源中的公开文件路径。参数: 来源副本、声明目录、字段路径、原文档。返回: None。异常: 字段错误。2728 只改本次副本，metadata内任意path不解释为文件；不调用stat或resolve。29 """30 from ..planning.checks import absolute_path, array_at, PlanValidationError31 try:32 # 路径文字、列表和带offset的文件对象共用路径定位，不展开输入目录。33 from ..planning.input_paths import source_path_fields34 for owner,key in source_path_fields(source,field_path_text(path)):35 owner[key]=str(absolute_path(owner[key],base,field_path_text((*path,key))))36 if 'dbc_by_channel' in source:37 values=source['dbc_by_channel']38 if type(values) is not dict:39 _reject(document,(*path,'dbc_by_channel'),'DBC通道映射必须为对象')40 source['dbc_by_channel']={key:str(absolute_path(value,base,field_path_text((*path,'dbc_by_channel',key))))41 for key,value in values.items()}42 except PlanValidationError as error:43 raise locate_error(error,document)444546def _append_definitions(data, prefix, document, source_prefix, merged, locations, origins, names):47 """追加一个文件的定义并迁移字段位置。参数: 原定义/目标前缀/原文档/原前缀及本次容器。返回: None。异常: 定义/重复错误。4849 names保存首个同名声明的位置；所有修改仅发生在本次展开容器，无全局状态。50 """51 for section in SECTIONS:52 if section not in data:continue53 values=data[section]54 expected=dict if section=='derived' else list55 if type(values) is not expected:56 _reject(document,(*source_prefix,section),57 'derived必须为名称到表达式的对象' if section=='derived' else '来源/信号定义必须为数组')58 offset=len(merged[section])59 entries=values.items() if section=='derived' else enumerate(values)60 for index,value in entries:61 path=(*source_prefix,section,index)62 if section!='derived' and type(value) is not dict:63 _reject(document,path,'每项公共定义必须为对象')64 if section=='signals' and 'signals' in value:65 children=value['signals']66 if type(children) is not list or any(type(child) is not dict for child in children):67 _reject(document,(*path,'signals'),'组内signals必须为信号对象数组')68 declarations=[(child.get('name'),(*path,'signals',child_index,'name'))69 for child_index,child in enumerate(children)]70 else:71 name=index if section=='derived' else value.get('name')72 declarations=[(name,path if section=='derived' else (*path,'name'))]73 for name,name_path in declarations:74 if type(name) is not str or not name:75 _reject(document,name_path,'每项公共定义必须明确填写非空名称')76 # 集中声明和普通信号仍共用一个名字空间，组本身没有额外名称。77 identity=('sources' if section=='sources' else 'signals/derived',name)78 if identity in names:79 first_source,first_path=names[identity]80 _reject(document,name_path,'重复定义 '+name,'duplicate_definition',81 '删除或改名；首次声明在 '+first_source+' '+field_path_text(first_path))82 names[identity]=(document.source,name_path)83 copied=deepcopy(value)84 if section=='sources':_source_paths(copied,Path(document.source).parent,path,document)85 if section=='derived':merged[section][name]=copied86 else:merged[section].append(copied)87 # 整个数组只遍历一次位置表，主文件项追加后仍指向主文件原索引。88 original_prefix=(*source_prefix,section)89 for original,location in document.locations.items():90 if original[:len(original_prefix)]!=original_prefix:continue91 suffix=original[len(original_prefix):]92 if not suffix:continue93 target=(*prefix,section,suffix[0] if section=='derived' else offset+suffix[0],*suffix[1:])94 locations[target]=location95 origins[target]=ConfigurationOrigin(document.source,original,location)969798def expand_document_imports(document):99 """展开当前分析配置的公共定义。参数: 已解析文档。返回: 可直接静态检查的文档。异常: 配置/读取错误。100101 未声明imports时原样返回，不增加复制或文件读取。批量文件通过每个子计划102 自行使用imports；纯字典API不调用此函数。版本错误由原入口拒绝，不读其导入。103 """104 data=document.data105 if type(data) is not dict:return document106 prefix=()107 analysis=data108 if type(analysis) is not dict or 'imports' not in analysis:return document109 if type(analysis.get('config_version')) is not int or analysis['config_version'] != 4:return document110 entries=analysis['imports']111 if type(entries) is not list:112 _reject(document,(*prefix,'imports'),'imports必须为本地配置路径数组')113 # 定义数组重新构造；先清除被移动数组的旧位置，避免缺失字段误用旧索引行列。114 raw=deepcopy(data)115 target=raw116 merged={'sources':[],'signals':[],'derived':{}}117 locations={path:location for path,location in document.locations.items()118 if not (path[:len(prefix)]==prefix and len(path)>len(prefix)+1119 and path[len(prefix)] in SECTIONS)}120 origins={path:ConfigurationOrigin(document.source,path,location) for path,location in locations.items()}121 names={}122 for index,entry in enumerate(entries):123 path=(*prefix,'imports',index)124 if type(entry) is not str or not entry.strip() or '://' in entry or any(char in entry for char in '*?[]'):125 _reject(document,path,'导入路径必须为明确的本地JSON/YAML文件，不支持URL、通配符或空值')126 imported=load_configuration(Path(document.source).parent/entry)127 fragment=imported.data128 if type(fragment) is not dict or not fragment:129 _reject(imported,(),'公共定义文件必须是非空对象，只能包含sources、signals、derived')130 unknown=set(fragment)-set(SECTIONS)131 if unknown:132 key=next(key for key in fragment if key in unknown)133 _reject(imported,(key,),'公共定义只允许sources、signals、derived；不支持嵌套imports或版本包装')134 _append_definitions(fragment,prefix,imported,(),merged,locations,origins,names)135 _append_definitions(analysis,prefix,document,prefix,merged,locations,origins,names)136 # 展开后不保留imports，执行计划只见一份普通配置；来源位置只在加载期使用。137 target.pop('imports')138 for section in SECTIONS:139 if section!='derived' or merged[section] or section in target:140 target[section]=merged[section]141 from .overrides import apply_overrides142 apply_overrides(target,document,prefix,locations,origins)143 return ConfigurationDocument(raw,document.source,document.format,144 MappingProxyType(locations),MappingProxyType(origins))145