chronaxis.execution.record_integral
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在有限内存内累计已求值的驱动记录对，保留实际端点直到比例裁剪完成。23输入为已归一化并按驱动同刻政策归组的(time_ns, ScalarValue[, valid_until_ns])有限批；输出仅含4积分值、单位、完成状态和质量的内部结果。复用原两后端面积及原序误差求和，5一次只持当前批和一条跨批前值，不创建历史文件、原帧副本、贡献或未知范围列表。6调用方负责绑定/表达式/有效期判定，并在输入确实结束后调用finish；本组件不7读取文件或证明来源EOF。实际间隔使用必要更新间的真实差，固定/DBC使用8所选驱动记录对；有效期都按实际截止裁剪，没有完整记录对时不虚构尾段。9"""1011from dataclasses import dataclass1213from ..core.integral_sum import require_partial_limit14from ..core.integral_sum_reference import ReferenceIntegralSum15from ..core.integral_sum_numpy import NumpyIntegralSum16from ..core.integral_terms_reference import compute_integral_terms_reference17from ..core.integral_terms_numpy import compute_integral_terms_numpy18from ..core.conditional_integral_terms_reference import compute_conditional_integral_terms_reference19from ..core.conditional_integral_terms_numpy import compute_conditional_integral_terms_numpy20from ..core.calculations import CalculationDefinition, CalculationSeries, calculation_unit, classify_calculation_value21from ..core.calculations import classify_coverage22from ..core.models import Interval23from ..core.numeric import ScalarValue24from ..core.record_intervals import require_record_step25from ..core.types import SignalDType as D, QualityFlag as Q26from ..results.budget import ResourceBudget, _reserve_owned_ram, _release_owned_ram, _cleanup_owner27from ..results.errors import ResultClosedError28from .record_point_buffer import RecordPointBuffer29from .history import _metadata_bytes303132@dataclass(frozen=True)33class RecordIntegralResult:34 """内部积分结果，不附逐条输入、时间对照或默认贡献记录；不充当公开发布清单。"""3536 value: float | None # 已换到请求单位；未知与有效零分开。37 unit: str # 输出单位，仅最终转换一次。38 status: str # 沿既有complete/partial/indeterminate政策。39 reason: str # 正常、观测不足或数值失败等必要原因。40 quality: Q # 输入和数值计算质量，不包含逐条副本。41 coverage_status: str | None = None424344class RecordIntegral(RecordPointBuffer):45 """拥有有限驱动点缓冲和累计状态；失败后只能关闭，完成后只读结果。4647 consume借用外部有限批；留下的点只含整数时间和冻结数值，费用归本实例。48 同刻冲突须由上游既有同刻政策解决，不能在此任意覆盖或累计零跨度记录。49 """5051 def __init__(self, budget, spec, input_unit, dtype, step_ns, *, backend='numpy',52 batch_intervals=64, partial_limit=4096, conditioned=False):53 """预付有界状态及点缓冲，数值临时数组在每次归约前另预约。5455 参数: budget共享账本，spec为integral请求；input_unit/dtype为已56 求值列单位/类型；step_ns为None（实际差）或固定/已核实DBC正周期；后端独立选择；57 batch_intervals为1—4096区间数，partial_limit为误差项容量；conditioned指定显式BOOL条件。58 返回: 新owner。异常: 配置/预算/取消错误；构造清理失败通过异常的59 record_integral_cleanup_owner提供重试入口。60 """61 if not isinstance(budget, ResourceBudget) or not isinstance(spec, CalculationDefinition):62 raise TypeError('记录流积分需要共享预算和CalculationDefinition')63 if spec.operation != 'integral':64 raise ValueError('记录流积分需要integral')65 if backend not in ('reference', 'numpy') or dtype not in (D.INT64, D.UINT64, D.FLOAT64):66 raise ValueError('记录流积分需要明确后端和数值dtype')67 if type(batch_intervals) is not int or not 1 <= batch_intervals <= 4096:68 raise ValueError('记录流区间批上限需要1到4096')69 if type(conditioned) is not bool:70 raise TypeError('条件选择需要bool')71 require_record_step(step_ns, ())72 limit = require_partial_limit(partial_limit)73 self._unit, self._factor = calculation_unit(spec, input_unit)74 self._budget, self._spec, self._input_unit, self._dtype = budget, spec, input_unit, dtype75 self._step_ns, self._limit = step_ns, batch_intervals76 self._conditioned = conditioned77 self._held_cost = self._fixed_cost = self._workspace_cost = 078 self._closed = self._closing = self._failed = self._finished = False79 self._rows = self._sum = self._result = None80 self._last_time = None81 self._covered_ns = 082 self._known_ns = self._selected_ns = self._selected_unknown_ns = self._false_ns = 083 self._quality = Q.NONE84 self._transform_partial = False85 self._coverage_value_ns=0 if spec.minimum_coverage>0 else None86 try:87 # 固定费覆盖误差求和的有限候选及点缓冲，不按总记录数扩大。88 _reserve_owned_ram(self, 65536 + limit * 256 + (batch_intervals + 1) * 102489 + 16 * _metadata_bytes((spec, input_unit)), '_fixed_cost')90 self._rows = []91 self._sum = (ReferenceIntegralSum if backend == 'reference' else NumpyIntegralSum)(limit)92 self._compute = compute_integral_terms_reference if backend == 'reference' else compute_integral_terms_numpy93 if conditioned:94 self._compute = (compute_conditional_integral_terms_reference if backend == 'reference'95 else compute_conditional_integral_terms_numpy)96 except BaseException as exc:97 _cleanup_owner(self, exc, 'record_integral_cleanup_owner', '记录流积分')98 raise99100 def _flush(self):101 """消费当前完整记录对并留下一个共享端点，首条/末条不凭空增加周期。102103 参数: 无，使用当前自有缓冲。返回: None。异常: 数值结构/预算/取消；104 清理异常不覆盖原错误，临时申请保留在owner供close重试。105 """106 if len(self._rows) < 2:107 return108 series = terms = None109 primary = None110 try:111 _reserve_owned_ram(self, 65536 + len(self._rows) * 8192, '_workspace_cost')112 times = tuple(item[0] for item in self._rows)113 values = tuple(item[1] for item in self._rows)114 # 截止仅为本批计算所需；不保存来源副本或另一列累计时钟。115 deadlines = (tuple(item[2] for item in self._rows)116 if any(item[2] is not None for item in self._rows) else None)117 conditions = tuple(item[3] for item in self._rows) if self._conditioned else None118 condition_deadlines = (tuple(item[4] for item in self._rows)119 if self._conditioned and any(item[4] is not None for item in self._rows) else None)120 series = CalculationSeries(Interval(times[0], times[-1]), times, values,121 (True,) * len(times), self._input_unit, conditions=conditions)122 # 窗口仍由原spec裁剪，分母保留完整记录对。按请求省去未知范围对象。123 options = {'condition_valid_until_ns': condition_deadlines} if self._conditioned else {}124 terms = self._compute(series, self._spec, step_ns=self._step_ns,125 retain_unknown_ranges=False, valid_until_ns=deadlines, **options)126 self._sum.push_values(terms.areas, terms.valid)127 facts = terms.facts128 from .transform_partial import partial_in_block129 if partial_in_block(self._rows,self._spec.window,samples=False):130 self._transform_partial=True131 if self._coverage_value_ns is not None:132 self._coverage_value_ns+=facts.value_valid_duration_ns133 if facts.actual_window is not None:134 span = facts.actual_window.end_ns - facts.actual_window.start_ns135 self._covered_ns += span136 self._known_ns += facts.condition_known_duration_ns if self._conditioned else span137 self._selected_ns += facts.selected_duration_ns if self._conditioned else facts.value_valid_duration_ns138 self._selected_unknown_ns += (facts.selected_value_unknown_duration_ns if self._conditioned139 else span - facts.value_valid_duration_ns)140 if self._conditioned:141 self._false_ns += facts.false_duration_ns142 self._quality |= facts.quality143 self._rows[:] = self._rows[-1:]144 except BaseException as exc:145 primary = exc146 self._failed = True147 raise148 finally:149 # 断开有限临时列后再退款；释放失败不允许调用方取得成功结果。150 series = terms = times = values = facts = deadlines = conditions = condition_deadlines = options = None151 try:152 _release_owned_ram(self, self._workspace_cost, '_workspace_cost')153 except BaseException as cleanup:154 self._failed = True155 if primary is None:156 raise157 primary.add_note('记录流积分临时费用清理失败：' + type(cleanup).__name__)158159 def finish(self):160 """输入调用方确认结束后结算最后完整记录对，返回精简内部结果。161162 参数: 无；不能代替读取器EOF/关闭成功的确认。返回: RecordIntegralResult，163 重复调用返回同一结果。异常: 预算/数值结构/取消错误，不发布前缀。164 """165 self._require_open()166 if self._finished:167 return self._result168 try:169 self._flush()170 number = self._sum.finish(self._unit, self._factor)171 value, status, reason = classify_calculation_value(self._spec, number.value, self._covered_ns,172 self._known_ns, self._selected_ns, self._selected_unknown_ns,173 calculation_failed=number.calculation_failed,174 all_false=self._conditioned and self._covered_ns > 0 and self._false_ns == self._covered_ns)175 coverage=None if self._coverage_value_ns is None else classify_coverage(176 self._spec,self._covered_ns,self._coverage_value_ns,self._known_ns)177 self._result = RecordIntegralResult(value, self._unit, status, reason, self._quality | number.quality,coverage)178 if self._transform_partial:179 from .transform_partial import partial_result180 self._result=partial_result(self._result,self._spec.partial_policy)181 self._finished = True182 return self._result183 except BaseException:184 self._failed = True185 raise186187 @property188 def closed(self):189 """读取自有费用完全释放状态。参数: 无。返回: bool。异常: 无。"""190 return self._closed191192 def close(self):193 """释放自有缓冲/数值状态及费用，不操作外部输入和目录。194195 参数: 无。返回: None；重复关闭安全。异常: 预算释放失败可重试，不假报已关闭。196 """197 if self._closed:198 return199 self._closing = True200 self._rows = self._sum = self._result = None201 _release_owned_ram(self, self._workspace_cost, '_workspace_cost')202 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')203 self._closed = True204