chronaxis.execution.current_message_calculation
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用三格式共用的消息计算一项逐报文累计，不建立消息历史或原帧副本。23输入是CAN读取进程投影后的有限LegacyDecodedUpdate批，公式使用本条值，不保持上一条。4复用既有归一化、参考/NumPy表达式、正负分桶及原序求和；period已经在公式5中时不再乘dt。结果是累计快照；来源是否完整读完由拥有读取器的调用方确认。6此组件由当前记录执行器调用，不创建旧绑定证明或额外核验扫描。7"""89from ..core import expression_reference, expression_numpy10from ..core.expression import validate_expression11from ..core.models import Interval12from ..core.numeric import to_float_reference13from ..core.numeric_numpy import to_float_numpy14from ..core.types import QualityFlag as Q15from ..core.where_updates import contains_where, numpy_active, reference_active16from ..core.enum_codes import code_dependencies17from dataclasses import replace18from ..legacy.sum import LegacySumAccumulator, LegacySumContribution, LegacySkipReason19from ..legacy.updates import LegacyDecodedUpdate20from ..planning.message_calculations import require_message_calculation21from ..results.budget import ResourceBudget, _reserve_owned_ram, _release_owned_ram, _cleanup_owner22from ..results.errors import ResultClosedError23from ..sources.can_spec import can_field_from_key24from ..sources.packing import pack_message_values25from .calculation_values import scalar_from_column, direction_bucket_state, direction_skip_reason26from .history import _metadata_bytes27from ..core.lookup import pack_requested_lookups282930class CurrentMessageCalculation:31 """持有一项公式与固定累计状态；输入批及明细回调都只在consume期间借用。"""3233 def __init__(self, calculation, sources, bindings, session_id, budget, *, backend='numpy',34 batch_rows=64, domain=None, contribution_consumer=None):35 """先核有限声明，再预约公式和累计状态，不读输入或DBC。3637 参数: calculation/sources/bindings为同一计划声明；session_id为会话；budget为38 共享预算；backend为独立数值核；batch_rows为批上限；domain为可选39 实际时间筛选；contribution_consumer仅显式请求明细时接收(贡献,本次质量)。40 返回: 新累计组件。异常: 声明/预算/取消错误；失败构造携带可重试清理句柄。41 """42 require_message_calculation(calculation, sources, bindings)43 if not isinstance(budget, ResourceBudget) or backend not in ('reference', 'numpy'):44 raise TypeError('逐消息计算需要共享预算及明确数值后端')45 if type(batch_rows) is not int or not 1 <= batch_rows <= 4096:46 raise ValueError('逐消息计算批上限需要1到4096')47 if domain is not None and not isinstance(domain, Interval):48 raise TypeError('消息时间筛选需要Interval或None')49 if contribution_consumer is not None and not callable(contribution_consumer):50 raise TypeError('贡献消费者需要可调用对象或None')51 self._budget = budget52 self._held_cost = self._fixed_cost = self._batch_cost = 053 self._accumulator = self._compiled = self._fields = self._match_fields = None54 self._closed = self._failed = self._closing = False55 self._calculation, self._domain = calculation, domain56 self._session, self._backend, self._batch_rows = session_id, backend, batch_rows57 self._consumer = contribution_consumer58 self._quality = Q.NONE59 try:60 # 配置仅保存一次；批内列空间另按当前实际条数预约，不保逐条原文或证明。61 _reserve_owned_ram(self, 65536 + 32 * _metadata_bytes(calculation), '_fixed_cost')62 self._compiled = validate_expression(calculation.expr.text, calculation.expr.bindings, unit=calculation.unit)63 self._has_where = contains_where(self._compiled.root)64 if (self._compiled.normalized_ast != calculation.expr.normalized_ast65 or repr(self._compiled.root) != repr(calculation.expr.root)):66 raise ValueError('当前消息公式与已校验声明不同')67 if backend == 'numpy':68 self._compiled = pack_requested_lookups(self._compiled)69 self._fields = tuple((name, bindings.resolve(name), can_field_from_key(bindings.resolve(name).field))70 for name in self._compiled.dependencies)71 requested_codes = code_dependencies(self._compiled.root)72 if requested_codes:73 self._fields = tuple((name, replace(item, capture_code=True) if name in requested_codes else item, field)74 for name, item, field in self._fields)75 # 树只在初始化数一次；运行批次不为计费重新遍历或编码公式。76 pending, node_count = [self._compiled.root], 077 while pending:78 node = pending.pop()79 node_count += 180 pending.extend(node.args)81 self._row_cost = 2048 + 2048 * len(self._fields) + 512 * node_count82 self._direction = (None if calculation.disjunction is None else83 can_field_from_key(bindings.resolve(calculation.disjunction.signal).field))84 # 固定类型约束只整理一次，不在每帧构造新的依赖列表。85 self._match_fields = tuple(field for _, _, field in self._fields)86 if self._direction is not None:87 self._match_fields += (self._direction,)88 self._actual_identity = None89 # 单条累计需要的固定工作区在读输入前预留；多条批仅申请超出部分。90 self._accumulator = LegacySumAccumulator(budget, session_id, calculation.source,91 legacy_policy=calculation.message_policy, backend=backend, max_batch_rows=batch_rows,92 prepay_single=True)93 except BaseException as exc:94 child = getattr(exc, 'legacy_sum_cleanup_owner', None)95 if child is not None:96 self._accumulator = child97 _cleanup_owner(self, exc, 'current_message_calculation_cleanup_owner', '共同消息计算')98 raise99100 def _require_open(self):101 """阻止失败后继续交付结果。参数: 无。返回: None。异常: ResultClosedError。"""102 if self._closed or self._closing or self._failed:103 raise ResultClosedError('共同消息计算已关闭、失败或正在清理')104105 def _matches(self, update):106 """核真实消息来源后选择本指标帧。参数: update。返回: bool。异常: 类型/身份错。"""107 if not isinstance(update, LegacyDecodedUpdate):108 raise TypeError('共同消息计算需要已投影的真实更新')109 if update.source != self._calculation.source or update.source_ref.session_id != self._session:110 raise ValueError('消息不属于当前来源或会话')111 selector = self._calculation.update_message112 if (update.is_remote or update.is_error or113 (update.channel, update.frame_id) != (selector.channel, selector.frame_id)114 or selector.is_extended is not None and selector.is_extended != update.is_extended115 or selector.is_fd is not None and selector.is_fd != update.is_fd):116 return False117 # 选择器未限制类型时，各依赖仍须来自本条实际帧，不能拿另一帧的同名信号。118 if any(field.is_extended is not None and field.is_extended != update.is_extended119 or field.is_fd is not None and field.is_fd != update.is_fd for field in self._match_fields):120 return False121 # 自动绑定的投影只保留本帧所属信号；真正解码缺值会显式保留None。122 # 显式类型的低层调用仍允许缺键表示缺值，保持原有输入合同。123 if any((field.is_extended is None or field.is_fd is None)124 and field.signal not in update.values for field in self._match_fields):125 return False126 identity = (update.is_extended, update.is_fd)127 if self._actual_identity is not None and self._actual_identity != identity:128 raise ValueError('逐消息选择匹配多种帧类型，请补充is_extended或is_fd')129 self._actual_identity = identity130 if self._domain is not None:131 if update.time_ns is None:132 raise ValueError('有时间筛选的消息不能省略时间')133 return self._domain.start_ns <= update.time_ns < self._domain.end_ns134 return True135136 def consume(self, updates):137 """按原序计算有限批；默认只累计，明细仅交给显式回调，不保存历史。138139 参数: updates为最多batch_rows条更新的tuple。返回: None。140 异常: 输入/数值/预算/回调/取消错误原样传播；失败后只能close。141 """142 self._require_open()143 selected = columns = context = arrays = floats = missing = nonfinite = contributions = qualities = None144 primary = None145 try:146 if type(updates) is not tuple or len(updates) > self._batch_rows:147 raise ValueError('共同消息计算需要不超批上限的tuple')148 if not updates:149 return150 # 借用输入值，不复制原消息；预付筛选、归一化列、表达式临时列和贡献包装。151 cost = 16384 + len(updates) * self._row_cost152 _reserve_owned_ram(self, cost, '_batch_cost')153 selected = tuple(update for update in updates if self._matches(update))154 if not selected:155 return156 columns = {name: pack_message_values(definition, tuple(update.values.get(field.signal)157 for update in selected), self._backend) for name, definition, field in self._fields}158 if self._backend == 'numpy':159 arrays = expression_numpy._evaluate(self._compiled.root, columns, len(selected), {})160 floats = to_float_numpy(arrays.value)161 if self._has_where and not floats.valid.all():162 # 输入错误与算术错误必须分开；只看每条消息实际选到的输入。163 missing, nonfinite = (numpy_active(self._compiled.root, columns,164 {name: (col.quality & int(flag)) != 0 for name, col in columns.items()},165 len(selected)) for flag in (Q.MISSING, Q.NONFINITE))166 contributions = []167 qualities = [] if self._consumer is not None else None168 for index, update in enumerate(selected):169 if self._backend == 'reference':170 context = {name: scalar_from_column(column, index) for name, column in columns.items()}171 value = to_float_reference(expression_reference._evaluate(self._compiled.root, context, {}).value)172 else:173 value = scalar_from_column(floats, index)174 # 保旧原因优先级，缺值明确跳过，不能恢复上一条有效值。175 reason = None176 if not value.valid:177 if self._has_where and self._backend == 'reference':178 has_missing, has_nonfinite = (reference_active(self._compiled.root, context,179 {name: bool(item.quality & flag) for name, item in context.items()})180 for flag in (Q.MISSING, Q.NONFINITE))181 elif self._has_where:182 has_missing, has_nonfinite = missing[index], nonfinite[index]183 else:184 flags = [int(column.quality[index]) for column in columns.values()]185 has_missing = any(flag & int(Q.MISSING) for flag in flags)186 has_nonfinite = any(flag & int(Q.NONFINITE) for flag in flags)187 reason = (LegacySkipReason.MISSING_INPUT if has_missing188 else LegacySkipReason.NONFINITE_INPUT if has_nonfinite189 else LegacySkipReason.EVALUATION_ERROR)190 contribution_quality = update.quality | value.quality191 if self._direction is not None:192 quality, keep = direction_bucket_state(update.values.get(self._direction.signal), self._calculation.disjunction.bucket)193 reason = direction_skip_reason(value, reason, quality, keep)194 contribution_quality |= quality195 # 业务结果需要的质量直接来自本次求值；未选中的另一桶不制造缺失诊断。196 if reason is not LegacySkipReason.CONDITION_FALSE:197 self._quality |= value.quality198 if self._direction is not None:199 self._quality |= quality200 contributions.append(LegacySumContribution(update,201 value.value if value.valid and reason is None else None, reason))202 if qualities is not None:203 qualities.append(contribution_quality)204 self._accumulator.consume(tuple(contributions))205 if self._consumer is not None:206 for contribution, quality in zip(contributions, qualities):207 self._consumer(contribution, quality)208 except BaseException as exc:209 primary = exc210 self._failed = True211 raise212 finally:213 # 清理先断当前有限列；释放失败仍保owner费用，不能继续累计或返回快照。214 selected = columns = context = arrays = floats = missing = nonfinite = contributions = qualities = None215 try:216 _release_owned_ram(self, self._batch_cost, '_batch_cost')217 except BaseException as cleanup:218 self._failed = True219 if primary is None:220 raise221 primary.add_note('共同消息批费用清理失败：' + type(cleanup).__name__)222223 def snapshot(self):224 """读取当前累计，不冒充文件EOF。参数: 无。返回: LegacySumSnapshot。异常: 状态错。"""225 self._require_open()226 return self._accumulator.snapshot()227228 @property229 def quality(self):230 """读取实际参与计算的质量汇总，不构造逐条诊断。参数: 无。返回: QualityFlag。异常: 状态错。"""231 self._require_open()232 return self._quality233234 @property235 def closed(self):236 """读取释放完成状态。参数: 无。返回: bool。异常: 无。"""237 return self._closed238239 def close(self):240 """仅关闭自有累计与费用。参数: 无。返回: None。异常: 释放失败保留句柄供重试。"""241 if self._closed:242 return243 self._closing = True244 if self._accumulator is not None:245 self._accumulator.close()246 self._compiled = self._fields = self._match_fields = self._calculation = self._consumer = self._direction = None247 self._actual_identity = None248 _release_owned_ram(self, self._batch_cost, '_batch_cost')249 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')250 self._closed = True251