chronaxis.planning.dbc_defaults
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""仅在信号省略类型/单位或enum映射时读取DBC元数据，不读取报文。23每次计划编译按DBC路径共用一次加载；结果冻结为已有SignalDefinition字段。4数值信号不会因存在VAL表而变成文字枚举。重新加载配置才重新推导默认值。5"""6from contextlib import contextmanager7from contextvars import ContextVar8from pathlib import Path910from .checks import PlanValidationError11from ..legacy.binding import dbc_candidates12from ..can_dependencies import CANDependencyError, import_can_dependency13from ..dbc_loading import load_dbc_file141516_PREPARED_DBC = ContextVar('chronaxis_prepared_dbc', default=None)171819@contextmanager20def scoped_dbc_cache():21 """在一次 fast 准备期间共享 DBC，嵌套子计划沿用外层缓存，退出即释放。"""22 if _PREPARED_DBC.get() is not None:23 yield24 return25 token = _PREPARED_DBC.set({})26 try:27 yield28 finally:29 _PREPARED_DBC.reset(token)303132def load_dbc(path):33 """返回当前准备范围的数据库；普通单任务保持原加载行为。"""34 cache = _PREPARED_DBC.get()35 cantools = import_can_dependency('cantools')36 if cache is None:37 return load_dbc_file(path, cantools)38 key = Path(path).resolve()39 if key not in cache:40 cache[key] = load_dbc_file(path, cantools)41 return cache[key]424344def requested_enum_codes(config):45 """在加载 DBC 前定位 enum_code 使用的信号，只解析表达式位置和简单派生别名。4647 参数: config 为已归组的分析配置，可能含尚待完整校验的节点。48 返回: 原码别名集合。异常: 完整语法诊断由稍后的统一编译器报告。49 """50 import ast51 derived = config.get('derived', {})52 pending = [config.get(group, ()) for group in ('derived', 'rules', 'calculations', 'message_calculations')]53 names = set()54 while pending:55 item = pending.pop()56 if isinstance(item, (tuple, list)):57 pending.extend(item)58 elif isinstance(item, dict):59 for key, value in item.items():60 if isinstance(value, (list, dict)) and key not in ('metadata', 'parameters'):61 pending.append(value)62 elif isinstance(value, str) and 'enum_code' in value and key in (63 'expr', 'signal', 'condition', 'end_condition', 'start_condition', 'exit_condition', 'enter_condition'):64 try:65 tree = ast.parse(value, mode='eval')66 except (SyntaxError, ValueError):67 continue68 for node in ast.walk(tree):69 if (isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and70 node.func.id == 'enum_code' and len(node.args) == 1 and isinstance(node.args[0], ast.Name)):71 names.add(node.args[0].id)72 for name in tuple(names):73 seen = set()74 while isinstance(derived, dict) and isinstance(derived.get(name), dict) and name not in seen:75 seen.add(name)76 text = derived[name].get('expr')77 if not isinstance(text, str):78 break79 try:80 expression = ast.parse(text, mode='eval').body81 except (SyntaxError, ValueError):82 break83 if not isinstance(expression, ast.Name):84 break85 name = expression.id86 names.add(name)87 return names888990def _numeric_dtype(signal):91 """按解码后的数值范围选类型；有浮点缩放就使用float64，不丢弃小数。9293 参数: signal为DBC信号定义，包含原位宽、符号及解码倍率/偏移。94 返回: float64、int64或uint64文本。95 异常: ValueError为缩放后整数不能用64位类型表示。96 """97 if signal.is_float or type(signal.scale) is not int or type(signal.offset) is not int:98 return 'float64'99 low = -(2**(signal.length-1)) if signal.is_signed else 0100 high = 2**(signal.length-1)-1 if signal.is_signed else 2**signal.length-1101 limits = sorted((low*signal.scale+signal.offset, high*signal.scale+signal.offset))102 if -(2**63) <= limits[0] and limits[1] < 2**63:103 return 'int64'104 if 0 <= limits[0] and limits[1] < 2**64:105 return 'uint64'106 raise ValueError('DBC缩放后的整数超出64位，请显式选择dtype并核对数值范围')107108109def _enum_choices(signal):110 """把DBC原码表转换到既有解码值域，不对实际记录再施加一次DBC缩放。111112 参数: signal为含choices的DBC信号定义。113 返回: [已解码整数码, 标签]列表。114 异常: ValueError为缺少枚举、解码码值含小数或标签为空。115 """116 if not signal.choices:117 raise ValueError('DBC没有该信号的枚举表，请提供choices或选择数值dtype')118 result = []119 for code, label in signal.choices.items():120 value = signal.conversion.raw_to_scaled(code, decode_choices=False)121 # 核心枚举编码保持int/bool/str契约；不把浮点物理值截断成整数码。122 if type(value) is not int or signal.is_float:123 raise ValueError('DBC枚举缩放产生浮点值，当前枚举编码不支持；请使用数值dtype，不会截断或重复缩放')124 text = str(label)125 if not text:126 raise ValueError('DBC枚举标签为空，请提供明确choices')127 result.append([value,text])128 return result129130131class DBCDefaults:132 """一次计划编译内的临时数据库缓存，显式完整声明不打开DBC。"""133 def __init__(self):134 """创建空缓存，不访问磁盘。135136 参数: 无。返回: None。异常: 内存分配错误传播。137 """138 self.databases = {}139 self.failures = {}140141 def fill(self, raw, config, field, path, *, config_version=4, need_code=False):142 """补全DBC可证明的声明，不重复应用原解码缩放。143144 参数: raw为配置；config/field为来源与定位；path为诊断位置；config_version选择规则。145 返回: 独立dict，v4保留DBC原单位作为input_unit。146 异常: PlanValidationError为缺项、歧义、读取或依赖错误。147 """148 from ..core.unit_conversion import require_config_version149 require_config_version(config_version)150 result = dict(raw)151 # 手工映射已明确标签类型，可以直接得出逻辑dtype，不读取DBC作第二次猜测。152 choices = result.get('choices')153 if 'dtype' not in result and type(choices) is list and choices:154 if all(type(pair) is list and len(pair)==2 and type(pair[1]) is str for pair in choices):155 result['dtype']='enum'156 elif all(type(pair) is list and len(pair)==2 and type(pair[1]) is bool for pair in choices):157 result['dtype']='bool'158 needed = [key for key in ('dtype','unit') if key not in result]159 # 目标单位显式填写仍不代表DBC原值已经处于该尺度。160 if 'input_unit' not in result:161 needed.append('input_unit')162 if result.get('dtype') == 'enum' and 'choices' not in result:163 needed.append('choices')164 if need_code:165 needed.append('code_dtype')166 if not needed:167 return result168 key = str(config.dbc_path)169 if key in self.failures:170 raise PlanValidationError(path, 'DBC加载已失败，先修复 ' + self.failures[key], 'blocked_dependency')171 try:172 if key not in self.databases:173 try:174 self.databases[key] = load_dbc(key)175 except Exception:176 # 同一次失败文件不因另一个信号再次加载；不缓存跨任务的失败。177 self.failures[key] = path178 raise179 candidates = dbc_candidates(self.databases[key],field.frame_id,(field.signal,),field.is_extended,field.is_fd)180 signals = [next(s for s in message.signals if s.name==field.signal) for message in candidates]181 for name in needed:182 if name == 'dtype':183 values = [_numeric_dtype(signal) for signal in signals]184 elif name == 'code_dtype':185 if any(signal.is_float or signal.scale != 1 or signal.offset != 0 for signal in signals):186 raise ValueError('enum_code 需要原整数；该 DBC 已缩放或提供浮点解码值')187 values = ['int64' if signal.is_signed else 'uint64' for signal in signals]188 if 'code_dtype' in raw and any(value != raw['code_dtype'] for value in values):189 raise ValueError('code_dtype 与 DBC 原始整数有符号性不符')190 elif name in ('unit', 'input_unit'):191 # 无单位定义时标未知，不冒称已核实为V/A或无量纲。192 values = [signal.unit or ('1' if result.get('dtype') in ('enum','bool') else 'unknown') for signal in signals]193 else:194 values = [_enum_choices(signal) for signal in signals]195 if any(value != values[0] for value in values[1:]):196 raise ValueError(f'候选DBC的{name}不同，请补充帧类型或显式{name}')197 result[name] = values[0]198 except CANDependencyError as error:199 # 这是来源的安装环境缺项，不要求用户反复修改当前信号字段。200 raise PlanValidationError('$.sources', str(error), error.code,201 hint=error.hint, object_name='来源 ' + config.name) from error202 except Exception as error:203 # 不吞文件/解析错误，不用猜测值替代无法读取的元数据。204 raise PlanValidationError(path,'无法从DBC补充信号声明：'+str(error)) from error205 return result206