chronaxis.core.rolling_values

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""对已裁剪的有限保持区间求一个滚动网格值。23权重是实际纳秒或原周期按实际重叠分摊的计量纳秒。独立参考路径使用精确4有理数累计，数组路径对当次窗口执行 NumPy 掩码和缩放加权归约。历史维护、5覆盖门槛、网格原点和事件均由调用者处理；本模块不保存上一个窗口。6"""7from fractions import Fraction8from math import frexp, isfinite, ldexp9from .numeric import ScalarValue10from .types import SignalDType as D, QualityFlag as Q111213def rolling_value(operation, backend, values, weights, conversion, dtype):14    """计算选中窗口值。参数: 操作、后端、有效数列、正权重、目标换算、类型。返回: ScalarValue。异常: 配置错误。"""15    if operation in ('rolling_minimum','rolling_maximum'):16        if backend=='reference':17            from .affine import affine_reference18            value=(min if operation=='rolling_minimum' else max)(values)19            return affine_reference(ScalarValue(value,dtype),conversion.factor,conversion.offset,conversion.dtype(dtype))20        import numpy as np21        from .numeric_numpy import DTYPES, NumericArray22        from .affine_numpy import affine_numpy23        col=np.asarray(values,dtype=DTYPES[dtype])24        result=(np.min if operation=='rolling_minimum' else np.max)(col)25        column=NumericArray(dtype,np.array([result],dtype=DTYPES[dtype]),np.ones(1,dtype=bool),np.zeros(1,dtype='uint16'))26        converted=affine_numpy(column,conversion.factor,conversion.offset,conversion.dtype(dtype))27        valid=bool(converted.valid[0])28        return ScalarValue(converted.values[0].item() if valid else None,converted.dtype,valid,Q(int(converted.quality[0])))29    # 整数进入浮点统计须无损。无效样本不能经插值或舍入变为普通值。30    if dtype in (D.INT64,D.UINT64) and any(int(float(v))!=v for v in values):31        return ScalarValue(None,D.FLOAT64,False,Q.PRECISION_LOSS)32    if backend=='reference':33        total=sum(Fraction(v)*w for v,w in zip(values,weights))34        measure=sum(weights) if operation=='rolling_time_mean' else 10**935        exact=total*Fraction(conversion.factor)/measure+Fraction(conversion.offset)36        try:37            number=float(exact)38        except OverflowError:39            return ScalarValue(None,D.FLOAT64,False,Q.NONFINITE)40        lost=number==0 and exact!=041    else:42        import numpy as np43        numbers=np.asarray(values,dtype=np.float64)44        w=np.asarray(weights,dtype=np.float64)45        from .calculations_numpy import _sum_finite_numpy46        total_weight=float(np.sum(w))47        normalized=w/total_weight48        with np.errstate(under='ignore'):49            terms=numbers*normalized50        # 对极小项先放大再乘权重。真实纳秒和原驱动比例均有int64界限，51        # 2^512足以使这些暂存项不会因单项落入次正规范围就丢掉其合计。52        tiny=(np.abs(terms)<np.finfo('float64').tiny)&(numbers!=0)53        total,failed,quality=(_sum_finite_numpy(terms[~tiny]) if np.any(~tiny) else (0.,False,Q.NONE))54        if failed:55            return ScalarValue(None,D.FLOAT64,False,quality)56        extra_exponent=057        if np.any(tiny):58            residual,failed,quality=_sum_finite_numpy(np.ldexp(numbers[tiny],512)*normalized[tiny])59            if failed:60                return ScalarValue(None,D.FLOAT64,False,quality)61            if abs(total)<=1:62                total=ldexp(total,512)+residual63                extra_exponent=-51264            else:65                total+=ldexp(residual,-512)66        mantissa,exponent=frexp(total)67        factor,ef=frexp(conversion.factor)68        scale,es=frexp(1 if operation=='rolling_time_mean' else total_weight/1e9)69        try:70            number=ldexp(mantissa*factor*scale,exponent+extra_exponent+ef+es)+conversion.offset71        except OverflowError:72            return ScalarValue(None,D.FLOAT64,False,Q.NONFINITE)73        lost=number==0 and total!=0 and not conversion.offset74    if not isfinite(number) or lost:75        return ScalarValue(None,D.FLOAT64,False,Q.NONFINITE if not isfinite(number) else Q.PRECISION_LOSS)76    return ScalarValue(number,D.FLOAT64)77