chronaxis.execution.driver_expression
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""从已提交的同刻信号状态取得所选驱动的数值或BOOL表达式，不保存输入历史。23输入是经过来源校验、归一化和同刻政策处理的LatestSignalState，以及静态4表达式声明；每个驱动更新最多返回一个(time, value, valid_until)计算点。5按实际更新时间判断valid_for，参考及生产逐点数值核独立；真正批量数组入口不变。固定周期6不进入此组件。最新值归state持有，临时求值空间在共享预算中预付。7这是新按需执行的内部接点，不生成CompiledExpression或伪造EOF/绑定证据，8不改变公开表达式入口拒绝StaticExpression的契约。来源读取/同刻归组由调用方负责。9"""1011from dataclasses import replace12from ..core import expression_reference13from ..core.expression import StaticExpression, validate_expression14from ..core.expression_point import PointProgram15from ..core.lookup import pack_requested_lookups16from ..core.where_updates import contains_where17from ..core.enum_codes import expire_value18from ..core.declarations import DeclarationBindings19from ..core.numeric import ScalarValue20from ..core.types import SignalDType as D, QualityFlag as Q21from ..results.budget import _reserve_owned_ram, _release_owned_ram, _cleanup_owner22from ..results.errors import ResultClosedError23from ..timebase.conversion import require_int64_ns24from .history import _metadata_bytes25from .static_metadata import static_metadata_bytes26from .state import LatestSignalState27from .where_state import before_context, group_updates, group_active282930class DriverExpression:31 """借用最新状态，持有已核实的表达式树与依赖；不跨组缓存计算值。"""3233 def __init__(self, static, state, driver, *, backend='numpy', derived=(), observation_ends=None):34 """核对声明与实际状态映射，预付树和依赖引用费用。3536 参数: static为StaticExpression；state为已校验来源的共享最新状态；37 driver为所选原信号别名；None用于实际间隔，每个完整必要更新组求值。38 backend选择数值核；derived为原派生声明。39 返回: 新owner。异常: 声明不一致、非数值/BOOL表达式、预算或取消错误；40 构造清理失败提供driver_expression_cleanup_owner重试句柄。41 """42 if (type(static) is not StaticExpression or type(static.bindings) is not DeclarationBindings43 or not isinstance(state, LatestSignalState)):44 raise TypeError('驱动求值需要静态表达式及真实最新状态')45 if backend not in ('reference', 'numpy'):46 raise ValueError('驱动求值后端需要reference或numpy')47 if static.root.dtype not in (D.FLOAT64, D.INT64, D.UINT64, D.BOOL,D.ENUM):48 raise ValueError('驱动求值需要数值、BOOL或ENUM表达式')49 self._budget, self._state = state.budget, state50 self._held_cost = self._fixed_cost = self._work_cost = 051 self._closed = self._closing = self._failed = False52 self._static = self._definitions = self._point_program = None53 self._last_revision = 054 self._last_group_time = None55 self._invalidated_at = None56 self._last_value = None57 self._driver, self._backend = driver, backend58 self._observation_ends = observation_ends59 try:60 # 新校验树、派生展开、点求值索引仍预付；bindings借用state/原计划，61 # 不再为每个指标复制计入整份信号声明。32倍只覆盖本次真实构造的树和临时结构。62 _reserve_owned_ram(self, 65536 + 32 * static_metadata_bytes(63 (static.root, static.text, static.normalized_ast, static.dependencies, derived)), '_fixed_cost')64 checked = validate_expression(static.text, static.bindings, derived=derived, unit=static.root.unit)65 if checked != static:66 raise ValueError('驱动表达式与原声明不一致')67 if static.bindings.session_id != state.session_id:68 raise ValueError('驱动表达式不属于当前会话')69 names = set(static.dependencies) | ({driver} if driver is not None else set())70 for name in names:71 declared = static.bindings.resolve(name)72 actual = state.definitions.get(name)73 # 唯一允许的启动补全是待DBC的有效期；字段、转换及其他身份仍逐项相同。74 if declared.valid_for_ns is None and actual is not None:75 declared = replace(declared, valid_for_ns=actual.valid_for_ns)76 if actual != declared:77 raise ValueError('驱动表达式与实际信号定义不同：' + str(name))78 static.bindings.require_synchronized(tuple(sorted(names)))79 # 固定申请已覆盖完整点对及每点16字节的只读列，参考后端直接二分原点对。80 self._static = pack_requested_lookups(checked) if backend == 'numpy' else checked81 # 内建单点直接计算；NumPy宿主只在自己的列接口边界装箱。82 # 原固定预算已预付节点/常量/查表引用，准备结果随当前owner关闭。83 self._point_program = PointProgram(self._static.root) if backend == 'numpy' else None84 self._has_where = contains_where(checked.root)85 self._definitions = tuple((name, state.definitions[name]) for name in static.dependencies)86 # 费用随表达式复杂度和依赖数增长，不随输入记录数增长。87 nodes, pending = 0, [checked.root]88 self._boolean_short_circuit = False89 while pending:90 node = pending.pop()91 nodes += 192 self._boolean_short_circuit |= node.operator in ('and', 'or')93 pending.extend(node.args)94 self._workspace = 65536 + nodes * 4096 + len(self._definitions) * 409695 except BaseException as exc:96 _cleanup_owner(self, exc, 'driver_expression_cleanup_owner', '驱动表达式')97 raise98 finally:99 # 构造失败保留异常栈时，不让临时编译树在退款后继续存活。100 checked = pending = node = static = state = derived = None101102 def _require_open(self):103 """拒绝失败或关闭后的求值。参数: 无。返回: None。异常: ResultClosedError。"""104 if self._closed or self._closing or self._failed or self._state.closed:105 raise ResultClosedError('驱动表达式或借用状态已关闭/失败')106107 def _context_at(self, time_ns, *, continuous=False):108 """读取当前持有值在实际时刻的有效性。参数: time_ns整数ns。返回: context及未来到期点。异常: 模型/分配错误。"""109 context, deadlines = {}, []110 for name, definition in self._definitions:111 sample = self._state.get(name)112 end = None if self._observation_ends is None else self._observation_ends.get(definition.source)113 if sample is None:114 # bool 的 choices 是输入编码到真假值的映射，只有 enum 才携带文字域。115 # 首样本未到时保持 MISSING，不把配置映射误当成观测值或枚举标签。116 value = ScalarValue(None, definition.dtype, False, Q.MISSING,117 tuple(dict.fromkeys(label for _, label in definition.choices))118 if definition.dtype is D.ENUM else ())119 elif end is not None and (time_ns >= end if continuous else time_ns > end):120 value = expire_value(sample.value, Q.MISSING)121 elif time_ns - sample.time_ns >= definition.valid_for_ns:122 value = expire_value(sample.value, Q.STALE)123 else:124 value = sample.value125 expires = sample.time_ns + definition.valid_for_ns126 available = value.valid or value.code is not None and value.code.valid127 if available and expires < (1 << 63):128 deadlines.append(expires)129 if available and end is not None:130 deadlines.append(end)131 context[name] = value132 return context, deadlines133134 def _evaluate_values(self, context, unobserved=None):135 """使用独立逐点实现。参数: typed值及可选规则缺项集合。返回: ScalarValue。异常: 原计算/分配错误。"""136 value = names = None137 cache = {}138 try:139 if self._backend == 'reference':140 value = expression_reference._evaluate(self._static.root, context, cache).value141 names = (node.name for node in cache if node.operator == 'name')142 else:143 value = self._point_program.evaluate(context, cache).value144 names = self._point_program.used_names(cache)145 # 缺项和部分质量只来自实际访问的分支，不能扫描静态依赖全集。146 partial = value.partial147 for name in names:148 if unobserved is not None and self._state.get(name) is None:149 unobserved.add(name)150 partial |= context[name].partial151 if value.valid and not value.partial and partial:152 value = replace(value,partial=True)153 return value154 finally:155 context = value = cache = names = None156157 def actual_update_at(self, time_ns):158 """检查真实数值更新。参数: 完整组纳秒。返回: bool。异常: 状态/预算/后端错误。"""159 return self._point_evidence(time_ns, 'updated')160161 def active_end_at(self, time_ns):162 """检查活动来源末端。参数: 查询纳秒。返回: bool。异常: 状态/预算/后端错误。"""163 return self._point_evidence(time_ns, 'source_end')164165 def actual_unknown_at(self, time_ns, point):166 """区分真实中断与并列 TTL。参数: 完整组纳秒、已知无效点值。返回: bool。异常: 状态/求值错误。"""167 return self._point_evidence(time_ns, 'invalid', point)168169 def _point_evidence(self, time_ns, kind, point=None):170 """检查点上的活动更新、未知原因或活动来源末端，不推进驱动。171172 参数: 查询纳秒、证据种类，invalid 另传当前实际点。返回: bool。173 异常: 状态、预算及后端求值错误；失败后只允许关闭。174 """175 self._require_open()176 source_end = kind == 'source_end'177 if not source_end and self._state.last_time_ns != time_ns:178 return False179 names = self._static.dependencies180 ends = self._observation_ends or {}181 if source_end and not any(ends.get(d.source) == time_ns for _, d in self._definitions):182 return False183 if not self._has_where:184 return source_end or any((sample := self._state.get(name)) is not None and sample.time_ns == time_ns for name in names)185 primary = None186 try:187 _reserve_owned_ram(self, self._workspace, '_work_cost')188 current, _ = self._context_at(time_ns)189 if source_end:190 return group_active(self._static.root, current,191 {name: ends.get(d.source) == time_ns for name, d in self._definitions}, self._backend,192 point_program=self._point_program)193 flags = {name: (sample := self._state.get(name)) is not None and sample.time_ns == time_ns for name in names}194 if kind == 'invalid':195 # 判定实际中断使用“同一 t 排除真实更新”而非左侧保持；否则 TTL 到期196 # 与旁列真实有效更新恰好同刻，会被误判为样本导致的中断。197 invalid = {n: real and not self._state.get(n).value.valid for n, real in flags.items()}198 if group_active(self._static.root, current, invalid, self._backend, point_program=self._point_program):199 return True200 prior = before_context(self._state, names, time_ns, ends=self._observation_ends)201 old = self._evaluate_values(prior)202 return old.valid or bool(point.quality & ~old.quality)203 previous = before_context(self._state, names, time_ns, ends=self._observation_ends, left_limit=True)204 return group_updates(self._static.root, current, previous, flags, self._backend,205 point_program=self._point_program)206 except BaseException as error:207 primary, self._failed = error, True208 raise209 finally:210 try:211 _release_owned_ram(self, self._work_cost, '_work_cost')212 except BaseException as cleanup:213 self._failed = True214 if primary is None:215 raise216 primary.add_note('活动依赖检查费用释放失败：' + type(cleanup).__name__)217218 def _output_deadline(self, time, output, deadlines):219 """选取首次可见失效。参数: 当前纳秒、输出、TTL/来源候选末端。220 返回: 整数截止或 None。异常: 候选边界上的求值错误。221 """222 if not self._has_where and not (output.dtype is D.BOOL and self._boolean_short_circuit):223 return min(deadlines, default=None)224 if output.valid:225 for candidate in sorted(set(deadlines)):226 if candidate >= time:227 future_context, _ = self._context_at(candidate, continuous=True)228 future = self._evaluate_values(future_context)229 if not future.valid or future.value != output.value:230 return candidate231 return None232233 def evaluate(self):234 """仅在最新完整组包含驱动更新时求值一次；其他组仅由state保存最新值。235236 参数: 无，读取借用state的当前完整组。返回: None或(time_ns, ScalarValue,237 实际有效截止)；None截止表示所有依赖在int64时间域内没有未来过期点。238 异常: 跳过未消费组、预算、计算结构或取消错误；失败后不再交付结果。239 约束: 必须在每次完整组提交后调用，不能用最新状态补算丢掉的历史组。240 """241 self._require_open()242 context = result = output = sample = None243 primary = None244 try:245 revision = self._state.revision246 if revision == self._last_revision:247 return None248 if revision != self._last_revision + 1:249 raise ValueError('驱动求值不能跳过已提交的完整组')250 time = self._state.last_time_ns251 previous_time, self._last_group_time = self._last_group_time, time252 self._invalidated_at = None253 # 非驱动依赖的无效更新也立即停止旧值保持，不能等下一条驱动才处理。254 # 只传必要的截止，不制造一个会参与N−1计数的驱动点。255 for name, _ in self._definitions:256 changed = self._state.get(name)257 definition = self._state.definitions[name]258 ended = None if self._observation_ends is None else self._observation_ends.get(definition.source)259 if ended is not None and ended <= time and (previous_time is None or ended > previous_time):260 self._invalidated_at = ended if self._invalidated_at is None else min(ended,self._invalidated_at)261 if changed is not None and changed.time_ns == time and not changed.value.valid:262 self._invalidated_at = time if self._invalidated_at is None else min(time,self._invalidated_at)263 driver_sample = self._state.get(self._driver) if self._driver is not None else None264 is_driver = self._driver is None or driver_sample is not None and driver_sample.time_ns == time265 # 已知FALSE and UNKNOWN等不能因为无关分支失效而被整体改成UNKNOWN。266 check_condition = (self._static.root.dtype is D.BOOL and self._boolean_short_circuit267 and self._invalidated_at is not None)268 if not is_driver and not check_condition and not self._has_where:269 self._last_revision = revision270 return None271 _reserve_owned_ram(self, self._workspace, '_work_cost')272 context, deadlines = self._context_at(time, continuous=not is_driver)273 output = self._evaluate_values(context)274 if self._has_where:275 # EOF 可能随下一条真实记录才确定。按该时间上仍生效的旧组求值，276 # 避免把新分支的失效倒填到旧区间，或让未选来源提前截止。277 self._invalidated_at = None278 ends = self._observation_ends or {}279 cutoffs = {ends[d.source] for _, d in self._definitions if d.source in ends and280 previous_time is not None and previous_time <= ends[d.source] <= time}281 for cutoff in sorted(cutoffs):282 point = before_context(self._state, self._static.dependencies, cutoff, ends=ends)283 if not group_active(self._static.root, point,284 {n: ends.get(d.source) == cutoff for n, d in self._definitions}, self._backend,285 point_program=self._point_program):286 continue287 prior = before_context(self._state, self._static.dependencies, cutoff, ends=ends, continuous=True)288 if not self._evaluate_values(prior).valid:289 self._invalidated_at = cutoff290 break291 if self._invalidated_at is None and not output.valid:292 self._invalidated_at = time293 if not is_driver:294 if (self._last_value is not None and output.valid and self._last_value.valid295 and output.value == self._last_value.value):296 self._invalidated_at = None297 self._last_revision = revision298 return None299 deadline = self._output_deadline(time, output, deadlines)300 if output.dtype is D.BOOL:301 if output.valid and self._last_value is not None and self._last_value.valid and output.value == self._last_value.value:302 self._invalidated_at = None303 self._last_value = output if output.dtype is D.BOOL else None304 self._last_revision = revision305 return time, output, deadline306 except BaseException as exc:307 primary = exc308 self._failed = True309 raise310 finally:311 context = columns = result = output = sample = driver_sample = value = changed = None312 deadlines = future_context = future = None313 try:314 _release_owned_ram(self, self._work_cost, '_work_cost')315 except BaseException as cleanup:316 self._failed = True317 if primary is None:318 raise319 primary.add_note('驱动表达式清理失败：' + type(cleanup).__name__)320321 @property322 def invalidated_at(self):323 """当前已消费组使数值依赖失效的实际时刻。参数: 无。返回: int或None。异常: 关闭/失败。"""324 self._require_open()325 return self._invalidated_at326327 def evaluate_at(self, time_ns, *, continuous=True, unobserved=None):328 """在当前最新状态上查询实际边界，供规则的TTL和等待调度使用。329330 参数: time_ns不得早于已提交组；continuous=False查询实际点，True查询右侧保持；331 unobserved可收集当前活动路径从未观测的别名，不记录未来推算中的缺项。332 返回: (ScalarValue, 下一次可能改变已知结果的实际截止或None)。不提交组、333 不改变驱动游标，不制造报文，也不保存查询轨迹。334 异常: 时间/状态/预算/数值错误；失败后仅可关闭，费用释放失败仍保原责任。335 """336 self._require_open()337 context = output = future_context = future = deadlines = None338 primary = None339 try:340 time = require_int64_ns(time_ns,'expression.query_time')341 if type(continuous) is not bool or self._state.last_time_ns is not None and time < self._state.last_time_ns:342 raise ValueError('边界查询需要bool视图且不能回读已覆盖的旧状态')343 _reserve_owned_ram(self,self._workspace,'_work_cost')344 context,deadlines = self._context_at(time,continuous=continuous)345 output = self._evaluate_values(context, unobserved)346 deadline = None347 if output.valid:348 for candidate in sorted(set(value for value in deadlines if value > time)):349 if not self._boolean_short_circuit and not self._has_where:350 deadline = candidate351 break352 future_context,_ = self._context_at(candidate,continuous=True)353 future = self._evaluate_values(future_context)354 if not future.valid or future.value != output.value:355 deadline = candidate356 break357 return output,deadline358 except BaseException as error:359 primary,self._failed = error,True360 raise361 finally:362 context = output = future_context = future = deadlines = None363 try:364 _release_owned_ram(self,self._work_cost,'_work_cost')365 except BaseException as cleanup:366 self._failed=True367 if primary is None:368 raise369 primary.add_note('规则边界求值费用释放失败：'+type(cleanup).__name__)370371 def close(self):372 """释放自有树/费用，借用state保持原所有者负责。参数: 无。返回: None。异常: 释放失败可重试。"""373 if self._closed:374 return375 self._closing = True376 self._static = self._definitions = self._state = self._last_value = self._observation_ends = self._point_program = None377 _release_owned_ram(self, self._work_cost, '_work_cost')378 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')379 self._closed = True380