chronaxis.core.lookup
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""校验 v4 一维标定表声明，并按已请求的插值方式生成只读计算表。23声明只保存配置点对和内容身份。被表达式引用时才将两列写为固定小端字节，4支持标量随机读取或 NumPy 零复制视图。无文件、全局缓存或第三方依赖。5"""67from dataclasses import dataclass, field, replace8from hashlib import sha2569from math import isfinite10from struct import Struct1112from .types import SignalDType as D13from .unit_registry import resolve_unit14from .validation import require_identifier151617MAX_TABLES = 6418MAX_POINTS = 409619_FORMATS = {D.INT64: Struct('<q'), D.UINT64: Struct('<Q'), D.FLOAT64: Struct('<d')}202122def _exact_f64(value):23 """校验 float64 转换。参数: 表内原生数。返回: float。异常: 非有限、整数超界或精度损失。"""24 if type(value) not in (int, float):25 raise TypeError('查表点必须为原生数值，不能是BOOL或文本')26 if type(value) is int and not -(1 << 63) <= value < (1 << 64):27 raise ValueError('查表整数超出64位范围')28 converted = float(value)29 if not isfinite(converted) or converted != value:30 raise ValueError('查表点必须可由有限float64精确表示')31 return converted323334@dataclass(frozen=True)35class LookupTablePlan:36 """保存经校验的有限标定点。身份保留原生数字类型、单位和每一点的次序。"""3738 name: str39 points: tuple[tuple[int | float, int | float], ...]40 input_unit: str41 unit: str42 dtype: D = field(init=False)43 identity: str = field(init=False)4445 def __post_init__(self):46 """校验轴和阶梯输出的共同类型。参数: 无。返回: 无。异常: 类型、单位、次序、大小或精度错误。"""47 require_identifier(self.name, 'lookup.name')48 if resolve_unit(self.input_unit) is None or resolve_unit(self.unit) is None:49 raise ValueError('查表输入/输出须声明已登记的数值单位')50 if type(self.points) is not tuple or not 2 <= len(self.points) <= MAX_POINTS:51 raise ValueError(f'查表points需要2至{MAX_POINTS}个冻结点对')52 previous, low, high, floating = None, 0, 0, False53 for row in self.points:54 if type(row) is not tuple or len(row) != 2:55 raise ValueError('查表points须为输入/输出的二元点对')56 x, y = row57 axis = _exact_f64(x)58 if previous is not None and axis <= previous:59 raise ValueError('查表轴须严格递增且不允许重复')60 previous = axis61 if type(y) is float:62 _exact_f64(y)63 floating = True64 elif type(y) is int and -(1 << 63) <= y < (1 << 64):65 low, high = min(low, y), max(high, y)66 else:67 raise ValueError('查表输出须为有限浮点数或64位整数')68 # 整数阶梯不经过 float64，可以无损保存 uint64 最大值。69 if floating:70 for _, y in self.points:71 _exact_f64(y)72 dtype = D.FLOAT6473 elif high < (1 << 63):74 dtype = D.INT6475 elif low >= 0:76 dtype = D.UINT6477 else:78 raise ValueError('查表输出负数与大uint64无法共用定宽整数类型')79 object.__setattr__(self, 'dtype', dtype)80 # 延迟导入避免基础绑定依赖查表声明时循环加载。81 from .binding import _identity82 payload = ('lookup-table-v1', self.name, self.input_unit, self.unit, self.points)83 object.__setattr__(self, 'identity', sha256(_identity(payload).encode('utf-8')).hexdigest())8485 def prepare(self, mode, bounds):86 """校验查表调用，不生成数值列。参数: 插值/越界模式。返回: PreparedLookup。异常: 模式或精度错误。"""87 if mode not in ('linear', 'previous', 'next') or bounds not in ('invalid', 'clamp'):88 raise ValueError('lookup仅支持linear/previous/next以及invalid/clamp')89 dtype = D.FLOAT64 if mode == 'linear' else self.dtype90 if dtype is D.FLOAT64 and self.dtype is not D.FLOAT64:91 for _, y in self.points:92 _exact_f64(y)93 return PreparedLookup(self.identity, mode, bounds, dtype, self.points)949596@dataclass(frozen=True)97class PreparedLookup:98 """保存已校验的模式和冻结点引用；实际数组执行者可单独持有只读打包列。"""99100 identity: str101 mode: str102 bounds: str103 dtype: D104 points: tuple = field(repr=False, compare=False)105 axis: bytes | None = field(default=None, repr=False, compare=False)106 values: bytes | None = field(default=None, repr=False, compare=False)107108 def packed(self):109 """产生只读数组视图的底层列。参数: 无。返回: 定宽两列的执行副本。异常: 分配错误。"""110 if self.axis is not None:111 return self112 axis, values = bytearray(8 * len(self.points)), bytearray(8 * len(self.points))113 for index, (x, y) in enumerate(self.points):114 _FORMATS[D.FLOAT64].pack_into(axis, 8 * index, x)115 _FORMATS[self.dtype].pack_into(values, 8 * index, y)116 return replace(self, axis=bytes(axis), values=bytes(values))117118 def axis_at(self, index):119 """按序号读取物理坐标。参数: 合法索引。返回: float。异常: 非法索引。"""120 return float(self.points[index][0])121122 def value_at(self, index):123 """按序号读取定宽值。参数: 合法索引。返回: 原生int或float。异常: 非法索引。"""124 y = self.points[index][1]125 return float(y) if self.dtype is D.FLOAT64 else y126127128def pack_requested_lookups(expression):129 """只为实际执行的根建立数组查表列。参数: 执行者的表达式。返回: 局部执行副本。异常: 分配错误。130131 共用同一配置及模式的调用只打包一次。未消费的派生定义不会进入此根，原声明保持不可变。132 """133 if not expression.bindings.lookups:134 return expression135 nodes, tables = {}, {}136137 def visit(node):138 """改写本根的查表节点。参数: 原节点。返回: 原节点或带执行列的副本。异常: 分配错误。"""139 if id(node) not in nodes:140 args = tuple(visit(child) for child in node.args)141 if node.operator == 'lookup':142 if node.value not in tables:143 tables[node.value] = node.value.packed()144 changed = replace(node, args=args, value=tables[node.value])145 elif any(a is not b for a, b in zip(args, node.args)):146 changed = replace(node, args=args)147 else:148 changed = node149 nodes[id(node)] = changed150 return nodes[id(node)]151152 root = visit(expression.root)153 return expression if root is expression.root else replace(expression, root=root)154155156def validate_lookups(tables, version):157 """检查有限表集合。参数: 冻结表元组和配置版本。返回: 无。异常: 非v4、重复名或表过多。"""158 if type(tables) is not tuple or any(type(table) is not LookupTablePlan for table in tables):159 raise TypeError('lookups需为LookupTablePlan tuple')160 if tables and version != 4:161 raise ValueError('lookups需要config_version=4')162 if len(tables) > MAX_TABLES or len({t.name for t in tables}) != len(tables):163 raise ValueError(f'lookups表名须唯一且最多{MAX_TABLES}张')164