chronaxis.core.sample_reduction_reference
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""独立标量归约有限真实样本，保留原dtype极值与原序样本算术均值。23push分别计算值/条件覆盖，只将半开窗中有效/TRUE原值真实更新计入候选。极值4保留原整数；mean先检查float精度，原单位样本按原序推进有界误差核，finish完成5原量纲总和后除真实样本数，再换输出单位一次。原序中间溢出沿既有故障政策保持6失败，不改为time_weighted_mean的规范化权重；除法/换算故障不改输入覆盖。不调用NumPy、7不读来源。精确分位数按专用配额保留被选中的原dtype样本，异常后由owner清理。8"""910# 标量覆盖及候选独立于数组后端；仅复用本参考原序误差状态。11from .moments import SAMPLE_MOMENTS12from .quantiles import SAMPLE_QUANTILES, ExactQuantileBuffer13from .moments_reference import ReferenceMoments14from math import isfinite15from .calculation_output_reference import convert_result_reference16from .integral_sum_reference import ReferenceIntegralSum17from .calculations import merge_ranges18from .models import Interval19from .numeric import ScalarValue, to_float_reference20from .sample_reduction import SampleCalculationFacts, SampleReductionResult, require_sample_config, require_sample_series21from .types import QualityFlag as Q, SignalDType as D222324class ReferenceSampleReduction:25 """累计当前范围的真实样本；分位数仅在请求时按共享预算保留原dtype样本。"""2627 def __init__(self, spec, input_unit, dtype, *, partial_limit=4096, quantile_budget=None):28 """核对原配置并建立本参考后端有限状态。2930 参数: spec/input_unit/dtype为原声明，partial_limit为正原生int误差容量。31 quantile_budget为精确分位数单独申请存储/排序空间时使用的共享预算。32 返回: None。异常: 配置或分配错误传播。33 """34 # 共享静态门槛先检查明确样本能力、数值类型和输出单位。35 self._conversion = require_sample_config(spec, input_unit, dtype, partial_limit)36 self._unit, self._factor = self._conversion.unit, self._conversion.factor37 self._spec, self._input_unit, self._dtype = spec, input_unit, dtype38 self._identity = self._candidate = None39 # 精确分位数单独保存被选中的原dtype；其他样本指标无需创建存储。40 self._quantile = (ExactQuantileBuffer(spec.operation,spec.q,dtype,'reference',quantile_budget)41 if spec.operation in SAMPLE_QUANTILES else None)42 self._moments = ReferenceMoments(spec.operation) if spec.operation in SAMPLE_MOMENTS else None43 self._value_duration = self._selected_duration = self._samples = 044 self._sum = ReferenceIntegralSum(partial_limit) if spec.operation == 'sample_mean' else None4546 @property47 def selected_duration_ns(self):48 """读取累计TRUE值有效时长。参数: 无。返回: 原生int纳秒。异常: 无。"""49 return self._selected_duration # 不作为样本均值分母。5051 @property52 def value_valid_duration_ns(self):53 """读取累计值有效时长。参数: 无。返回: 原生int纳秒。异常: 无。"""54 return self._value_duration # 与条件TRUE/FALSE/UNKNOWN独立。5556 @property57 def valid_sample_count(self):58 """读取有效真实样本数。参数: 无。返回: 原生int。异常: 无。"""59 return self._samples # TTL/条件更新或裁剪保持不增加此数。6061 def push(self, series, *, sample_consumer=None):62 """消费有限块双覆盖与真实样本，不以保持区间代替样本。6364 参数: series为同unit/dtype有限CalculationSeries，条件可省略为全TRUE。65 返回: SampleCalculationFacts；mean精度失败进入输入覆盖，求和故障不改覆盖。66 异常: 结构、身份、容量、分配或取消传播；外层异常后不能接着使用前缀状态。67 """68 # 输入结构和交集先核对，跨块原表达式绑定身份必须保持。69 actual = require_sample_series(series, self._spec, self._input_unit, self._dtype)70 if self._identity is not None and self._identity != series.binding_identity:71 raise ValueError('样本归约不能跨绑定身份累计')72 self._identity = series.binding_identity73 # 只有mean走浮点精度门槛，min/max不把大整数舍入为float。74 values = tuple(to_float_reference(value) for value in series.values) if self._spec.operation == 'sample_mean' or self._moments is not None else series.values75 unconditional = ScalarValue(True, D.BOOL)76 start, end = (actual.start_ns, actual.end_ns) if actual is not None else (0, 0)77 value_duration = known_duration = selected_duration = selected_unknown = false_duration = samples = 078 value_unknown, condition_unknown, chosen = [], [], []79 quality = Q.NONE80 # 独立逐正区间计算双覆盖；这只说明保持事实，不向数值样本归约添加值。81 for index in range(len(series.times_ns)-1):82 left, right = max(series.times_ns[index], start), min(series.times_ns[index+1], end)83 if left >= right:84 continue85 value = values[index]86 condition = series.conditions[index] if series.conditions is not None else unconditional87 dt = right-left88 quality |= series.diagnostics[index] | value.quality | condition.quality89 if value.valid:90 value_duration += dt91 else:92 value_unknown.append(Interval(left, right))93 if condition.valid:94 known_duration += dt95 if not condition.value:96 false_duration += dt97 elif value.valid:98 selected_duration += dt99 else:100 selected_unknown += dt101 else:102 condition_unknown.append(Interval(left, right))103 # 半开窗口真实原值更新才进入候选；外条件变化和共享右端不产生样本。104 for index, time in enumerate(series.times_ns):105 if not start <= time < end or not series.actual_value_updates[index]:106 continue107 value = values[index]108 condition = series.conditions[index] if series.conditions is not None else unconditional109 quality |= series.diagnostics[index] | value.quality | condition.quality110 if not value.valid or not condition.valid or not condition.value:111 continue112 samples += 1113 if sample_consumer is not None:114 chosen.append(value)115 elif self._spec.operation == 'sample_mean' or self._moments is not None:116 chosen.append(value.value)117 elif self._quantile is not None:118 # 虽可在同一批中见到许多样本，先预付每个被保留的值，再追加。119 self._quantile.append((value.value,))120 elif self._spec.operation != 'sample_count':121 # 极值只存一个原dtype候选，不按时长加权或保留跨块样本集合。122 self._candidate = value.value if self._candidate is None else (min if self._spec.operation == 'sample_minimum' else max)(self._candidate, value.value)123 # raw原单位样本按原序推进；中间溢出不会因后续抵消恢复正常。124 if sample_consumer is not None and chosen:125 sample_consumer(tuple(chosen))126 if self._sum is not None:127 self._sum.push_values(tuple(chosen), (True,)*len(chosen))128 elif self._moments is not None:129 self._moments.push(chosen)130 # 跨块只留精确计数，返回的有限双未知和来源仍需外层owner计费及管理生命周期。131 self._value_duration += value_duration132 self._selected_duration += selected_duration133 self._samples += samples134 return SampleCalculationFacts(self._spec, actual, value_duration, known_duration, selected_duration,135 selected_unknown, false_duration, samples, merge_ranges(value_unknown), merge_ranges(condition_unknown),136 quality, self._input_unit, self._dtype, series.binding_identity, series.source_refs)137138 def finish(self):139 """完成原单位总和、除真实样本数，再仅一次转换输出单位。140141 参数: 无。返回: SampleReductionResult快照，不清空状态或分类覆盖政策。142 异常: 配置/分配错误传播；原序和、除法、精度或换算故障返回None及质量。143 """144 # 空真实样本没有数值证据，即使保持值长期有效也不能补正常零。145 if self._spec.operation == 'sample_count' and self._samples:146 # 整数计数已跨块累计；不需要任何原数值候选或单位换算。147 return SampleReductionResult(self._samples, '1', False, Q.NONE, self._samples)148 if self._quantile is not None:149 value, quality = self._quantile.finish()150 if value is not None:151 value, quality = convert_result_reference(value, self._dtype, self._conversion)152 return SampleReductionResult(value, self._unit, quality != Q.NONE, quality, self._samples)153 if self._moments is not None:154 value, quality = self._moments.finish(self._spec.ddof, self._conversion.factor)155 return SampleReductionResult(value, self._unit, quality != Q.NONE, quality, self._samples,156 insufficient_samples=self._spec.ddof == 1 and self._samples == 1)157 if self._samples == 0:158 return SampleReductionResult(None, self._unit, False, Q.NONE, 0)159 quality = Q.NONE160 if self._sum is not None:161 # 此处原单位factor=1先完成总和，不能先输出换单位或先归约块均值。162 number = self._sum.finish(self._input_unit)163 if number.calculation_failed:164 return SampleReductionResult(None, self._unit, True, number.quality, self._samples)165 value = number.value / self._samples166 if not isfinite(value) or value == 0 and number.value != 0:167 quality |= Q.NONFINITE if not isfinite(value) else Q.PRECISION_LOSS168 else:169 value = self._candidate170 # v4换算包含偏移；输入/归约故障不能被正常单位结果覆盖。171 if quality == Q.NONE:172 value, quality = convert_result_reference(value, self._dtype, self._conversion)173 # 成功浮点零统一+0.0；所有数值故障保留真实样本数而不改输入覆盖。174 if quality == Q.NONE and type(value) is float and value == 0:175 value = 0.0176 return SampleReductionResult(None if quality != Q.NONE else value, self._unit, quality != Q.NONE, quality, self._samples)177178 def close(self):179 """退回按需分位数存储；无动态存储的样本归约无需释放费用。180181 参数: 无。返回: None。异常: 预算释放失败时保留可重试的子缓冲。182 """183 if self._quantile is not None:184 self._quantile.close()185