chronaxis.core.lookup_numpy
源码基线:8904c2f744951857659abd40aaa3535d57faf53c
1"""通过 NumPy searchsorted 为整批活动样本查表并计算线性或阶梯输出。23从冻结字节借用小端定宽列,不为每个样本或批次复制标定表。输入先按4既有无损规则转为浮点轴,只有有效且位于开区间内部时才执行插值。5临时列随本批行数增长,不调用标量核,也不改变输入与其质量。6"""78import numpy as np9from .numeric_numpy import DTYPES, NumericArray, to_float_numpy10from .types import QualityFlag as Q111213def lookup_numpy(table, value):14 """批量查表。参数: 已准备表和一维数值列。返回: 同长的表值列。异常: 类型、分配或布局错误。"""15 x = to_float_numpy(value)16 # 低层直接数组 API 没有流式 owner 时只持有本批打包列;驱动所有者复用自己的执行副本。17 table = table.packed()18 axis = np.frombuffer(table.axis, dtype='<f8')19 nodes = np.frombuffer(table.values, dtype=DTYPES[table.dtype].newbyteorder('<'))20 quality = x.quality.copy()21 values = np.zeros(len(x.values), dtype=DTYPES[table.dtype])22 # 输入缺失时不读取其占位值。精确节点包括闭端点,三种模式使用同一组比较。23 rows = np.flatnonzero(x.valid)24 query = x.values[rows]25 hi = np.searchsorted(axis, query, side='left')26 safe_hi = np.minimum(hi, len(axis) - 1)27 hit = query == axis[safe_hi]28 values[rows[hit]] = nodes[safe_hi[hit]]29 outside = ~hit & ((hi == 0) | (hi == len(axis)))30 if table.bounds == 'clamp':31 values[rows[outside]] = nodes[safe_hi[outside]]32 else:33 quality[rows[outside]] |= int(Q.OUT_OF_RANGE)34 inside = ~hit & ~outside35 rows, query, hi = rows[inside], query[inside], hi[inside]36 if table.mode != 'linear':37 values[rows] = nodes[hi - 1 if table.mode == 'previous' else hi]38 else:39 # 仅运算内部点;先标记非有限轴差,再区分内部点坐标折叠与正常端点。40 left, right = axis[hi - 1], axis[hi]41 y0, y1 = nodes[hi - 1], nodes[hi]42 flat = y0 == y143 values[rows[flat]] = y0[flat]44 rows, query, left, right = rows[~flat], query[~flat], left[~flat], right[~flat]45 y0, y1 = y0[~flat], y1[~flat]46 with np.errstate(over='ignore', invalid='ignore', under='ignore'):47 span = right - left48 fraction = (query - left) / span49 same_sign = (y0 >= 0) == (y1 >= 0)50 result = np.empty_like(fraction)51 result[same_sign] = y0[same_sign] + fraction[same_sign] * (y1[same_sign] - y0[same_sign])52 result[~same_sign] = (1.0 - fraction[~same_sign]) * y0[~same_sign] + fraction[~same_sign] * y1[~same_sign]53 finite = np.isfinite(span) & np.isfinite(result)54 loss = finite & ((fraction == 0.0) | (fraction == 1.0))55 quality[rows[~finite]] |= int(Q.NONFINITE)56 quality[rows[loss]] |= int(Q.PRECISION_LOSS)57 keep = finite & ~loss58 values[rows[keep]] = result[keep]59 return NumericArray(table.dtype, values, quality == 0, quality)60