chronaxis.core.sample_reduction_numpy
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用真实NumPy样本与覆盖掩码归约有限min/max/sample_mean。23输入值及可选BOOL为(N,)typed列，覆盖为(N-1,)精确int64差与bool掩码；真实样本4按半开窗口原值actual更新、有效和条件TRUE选择。极值保持原dtype，mean只转换5可核实值，并将原序样本推进本后端有界误差核。finish先完成原单位总和、除真实6样本数，再转换输出一次；原量纲中间溢出不改为时间规范权重。输入覆盖与数值7故障分开。精确分位数按专用配额保留被选中的原dtype样本；其他操作保持有限状态。8"""910# 数值与覆盖均用独立真实数组运算，仅共享typed模型与本数组原序误差状态。11import numpy as np12from .moments import SAMPLE_MOMENTS13from .quantiles import SAMPLE_QUANTILES, ExactQuantileBuffer14from .moments_numpy import NumpyMoments15from .calculation_output_numpy import convert_result_numpy16from .integral_sum_numpy import NumpyIntegralSum17from .calculations import merge_ranges18from .models import Interval19from .numeric_numpy import DTYPES, NumericArray, to_float_numpy20from .sample_reduction import SampleCalculationFacts, SampleReductionResult, require_sample_config, require_sample_series21from .types import QualityFlag as Q222324class NumpySampleReduction: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,'numpy',quantile_budget)41 if spec.operation in SAMPLE_QUANTILES else None)42 self._moments = NumpyMoments(spec.operation) if spec.operation in SAMPLE_MOMENTS else None43 self._value_duration = self._selected_duration = self._samples = 044 self._sum = NumpyIntegralSum(partial_limit) if spec.operation == 'sample_mean' else None4546 @property47 def selected_duration_ns(self):48 """读取累计TRUE有效保持时长。参数: 无。返回: 原生int纳秒。异常: 无。"""49 return self._selected_duration # 不是样本mean分母。5051 @property52 def value_valid_duration_ns(self):53 """读取累计值有效时长。参数: 无。返回: 原生int纳秒。异常: 无。"""54 return self._value_duration # 与选择条件独立。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，可选条件None表示全TRUE。65 返回: SampleCalculationFacts；mean精度入值覆盖，数值归约故障不改覆盖。66 异常: 结构、身份、容量、分配或取消传播；异常后外层只能清理owner。67 内存: 新列形状N/N-1或当前所选样本数，当前push存活，不修改输入列。68 """69 # 有限输入结构和整数交集先验证，禁止跨块更换原值绑定。70 actual = require_sample_series(series, self._spec, self._input_unit, self._dtype)71 if self._identity is not None and self._identity != series.binding_identity:72 raise ValueError('样本归约不能跨绑定身份累计')73 self._identity = series.binding_identity74 # N个typed原值、有效性与uint16质量独立装箱；无效0只是掩码约束的存储占位。75 column = NumericArray(self._dtype,76 np.array([value.value if value.valid else 0 for value in series.values], dtype=DTYPES[self._dtype]),77 np.array([value.valid for value in series.values], dtype=bool),78 np.array([int(value.quality) for value in series.values], dtype=np.uint16))79 # mean独立数组精度门槛，极值原int64/uint64保持；无条件显式全TRUE而非UNKNOWN。80 values = to_float_numpy(column) if self._spec.operation == 'sample_mean' or self._moments is not None else column81 size = len(series.times_ns)82 known = np.ones(size, dtype=bool) if series.conditions is None else np.array([condition.valid for condition in series.conditions], dtype=bool)83 truth = np.ones(size, dtype=bool) if series.conditions is None else np.array([condition.value if condition.valid else False for condition in series.conditions], dtype=bool)84 condition_quality = np.zeros(size, dtype=np.uint16) if series.conditions is None else np.array([int(condition.quality) for condition in series.conditions], dtype=np.uint16)85 times = np.array(series.times_ns, dtype=np.int64)86 start, end = (actual.start_ns, actual.end_ns) if actual is not None else (0, 0)87 # 覆盖只对正裁剪区间求int64差，绝对时间不转float，域外反序差不wrap。88 left, right = np.maximum(times[:-1], np.int64(start)), np.minimum(times[1:], np.int64(end))89 observed = right > left90 durations = np.zeros(len(left), dtype=np.int64)91 np.subtract(right, left, out=durations, where=observed)92 value_valid = observed & values.valid[:-1]93 selected = value_valid & known[:-1] & truth[:-1]94 value_duration = int(np.sum(durations, where=value_valid, dtype=np.int64))95 known_duration = int(np.sum(durations, where=observed & known[:-1], dtype=np.int64))96 selected_duration = int(np.sum(durations, where=selected, dtype=np.int64))97 selected_unknown = int(np.sum(durations, where=observed & known[:-1] & truth[:-1] & ~values.valid[:-1], dtype=np.int64))98 false_duration = int(np.sum(durations, where=observed & known[:-1] & ~truth[:-1], dtype=np.int64))99 # 样本(N,)掩码单独使用actual更新，不能由区间selected或条件变化推算。100 actual_samples = (times >= start) & (times < end) & np.array(series.actual_value_updates, dtype=bool)101 samples = actual_samples & values.valid & known & truth102 if sample_consumer is not None and np.any(samples):103 sample_consumer(values.take(samples))104 sample_count = int(np.count_nonzero(samples))105 relevant = actual_samples.copy()106 relevant[:-1] |= observed107 diagnostics = np.array([int(flag) for flag in series.diagnostics], dtype=np.uint16) | values.quality | condition_quality108 quality = Q(int(np.bitwise_or.reduce(diagnostics, where=relevant, initial=0)))109 # Python仅恢复有限未知定位；时长和样本计数均已由真实数组掩码归约。110 value_unknown = tuple(Interval(int(left[index]), int(right[index])) for index in np.flatnonzero(observed & ~values.valid[:-1]))111 condition_unknown = tuple(Interval(int(left[index]), int(right[index])) for index in np.flatnonzero(observed & ~known[:-1]))112 if self._sum is not None:113 # 按原(N,)位置顺序选择原单位样本，不先块求和或块均值；只读列交给本后端误差核。114 chosen = values.values[samples]115 keep = np.ones(len(chosen), dtype=bool)116 chosen.setflags(write=False)117 keep.setflags(write=False)118 self._sum.push_values(chosen, keep)119 elif self._moments is not None:120 self._moments.push(values.values[samples])121 elif self._quantile is not None:122 self._quantile.append(values.values,samples)123 elif sample_count and self._spec.operation != 'sample_count':124 # 极值当前样本列用原dtype数组归约，跨块仅保留一个同dtype候选。125 candidate = (np.min if self._spec.operation == 'sample_minimum' else np.max)(values.values[samples])126 if self._candidate is not None:127 previous = DTYPES[self._dtype].type(self._candidate)128 candidate = (np.minimum if self._spec.operation == 'sample_minimum' else np.maximum)(previous, candidate)129 self._candidate = candidate.item()130 # 跨块仅累计三个原生计数；冻结事实保留原请求、双未知与完整真实来源。131 self._value_duration += value_duration132 self._selected_duration += selected_duration133 self._samples += sample_count134 return SampleCalculationFacts(self._spec, actual, value_duration, known_duration, selected_duration,135 selected_unknown, false_duration, sample_count, 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 # 没有真实有效样本只能无候选，保持时长不能替代此计数证明0。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_numpy(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先完成raw总和，原量纲中途溢出保持失败，不预规范化样本。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 with np.errstate(over='ignore', under='ignore', invalid='ignore'):166 value = np.divide(np.float64(number.value), np.float64(self._samples)).item()167 if not np.isfinite(value) or value == 0 and number.value != 0:168 quality |= Q.NONFINITE if not np.isfinite(value) else Q.PRECISION_LOSS169 else:170 value = self._candidate171 # v4换算包含偏移；输入/归约故障不能被正常单位结果覆盖。172 if quality == Q.NONE:173 value, quality = convert_result_numpy(value, self._dtype, self._conversion)174 # 只规范成功浮点零，不清除数值故障或改写真实样本/输入覆盖事实。175 if quality == Q.NONE and type(value) is float and value == 0:176 value = 0.0177 return SampleReductionResult(None if quality != Q.NONE else value, self._unit, quality != Q.NONE, quality, self._samples)178179 def close(self):180 """退回按需分位数存储；无动态存储的样本归约无需释放费用。181182 参数: 无。返回: None。异常: 预算释放失败时保留可重试的子缓冲。183 """184 if self._quantile is not None:185 self._quantile.close()186