chronaxis.core.enum_codes
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在 v4 表达式中读取按需保留的枚举原始整数列，标签索引没有原码含义。"""23from .numeric import INTEGER_BOUNDS, ScalarValue4from .types import QualityFlag as Q, SignalDType as D5from math import isfinite678def validate_code_dtype(dtype, choices):9 """检查声明的整数存储类型及其可表示的 choices 键。1011 参数: dtype 为 INT64 或 UINT64，choices 为原代码到逻辑标签的映射对。12 返回: None。异常: ValueError 为非整数类型、非整数键或定宽范围不符。13 同一标签可由多个不同代码表示，此处不反向查找标签。14 """15 if dtype not in (D.INT64, D.UINT64):16 raise ValueError('枚举 code_dtype 必须是 int64 或 uint64')17 lower, upper = INTEGER_BOUNDS[dtype]18 if any(type(code) is not int or not lower <= code <= upper for code, _ in choices):19 raise ValueError('枚举 choices 原码必须是 code_dtype 可表示的整数')202122def compile_enum_code(compiler, call):23 """从已编译的枚举来源取得原码 dtype 并登记本节点的真实依赖。2425 参数: compiler 为 v4 编译器，call 为 enum_code 的 AST 调用。26 返回: 输出单位 1 的整数节点。异常: 参数数量、来源或原码类型不明时报编译错误。27 """28 from .expression import _node29 if len(call.args) != 1 or call.keywords:30 compiler._error(call, 'enum_code 需要一个枚举信号参数')31 operand = compiler.visit(call.args[0])32 if operand.operator != 'name' or operand.dtype is not D.ENUM:33 compiler._error(call, 'enum_code 需要来自声明信号的枚举值')34 dtype = compiler.signals[operand.name].code_dtype35 if dtype is None:36 compiler._error(call, 'enum_code 需要已核实的整数原码类型，请声明 code_dtype: int64 或 uint64')37 return _node('enum_code', dtype, '1', (operand,))383940def code_dependencies(root):41 """收集参与本次计算或其派生子树的枚举原码，忽略单纯使用标签的名称。4243 参数: root 为已编译表达式根。返回: 需要同时读取原码的别名集合。44 异常: 输入不是表达式节点时传播属性错误。45 """46 found, seen, pending = set(), set(), [root]47 while pending:48 node = pending.pop()49 if id(node) in seen:50 continue51 seen.add(id(node))52 if node.operator == 'enum_code':53 found.add(node.args[0].name)54 pending.extend(node.args)55 return frozenset(found)565758def original_code(definition, original, quality=Q.NONE):59 """把已解码原整数装入声明的定宽类型，不经标签或浮点中转。6061 参数: definition 为有 code_dtype 的枚举，original 为映射前原值，quality 为来源质量。62 返回: 整数 ScalarValue。异常: 缺少有效类型声明时报错。未知映射不使整数本身无效。63 """64 dtype = definition.code_dtype65 lower, upper = INTEGER_BOUNDS[dtype]66 if quality == Q.NONE:67 if original is None:68 quality = Q.MISSING69 elif type(original) is float and not isfinite(original):70 quality = Q.NONFINITE71 elif type(original) is not int:72 quality = Q.DECODE_ERROR73 elif not lower <= original <= upper:74 quality = Q.OVERFLOW75 elif any(type(value) is int and value == original for value in definition.invalid_values):76 quality = Q.OUT_OF_RANGE77 return ScalarValue(original if not quality else None, dtype, not quality, quality)787980def code_column(definition, originals):81 """按原生输入类型装箱有限原码批；未请求不分配列。8283 参数: definition 为运行投影，originals 为映射前异构 Python 输入。84 返回: None 或只读的定宽整数 NumericArray。异常: 声明或输入违反模型时传播。85 """86 if not definition.capture_code:87 return None88 import numpy as np89 from .numeric_numpy import NumericArray, DTYPES90 values = np.zeros(len(originals), dtype=DTYPES[definition.code_dtype])91 quality = np.zeros(len(originals), dtype=np.uint16)92 for index, original in enumerate(originals):93 code = original_code(definition, original)94 if code.valid:95 values[index] = code.value96 quality[index] = int(code.quality)97 return NumericArray(definition.code_dtype, values, quality == 0, quality)9899100def memory_histories(compiled, names, factory, condition=None):101 """只为实际消费码视图的表达式建立附原码的历史。102103 参数: compiled/condition 为值和可选条件，names 为实际联合依赖，factory 为独立后端。104 返回: 名称到 SignalSeries 的映射。异常: 原输入转换或绑定错误直接传播。105 """106 required = code_dependencies(compiled.root)107 if condition is not None:108 required |= code_dependencies(condition.root)109 return {name: factory(compiled.bindings, name, capture_code=name in required) for name in names}110111112def aligned_scalar(definition, state):113 """恢复对齐点的标签和独立原码，过期和域外状态同时作用于两者。114115 参数: definition 为信号声明，state 为参考保持结果。116 返回: ScalarValue。异常: 质量或输入布局非法时传播。117 """118 labels = tuple(dict.fromkeys(label for _, label in definition.choices)) if definition.dtype is D.ENUM else ()119 code = state.sample.code if state.sample is not None else None120 boundary = state.quality & (Q.MISSING | Q.STALE)121 if code is not None and boundary:122 code = ScalarValue(None, code.dtype, False, code.quality | boundary)123 return ScalarValue(state.value, definition.dtype, state.valid, state.quality, labels, code)124125126def expire_value(value, reason):127 """把同一观测边界分别施加到值及附带原码，各自保留独立质量。128129 参数: value 为保存的最新值，reason 为本次缺观测或过期原因。130 返回: 当前时刻无效的 ScalarValue。异常: 错误 dtype 或质量组合由模型拒绝。131 """132 code = value.code133 if code is not None:134 code = ScalarValue(None, code.dtype, False, code.quality | reason)135 return ScalarValue(None, value.dtype, False, value.quality | reason, value.labels, code)136137138def aligned_column(state):139 """使用数组保持的相同样本索引同步原码及其自身质量。140141 参数: state 为 AlignedArray。返回: NumericArray；标签单独使用时不分配码列。142 异常: 不一致的历史布局或整数存储错误。143 """144 import numpy as np145 from .numeric_numpy import DTYPES, NumericArray146 history = state.history147 samples = history.series.samples148 definition = history.series.bindings.resolve(history.series.name)149 code = None150 if samples and samples[0].code is not None:151 dtype = definition.code_dtype152 source = np.asarray([s.code.value if s.code.valid else 0 for s in samples], dtype=DTYPES[dtype])153 reasons = np.asarray([int(s.code.quality) for s in samples], dtype=np.uint16)154 present = state.sample_indices >= 0155 values = np.zeros(state.values.shape, dtype=DTYPES[dtype])156 quality = state.quality & int(Q.MISSING | Q.STALE)157 indices = state.sample_indices[present]158 values[present] = source[indices]159 quality[present] |= reasons[indices]160 code = NumericArray(dtype, values, quality == 0, quality)161 return NumericArray(definition.dtype, state.values, state.valid, state.quality, history.labels, code)162163164def history_code(definition, sample):165 """读取历史中单独保存的原码，从原数据重查标签映射无关的质量。166167 参数: definition 为枚举声明，sample 为原历史观测，None 表示从未观测。168 返回: 原码 ScalarValue。异常: 缺少必要历史投影时明确失败。169 """170 if sample is None:171 return ScalarValue(None, definition.code_dtype, False, Q.MISSING)172 if sample.value.code is not None:173 return sample.value.code174 if not sample.original_present:175 raise ValueError('enum_code 需要回读时保留枚举原始整数')176 quality = sample.value.quality177 if sample.original_value is not None:178 quality &= ~Q.OUT_OF_RANGE179 return original_code(definition, sample.original_value, quality)180181182def historical_column(definition, arrays):183 """沿历史保持的有样本掩码及时间质量构建请求的原码列。184185 参数: definition 为枚举声明，arrays 为同一历史快照的有限保持批。186 返回: 与该批同长的整数 NumericArray。异常: 缺少原码投影或布局错误。187 """188 import numpy as np189 from .numeric_numpy import NumericArray, DTYPES190 code = history_code(definition, arrays.sample)191 quality = arrays.quality & int(Q.MISSING | Q.STALE)192 quality[arrays.has_sample] |= int(code.quality)193 values = np.full(arrays.values.shape, code.value if code.valid else 0, dtype=DTYPES[code.dtype])194 return NumericArray(code.dtype, values, quality == 0, quality)195196197def enum_code_reference(node, operand):198 """读取一条原码及其独立质量，保持有效的未映射代码。199200 参数: node 为已检查的 enum_code 节点，operand 为对应 ENUM 的 ScalarValue。201 返回: 整数 ScalarValue。异常: 已有观测却缺原码投影或类型不一致时报错。202 """203 if operand.code is not None and operand.code.dtype is node.dtype:204 return operand.code205 if operand.code is None and operand.quality & (Q.MISSING | Q.STALE):206 return ScalarValue(None, node.dtype, False, operand.quality)207 raise ValueError('枚举缺少与声明一致的原码投影，无法从标签还原代码')208209210def enum_code_numpy(node, operand):211 """读取当前批的原码列及独立有效性，原码与枚举有效性分别判断。212213 参数: node 为已检查的 enum_code 节点，operand 为形状 (N,) 的 ENUM 列。214 返回: 同形状整数 NumericArray。异常: 有观测而原码投影缺失或类型错误时报错。215 """216 import numpy as np217 from .numeric_numpy import NumericArray, DTYPES218 if operand.code is not None and operand.code.dtype is node.dtype:219 return operand.code220 if operand.code is None and np.all((operand.quality & int(Q.MISSING | Q.STALE)) != 0):221 return NumericArray(node.dtype, np.zeros(len(operand.values), dtype=DTYPES[node.dtype]),222 operand.valid, operand.quality)223 raise ValueError('枚举缺少与声明一致的原码投影，无法从标签还原代码')224