chronaxis.core.normalize_numpy
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""按信号声明转换一列原字段值，并保留有效性、质量和原来源身份。23直接入口输入SignalDefinition、同质的已解码原列及uint16上游质量，输出4NumericArray。执行额外线性转换、编码映射、无效值及业务范围检查，不重复5DBC已完成的缩放。整数仿射先用精确配置常量求合法输入区间，再用无符号6距离按列计算，避免定宽中间回绕，不使用object数组进行数值运算。7memory_series_numpy按原生类型分组装箱，批量转换后按原行顺序恢复样本。8Python逐行操作负责类型分组和身份还原，不执行逐样本数值参考计算。9本模块依赖NumPy和核心模型，不读取文件或旧来源，不修改调用方原列；10完整内存历史会分配与行数成比例的存储，不保证生产执行预算。11"""1213# 配置常量的精确区间算法与列数值计算分开。14from math import ceil, floor, isfinite15# NumPy显式加载，承担真实typed列/ufunc与掩码计算。16import numpy as np1718# 共享校验模型/布局；不导入参考normalize函数。19from .binding import SignalBindings, SignalDefinition20# 保留原样本来源与身份，不能重写原行。21from .models import SignalSample22# 仅共享dtype/定宽静态协议；数组路径不调用标量数值。23from .numeric import INTEGER_BOUNDS24# 实际独立数组入口/布局供列执行及测试对照。25from .numeric_numpy import DTYPES, NumericArray, _from_magnitude, _magnitude, _output, to_float_numpy26# 接入已校验保持模型，不改变原来源顺序。27from .state import SignalSeries28# 只读静态映射或显式质量/type枚举用于结构协议。29from .types import QualityFlag, SignalDType30# 时间offset使用已有精确整数加法协议。31from ..timebase.conversion import checked_add_ns323334def _affine_integer(raw, selected, dtype, scale, offset):35 """用逆映射范围与幅度分段实现精确整数线性转换。3637 参数: raw为int64/uint64列；selected为有效mask；dtype目标整数；38 scale/offset为任意精确Python整数配置常量。39 返回: 目标定宽列及OVERFLOW mask；允许中间数学值越界但最终值合法。40 异常: 调用侧校验类型，分配错误传播；不逐样本用Python算数。41 """42 # 目标上下界在精确Python配置域，不转float。43 lower, upper = INTEGER_BOUNDS[dtype]44 # 原输入只支持两种64位整数，边界来自真实物理类型。45 input_lower, input_upper = (-(2 ** 63), 2 ** 63 - 1) if raw.dtype == np.dtype('int64') else (0, 2 ** 64 - 1)46 # 输出同长目标列，invalid占位不正常使用。47 result = np.zeros(len(raw), dtype=DTYPES[dtype])48 # 零scale也是声明关系，但非法/非有限原输入早已被拒绝。49 if scale == 0:50 # 只有常数结果在目标范围内才正常。51 overflow = selected.copy() if not lower <= offset <= upper else np.zeros(len(raw), dtype=bool)52 # 合法结果为同一常量，不执行乘法。53 if lower <= offset <= upper:54 # target常量已证明可表示。55 result[selected] = offset56 # 上游原因不由占位污染。57 return result, overflow58 # 正scale和负scale分别求精确逆区间，向上取整使用ceil(n/d)=-floor(-n/d)。59 if scale > 0:60 # 求最小合法整数输入。61 lo = max(input_lower, -((-(lower - offset)) // scale))62 # 求最大合法整数输入。63 hi = min(input_upper, (upper - offset) // scale)64 else:65 # 负scale反转不等号，仍用精确配置整数。66 lo = max(input_lower, -((-(upper - offset)) // scale))67 # 最大合法输入对应目标下界。68 hi = min(input_upper, (lower - offset) // scale)69 # 不存在合法原整数时不能进行任何数值运算。70 if lo > hi:71 # 全部选中位置明确目标OVERFLOW。72 return result, selected.copy()73 # 逆区间已经与原dtype求交，比较常量可安全cast。74 accepted = selected & (raw >= raw.dtype.type(lo)) & (raw <= raw.dtype.type(hi))75 # 先标记不合法输入，不能先机器乘加再尝试检查。76 overflow = selected & ~accepted77 # 精确选取最小合法输入作为局部计算原点。78 base = lo * scale + offset79 # 子集所有距离非负且不超过原64位跨度。80 values = raw[accepted]81 # 到负pivot的距离用幅度相减或相加，不算signed x-lo。82 if lo < 0:83 # 此分支原输入一定signed。84 magnitude = _magnitude(values)85 # 负输入离pivot更近，幅度≤abs(pivot)。86 negative = values < 087 # 距离为uint64，一维有效子集M。88 distance = np.empty(len(values), dtype=np.uint64)89 # abs(pivot)-abs(x)安全非负。90 distance[negative] = np.uint64(-lo) - magnitude[negative]91 # abs(pivot)+x≤uint64 max，已由原输入跨度证明。92 distance[~negative] = np.uint64(-lo) + magnitude[~negative]93 else:94 # 非负pivot意味着选中输入非负，可安全cast。95 distance = values.astype(np.uint64) - np.uint64(lo)96 # 如果系数超过uint64 max，逆区间最多一个输入，所有距离为0。97 if abs(scale) > 2 ** 64 - 1:98 # 无需cast巨大系数；静态范围已证明合法结果都是base。99 result[accepted] = base100 # 正常结果完全精确。101 return result, overflow102 # distance*abs(scale)≤目标跨度≤uint64 max，由逆区间证明安全。103 change = distance * np.uint64(abs(scale))104 # unsigned目标只做已证明不溢出/下溢的相应方向运算。105 if dtype is SignalDType.UINT64:106 # 最终合法范围已在运算前证明。107 result[accepted] = np.uint64(base) + change if scale > 0 else np.uint64(base) - change108 else:109 # signed结果跨过零时使用幅度与符号分段，避免signed中间超界。110 magnitude = np.empty(len(change), dtype=np.uint64)111 # 正scale增加，负base可越过零。112 if scale > 0:113 # base>=0全部非负且base+change≤max。114 if base >= 0:115 # unsigned内部结果也在signed max内。116 magnitude[:] = np.uint64(base) + change117 # 没有负结果。118 negative = np.zeros(len(change), dtype=bool)119 else:120 # change小于abs(base)的位置仍为负。121 negative = change < np.uint64(-base)122 # 不同方向分别执行安全unsigned减法。123 magnitude[negative] = np.uint64(-base) - change[negative]124 # 含零结果，幅度不超signed max。125 magnitude[~negative] = change[~negative] - np.uint64(-base)126 else:127 # base<=0再减少，负幅度加法≤2**63。128 if base <= 0:129 # 已证明目标范围，幅度加法无wrap。130 magnitude[:] = np.uint64(-base) + change131 # 零的符号在位解释函数中特别处理。132 negative = np.ones(len(change), dtype=bool)133 else:134 # 跨零分段，equal对应正常0。135 negative = change > np.uint64(base)136 # 负结果幅度≤2**63。137 magnitude[negative] = change[negative] - np.uint64(base)138 # 正数/零幅度≤signed max。139 magnitude[~negative] = np.uint64(base) - change[~negative]140 # 每个符号段已证明范围，位解释包含min，不做signed负号溢出。141 result[accepted] = _from_magnitude(magnitude, negative)142 # 数值与错误mask保持原行。143 return result, overflow144145146def _float_limit(value, *, lower):147 """将精确配置边界向区间内收至可表示float64端点。148149 参数: value为原生int/有限float；lower表示下界或上界。150 返回: float端点，大到不可表示的整数按方向得到Inf/最大有限值。151 异常: 配置类型已校验；数值转换溢出仅表示不可达端点。152 """153 # 配置整数不能先舍入后改变有效范围。154 try:155 # 这里只转换一次配置常量，不是逐样本数值。156 result = float(value)157 except OverflowError:158 # 超float范围的整数边界仍有确定区间含义。159 result = float('inf') if value > 0 else float('-inf')160 # 向合法方向收缩；Python的float/int比较保留精确整数边界。161 if (lower and result < value) or (not lower and result > value):162 # nextafter处理边界舍入及Inf收缩，避免误收工程值。163 result = float(np.nextafter(result, float('inf') if lower else float('-inf')))164 # 返回可直接批量比较的端点。165 return result166167168def normalize_numpy(definition: SignalDefinition, raw: np.ndarray, quality: np.ndarray | None = None) -> NumericArray:169 """按声明批量转换原列，检查编码映射、无效原值和转换后的业务范围。170171 参数:172 definition (SignalDefinition): 当前信号类型、转换、编码和范围声明。173 raw (ndarray, shape=(N,)): 同质原列，dtype为int64、uint64、float64、174 bool或Unicode文本；不接受object数组和隐式广播。175 quality (ndarray[uint16], shape=(N,) | None): 已知上游质量原因；默认176 None表示全部NONE。已有原因保留，无效位置不进入正常计算。177 返回:178 NumericArray: 目标声明类型的值、有效掩码和完整质量原因。无效值是179 存储占位；内存适配另外从原行保留诊断值。180 异常:181 TypeError或ValueError: 输入布局、质量位或声明结构不合法。182 MemoryError: 数组分配失败。183 数值溢出、非有限值或精度损失记录为质量原因，不输出正常值。184 副作用:185 输出和原因使用局部数组，不修改原列或调用者质量列。186 """187 # 不接受object列或隐藏dtype转换，原数值宽度明确。188 if not isinstance(definition, SignalDefinition) or type(raw) is not np.ndarray or raw.ndim != 1 or not (raw.dtype in (np.dtype('int64'), np.dtype('uint64'), np.dtype('float64'), np.dtype('bool')) or raw.dtype.kind == 'U'):189 # I/O适配应先按原类型装箱。190 raise TypeError('normalize需要声明及1维明确原列类型')191 # 默认上游质量NONE，真实缺失由适配写MISSING。192 reasons = np.zeros(len(raw), dtype=np.uint16) if quality is None else quality193 # 布局不自动广播，不接受未知质量位。194 if type(reasons) is not np.ndarray or reasons.ndim != 1 or reasons.dtype != np.dtype('uint16') or len(reasons) != len(raw) or np.any(reasons > 511):195 # 配置/结构问题不能伪装数据缺失。196 raise ValueError('上游质量需要同长uint16已知位')197 # 后续只改局部原因副本，不修改调用侧。198 reasons = reasons.copy()199 # 每一层只对还正常的位置新增诊断。200 selected = reasons == 0201 # 非有限先拒绝，scale0不能把Inf恢复正常。202 if raw.dtype == np.dtype('float64'):203 # 正常输入NaN/Inf明确NONFINITE。204 reasons[selected & ~np.isfinite(raw)] |= np.uint16(QualityFlag.NONFINITE)205 # 同类型invalid原值规则批量运行，bool/int不会互相匹配。206 native_type = {'i': int, 'u': int, 'f': float, 'b': bool, 'U': str}[raw.dtype.kind]207 # 配置规则循环次数与样本数无关，每次真实列比较。208 for value in definition.invalid_values:209 # 超原dtype整数范围不会存在，不让NumPy隐式promotion到float。210 if type(value) is native_type and (native_type is not int or np.iinfo(raw.dtype).min <= value <= np.iinfo(raw.dtype).max):211 # 仅正常尚未失败的位置拒绝原无效码。212 reasons[(reasons == 0) & (raw == value)] |= np.uint16(QualityFlag.OUT_OF_RANGE)213 # 目标列物理dtype明确，无效为仅存储占位。214 values = np.zeros(len(raw), dtype=DTYPES[definition.dtype])215 # enum域按显式映射第一次标签顺序固定。216 labels = tuple(dict.fromkeys(label for _, label in definition.choices)) if definition.dtype is SignalDType.ENUM else ()217 # bool/enum映射与数值转换完全分开。218 if definition.dtype in (SignalDType.BOOL, SignalDType.ENUM):219 # 原生bool且没有额外choices可以保留真实False。220 if definition.dtype is SignalDType.BOOL and not definition.choices and raw.dtype == np.dtype('bool'):221 # 不使用truthiness推断其他原类型。222 values[reasons == 0] = raw[reasons == 0]223 # 正常False与invalid分开。224 return _output(definition.dtype, values, reasons)225 # 命中域掩码一维N。226 matched = np.zeros(len(raw), dtype=bool)227 # 逐配置编码向整列批量映射，不逐样本回调。228 for code, label in definition.choices:229 # 原生类型必须精确相同，整数编码另检查可表示。230 if type(code) is native_type and (native_type is not int or np.iinfo(raw.dtype).min <= code <= np.iinfo(raw.dtype).max):231 # 只有合法输入能命中声明码。232 mask = (reasons == 0) & (raw == code)233 # enum显式标签编码，bool保留类型。234 values[mask] = labels.index(label) if labels else label235 # 后续未匹配原码将明确无效。236 matched |= mask237 # 不自动取第一项，不转任意整数为bool。238 reasons[(reasons == 0) & ~matched] |= np.uint16(QualityFlag.OUT_OF_RANGE)239 # 返回固定逻辑域及完整质量。240 return _output(definition.dtype, values, reasons, labels)241 # 数值输入拒绝bool/字符串，而非解析/恢复正常。242 if raw.dtype.kind in ('b', 'U'):243 # upstream已有原因保持，不通过错误原码增加算术诊断。244 reasons[reasons == 0] |= np.uint16(QualityFlag.DECODE_ERROR)245 # 所有结果均有明确原因。246 return _output(definition.dtype, values, reasons)247 # 整数目标和整数配置在精确仿射路径处理。248 if definition.dtype in INTEGER_BOUNDS and raw.dtype.kind in ('i', 'u') and type(definition.scale) is int and type(definition.offset) is int:249 # 原整数不先经float，不因中间数学值超界误判最终合法结果。250 values, overflow = _affine_integer(raw, reasons == 0, definition.dtype, definition.scale, definition.offset)251 # 失败输入在计算之前识别。252 reasons[overflow] |= np.uint16(QualityFlag.OVERFLOW)253 else:254 # 原数值列封装只作模型校验，浮点精度核独立。255 raw_dtype = {'i': SignalDType.INT64, 'u': SignalDType.UINT64, 'f': SignalDType.FLOAT64}[raw.dtype.kind]256 # 已标记非有限的位置不是正常float，因此可保留原诊断。257 original_valid = reasons == 0258 # 原值和常量转换失败需要独立合并，不能掩盖同时发生的质量。259 converted = to_float_numpy(_output(raw_dtype, raw, reasons))260 # 合并真实精度失败位置。261 reasons = converted.quality.copy()262 # 配置常量只转换一次，但精度仍严格检查。263 constants = []264 # 系数/offset不应因为转换无穷而吞配置语义。265 for constant in (definition.scale, definition.offset):266 # Python配置常量转float有限检查，与reference核独立。267 try:268 # 常量次数固定，不是行循环。269 number = float(constant)270 except OverflowError:271 # 表示float数值失败，不能生成正常Inf。272 number = float('inf')273 # float64可精确表示常量才可参与列乘加。274 flag = QualityFlag.NONFINITE if not isfinite(number) else QualityFlag.PRECISION_LOSS if type(constant) is int and int(number) != constant else QualityFlag.NONE275 # 数值转换失败只影响原本有效输入。276 reasons[original_valid] |= np.uint16(flag)277 # 局部常量仅在全部检查后用于算术。278 constants.append(number)279 # 只有原值与常量都正常才执行线性转换。280 selected = reasons == 0281 # 局部float输出一维N，invalid占位不参与计算。282 transformed = np.zeros(len(raw), dtype=np.float64)283 # 单独乘再加，与参考float流程一致，不使用融合乘加改变两步计算的舍入。284 with np.errstate(over='ignore', invalid='ignore'):285 # where保证无效输入/非法常量未执行乘加。286 np.multiply(converted.values, constants[0], out=transformed, where=selected)287 # 第二步同样只对原先正常执行位置。288 np.add(transformed, constants[1], out=transformed, where=selected)289 # 算术非有限明确失败，不通过targetcast隐藏。290 reasons[selected & ~np.isfinite(transformed)] |= np.uint16(QualityFlag.NONFINITE)291 # 整数目标还需要拒绝小数和端点外float。292 if definition.dtype in INTEGER_BOUNDS:293 # 保留前序失败后再检测小数。294 selected = reasons == 0295 # 不截断合法小数冒充整型。296 reasons[selected & (transformed != np.trunc(transformed))] |= np.uint16(QualityFlag.PRECISION_LOSS)297 # 上界采用exclusive 2**63/2**64，避免max转float向上舍入。298 lo, hi = INTEGER_BOUNDS[definition.dtype]299 # 下界和exclusive上界均能精确表示。300 outside = (transformed < float(lo)) | (transformed >= float(hi + 1))301 # 小数精度失败优先级与参考一致，不添加额外OVERFLOW。302 reasons[(reasons == 0) & outside] |= np.uint16(QualityFlag.OVERFLOW)303 # 证明可表示后才能实际astype。304 selected = reasons == 0305 # 不对NaN/Inf/超界值执行cast。306 values[selected] = transformed[selected].astype(DTYPES[definition.dtype])307 else:308 # float结果直接保持真实计算值和质量。309 values = transformed310 # 业务范围针对最终有效转换值，严禁先舍入配置整数边界。311 if definition.valid_range is not None:312 # 固定两个配置端点，不是逐样本检查。313 lo, hi = definition.valid_range314 # float边界向合法区间内收后批量比较。315 if definition.dtype is SignalDType.FLOAT64:316 # 端点转换是精确边界而非隐藏工程值修正。317 outside = (values < _float_limit(lo, lower=True)) | (values > _float_limit(hi, lower=False))318 else:319 # 整型工程边界先取ceil/floor再求目标域交集。320 lower, upper = INTEGER_BOUNDS[definition.dtype]321 # 配置Python整数保持完整；float有限边界取整精确。322 lo, hi = max(lower, ceil(lo)), min(upper, floor(hi))323 # 空域全部工程失败，否则安全cast边界。324 outside = np.ones(len(raw), dtype=bool) if lo > hi else ((values < DTYPES[definition.dtype].type(lo)) | (values > DTYPES[definition.dtype].type(hi)))325 # 先前数值失败不被业务范围0占位污染。326 reasons[(reasons == 0) & outside] |= np.uint16(QualityFlag.OUT_OF_RANGE)327 # 最终结果独立只读，valid只由完整质量决定。328 return _output(definition.dtype, values, reasons)329330331def memory_series_numpy(bindings: SignalBindings, name: str, *, capture_code=False) -> SignalSeries:332 """按原字段类型分组装箱，批量转换后恢复原行顺序和来源身份。333334 参数:335 bindings (SignalBindings): 当前会话的已校验绑定。336 name (str): 已绑定的原信号别名。337 返回:338 SignalSeries: 正常值使用转换结果，无效值保留原字段诊断；时间只应用339 声明的固定偏移，原序号和来源引用保持不变。340 异常:341 TypeError、ValueError或OverflowError: 定义、布局、同时间冲突、原行归属342 或时间映射检查失败。343 MemoryError: 装箱、结果或数组分配失败。344 副作用:345 分配本次完整历史，不读取旧来源或修改原行。346 """347 # 当前定义与真实来源唯一，不继承上次输入。348 definition = bindings.resolve(name)349 # 来源名称已在绑定层校验。350 source = next(item for item in bindings.session.sources if item.name == definition.source)351 # 原行顺序，只选已有目标字段，不增加观测或解码。352 rows = tuple(row for row in source.rows if definition.field in row.values)353 # 按原生物理类型分组装箱，不使用object算术。354 groups = {}355 # 空间为原输入行数；当前memory核心不声称生产资源门槛。356 results = [None] * len(rows)357 # Python逐行仅作原生类型/宽度检查与装箱身份。358 for index, row in enumerate(rows):359 # 原值来自已解码memory字段。360 value = row.values[definition.field]361 # 缺失无法在正常typed列表示，先包装质量而不是补成正常0。362 if value is None:363 # 模拟typed原列占位但携带MISSING，转换核不会使用。364 key, packed, flag = 'i', 0, QualityFlag.MISSING365 # 原生整数先确定原64位存储域，不能从目标dtype倒推源宽度。366 elif type(value) is int:367 # 原数值协议仅支持int64/uint64；宽度超出明确OVERFLOW。368 flag = QualityFlag.NONE if -(2 ** 63) <= value <= 2 ** 64 - 1 else QualityFlag.OVERFLOW369 # 超界原整数占位不参与运算，原诊断最后恢复。370 key, packed = ('i' if value <= 2 ** 63 - 1 else 'u'), value if flag == QualityFlag.NONE else 0371 else:372 # bool和str保持各自独立物理域，float不提前过滤非有限。373 key, packed, flag = {float: 'f', bool: 'b', str: 'U'}[type(value)], value, QualityFlag.NONE374 # 每组保存输出位置/原数据/质量，只进行存储布局操作。375 groups.setdefault(key, []).append((index, packed, int(flag)))376 # 不同原类型分别走同质真实数组内核，循环次数至多5。377 for key, items in groups.items():378 # 类型由装箱规则明确，不采用跨类型隐式promotion。379 dtype = {'i': np.int64, 'u': np.uint64, 'f': np.float64, 'b': bool, 'U': str}[key]380 # 每组批量形成原值/质量typed数组。381 raw = np.array([value for _, value, _ in items], dtype=dtype)382 # None/超界输入携带质量，因此占位不能恢复正常。383 quality = np.array([flag for _, _, flag in items], dtype=np.uint16)384 # 核心计算真实批量执行，不调用reference归一化。385 normalized = normalize_numpy(definition, raw, quality)386 # Python遍历仅还原公开原生结果与身份，不进行数值计算。387 for position, (index, _, _) in enumerate(items):388 # 正常enum编码按冻结域恢复标签。389 value = normalized.labels[int(normalized.values[position])] if normalized.labels and normalized.valid[position] else normalized.values[position].item()390 # 保存值/有效性/完整质量，原诊断稍后按行恢复。391 results[index] = (value, bool(normalized.valid[position]), QualityFlag(int(normalized.quality[position])))392 # 输出按原行顺序恢复，原时间offset明确且安全。393 from .enum_codes import original_code394 samples = tuple(SignalSample(name, checked_add_ns(row.time_ns, source.offset_ns), value if valid else row.values[definition.field], valid, quality, bindings.source_ref(name), row.sequence,395 original_code(definition, row.values[definition.field]) if capture_code else None)396 for row, (value, valid, quality) in zip(rows, results))397 # 同时间及来源约束由SignalSeries检查，不改变原来源消息数量。398 return SignalSeries(bindings, name, samples)399