chronaxis.core.window_delta_reference

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""用独立Python遍历计算完整窗口变化量，作为数组实现的数值参考。23输入是有界ZOH片段和一个查询时刻，输出数值、三值条件和覆盖率。4首尾使用点值，窗内使用实际观测及连续片段；缺口不插值、不延长有效期。5本模块不依赖NumPy，不读文件，不保存跨调用状态或触发锁存。6"""78from math import isfinite9from .types import QualityFlag as Q, SignalDType as D, Truth10from .window_delta import WindowDeltaResult, validate_delta_input111213def window_delta_reference(spec, spans, time_ns, dtype):14    """逐段求完整窗口首尾差和方向单调性。1516    参数: spec配置、spans有序DeltaSpan tuple、time_ns纳秒、dtype数值类型。17    返回: WindowDeltaResult；阈值比较包含等于，UNKNOWN不伪装False。18    异常: 输入结构错误；定宽数值溢出以质量结果返回。19    """20    time_ns = validate_delta_input(spec, spans, time_ns, dtype)21    left = time_ns - spec.window_ns22    duration, quality, first, last, observations = 0, Q.NONE, None, None, []23    first_found = last_found = False24    for item in spans:25        # 左端在边界上优先用该边界真实点；内部位置沿有限ZOH。26        if item.start_ns == left or item.start_ns < left < item.end_ns:27            first_found = True28            first = item.value29            quality |= item.point_quality if item.start_ns == left else item.quality30        if item.start_ns == time_ns or item.start_ns < time_ns < item.end_ns:31            last_found = True32            last = item.value33            quality |= item.point_quality if item.start_ns == time_ns else item.quality34        lo, hi = max(left, item.start_ns), min(time_ns, item.end_ns)35        if lo < hi:36            if item.quality == Q.NONE:37                duration += hi - lo38            else:39                quality |= item.quality40        # 边界值与窗内真实观测都参与单调性，不能只比较两个端点。41        if left < item.start_ns < time_ns:42            quality |= item.point_quality43            observations.append(item.value)44    if not first_found or not last_found or duration != spec.window_ns:45        quality |= Q.MISSING46    if quality != Q.NONE:47        return WindowDeltaResult(time_ns, left, None, Truth.UNKNOWN, quality,48                                 duration, duration / spec.window_ns, None)49    values = [first, *observations, last]50    nondecreasing = all(a <= b for a, b in zip(values, values[1:]))51    nonincreasing = all(a >= b for a, b in zip(values, values[1:]))52    monotonic = (nondecreasing if spec.direction == 'rise' else nonincreasing53                 if spec.direction == 'fall' else nondecreasing or nonincreasing)54    # Python整数先算真实差值再检查定宽范围，避免溢出后拿回绕值判断阈值。55    value = first - last if spec.direction == 'fall' else last - first56    if spec.direction == 'any':57        value = abs(value)58    if dtype is D.FLOAT64 and not isfinite(value):59        quality = Q.NONFINITE60    elif dtype is D.INT64 and not -(2**63) <= value < 2**63:61        quality = Q.OVERFLOW62    elif dtype is D.UINT64 and not 0 <= value < 2**64:63        quality = Q.OVERFLOW64    if quality != Q.NONE:65        return WindowDeltaResult(time_ns, left, None, Truth.UNKNOWN, quality,66                                 duration, 1.0, monotonic)67    satisfied = value >= spec.threshold and (not spec.monotonic or monotonic)68    return WindowDeltaResult(time_ns, left, value, Truth.TRUE if satisfied else Truth.FALSE,69                             Q.NONE, duration, 1.0, monotonic)70