chronaxis.core.time_reduction_numpy
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用真实数组处理有限块时间极值、规范化时间均值及BOOL TRUE时长。23push把有限CalculationSeries装成(N,)typed列，区间使用(N-1,)int64纳秒差与bool掩码。4min/max保留原dtype；duration用TRUE掩码精确累计纳秒；mean先独立检查整数5浮点精度，第二遍以全窗有效时长规范化权重，再推进本数组原序误差归约。6最终单位仅finish转换一次，不读旧来源、不保存全轴，也不调用参考数值核。7装箱和未知Interval恢复循环只处理协议；数值裁剪、掩码、权重和极值用数组。8状态驱动累计少量标量事实；数组及误差状态预算由外层owner提前预约。9指定step_ns只接收所选驱动真实点；时长/均值按实际交集折算，另有一份有限权重10求和状态。接线者须预约这份状态和当前权重数组，不能沿用actual的费用冒充已计费。11"""1213# 本后端只复用数组原序核与静态模型，数值不委托参考实现。14import numpy as np15from .calculation_output_numpy import convert_result_numpy16from .integral_sum_numpy import NumpyIntegralSum17from .moments import TIME_MOMENTS18from .moments_numpy import NumpyMoments19from .models import Interval20from .numeric_numpy import NumericArray, DTYPES, to_float_numpy21from .types import QualityFlag as Q22from .time_reduction import (TimeCalculationFacts, TimeReductionResult,23 require_time_config, require_time_series, require_time_options)2425from .time_reduction import MeanTimeBlock26from .mean_weights_numpy import normalized_mean_terms272829class NumpyTimeReduction:30 """保留单候选、精确时长计数及有限误差数组，不保留历史输入列。3132 mean分母为None时只处理第一遍事实，不能finish；第二遍须逐块消费同一历史。33 原序浮点累计属于state_driver，每项误差推进与块掩码均是实际数组运算。34 """3536 def __init__(self, spec, input_unit, dtype, *, selected_duration_ns=None, partial_limit=4096,37 step_ns=None, selected_weight_seconds=None):38 """检查无条件时间配置并建立本数组后端状态。3940 参数: spec/单位/dtype为原配置；selected_duration_ns为均值全窗有效时长；41 partial_limit为误差容量；step_ns可选周期，周期均值第二遍改用42 selected_weight_seconds。返回: None。异常: 配置、数组分配错误传播。43 """44 self._conversion = require_time_config(spec, input_unit, dtype, selected_duration_ns, partial_limit,45 step_ns=step_ns, selected_weight_seconds=selected_weight_seconds)46 self._unit, self._factor = self._conversion.unit, self._conversion.factor47 self._moments = NumpyMoments(spec.operation) if spec.operation in TIME_MOMENTS else None48 self._spec, self._input_unit, self._dtype = spec, input_unit, dtype49 self._step_ns = step_ns50 self._denominator = selected_duration_ns if step_ns is None else selected_weight_seconds51 # 固定状态不随输入块数增长，输入列生命周期只限当前push。52 self._identity = self._candidate = None53 self._value_duration = self._selected_duration = self._terms = 054 self._numeric_quality = Q.NONE55 self._sum = NumpyIntegralSum(partial_limit) if spec.operation == 'time_weighted_mean' and self._denominator is not None else None56 self._weight_sum = NumpyIntegralSum(partial_limit) if step_ns is not None and spec.operation in ('duration', 'time_weighted_mean') else None5758 @property59 def selected_weight_seconds(self):60 """读取必要的周期累计权重。参数: 无。返回: float或不适用None。异常: 数值故障。"""61 if self._weight_sum is None:62 return None63 result = self._weight_sum.finish('s')64 if result.calculation_failed:65 raise OverflowError('周期权重累计失败')66 return 0.0 if result.value is None else result.value6768 @property69 def selected_duration_ns(self):70 """读取累计选择时长。参数: 无。返回: 原生int纳秒。异常: 无。"""71 return self._selected_duration7273 @property74 def value_valid_duration_ns(self):75 """读取累计有效时长。参数: 无。返回: 原生int纳秒。异常: 无。"""76 return self._value_duration7778 def push(self, series, *, valid_until_ns=None, retain_unknown_ranges=True, retain_mean_terms=False):79 """数组计算有限块事实，并推进对应原单位数值状态。8081 参数: series为无条件、同dtype和原单位CalculationSeries；末点不贡献区间或样本。82 valid_until_ns按实际时间截断；retain_unknown_ranges=False省去未知位置构造。83 retain_mean_terms仅均值首遍返回MeanTimeBlock，其他调用返回事实。84 返回: TimeCalculationFacts；数值权重故障与输入覆盖有效性分别记录。85 异常: 结构、身份、容量或分配错误传播；外层异常后只能清理owner。86 内存: 新数组都是当前(N,)或(N-1,)有限列，不修改输入，也不存入跨块状态。87 """88 actual = require_time_series(series, self._spec, self._input_unit, self._dtype)89 require_time_options(series, self._step_ns, valid_until_ns, retain_unknown_ranges)90 if type(retain_mean_terms) is not bool or retain_mean_terms and (self._spec.operation != 'time_weighted_mean' or self._denominator is not None):91 raise ValueError('仅均值首遍可请求必要回读项')92 if self._identity is not None and self._identity != series.binding_identity:93 raise ValueError('时间归约不能跨绑定身份累计')94 self._identity = series.binding_identity95 # 装箱仅恢复typed列；无效位置0需要显式valid和quality，不能作为正常值。96 column = NumericArray(self._dtype,97 np.array([value.value if value.valid else 0 for value in series.values], dtype=DTYPES[self._dtype]),98 np.array([value.valid for value in series.values], dtype=bool),99 np.array([int(value.quality) for value in series.values], dtype=np.uint16))100 values = to_float_numpy(column) if self._spec.operation == 'time_weighted_mean' or self._moments is not None else column101 times = np.array(series.times_ns, dtype=np.int64)102 start, end = (actual.start_ns, actual.end_ns) if actual is not None else (0, 0)103 # 先裁剪整数端点，仅在正交集做差，避免域外反序减法wrap。104 left = np.maximum(times[:-1], np.int64(start))105 right = np.minimum(times[1:], np.int64(end))106 observed = right > left107 deadlines = np.array([2**63-1 if item is None else item for item in valid_until_ns], dtype=np.int64) if valid_until_ns is not None else None108 value_end = right.copy() if deadlines is None else np.minimum(right, deadlines[:-1])109 value_end = np.maximum(left, np.where(values.valid[:-1], value_end, left))110 durations = np.zeros(len(left), dtype=np.int64)111 usable = observed & (value_end > left)112 np.subtract(value_end, left, out=durations, where=usable)113 selected = usable & values.values[:-1] if self._spec.operation == 'duration' else usable114 # 非负时长总和不超过已校验观测跨度，int64不会wrap；全窗由Python整数累加。115 value_duration = int(np.sum(durations[usable], dtype=np.int64))116 selected_duration = int(np.sum(durations[selected], dtype=np.int64))117 seconds = None118 if self._weight_sum is not None or (self._step_ns is not None and self._moments is not None):119 # 只计算当前有限块；求和后列只读，后续归一化另生成数值结果。120 seconds = durations[selected].astype(np.float64)121 seconds /= np.diff(times)[selected]122 seconds *= self._step_ns / 1_000_000_000123 weight_valid = np.ones(len(seconds), dtype=bool)124 seconds.setflags(write=False)125 weight_valid.setflags(write=False)126 if self._weight_sum is not None:127 self._weight_sum.push_values(seconds, weight_valid)128 terms = int(np.count_nonzero(usable))129 samples = (times >= start) & (times < end) & np.array(series.actual_value_updates, dtype=bool) & values.valid130 if deadlines is not None:131 samples &= times < deadlines132 if self._spec.operation == 'duration':133 samples &= values.values134 # 依赖诊断覆盖正区间与实际更新点；EOF右端不凭点增加时长或样本。135 relevant = np.concatenate((observed, np.array([False]))) | ((times >= start) & (times < end) & np.array(series.actual_value_updates, dtype=bool))136 diagnostics = np.array([int(item) for item in series.diagnostics], dtype=np.uint16) | values.quality137 quality = Q(int(np.bitwise_or.reduce(diagnostics[relevant], initial=np.uint16(0))))138 if np.any(observed & values.valid[:-1] & (value_end < right)):139 quality |= Q.STALE140 unknown = (tuple(Interval(int(value_end[index]), int(right[index])) for index in np.flatnonzero(observed & (value_end < right)))141 if retain_unknown_ranges else ())142 if self._spec.operation in ('min_time', 'max_time') and terms:143 # 每块极值由原dtype数组归约，跨块仅保存一个精确原生候选。144 candidate = (np.min if self._spec.operation == 'min_time' else np.max)(values.values[:-1][usable])145 if self._candidate is not None:146 previous = DTYPES[self._dtype].type(self._candidate)147 candidate = (np.minimum if self._spec.operation == 'min_time' else np.maximum)(previous, candidate)148 self._candidate = candidate.item()149 elif self._moments is not None:150 measures = durations[usable] if self._step_ns is None else seconds151 self._moments.push(values.values[:-1][usable], measures)152 elif self._spec.operation == 'time_weighted_mean' and self._denominator is not None:153 chosen = values.values[:-1][usable]154 measures = durations[usable].astype(np.float64) if self._step_ns is None else seconds155 weighted, weight_quality = normalized_mean_terms(chosen, measures, self._denominator)156 self._numeric_quality |= weight_quality157 if self._numeric_quality == Q.NONE:158 keep = np.ones(len(weighted), dtype=bool)159 keep.setflags(write=False)160 self._sum.push_values(weighted, keep)161 self._value_duration += value_duration162 self._selected_duration += selected_duration163 self._terms += terms164 facts = TimeCalculationFacts(self._spec, actual, value_duration, selected_duration,165 int(np.count_nonzero(samples)), unknown, quality, self._input_unit,166 self._dtype, series.binding_identity, series.source_refs, retain_unknown_ranges)167 if retain_mean_terms:168 selected_values = values.values[:-1][usable]169 selected_weights = durations[usable].copy() if self._step_ns is None else seconds170 selected_values.setflags(write=False)171 selected_weights.setflags(write=False)172 return MeanTimeBlock(facts, selected_values, selected_weights)173 return facts174175 def finish(self):176 """正确舍入并做一次最终转换，生成本后端数值快照。177178 参数: 无。返回: TimeReductionResult，不执行覆盖状态分类。179 异常: RuntimeError为mean第一遍；ValueError为有效时长与全窗分母不符；180 分配或取消传播，数值溢出与非零下溢写成计算失败。181 """182 if self._moments is not None:183 value, quality = self._moments.finish(self._spec.ddof, self._conversion.factor)184 return TimeReductionResult(value, self._unit, quality != Q.NONE, quality, self._terms)185 if self._spec.operation == 'time_weighted_mean':186 if self._denominator is None:187 raise RuntimeError('时间均值第一遍只收事实，不能生成数值')188 measured = self._selected_duration if self._step_ns is None else self.selected_weight_seconds189 if measured != self._denominator:190 raise ValueError('时间均值第二遍有效时长与整窗分母不同')191 if self._numeric_quality != Q.NONE:192 return TimeReductionResult(None, self._unit, True, self._numeric_quality, self._terms)193 # 数组原序核直接执行一次输出转换；映射冻结结果不再次做数值处理。194 number = self._sum.finish(self._input_unit,195 1.0)196 if not number.calculation_failed:197 value, quality = convert_result_numpy(number.value, self._dtype, self._conversion)198 return TimeReductionResult(value, self._unit, quality != Q.NONE, quality, self._terms)199 return TimeReductionResult(number.value, self._unit, number.calculation_failed, number.quality, self._terms)200 if self._terms == 0:201 return TimeReductionResult(None, self._unit, False, Q.NONE, 0)202 # BOOL时长先保持精确整数纳秒，再最终换成秒；极值保持原native dtype。203 value = ((float(np.divide(np.float64(self._selected_duration), np.float64(1_000_000_000)))204 if self._step_ns is None else self.selected_weight_seconds)205 if self._spec.operation == 'duration' else self._candidate)206 quality = Q.NONE207 # v4换算包含偏移；输入/归约故障不能被正常单位结果覆盖。208 if quality == Q.NONE:209 value, quality = convert_result_numpy(value, self._dtype, self._conversion)210 return TimeReductionResult(None if quality != Q.NONE else value, self._unit,211 quality != Q.NONE, quality, self._terms)212