chronaxis.execution.same_frame_integrals
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""把同一物理报文的简单数值表达式直接交给现役实际时间积分器。23本对象只保存一个待完成同刻组；不构造信号数组、历史或通用最新状态。4读取、DBC投影、范围推进、积分器及结果发布仍由原会话负责。只有首帧前5证明绑定一致的内部调用方使用本入口，运行错误不切换路径或重读输入。6"""78from __future__ import annotations910from dataclasses import dataclass, replace11from math import isnan12from typing import Callable, Mapping, Sequence, TYPE_CHECKING1314from ..core.binding import SignalDefinition15from ..core.expression import validate_expression16from ..core.expression_reference import _evaluate17from ..core.numeric import normalize_reference, ScalarValue18from ..core.scalar import _scalar_from_validated_fields19from ..core.types import SignalDType as D20from ..legacy.records import field_from_dict21from ..legacy.updates import LegacyDecodedUpdate22from ..planning.backend_selection import simple_numeric_expression23from ..results.budget import ResourceBudget, _reserve_owned_ram, _release_owned_ram, _cleanup_owner24from ..results.errors import ResultClosedError25from ..sources.can_spec import can_field_from_key26from .history import _metadata_bytes27from .static_metadata import static_metadata_bytes2829if TYPE_CHECKING:30 from ..core.expression import StaticExpression31 from ..planning.plan import AnalysisPlan32 from .record_integral import RecordIntegral333435def direct_integral_inputs(plan: AnalysisPlan) -> tuple[SignalDefinition, ...] | None:36 """检查直接积分的完整静态边界，不读取DBC或数据。3738 参数: 已选reference的计划。返回: 本次必要定义tuple；不适用为None。39 异常: 非法typed结构，不把运行错误当成选择其他路径的理由。40 """41 if (plan.execution.backend != 'reference' or plan.extensions is not None42 or plan.rules or plan.windows or plan.sequences or plan.transforms or plan.aggregations43 or plan.message_calculations or not plan.calculations44 or plan.outputs.signal_csv is not None or plan.outputs.can_frames is not None45 or plan.diagnostics.trace or plan.diagnostics.integrity):46 return None47 choices, used = dict(plan.intervals.resolved), set()48 for item in plan.calculations:49 if (item.operation != 'integral' or item.scope != 'session' or item.method != 'zoh'50 or item.condition is not None or not simple_numeric_expression(item.signal)51 or item.calculation_id not in choices or choices[item.calculation_id].mode != 'actual'52 or choices[item.calculation_id].driver is not None):53 return None54 used.update(item.signal.dependencies)55 definitions = tuple(item.definition for item in plan.signals if item.definition.name in used)56 if {item.name for item in definitions} != used or len({item.source for item in definitions}) != 1:57 return None58 source = next((item for item in plan.sources if item.name == definitions[0].source), None)59 if source is None or source.source_type not in ('can_blf', 'can_asc', 'asc_or_blf', 'raw_can_csv'):60 return None61 fields = tuple(can_field_from_key(item.field) for item in definitions)62 if len({(f.channel, f.frame_id, f.is_extended, f.is_fd) for f in fields}) != 1:63 return None64 if any(item.dtype not in (D.INT64, D.UINT64, D.FLOAT64) or item.capture_code for item in definitions):65 return None66 return definitions676869@dataclass(frozen=True)70class SameFrameBinding:71 """保存同次DBC证明的共同定位；fd可为None，真实歧义继续由原投影器检查。"""7273 source: str # 当前计划中的来源名称。74 channel: int # 已转换为内部零起序号的通道。75 frame_id: int # DBC和原配置共同定位的报文编号。76 extended: bool # 同次DBC字段事实已经确定的标准/扩展标记。77 fd: bool | None # 原配置的CAN/FD约束；None保留原投影器的自动识别。78 dbc_fd: bool # DBC声明的FD属性，沿用经典CAN兼容规则。7980 def matches(self, update: LegacyDecodedUpdate) -> bool:81 """匹配已投影更新。参数: 更新。返回: bool。异常: 非法内部对象属性错误。"""82 return (update.source == self.source and update.channel == self.channel83 and update.frame_id == self.frame_id and update.is_extended == self.extended84 and (self.fd is None or update.is_fd == self.fd)85 and (self.dbc_fd or not update.is_fd)86 and not update.is_remote and not update.is_error)878889def same_frame_binding(definitions: tuple[SignalDefinition, ...], facts: Sequence[Mapping]) -> SameFrameBinding | None:90 """复用同次DBC字段事实证明唯一共同绑定，不加载文件。9192 参数: 实际所需信号定义和metadata.fields。返回: 共同定位或None（交回普通路径）。93 异常: 已有元数据结构损坏；不能将损坏当不支持而吞错。94 """95 if not definitions:96 return None97 by_field = {field_from_dict(dict(fact['locator'])).key: fact for fact in facts}98 chosen = None99 for definition in definitions:100 if definition.dtype not in (D.FLOAT64, D.INT64, D.UINT64) or definition.capture_code:101 return None102 field = can_field_from_key(definition.field)103 fact = by_field.get(definition.field)104 if fact is None:105 return None106 if 'dbc_frames' in fact:107 candidates = tuple(tuple(pair) for pair in fact['dbc_frames'])108 elif field.is_extended is not None and field.is_fd is not None:109 # 显式帧型沿用现有READY结构，不要求读取器增加重复候选字段。110 # 字段事实已由同次DBC核对；仅单向FD兼容时使用已有兼容提示。111 dbc_fd = field.is_fd or 'frame_format_compatibility' in fact112 candidates = ((field.is_extended, dbc_fd),)113 else:114 return None115 if len(candidates) != 1:116 return None117 extended, fd = candidates[0]118 if field.is_extended is not None and field.is_extended != extended:119 return None120 binding = SameFrameBinding(definition.source, field.channel, field.frame_id, extended, field.is_fd, fd)121 if chosen is not None and chosen != binding:122 return None123 chosen = binding124 return chosen125126127def _same_value(left: ScalarValue, right: ScalarValue) -> bool:128 """沿用同刻事实比较。参数: 归一化标量。返回: bool。异常: 无外部输入转换。"""129 return (left.valid == right.valid and left.quality == right.quality and left.code == right.code130 and type(left.value) is type(right.value)131 and (left.value == right.value or type(left.value) is float and isnan(left.value) and isnan(right.value)))132133134class SameFrameIntegrals:135 """拥有同刻选择和表达式上下文；借用积分器，关闭时不代替父会话关闭它们。"""136137 def __init__(self, budget: ResourceBudget, definitions: tuple[SignalDefinition, ...],138 expressions: tuple[StaticExpression, ...], calculations: tuple[RecordIntegral, ...],139 binding: SameFrameBinding, session_id: str, before_group: Callable[[], None], *, derived: tuple = ()) -> None:140 """检查静态表达式与实定义，预付有界状态，不读取输入。141142 参数: 共享预算、已补齐TTL的定义、静态表达式tuple、对应积分器tuple、143 已证明共同定位、会话ID、积分缓冲结算前刷新已观测范围的同步回调；derived为原派生声明。144 返回: owner。异常: 声明/预算/取消；未返回时附same_frame_cleanup_owner。145 """146 if not isinstance(budget, ResourceBudget) or type(definitions) is not tuple:147 raise TypeError('同帧积分需要预算和定义tuple')148 if (type(expressions) is not tuple or type(calculations) is not tuple149 or not expressions or len(expressions) != len(calculations)150 or type(binding) is not SameFrameBinding or not callable(before_group)):151 raise TypeError('同帧积分需要对应表达式/积分器、已证明定位和范围回调')152 self._budget = budget153 self._held_cost = self._fixed_cost = self._group_cost = self._work_cost = 0154 self._closed = self._closing = self._failed = self._finished = False155 self._definitions = self._fields = self._expressions = self._calculations = None156 self._chosen = self._seen = self._before_group = None157 self._time = self._last_time = None158 self._binding, self._session = binding, session_id159 try:160 # bindings由父计划持有；这里只计校验树、当前字段索引和必要同刻槽。161 trees = tuple((expr.root, expr.text, expr.normalized_ast, expr.dependencies) for expr in expressions)162 _reserve_owned_ram(self, 131072 + 32 * static_metadata_bytes((definitions, trees, derived)), '_fixed_cost')163 by_name = {item.name: item for item in definitions}164 if len(by_name) != len(definitions) or any(type(item) is not SignalDefinition165 or item.dtype not in (D.INT64, D.UINT64, D.FLOAT64) or item.capture_code166 or item.valid_for_ns is None for item in definitions):167 raise ValueError('同帧积分需要唯一数值定义及已知有效期')168 fields = tuple((item, can_field_from_key(item.field).signal) for item in definitions)169 for definition in definitions:170 field = can_field_from_key(definition.field)171 if (definition.source != binding.source or field.channel != binding.channel172 or field.frame_id != binding.frame_id173 or field.is_extended is not None and field.is_extended != binding.extended174 or field.is_fd != binding.fd):175 raise ValueError('同帧实际定义与已证明定位不一致')176 for expression, calculation in zip(expressions, calculations):177 if not simple_numeric_expression(expression):178 raise ValueError('同帧积分仅接受简单内建数值表达式')179 checked = validate_expression(expression.text, expression.bindings, derived=derived, unit=expression.root.unit)180 if checked != expression or expression.bindings.session_id != session_id:181 raise ValueError('同帧表达式与当前声明或会话不一致')182 for name in expression.dependencies:183 declared = expression.bindings.resolve(name)184 actual = by_name.get(name)185 if declared.valid_for_ns is None and actual is not None:186 declared = replace(declared, valid_for_ns=actual.valid_for_ns)187 if declared != actual:188 raise ValueError('同帧表达式与实际定义不一致：' + name)189 expression.bindings.require_synchronized(expression.dependencies)190 if (calculation._step_ns is not None or calculation._conditioned191 or calculation._dtype != expression.root.dtype192 or calculation._input_unit != expression.root.unit):193 raise ValueError('同帧表达式与实际时间积分器不一致')194 self._definitions, self._fields = by_name, fields195 self._expressions, self._calculations = expressions, calculations196 self._chosen, self._seen = {}, set()197 self._before_group = before_group198 # 输入原值仍由借用的消息持有；这里只构造数值标量和有限表达式工作区。199 self._workspace = 65536 + 32 * static_metadata_bytes(tuple(expr.root for expr in expressions))200 except BaseException as error:201 _cleanup_owner(self, error, 'same_frame_cleanup_owner', '同帧积分构造')202 raise203 finally:204 definitions = expressions = calculations = fields = by_name = checked = expression = calculation = trees = None205 definition = declared = actual = field = binding = before_group = derived = None206207 def _require_open(self) -> None:208 """失败后仅允许关闭。参数: 无。返回: None。异常: ResultClosedError。"""209 if self._closed or self._closing or self._failed or self._finished:210 raise ResultClosedError('同帧积分已关闭、失败或完成')211212 def consume(self, update: LegacyDecodedUpdate) -> None:213 """接收一条已投影报文，完整同刻组才推进积分。214215 参数: 当前LegacyDecodedUpdate。返回: None。异常: 身份、同刻冲突、预算、计算原错。216 无效字段覆盖本次值，不沿用前条。未选择的帧只由原读取器维护观测范围。217 """218 self._require_open()219 values = value = None220 primary = None221 try:222 if not isinstance(update, LegacyDecodedUpdate) or update.source_ref.session_id != self._session:223 raise ValueError('同帧更新不属于当前会话')224 if not self._binding.matches(update):225 return226 if update.time_ns is None or self._last_time is not None and update.time_ns < self._last_time:227 raise ValueError('同帧积分需要非递减实际时间')228 if self._time is not None and update.time_ns > self._time:229 self._flush()230 if update.sequence in self._seen:231 raise ValueError('完整组同别名原sequence重复')232 # 共同投影器对本帧解码失败会插入None；缺少键说明绑定证明失效，不能偷用旧值。233 if any(name not in update.values for _, name in self._fields):234 raise ValueError('同帧投影没有交付全部已证明字段')235 _reserve_owned_ram(self, self._workspace, '_work_cost')236 values = {}237 for definition, name in self._fields:238 value = normalize_reference(definition, update.values[name])239 if not value.valid:240 value = _scalar_from_validated_fields(None, value.dtype, False, value.quality)241 previous = self._chosen.get(definition.name)242 if previous is not None and definition.same_time_policy == 'error' and not _same_value(previous[1], value):243 raise ValueError(f'同信号同时间冲突：{definition.name}@{update.time_ns}ns')244 values[definition.name] = (update.sequence, value) if previous is None or update.sequence > previous[0] else previous245 # 只保一套选定值及同刻原序号；集合每新增一项先付费，不随总文件长度增长。246 _reserve_owned_ram(self, 512 + 16 * _metadata_bytes(update.sequence), '_group_cost')247 self._seen.add(update.sequence)248 self._chosen = values249 values = None250 self._last_time = self._time = update.time_ns251 except BaseException as error:252 primary, self._failed = error, True253 raise254 finally:255 values = value = update = previous = definition = None256 try:257 _release_owned_ram(self, self._work_cost, '_work_cost')258 except BaseException as cleanup:259 self._failed = True260 if primary is None:261 raise262 primary.add_note('同帧积分临时费用清理失败：' + type(cleanup).__name__)263264 def _flush(self) -> None:265 """提交完整同刻值，复用原表达式核及积分器。参数: 无。返回: None。异常: 原计算/预算错误。"""266 if self._time is None:267 return268 context = cache = value = None269 primary = None270 try:271 # 点缓冲只在满批时使用范围计算面积；不为每条报文重建整份计划。272 # 即将结算前必须先取得已观测范围，尾部则由finish统一更新。273 if any(len(calculation._rows) >= calculation._limit for calculation in self._calculations):274 self._before_group()275 _reserve_owned_ram(self, self._workspace, '_work_cost')276 context = {name: item[1] for name, item in self._chosen.items()}277 cache = {}278 for expression, calculation in zip(self._expressions, self._calculations):279 value = _evaluate(expression.root, context, cache).value280 deadlines = (self._time + self._definitions[name].valid_for_ns281 for name in expression.dependencies if context[name].valid)282 deadline = min((time for time in deadlines if time < (1 << 63)), default=None)283 calculation.consume(((self._time, value, deadline),))284 self._chosen.clear()285 self._seen.clear()286 self._time = None287 _release_owned_ram(self, self._group_cost, '_group_cost')288 except BaseException as error:289 primary, self._failed = error, True290 raise291 finally:292 context = cache = value = deadlines = expression = calculation = None293 try:294 _release_owned_ram(self, self._work_cost, '_work_cost')295 except BaseException as cleanup:296 self._failed = True297 if primary is None:298 raise299 primary.add_note('同帧表达式工作区清理失败：' + type(cleanup).__name__)300301 def finish(self, observed_end: int | None) -> None:302 """在原读取器确认EOF并关闭后，补算到来源真实末端的保持尾段。303304 参数: 来源实际末端或None。返回: None。异常: 原数值、预算或取消错误。305 不替代EOF证明，不把domain.end当采集终点，不虚构固定周期。306 """307 self._require_open()308 try:309 self._before_group()310 self._flush()311 if observed_end is not None:312 for calculation in self._calculations:313 if calculation._rows and calculation._last_time < observed_end:314 time, value, deadline, _, _ = calculation._rows[-1]315 deadline = observed_end if deadline is None else min(deadline, observed_end)316 calculation._rows[-1] = (time, value, deadline, None, None)317 calculation.consume(((observed_end, value, deadline),))318 self._finished = True319 except BaseException:320 self._failed = True321 raise322 finally:323 calculation = value = None324325 def close(self) -> None:326 """释放自有上下文及费用，借用积分器由父会话关闭。参数: 无。返回: None。异常: 释放失败可重试。"""327 if self._closed:328 return329 self._closing = True330 self._chosen = self._seen = self._definitions = self._fields = None331 self._expressions = self._calculations = self._before_group = self._binding = None332 _release_owned_ram(self, self._work_cost, '_work_cost')333 _release_owned_ram(self, self._group_cost, '_group_cost')334 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')335 self._closed = True336