chronaxis.planning.message_calculations
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""静态解析当前消息legacy_sum声明，保持解码profile和新计算policy分开。23输入为message_calculations原生JSON数组、当前来源/绑定和计划ID集合；输出冻结的4MessageCalculationPlan tuple。frame_id必填，channel默认采用来源通道，帧类型可省略；参数按FLOAT64处理，表达式5只访问本来源、本消息的完整CAN定位，不沿用全局派生、上一条值或普通TTL。6当前模式明确拒绝cache_policy，旧完整触发/缓存未接线不能靠同名字段冒充。7可选disjunction另声明方向别名和正负桶；方向可不在expr中，仍须有同一原消息8的完整定位。它选择本条原解码值，不改表达式、信号转换或旧读写调用。910require_message_calculation只检查已有typed计划与当前来源/声明的关系，供解析和11validate/explain共用；它不重新解析AST或raw_config。所有操作只处理有限配置，12不open/stat/resolve，不加载DBC、旧reader或执行流，不读取任何分析输入。13"""1415# 只导入冻结模型、受限表达式及静态字段校验，不加载来源执行组件。16from ..core.declarations import DeclarationBindings17from ..core.message_binding import ExpressionParameter, MessageDeclarations18from ..legacy.records import PROFILE_ID19from ..legacy.disjunction import LegacyDisjunction20from ..core.types import SignalDType21from ..sources.can_spec import LegacyCANSpec, can_field_from_key22from ..sources.text_can_spec import CANTextSpec23from .diagnostic_collection import usable_node, collect_node_error24from .checks import PlanValidationError, array_at, choice_at, model_at, object_at, text_at25from .models import LegacyMessageSelector, MessageCalculationPlan26from .nodes import expression_at, register_id272829def require_message_calculation(calculation, sources, bindings: DeclarationBindings) -> None:30 """核已有typed逐消息指标与当前计划一致，不借旧配置覆盖typed替换。3132 参数: calculation为MessageCalculationPlan；sources为当前SourcePlan tuple；bindings33 为当前DeclarationBindings。34 返回: None，来源、帧定位和原声明均一致。35 异常: TypeError为指标/绑定模型错误；ValueError为来源能力、声明或定位冲突。36 边界: 不重编译AST，不读DBC；字段存在性和单位实况仍由同次READY另行验证。37 """38 if type(calculation) is not MessageCalculationPlan or not isinstance(bindings, DeclarationBindings):39 raise TypeError('逐消息指标需要MessageCalculationPlan及当前DeclarationBindings')40 matching = tuple(source for source in sources if source.name == calculation.source)41 if len(matching) != 1:42 raise ValueError('逐消息指标必须引用当前唯一来源')43 source = matching[0]44 # 计算只取当条解码值；格式仍由各自真实读取profile标识，不能冒充旧解码器。45 if source.source_type == 'can_blf' and isinstance(source.config, LegacyCANSpec):46 if (source.config.legacy_profile != PROFILE_ID or source.config.unit != 's'47 or source.time_unit != 's' or source.timezone_text is not None):48 raise ValueError('BLF逐消息来源需保原profile和秒时间声明')49 profile = PROFILE_ID50 elif (source.source_type in ('can_asc', 'asc_or_blf', 'raw_can_csv')) and (isinstance(source.config, CANTextSpec)):51 if source.source_type != source.config.source_type:52 raise ValueError('文本报文来源种类与解析声明不同')53 profile = 'can-text-v1'54 else:55 raise ValueError('逐消息计算需要BLF、ASC或原始报文CSV来源')56 if source.config.name != source.name or calculation.profile != profile:57 raise ValueError('逐消息指标profile和名称必须等于实际来源')58 declared = next((item for item in bindings.sources if item.name == calculation.source), None)59 if (declared is None or declared.input_identity != source.input_identity60 or declared.timebase != source.config.timebase or declared.offset_ns != source.config.offset_ns):61 raise ValueError('逐消息来源定位、原时基或offset与当前完整声明不符')62 if calculation.update_message.channel != source.config.channel:63 raise ValueError('逐消息选择器channel与来源声明不一致')64 declaration = calculation.expr.bindings65 if (declaration.declarations.identity != bindings.identity or declaration.source != calculation.source66 or declaration.parameters != calculation.parameters or declaration.derived_names67 or declaration.message_policy != calculation.message_policy):68 raise ValueError('逐消息表达式必须属于当前完整原声明、来源和参数')69 # 依赖必须是编译树的真实别名，不能由外层投影遗漏或偷偷增加其他消息。70 if type(calculation.expr.dependencies) is not tuple or calculation.expr.dependencies != calculation.expr.root.aliases:71 raise ValueError('逐消息表达式依赖与静态树不一致')72 selector = calculation.update_message73 required_flags = [selector.is_extended, selector.is_fd]74 # 方向可以独立于公式，但仍必须来自本条同一原消息，不能借缓存跨消息取值。75 dependencies = calculation.expr.dependencies76 if calculation.disjunction is not None:77 direction = bindings.resolve(calculation.disjunction.signal)78 if direction.dtype not in (SignalDType.INT64, SignalDType.UINT64, SignalDType.FLOAT64, SignalDType.BOOL):79 raise ValueError('disjunction方向仅接受数值或bool声明，不接受enum')80 dependencies = tuple(dict.fromkeys((*dependencies, calculation.disjunction.signal)))81 for alias in dependencies:82 definition = bindings.resolve(alias)83 if definition.source != calculation.source or declaration.resolve(alias) is not definition:84 raise ValueError('逐消息依赖须保当前来源的同一原SignalDefinition')85 field = can_field_from_key(definition.field)86 # 静态声明允许省略类型；明确给出的约束不能冲突，实际唯一性由读取阶段检查。87 if ((field.channel, field.frame_id) != (selector.channel, selector.frame_id)88 or any(left is not None and right is not None and left != right for left, right in89 ((field.is_extended, selector.is_extended), (field.is_fd, selector.is_fd)))):90 raise ValueError('逐消息依赖的帧定位与update_message不一致')91 # 未限制的选择器不能掩盖两个依赖相互矛盾的明确帧类型。92 for index, value in enumerate((field.is_extended, field.is_fd)):93 if value is not None:94 if required_flags[index] is not None and required_flags[index] != value:95 raise ValueError('逐消息表达式的依赖必须属于同一实际帧类型')96 required_flags[index] = value979899def parse_message_calculations(value, sources, bindings, ids) -> tuple[MessageCalculationPlan, ...]:100 """按声明顺序解析独立逐消息指标，ID与原规则/窗口/指标共用集合。101102 参数: value为原生JSON数组；sources/bindings为原计划静态模型；ids为本次103 已使用ID集合。返回: MessageCalculationPlan tuple。104 异常: PlanValidationError保具体JSON路径；cache_policy报旧完整缓存能力未接线。105 副作用: 只向本次局部ids登记名称，不读输入、DBC或执行表达式。106 """107 result = []108 for index, raw in enumerate(array_at(value, '$.message_calculations')):109 path = f'$.message_calculations[{index}]'110 if not usable_node(path, raw):111 continue112 try:113 path = f'$.message_calculations[{index}]'114 # 旧配置即使尚无新policy字段，也先说明旧完整cache模式为何不能运行。115 if type(raw) is dict and 'cache_policy' in raw:116 raise PlanValidationError(path + '.cache_policy',117 '旧profile完整触发/cache模式尚未接线；current_message-v1仅用本条消息，不接受cache_policy',118 'capability_unavailable')119 raw = object_at(raw, path, ('name', 'update_message', 'expr', 'unit'),120 ('parameters', 'split_by_sign', 'source', 'profile', 'operation', 'message_policy'))121 calculation_id = register_id(raw['name'], path + '.name', ids)122 from .input_defaults import source_reference123 source_name = source_reference(raw, sources, path, can_only=True)124 source = next((item for item in sources if item.name == source_name), None)125 if source is None:126 raise PlanValidationError(path + '.source', '逐消息指标来源不存在')127 text_source = ((source.source_type in ('can_asc', 'asc_or_blf', 'raw_can_csv')) and (isinstance(source.config, CANTextSpec)))128 if not text_source and (source.source_type != 'can_blf' or not isinstance(source.config, LegacyCANSpec)):129 raise PlanValidationError(path + '.source', '逐消息计算需要can_blf、can_asc或raw_can_csv来源', 'capability_unavailable')130 expected_profile = 'can-text-v1' if text_source else source.config.legacy_profile131 profile = choice_at(raw.get('profile', expected_profile), (expected_profile,), path + '.profile')132 if profile != expected_profile:133 raise PlanValidationError(path + '.profile', 'profile必须等于来源legacy_profile')134 # 当前只有一种逐消息累计策略；省略采用它，显式未知策略仍拒绝。135 operation = choice_at(raw.get('operation', 'legacy_sum'), ('legacy_sum',), path + '.operation')136 policy = choice_at(raw.get('message_policy', 'current_message-v1'), ('current_message-v1',), path + '.message_policy')137 selector_raw = object_at(raw['update_message'], path + '.update_message',138 ('frame_id',), ('channel', 'is_extended', 'is_fd'))139 for flag in ('is_extended', 'is_fd'):140 if flag in selector_raw and type(selector_raw[flag]) is not bool:141 raise PlanValidationError(path + '.update_message.' + flag, '填写时需要bool，不能为null')142 from .channels import channel_index143 channel = (channel_index(selector_raw['channel'], path + '.update_message.channel')144 if 'channel' in selector_raw else source.config.channel)145 selector = model_at(LegacyMessageSelector, path + '.update_message',146 channel, selector_raw['frame_id'], selector_raw.get('is_extended'), selector_raw.get('is_fd'))147 if selector.channel != source.config.channel:148 raise PlanValidationError(path + '.update_message.channel', '选择器channel必须等于来源通道')149 parameters_raw = raw.get('parameters', {})150 # 参数名称可由用户声明，但容器及每个value/unit对象仍严格校验。151 parameters_raw = object_at(parameters_raw, path + '.parameters', (),152 tuple(parameters_raw) if type(parameters_raw) is dict else ())153 parameters = []154 for name, item in parameters_raw.items():155 parameter_path = path + '.parameters.' + name156 item = object_at(item, parameter_path, ('value', 'unit'))157 # 参数使用当前分析版本的单位规则，空信号消息域也不能退回旧版本。158 parameters.append(model_at(ExpressionParameter, parameter_path, name, item['value'], item['unit'],159 config_version=bindings.config_version))160 parameters = tuple(parameters)161 # 不接普通全局派生；当前表达式仅原别名与本指标显式参数，没有保持语义。162 declaration = model_at(MessageDeclarations, path + '.parameters', bindings, source_name, parameters,163 message_policy=policy)164 unit = text_at(raw['unit'], path + '.unit')165 expression = expression_at(raw['expr'], path + '.expr', declaration, (), unit=unit)166 # 三字段显式选择原值政策，不把配置缺失、null或自由条件当作分桶。167 disjunction = None168 if 'split_by_sign' in raw:169 item = object_at(raw['split_by_sign'], path + '.split_by_sign', ('signal', 'bucket'), ('value_policy',))170 disjunction = model_at(LegacyDisjunction, path + '.split_by_sign',171 item['signal'], item['bucket'], item.get('value_policy', 'raw_decoded-v1'))172 model_at(bindings.resolve, path + '.split_by_sign.signal', disjunction.signal)173 calculation = model_at(MessageCalculationPlan, path, calculation_id, source_name, profile, operation,174 selector, expression, unit, parameters, policy, disjunction)175 # 校验共享于validate/explain，typed替换不能绕过跨对象完整定位关系。176 model_at(require_message_calculation, path + '.expr', calculation, sources, bindings)177 result.append(calculation)178 except PlanValidationError as error:179 collect_node_error(error, path, raw)180 return tuple(result)181