chronaxis.configuration.overrides
源码基线:8904c2f744951857659abd40aaa3535d57faf53c
1"""只覆盖公共定义中的明确配置项,不允许隐藏改名、添加对象或递归合并自由数据。23覆盖值按主文件定位,随后仍由普通计划校验器检查;来源表同时指回真正的覆盖位置。4"""5from copy import deepcopy6from .models import ConfigurationOrigin78ALLOWED={9 'sources':{'path','paths','dbc_by_channel','time','reader'},10 'signals':{'dtype','unit','valid_for','period','conversion','choices','valid_range','invalid_values','sample_policy','same_time_policy'},11 'derived':{'expr','unit'},12}131415def apply_overrides(target, document, prefix, locations, origins):16 """修改本次独立副本;参数含原文位置。返回None;非法对象/字段经_imports._reject定位。"""17 from .imports import _reject, _source_paths18 from pathlib import Path19 raw=target.pop('overrides',{})20 if type(raw) is not dict or set(raw)-set(ALLOWED):21 _reject(document,(*prefix,'overrides'),'overrides仅允许sources/signals/derived对象')22 for section,items in raw.items():23 where=(*prefix,'overrides',section)24 if type(items) is not dict:_reject(document,where,'覆盖项须按已有名称索引')25 catalog={}26 if section=='derived':27 catalog={name:(value,(*prefix,section,name)) for name,value in target.get(section,{}).items()}28 else:29 for index,value in enumerate(target.get(section,[])):30 if section=='signals' and 'signals' in value:31 for ci,child in enumerate(value['signals']):catalog[child['name']]=(child,(*prefix,section,index,'signals',ci))32 else:catalog[value['name']]=(value,(*prefix,section,index))33 for name,changes in items.items():34 path=(*where,name)35 if name not in catalog:_reject(document,path,'覆盖目标不存在,不允许新增定义')36 if type(changes) is not dict or not changes or set(changes)-ALLOWED[section]:37 _reject(document,path,'覆盖字段不在允许列表;不能改名、改绑定或修改自由metadata')38 value,target_path=catalog[name]39 # path/paths是同一输入声明的两种形状,只保留用户本次选定的一种。40 if section=='sources':41 if 'path' in changes and 'paths' not in changes:value.pop('paths',None)42 if 'paths' in changes and 'path' not in changes:value.pop('path',None)43 # 被整体替换的字段不再保留旧叶子来源,例如旧time.clock已被省略。44 replaced=set(changes)45 if section=='sources' and replaced & {'path','paths'}:replaced.update(('path','paths'))46 for effective in tuple(origins):47 if effective[:len(target_path)]==target_path and len(effective)>len(target_path) and effective[len(target_path)] in replaced:48 origins.pop(effective,None);locations.pop(effective,None)49 value.update(deepcopy(changes))50 if section=='sources':_source_paths(value,Path(document.source).parent,path,document)51 for original,location in document.locations.items():52 if original[:len(path)]==path and len(original)>len(path):53 effective=(*target_path,*original[len(path):])54 locations[effective]=location55 origins[effective]=ConfigurationOrigin(document.source,original,location)56