chronaxis.execution.record_time_calculation
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用有限记录计算时间统计和真实样本统计；仅时间均值暂存必要值/权重。23输入与RecordIntegral共享点缓冲、递增时间和失效处理。复用原时间/样本数值核；4样本统计保真实更新标志，不把条件/TTL更新当新样本，样本均值使用原序求和。5均值首遍仅写值与权重，实际权重为int64纳秒，周期权重为float秒；6次遍按已知总权重规范化，避免先求面积溢出。7临时文件由本owner关闭时删除，不默认留现场；结果不附历史、贡献或时间对照。8调用者必须先确认来源EOF及关闭成功再finish，本对象不替代公共AnalysisRunner发布。9"""1011from ..core.moments import SAMPLE_MOMENTS12from ..core.quantiles import SAMPLE_QUANTILES13from dataclasses import dataclass14from pathlib import Path15from struct import Struct16from tempfile import TemporaryFile1718import numpy as np1920from ..core.calculations import CalculationDefinition, CalculationSeries, classify_calculation_value21from ..core.calculations import classify_coverage22from ..core.models import Interval23from ..core.types import SignalDType as D, QualityFlag as Q24from ..core.time_reduction import require_time_config, TimeReductionResult25from ..core.time_reduction_reference import ReferenceTimeReduction26from ..core.time_reduction_numpy import NumpyTimeReduction27from ..core.conditional_time_reduction_reference import ReferenceConditionalTimeReduction28from ..core.conditional_time_reduction_numpy import NumpyConditionalTimeReduction29from ..core.integral_sum_reference import ReferenceIntegralSum30from ..core.integral_sum_numpy import NumpyIntegralSum31from ..core.mean_weights_reference import normalized_mean_terms as reference_mean32from ..core.mean_weights_numpy import normalized_mean_terms as numpy_mean33from ..results.budget import ResourceBudget, _reserve_owned_ram, _release_owned_ram, _cleanup_owner34from ..results.errors import BudgetExceededError35from .history import _metadata_bytes36from .record_point_buffer import RecordPointBuffer37from ..core.sample_reduction import require_sample_config38from ..core.calculation_output_reference import convert_result_reference39from ..core.calculation_output_numpy import convert_result_numpy40from ..core.sample_reduction_reference import ReferenceSampleReduction41from ..core.sample_reduction_numpy import NumpySampleReduction42from .record_sample_series import sample_series434445@dataclass(frozen=True)46class RecordTimeResult:47 """保存时间统计的必要输出，整数极值不转成浮点数。"""4849 value: int | float | None50 unit: str51 status: str52 reason: str53 quality: Q54 coverage_status: str | None = None555657class RecordTimeCalculation(RecordPointBuffer):58 """拥有一份点缓冲、数值状态及可选均值临时文件，失败后只能清理。"""5960 def __init__(self, budget, spec, input_unit, dtype, step_ns, *, backend='numpy',61 batch_intervals=64, partial_limit=4096, conditioned=False, work_dir=None,hosted=None):62 """预付有限状态，均值的文件延迟到首批有效项才创建。6364 参数: 共享budget、时间统计spec、已求值类型/单位；step_ns为None实际差或正周期；65 后端和批上限明确；66 conditioned表示外BOOL条件，time_weighted_mean需要既存work_dir。67 返回: 新owner。异常: 配置/预算失败；持续清理故障附record_time_cleanup_owner。68 """69 if not isinstance(budget, ResourceBudget) or not isinstance(spec, CalculationDefinition):70 raise TypeError('记录时间统计需要共享预算和CalculationDefinition')71 if backend not in ('reference', 'numpy') or type(conditioned) is not bool:72 raise ValueError('时间统计需要明确后端及bool条件开关')73 if type(batch_intervals) is not int or not 1 <= batch_intervals <= 4096:74 raise ValueError('记录批上限需要1到4096')75 self._samples=spec.operation in ('sample_count','sample_minimum','sample_maximum','sample_mean', *SAMPLE_MOMENTS, *SAMPLE_QUANTILES)76 # 宿主归约自行管理其状态；只有内建精确分位数在本层预估样本保留费用。77 self._builtin_quantile = hosted is None and spec.operation in SAMPLE_QUANTILES78 if hosted is not None:79 from ..core.calculations import calculation_conversion80 from ..core.unit_registry import resolve_unit,conversion81 scale,shift=conversion(resolve_unit(input_unit),resolve_unit(hosted.input.unit))82 if (not self._samples or step_ns is not None or dtype is not hosted.input.dtype or83 scale.factor!=1 or scale.pi_power or shift):84 raise ValueError('宿主样本归约需要声明的输入dtype/单位和真实更新')85 self._conversion=calculation_conversion(spec,hosted.output.unit)86 elif self._samples:87 if step_ns is not None:raise ValueError('样本统计不使用固定或DBC周期')88 self._conversion=require_sample_config(spec,input_unit,dtype,partial_limit)89 else:90 self._conversion = require_time_config(spec, input_unit, dtype, None, partial_limit, step_ns=step_ns)91 self._unit,self._factor=self._conversion.unit,self._conversion.factor92 self._work_dir = None93 if spec.operation == 'time_weighted_mean':94 if work_dir is None or not Path(work_dir).is_dir():95 raise ValueError('均值需要已存在的专用工作目录')96 self._work_dir = Path(work_dir)97 self._budget, self._spec, self._input_unit, self._dtype = budget, spec, input_unit, dtype98 self._step_ns, self._limit, self._conditioned = step_ns, batch_intervals, conditioned99 self._backend, self._partial_limit = backend, partial_limit100 self._held_cost = self._fixed_cost = self._workspace_cost = 0101 self._closed = self._closing = self._failed = self._finished = False102 self._rows = self._reduction = self._result = self._file = self._mean_sum = None103 self._last_time = None104 self._covered_ns = self._known_ns = self._selected_ns = self._selected_unknown_ns = self._false_ns = 0105 self._quality = Q.NONE106 self._transform_partial = False107 self._coverage_value_ns=0 if spec.minimum_coverage>0 else None108 self._record_count = 0109 self._codec = Struct('>dq' if step_ns is None else '>dd') # 仅值/权重，实际纳秒不降为浮点。110 try:111 _reserve_owned_ram(self, 65536 + partial_limit * 512 + (batch_intervals+1)*1024112 + 16*_metadata_bytes((spec, input_unit)), '_fixed_cost')113 self._rows = []114 kernel = ((ReferenceConditionalTimeReduction if backend == 'reference' else NumpyConditionalTimeReduction)115 if conditioned else (ReferenceTimeReduction if backend == 'reference' else NumpyTimeReduction))116 if hosted is not None:117 from .hosted_sample_reduction import HostedSampleReduction118 self._reduction=HostedSampleReduction(spec,input_unit,dtype,hosted,self._conversion,backend,budget,partial_limit)119 else:120 self._reduction = ((ReferenceSampleReduction if backend=='reference' else NumpySampleReduction)121 (spec,input_unit,dtype,partial_limit=partial_limit,quantile_budget=budget) if self._samples else122 kernel(spec, input_unit, dtype, step_ns=step_ns, partial_limit=partial_limit))123 except BaseException as exc:124 self._reduction=getattr(exc,'hosted_sample_cleanup_owner',self._reduction)125 _cleanup_owner(self, exc, 'record_time_cleanup_owner', '记录时间统计')126 raise127128 def _append_mean(self, values, weights):129 """只写本批选中的必要均值项；费用先预约，不缓存完整输入。130131 参数: 已由核校验的有限数值/权重。返回: None。异常: 配额、短写、IO或取消传播。132 """133 if len(values) == 0:134 return135 if self._file is None:136 self._file = TemporaryFile(mode='w+b', buffering=0, dir=self._work_dir, prefix='mean-')137 payload = bytearray(len(values)*self._codec.size)138 view = None139 try:140 for index, (value, weight) in enumerate(zip(values, weights)):141 self._codec.pack_into(payload, index*self._codec.size, float(value),142 int(weight) if self._step_ns is None else float(weight))143 self._budget.reserve_spool(len(payload))144 view = memoryview(payload)145 position = 0146 while position < len(view):147 written = self._file.write(view[position:])148 if type(written) is not int or not 0 < written <= len(view)-position:149 raise OSError('均值暂存短写或无写入进展')150 position += written151 self._record_count += len(values)152 finally:153 if view is not None:154 view.release()155 values = weights = payload = view = None156157 def _flush(self):158 """在共享预算内分块处理样本统计，保留每块的共同右端。159160 参数: 无，使用已预付的点缓冲。返回: None。161 异常: 最小两点工作区也放不下时报告计算项与用途；数值、取消原错传播。162 不降低每行估算、不放宽限额，也不丢弃、重排或重复累计真实样本。163 """164 if not self._samples:165 return self._flush_block()166 remaining = None167 try:168 while len(self._rows) >= 2:169 # 读取队列和其他指标仍存活，不能假定本指标可独占整个预算。170 # 系数沿用原样本核对最多两个TTL插入点的保守预付。171 available = self._budget.memory_budget_bytes - self._budget.ram_bytes172 count = min(len(self._rows), (available - 65536) // 36864)173 if self._builtin_quantile:174 # 临时工作区与本块新增的持久样本同时存活。右端不属于半开块，175 # TTL插入点也不是真实样本，故至多保留count-1个值。按缓冲器176 # 的实际计费规则找最大可用块，不用固定百分比吞掉剩余预算。177 lower, upper = 0, max(0, count)178 while lower < upper:179 middle = (lower + upper + 1) // 2180 needed = (65536 + middle * 36864181 + self._reduction._quantile._append_memory_bytes(max(0, middle - 1)))182 if needed <= available:183 lower = middle184 else:185 upper = middle - 1186 count = lower187 if count < 2:188 minimum = 139264189 if self._builtin_quantile:190 minimum += self._reduction._quantile._append_memory_bytes(1)191 raise BudgetExceededError(192 f'计算项 {self._spec.calculation_id}（{self._spec.operation}）'193 f'样本统计工作区不足：最小两点及必要样本存储需要{minimum}字节，当前可用{available}字节；'194 '已尝试按可用预算分块，请减少同时运行的计算项或读取批量，'195 '核实必要状态后再调整memory_budget_bytes')196 if count == len(self._rows):197 self._flush_block()198 break199 # 两份短列表的引用槽由既有点缓冲1024字节/行覆盖；不复制值对象。200 # 右端在前块半开范围内不计样本，下一块再作为左端处理一次。201 remaining = self._rows[count - 1:]202 self._rows = self._rows[:count]203 self._flush_block()204 self._rows, remaining = remaining, None205 finally:206 # 中途失败不能持有已退款缓冲以外的尾部引用；外层负责标失败并关闭。207 remaining = None208209 def _flush_block(self):210 """计算完整记录对，只留共享右端；均值写选中项，其他指标不写盘。211212 参数: 无。返回: None。异常: 预算/数值结构/IO/取消，失败后不得交付前缀。213 """214 if len(self._rows) < 2:215 return216 series = block = facts = None217 primary = None218 try:219 _reserve_owned_ram(self, 65536 + len(self._rows)*(36864 if self._samples else 12288), '_workspace_cost')220 times = tuple(row[0] for row in self._rows)221 values = tuple(row[1] for row in self._rows)222 deadlines = tuple(row[2] for row in self._rows) if any(row[2] is not None for row in self._rows) else None223 conditions = tuple(row[3] for row in self._rows) if self._conditioned else None224 options = {'condition_valid_until_ns': tuple(row[4] for row in self._rows)} if self._conditioned else {}225 series = (sample_series(self._rows,self._input_unit,self._conditioned) if self._samples else226 CalculationSeries(Interval(times[0], times[-1]), times, values, (True,)*len(times), self._input_unit, conditions=conditions))227 mean = self._spec.operation == 'time_weighted_mean'228 block = (self._reduction.push(series) if self._samples else229 self._reduction.push(series, valid_until_ns=deadlines,230 retain_unknown_ranges=False, retain_mean_terms=mean, **options))231 facts = block.facts if mean else block232 from .transform_partial import partial_in_block233 if partial_in_block(self._rows,self._spec.window,samples=self._samples):234 self._transform_partial=True235 if self._coverage_value_ns is not None:236 self._coverage_value_ns+=facts.value_valid_duration_ns237 if mean:238 self._append_mean(block.values, block.weights)239 span = 0 if facts.actual_window is None else facts.actual_window.end_ns-facts.actual_window.start_ns240 self._covered_ns += span241 self._known_ns += facts.condition_known_duration_ns if self._conditioned else span242 self._selected_ns += facts.selected_duration_ns243 self._selected_unknown_ns += facts.selected_value_unknown_duration_ns if self._conditioned else span-facts.value_valid_duration_ns244 if self._conditioned:245 self._false_ns += facts.false_duration_ns246 self._quality |= facts.quality247 self._rows[:] = self._rows[-1:]248 except BaseException as exc:249 primary, self._failed = exc, True250 raise251 finally:252 series = block = facts = times = values = deadlines = conditions = options = None253 try:254 _release_owned_ram(self, self._workspace_cost, '_workspace_cost')255 except BaseException as cleanup:256 self._failed = True257 if primary is None:258 raise259 primary.add_note('时间统计临时费用释放失败：'+type(cleanup).__name__)260261 def _finish_mean(self):262 """分批回读必要数值/权重，复用原归一化及原序求和，最终只换单位一次。263264 参数: 无。返回: TimeReductionResult。异常: 临时数据短缺/多出、IO、预算或取消。265 """266 if self._record_count == 0:267 return TimeReductionResult(None, self._unit, False, Q.NONE, 0)268 denominator = (self._reduction.selected_duration_ns if self._step_ns is None269 else self._reduction.selected_weight_seconds)270 self._file.seek(0)271 self._mean_sum = (ReferenceIntegralSum if self._backend == 'reference' else NumpyIntegralSum)(self._partial_limit)272 remaining, quality = self._record_count, Q.NONE273 payload = pairs = values = weights = terms = keep = None274 try:275 while remaining:276 count = min(remaining, self._limit)277 payload = self._file.read(count*self._codec.size)278 if len(payload) != count*self._codec.size:279 raise ValueError('均值必要暂存数据不完整')280 if self._backend == 'reference':281 pairs = tuple(self._codec.iter_unpack(payload))282 values, weights = tuple(item[0] for item in pairs), tuple(item[1] for item in pairs)283 terms, failure = reference_mean(values, weights, denominator)284 keep = (True,)*len(terms)285 else:286 pairs = np.frombuffer(payload, dtype=[('value', '>f8'),287 ('weight', '>i8' if self._step_ns is None else '>f8')])288 values, weights = pairs['value'], pairs['weight']289 terms, failure = numpy_mean(values, weights, denominator)290 keep = np.ones(len(terms), dtype=bool)291 keep.setflags(write=False)292 quality |= failure293 if quality == Q.NONE:294 self._mean_sum.push_values(terms, keep)295 remaining -= count296 if self._file.read(1):297 raise ValueError('均值必要暂存含未登记数据')298 if quality != Q.NONE:299 return TimeReductionResult(None, self._unit, True, quality, self._record_count)300 # v4先完成原量纲均值，再由所选后端一次换单位；偏移不能随块数重复。301 number = self._mean_sum.finish(self._input_unit,302 1.0)303 if not number.calculation_failed:304 convert = convert_result_reference if self._backend == 'reference' else convert_result_numpy305 value, quality = convert(number.value, self._dtype, self._conversion)306 return TimeReductionResult(value, self._unit, quality != Q.NONE, quality, self._record_count)307 return TimeReductionResult(number.value, self._unit, number.calculation_failed, number.quality, self._record_count)308 finally:309 payload = pairs = values = weights = terms = keep = None310311 def finish(self):312 """上游确认EOF后结算，并关闭删除均值临时文件，再交付精简结果。313314 参数: 无。返回: RecordTimeResult，重复调用同一结果。异常: 失败拒绝前缀，close可重试。315 """316 self._require_open()317 if self._finished:318 return self._result319 primary = None320 try:321 self._flush()322 workspace = 65536 + (self._limit+1)*12288 if self._spec.operation == 'time_weighted_mean' else 4096323 _reserve_owned_ram(self, workspace, '_workspace_cost')324 number = self._finish_mean() if self._spec.operation == 'time_weighted_mean' else self._reduction.finish()325 value, status, reason = classify_calculation_value(self._spec, number.value, self._covered_ns,326 self._known_ns, self._selected_ns, self._selected_unknown_ns,327 all_false=self._conditioned and self._covered_ns>0 and self._false_ns==self._covered_ns,328 calculation_failed=number.calculation_failed,329 sample_count=number.valid_sample_count if self._samples else 0)330 if self._file is not None:331 self._file.close()332 self._file = None333 coverage=None if self._coverage_value_ns is None else classify_coverage(334 self._spec,self._covered_ns,self._coverage_value_ns,self._known_ns)335 self._result = RecordTimeResult(value, self._unit, status, reason, self._quality | number.quality,coverage)336 if self._transform_partial or getattr(self._reduction,'partial',False):337 from .transform_partial import partial_result338 self._result=partial_result(self._result,self._spec.partial_policy)339 self._finished = True340 return self._result341 except BaseException as exc:342 primary, self._failed = exc, True343 raise344 finally:345 number = None346 try:347 _release_owned_ram(self, self._workspace_cost, '_workspace_cost')348 except BaseException as cleanup:349 self._failed = True350 if primary is None:351 raise352 primary.add_note('时间统计结算费用释放失败：'+type(cleanup).__name__)353354 @property355 def closed(self):356 """读取自有文件/费用是否全部释放。参数: 无。返回: bool。异常: 无。"""357 return self._closed358359 def close(self):360 """清理自有数值状态和临时文件，释放失败保留重试入口，不碰输入文件。361362 参数: 无。返回: None。异常: 文件关闭/费用释放失败可重试；重复关闭安全。363 """364 if self._closed:365 return366 self._closing = True367 if self._samples and self._reduction is not None:368 self._reduction.close()369 self._rows = self._reduction = self._mean_sum = self._result = None370 if self._file is not None:371 self._file.close()372 self._file = None373 _release_owned_ram(self, self._workspace_cost, '_workspace_cost')374 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')375 self._closed = True376