chronaxis.core.window_delta_numpy
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用数组掩码计算窗口覆盖、首尾位置和整窗单调性。23输入和输出与独立参考一致；实际使用searchsorted、区间裁剪、质量掩码和4相邻数组比较，不调用参考核。时间在整数域裁剪后才建立数组，避免下限回绕。5数值保留int64/uint64/float64，不将大整数转换成浮点。无文件和状态副作用。6"""78import numpy as np9from .types import QualityFlag as Q, SignalDType as D, Truth10from .window_delta import WindowDeltaResult, validate_delta_input111213def window_delta_numpy(spec, spans, time_ns, dtype):14 """通过真实数组运算计算一个窗口，不复用参考算法的数值函数。1516 参数: spec配置、spans有序DeltaSpan tuple、time_ns纳秒、dtype类型。17 返回: WindowDeltaResult。异常: 输入错误；数值失败记录质量。18 """19 time_ns = validate_delta_input(spec, spans, time_ns, dtype)20 left = time_ns - spec.window_ns21 starts = np.asarray([item.start_ns for item in spans], dtype=np.int64)22 ends = np.asarray([item.end_ns for item in spans], dtype=np.int64)23 flags = np.asarray([int(item.quality) for item in spans], dtype=np.uint16)24 points = np.asarray([int(item.point_quality) for item in spans], dtype=np.uint16)25 # 每个交叠长度不超过window_ns，求和不超过同一窗口，定宽时间运算安全。26 lower = np.maximum(starts, max(left, -(2**63)))27 upper = np.minimum(ends, time_ns)28 overlap = lower < upper29 lengths = np.zeros(len(spans), dtype=np.int64)30 np.subtract(upper, lower, out=lengths, where=overlap)31 duration = int(np.sum(lengths, where=flags == 0, initial=0, dtype=np.int64))32 quality = Q(int(np.bitwise_or.reduce(flags[overlap], initial=0)))33 indices = []34 for boundary in (left, time_ns):35 # left可能低于int64下限，先处理后再交给searchsorted，防隐式浮点比较。36 index = -1 if boundary < -(2**63) else int(np.searchsorted(starts, boundary, side='right')) - 137 if index < 0 or not (starts[index] == boundary or boundary < ends[index]):38 quality |= Q.MISSING39 indices.append(None)40 else:41 indices.append(index)42 quality |= Q(int(points[index] if starts[index] == boundary else flags[index]))43 middle = (starts > max(left, -(2**63))) & (starts < time_ns)44 # 当left低于最小时间时，最小时间处仍属于窗口内观测。45 if left < -(2**63):46 middle = starts < time_ns47 quality |= Q(int(np.bitwise_or.reduce(points[middle], initial=0)))48 if duration != spec.window_ns:49 quality |= Q.MISSING50 if quality != Q.NONE:51 return WindowDeltaResult(time_ns, left, None, Truth.UNKNOWN, quality,52 duration, duration / spec.window_ns, None)53 first, last = (spans[index].value for index in indices)54 values = np.asarray([first, *(item.value for item, include in zip(spans, middle) if include), last],55 dtype=dtype.value)56 nondecreasing = bool(np.all(values[1:] >= values[:-1]))57 nonincreasing = bool(np.all(values[1:] <= values[:-1]))58 monotonic = (nondecreasing if spec.direction == 'rise' else nonincreasing59 if spec.direction == 'fall' else nondecreasing or nonincreasing)60 a, b = (values[0], values[-1]) if spec.direction == 'fall' else (values[-1], values[0])61 if spec.direction == 'any' and a < b:62 a, b = b, a63 # 减法前用范围不等式证明安全，防止numpy先回绕；无隐藏float提升。64 if dtype is D.INT64:65 aa, bb = int(a), int(b)66 if (bb > 0 and aa < -(2**63) + bb) or (bb < 0 and aa > 2**63 - 1 + bb):67 quality = Q.OVERFLOW68 elif dtype is D.UINT64 and a < b:69 quality = Q.OVERFLOW70 if quality != Q.NONE:71 return WindowDeltaResult(time_ns, left, None, Truth.UNKNOWN, quality, duration, 1.0, monotonic)72 with np.errstate(over='ignore', invalid='ignore'):73 value = np.subtract(a, b).item()74 if dtype is D.FLOAT64 and not np.isfinite(value):75 return WindowDeltaResult(time_ns, left, None, Truth.UNKNOWN, Q.NONFINITE, duration, 1.0, monotonic)76 satisfied = value >= spec.threshold and (not spec.monotonic or monotonic)77 return WindowDeltaResult(time_ns, left, value, Truth.TRUE if satisfied else Truth.FALSE,78 Q.NONE, duration, 1.0, monotonic)79