chronaxis.core.conditional_time_reduction_numpy
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用真实NumPy掩码归约有限条件时间极值、规范化均值和主BOOL时长。23typed值及外BOOL为(N,)列，正区间为(N-1,)精确int64差与独立bool掩码。极值在原4dtype上归约；duration同时保留外TRUE主FALSE零证据；mean首遍仅确认值精度和5TRUE有效选择，次遍计算x*(dt/D)，由本数组原序误差核推进。数值weight故障不改6双覆盖，最终单位只finish转换一次。装箱、恢复未知和固定标量状态驱动不冒充7数组数值核；不调用参考算法、不读来源、不存全轴，有限内存由外层owner预约。8"""910# 原序推进复用独立数组误差状态，只共享结构/typed协议，不调用标量后端。11import numpy as np12from .calculation_output_numpy import convert_result_numpy13from .conditional_time_reduction import ConditionalTimeCalculationFacts, require_conditional_time_series14from .integral_sum_numpy import NumpyIntegralSum15from .moments import TIME_MOMENTS16from .moments_numpy import NumpyMoments17from .calculations import merge_ranges18from .models import Interval19from .numeric_numpy import DTYPES, NumericArray, to_float_numpy20from .time_reduction import TimeReductionResult, require_time_config, require_time_options21from .integral_terms import require_valid_until22from .types import QualityFlag as Q2324from .time_reduction import MeanTimeBlock25from .mean_weights_numpy import normalized_mean_terms262728class NumpyConditionalTimeReduction:29 """保存单原dtype候选、整数时长与有界误差列，不存跨块输入数组。"""3031 def __init__(self, spec, input_unit, dtype, *, selected_duration_ns=None, partial_limit=4096,32 step_ns=None, selected_weight_seconds=None):33 """校验原配置并建立本数组后端固定状态。3435 参数: spec/input_unit/dtype为原配置；selected_duration_ns为mean次遍全TRUE有效分母；36 partial_limit为误差容量；step_ns选周期，周期均值次遍传selected_weight_seconds。返回: None。异常: 配置、数组分配错误传播。37 """38 # 静态门槛只检查明确配置，两个后端不共享后续数值或覆盖计算。39 self._conversion = require_time_config(spec, input_unit, dtype, selected_duration_ns, partial_limit,40 step_ns=step_ns, selected_weight_seconds=selected_weight_seconds)41 self._unit, self._factor = self._conversion.unit, self._conversion.factor42 self._moments = NumpyMoments(spec.operation) if spec.operation in TIME_MOMENTS else None43 self._spec, self._input_unit, self._dtype = spec, input_unit, dtype44 self._step_ns = step_ns45 self._denominator = selected_duration_ns if step_ns is None else selected_weight_seconds46 # mean首遍不建立误差列，固定候选/计数不随块数增长。47 self._identity = self._candidate = None48 self._value_duration = self._selected_duration = self._terms = 049 self._numeric_quality = Q.NONE50 self._sum = NumpyIntegralSum(partial_limit) if spec.operation == 'time_weighted_mean' and self._denominator is not None else None51 self._weight_sum = NumpyIntegralSum(partial_limit) if step_ns is not None and spec.operation in ('duration', 'time_weighted_mean') else None5253 @property54 def selected_weight_seconds(self):55 """读取必要周期权重。参数: 无。返回: float或不适用None。异常: 权重求和故障。"""56 if self._weight_sum is None:57 return None58 result = self._weight_sum.finish('s')59 if result.calculation_failed:60 raise OverflowError('条件周期权重累计失败')61 return 0.0 if result.value is None else result.value6263 @property64 def selected_duration_ns(self):65 """读取累计已选时长。参数: 无。返回: 原生int纳秒。异常: 无。"""66 return self._selected_duration # duration仅主TRUE；mean为规范化分母事实。6768 @property69 def value_valid_duration_ns(self):70 """读取累计值有效时长。参数: 无。返回: 原生int纳秒。异常: 无。"""71 return self._value_duration # 独立于外条件FALSE/UNKNOWN。7273 def push(self, series, *, valid_until_ns=None, condition_valid_until_ns=None, retain_unknown_ranges=True, retain_mean_terms=False):74 """计算有限块双覆盖并按原序推进本后端数值候选。7576 参数: series为同dtype/原单位且显式BOOL条件的有限CalculationSeries。77 valid_until_ns/condition_valid_until_ns分别截断实际有效期；78 retain_unknown_ranges=False省去两类未知范围。79 retain_mean_terms仅均值首遍返回MeanTimeBlock，其他调用返回事实。80 返回: ConditionalTimeCalculationFacts；权重故障仅改变数值失败状态，不改输入覆盖。81 异常: 结构、身份、容量、分配或取消错误传播，外层异常后只能清理owner。82 内存: 所有新输入/掩码列形状N或N-1，仅当前push存活，不修改调用者输入。83 """84 # 当前输入先核对显式条件/单位/dtype和交集，跨块原值绑定不能变化。85 actual = require_conditional_time_series(series, self._spec, self._input_unit, self._dtype)86 require_time_options(series, self._step_ns, valid_until_ns, retain_unknown_ranges)87 require_valid_until(series.times_ns, condition_valid_until_ns)88 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):89 raise ValueError('仅均值首遍可请求必要回读项')90 if self._identity is not None and self._identity != series.binding_identity:91 raise ValueError('条件时间归约不能跨绑定身份累计')92 self._identity = series.binding_identity93 # typed装箱仅恢复协议，值精度转换只对mean执行，极值保持int64/uint64原数组。94 column = NumericArray(self._dtype,95 np.array([value.value if value.valid else 0 for value in series.values], dtype=DTYPES[self._dtype]),96 np.array([value.valid for value in series.values], dtype=bool),97 np.array([int(value.quality) for value in series.values], dtype=np.uint16))98 values = to_float_numpy(column) if self._spec.operation == 'time_weighted_mean' or self._moments is not None else column99 known = np.array([condition.valid for condition in series.conditions], dtype=bool)100 truth = np.array([condition.value if condition.valid else False for condition in series.conditions], dtype=bool)101 condition_quality = np.array([int(condition.quality) for condition in series.conditions], dtype=np.uint16)102 times = np.array(series.times_ns, dtype=np.int64)103 start, end = (actual.start_ns, actual.end_ns) if actual is not None else (0, 0)104 # 整数端点先裁剪，只对正区间做差，域外反序不发生int64减法wrap。105 left, right = np.maximum(times[:-1], np.int64(start)), 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 condition_deadlines = np.array([2**63-1 if item is None else item for item in condition_valid_until_ns], dtype=np.int64) if condition_valid_until_ns is not None else None109 value_end = right.copy() if deadlines is None else np.minimum(right, deadlines[:-1])110 condition_end = right.copy() if condition_deadlines is None else np.minimum(right, condition_deadlines[:-1])111 value_end = np.maximum(left, np.where(values.valid[:-1], value_end, left))112 condition_end = np.maximum(left, np.where(known[:-1], condition_end, left))113 # 三种交集独立做安全整数差，原始记录对仍留作周期权重分母。114 value_dt = np.zeros(len(left), dtype=np.int64)115 condition_dt = np.zeros(len(left), dtype=np.int64)116 durations = np.zeros(len(left), dtype=np.int64)117 np.subtract(value_end, left, out=value_dt, where=observed)118 np.subtract(condition_end, left, out=condition_dt, where=observed)119 np.subtract(np.minimum(value_end, condition_end), left, out=durations, where=observed)120 usable = observed & (durations > 0) & truth[:-1]121 selected = usable & values.values[:-1] if self._spec.operation == 'duration' else usable122 selected_false_mask = usable & ~values.values[:-1] if self._spec.operation == 'duration' else np.zeros(len(left), dtype=bool)123 # 所有非负dt归约不超过已检查int64实际跨度，跨块由Python整数累加。124 value_duration = int(np.sum(value_dt, dtype=np.int64))125 known_duration = int(np.sum(condition_dt, dtype=np.int64))126 selected_duration = int(np.sum(durations, where=selected, dtype=np.int64))127 selected_unknown = int(np.sum(condition_dt - durations, where=truth[:-1], dtype=np.int64))128 false_duration = int(np.sum(condition_dt, where=~truth[:-1], dtype=np.int64))129 selected_false = int(np.sum(durations, where=selected_false_mask, dtype=np.int64))130 terms = int(np.count_nonzero(usable))131 seconds = None132 if self._weight_sum is not None or (self._step_ns is not None and self._moments is not None):133 seconds = durations[selected].astype(np.float64)134 seconds /= np.diff(times)[selected]135 seconds *= self._step_ns / 1_000_000_000136 keep_weights = np.ones(len(seconds), dtype=bool)137 seconds.setflags(write=False)138 keep_weights.setflags(write=False)139 if self._weight_sum is not None:140 self._weight_sum.push_values(seconds, keep_weights)141 # 半开原值真实更新且外TRUE才成为样本；duration另要求主TRUE，条件更新不计。142 actual_samples = (times >= start) & (times < end) & np.array(series.actual_value_updates, dtype=bool)143 samples = actual_samples & values.valid & known & truth144 if deadlines is not None:145 samples &= times < deadlines146 if condition_deadlines is not None:147 samples &= times < condition_deadlines148 if self._spec.operation == 'duration':149 samples &= values.values150 relevant = actual_samples.copy()151 relevant[:-1] |= observed152 diagnostics = np.array([int(flag) for flag in series.diagnostics], dtype=np.uint16) | values.quality | condition_quality153 quality = Q(int(np.bitwise_or.reduce(diagnostics, where=relevant, initial=0)))154 # 元数据循环只恢复两种未知Interval，覆盖和数值运算均已由真实数组完成。155 if np.any(observed & ((values.valid[:-1] & (value_end < right)) | (known[:-1] & (condition_end < right)))):156 quality |= Q.STALE157 value_unknown = (tuple(Interval(int(value_end[index]), int(right[index])) for index in np.flatnonzero(observed & (value_end < right)))158 if retain_unknown_ranges else ())159 condition_unknown = (tuple(Interval(int(condition_end[index]), int(right[index])) for index in np.flatnonzero(observed & (condition_end < right)))160 if retain_unknown_ranges else ())161 if self._spec.operation in ('min_time', 'max_time') and terms:162 # 当前块原dtype极值，跨块仅比较一个同dtype候选，避免uint64精度舍入。163 candidate = (np.min if self._spec.operation == 'min_time' else np.max)(values.values[:-1][usable])164 if self._candidate is not None:165 previous = DTYPES[self._dtype].type(self._candidate)166 candidate = (np.minimum if self._spec.operation == 'min_time' else np.maximum)(previous, candidate)167 self._candidate = candidate.item()168 elif self._moments is not None:169 measures = durations[usable] if self._step_ns is None else seconds170 self._moments.push(values.values[:-1][usable], measures)171 elif self._spec.operation == 'time_weighted_mean' and self._denominator is not None:172 chosen = values.values[:-1][usable]173 measures = durations[usable].astype(np.float64) if self._step_ns is None else seconds174 weighted, weight_quality = normalized_mean_terms(chosen, measures, self._denominator)175 self._numeric_quality |= weight_quality176 if self._numeric_quality == Q.NONE:177 keep = np.ones(len(weighted), dtype=bool)178 keep.setflags(write=False)179 self._sum.push_values(weighted, keep)180 self._value_duration += value_duration181 self._selected_duration += selected_duration182 self._terms += terms183 facts = ConditionalTimeCalculationFacts(self._spec, actual, value_duration, known_duration, selected_duration,184 selected_unknown, false_duration, selected_false, int(np.count_nonzero(samples)), merge_ranges(value_unknown),185 merge_ranges(condition_unknown), quality, self._input_unit, self._dtype, series.binding_identity, series.source_refs, retain_unknown_ranges)186 if retain_mean_terms:187 selected_values = values.values[:-1][usable]188 selected_weights = durations[usable].copy() if self._step_ns is None else seconds189 selected_values.setflags(write=False)190 selected_weights.setflags(write=False)191 return MeanTimeBlock(facts, selected_values, selected_weights)192 return facts193194 def finish(self):195 """生成原序统计快照并做一次最终换单位，双覆盖由外层分类。196197 参数: 无。返回: 既有TimeReductionResult。异常: mean首遍RuntimeError、次遍分母不符ValueError；198 配置/分配/取消传播，数值溢出和非零下溢返回计算失败。199 """200 # mean首遍禁交付数值；次遍需要实际TRUE有效累计时长与全窗D相等。201 if self._moments is not None:202 value, quality = self._moments.finish(self._spec.ddof, self._conversion.factor)203 return TimeReductionResult(value, self._unit, quality != Q.NONE, quality, self._terms)204 if self._spec.operation == 'time_weighted_mean':205 if self._denominator is None:206 raise RuntimeError('条件时间均值首遍只收事实，不能生成数值')207 measured = self._selected_duration if self._step_ns is None else self.selected_weight_seconds208 if measured != self._denominator:209 raise ValueError('条件均值次遍选择时长与全窗分母不同')210 if self._numeric_quality != Q.NONE:211 return TimeReductionResult(None, self._unit, True, self._numeric_quality, self._terms)212 number = self._sum.finish(self._input_unit,213 1.0)214 if not number.calculation_failed:215 value, quality = convert_result_numpy(number.value, self._dtype, self._conversion)216 return TimeReductionResult(value, self._unit, quality != Q.NONE, quality, self._terms)217 return TimeReductionResult(number.value, self._unit, number.calculation_failed, number.quality, self._terms)218 # 没有正区间数值证据不补0；全外FALSE空0由外层完整请求分类证明。219 if self._terms == 0:220 return TimeReductionResult(None, self._unit, False, Q.NONE, 0)221 # 主FALSE为有效duration项而时长0；极值未换单位时保持原生精确整数。222 value = ((float(np.divide(np.float64(self._selected_duration), np.float64(1_000_000_000)))223 if self._step_ns is None else self.selected_weight_seconds)224 if self._spec.operation == 'duration' else self._candidate)225 quality = Q.NONE226 # v4换算包含偏移；输入/归约故障不能被正常单位结果覆盖。227 if quality == Q.NONE:228 value, quality = convert_result_numpy(value, self._dtype, self._conversion)229 return TimeReductionResult(None if quality != Q.NONE else value, self._unit, quality != Q.NONE, quality, self._terms)230