chronaxis.fast.compute
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""把单帧数值公式准备成直接读取解码字典的函数，按原始到达顺序累计。23修复版 base 可传只含原生数值尺度的算式；配置入口另传公共编译器核过的4program，以保留 v4 在表达式内部插入的单位换算。两条路径都不执行用户代码。5"""67import ast8import math9import operator101112_LIMITS = {'int64': (-(1 << 63), (1 << 63) - 1), 'uint64': (0, (1 << 64) - 1)}13_BINARY = {'add': operator.add, 'sub': operator.sub, 'mul': operator.mul, 'div': operator.truediv}14_SYNTAX = {ast.Add: 'add', ast.Sub: 'sub', ast.Mult: 'mul', ast.Div: 'div'}151617def _float(value):18 """转换有限且可精确转双精度的数值；拒绝 bool 和非有限值。"""19 if type(value) not in (int, float):20 raise ValueError('需要物理解码后的数值')21 result = float(value)22 if not math.isfinite(result) or result != value:23 raise ValueError('非有限数值或整数转 float64 丢失精度')24 return result252627def _typed(value, dtype):28 """检查定宽整数范围、整数性或双精度可表示性，返回目标原生数值。"""29 if dtype == 'float64':30 return _float(value)31 if dtype not in _LIMITS or type(value) not in (int, float):32 raise ValueError('需要已声明的数值类型')33 low, high = _LIMITS[dtype]34 if not low <= value <= high or type(value) is float and not value.is_integer():35 raise ValueError(f'{dtype} 溢出或小数精度丢失')36 return int(value)373839def _number_type(value):40 """为已检查的数值常量确定低层公式的自然类型。"""41 if type(value) is float:42 _float(value)43 return 'float64'44 if type(value) is not int:45 raise ValueError('常量必须为数值')46 dtype = 'uint64' if value > _LIMITS['int64'][1] else 'int64'47 _typed(value, dtype)48 return dtype495051def _affine(value, scale, offset, dtype):52 """按一个完整仿射变换检查最终类型，允许摄氏到开尔文的合法抵消。"""53 if dtype == 'float64' or not all(type(v) is int for v in (value, scale, offset)):54 value, scale, offset = _float(value), _float(scale), _float(offset)55 total = value * scale + offset56 if total == 0 and offset == 0 and value != 0 and scale != 0:57 raise ValueError('单位换算下溢')58 return _typed(total, dtype)596061def encode_program(node):62 """将公共 ExprNode 的数值子集编码为普通 dict。其他操作准备时即拒绝。"""63 op = node.operator64 if op not in {*_BINARY, 'name', 'const', 'abs', 'neg', 'to_float', 'unit_convert'}:65 raise ValueError(f'fast 不支持表达式操作 {op}')66 result = {'op': op, 'dtype': node.dtype.value}67 if op == 'name':68 result['name'] = node.name69 elif op == 'const':70 result['value'] = node.value71 else:72 result['args'] = [encode_program(child) for child in node.args]73 if op == 'unit_convert':74 result['scale'], result['offset'] = node.value75 return result767778def _syntax(text, signals, parameters):79 """将受限的原尺度公式转换为节点；不解析属性、容器或任意函数。"""80 if type(text) is not str or not 0 < len(text) <= 4096:81 raise ValueError('逐帧公式必须是非空且不超过4096字符的文本')82 tree = ast.parse(text, mode='eval')83 if sum(1 for _ in ast.walk(tree)) > 512:84 raise ValueError('逐帧公式过长')8586 def literal(value):87 return dict(op='const', dtype=_number_type(value), value=value)8889 def visit(node):90 if isinstance(node, ast.Constant):91 return literal(node.value)92 if isinstance(node, ast.Name):93 if node.id in parameters:94 return literal(parameters[node.id])95 if node.id not in signals:96 raise ValueError(f'未知信号或参数 {node.id}')97 return dict(op='name', name=node.id, dtype=signals[node.id]['dtype'])98 if isinstance(node, ast.UnaryOp) and isinstance(node.op, (ast.UAdd, ast.USub)):99 if isinstance(node.op, ast.USub) and isinstance(node.operand, ast.Constant) and type(node.operand.value) in (int,float):100 return literal(-node.operand.value)101 child = visit(node.operand)102 return child if isinstance(node.op, ast.UAdd) else dict(op='neg', dtype=child['dtype'], args=[child])103 if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == 'abs' and len(node.args) == 1 and not node.keywords:104 child = visit(node.args[0])105 return dict(op='abs', dtype=child['dtype'], args=[child])106 if isinstance(node, ast.BinOp) and type(node.op) in _SYNTAX:107 left, right = visit(node.left), visit(node.right)108 op = _SYNTAX[type(node.op)]109 dtype = 'float64' if op == 'div' or 'float64' in (left['dtype'],right['dtype']) else left['dtype']110 return dict(op=op, dtype=dtype, args=[left, right])111 raise ValueError('fast 公式仅支持数值、名称、加减乘除、正负号和 abs')112 return visit(tree.body)113114115def _bind(node, signals, depth=0):116 """在准备阶段创建标量运算函数，运行时只传当帧的解码字典。"""117 if depth > 128 or type(node) is not dict:118 raise ValueError('无效或过深的逐帧数值指令')119 op, dtype = node['op'], node['dtype']120 if dtype not in ('float64', 'int64', 'uint64'):121 raise ValueError('逐帧累计需要数值输出')122 if op == 'const':123 constant = _typed(node['value'], dtype)124 return lambda values: constant125 if op == 'name':126 signal = signals[node['name']]127 if dtype != signal['dtype']:128 raise ValueError('指令中的信号类型与声明不符')129 key = signal['signal']130 scale, offset = signal.get('scale', 1), signal.get('offset', 0)131 _number_type(scale)132 _number_type(offset)133 original = signal.get('input_dtype', dtype)134 if scale == 1 and offset == 0 and original == dtype:135 return lambda values: _typed(values[key], dtype)136 return lambda values: _affine(_typed(values[key], original), scale, offset, dtype)137 args = [_bind(child, signals, depth + 1) for child in node['args']]138 if len(args) == 2 and op in _BINARY:139 left, right = args140 combine = _BINARY[op]141 if dtype == 'float64':142 return lambda values: _float(combine(_float(left(values)), _float(right(values))))143 return lambda values: _typed(combine(left(values), right(values)), dtype)144 if len(args) == 1:145 child, = args146 if op == 'to_float':147 return lambda values: _float(child(values))148 if op in ('neg', 'abs'):149 unary = abs if op == 'abs' else operator.neg150 return lambda values: _typed(unary(child(values)), dtype)151 if op == 'unit_convert':152 scale, offset = node['scale'], node['offset']153 _number_type(scale)154 _number_type(offset)155 return lambda values: _affine(child(values), scale, offset, dtype)156 raise ValueError(f'无效的数值指令 {op}')157158159class _Accumulator:160 """按原顺序累加已经求好的贡献，不解析公式或保存解码字段。"""161162 def __init__(self, spec):163 """读取结果名称和单位；新累计器没有观测值，不把空输入记为零。"""164 self.name, self.unit = spec['name'], spec['unit']165 self.matched = self.selected = 0166 self.total = 0.0167168 def add(self, contribution, keep):169 """提交一条已核验贡献；溢出时保留此前累计及计数，原异常传播。"""170 if keep:171 self.total = _float(self.total + contribution)172 self.selected += 1173 self.matched += 1174175 def result(self):176 """返回当前业务行；区分没有目标、空方向桶和贡献相消。"""177 reason = None if self.selected else 'no_selected_direction' if self.matched else 'no_target_frames'178 return dict(name=self.name, unit=self.unit, value=self.total if self.matched else None,179 status='complete' if self.matched else 'indeterminate', reason=reason)180181182class _Metric(_Accumulator):183 """当前文件的求值器和累计状态；先求公式和方向，再提交本条贡献。"""184185 def __init__(self, spec, formula):186 super().__init__(spec)187 self.evaluate = formula188 self.factor = _float(spec.get('output_scale', 1))189 self.direction = None190 if spec.get('direction') is not None:191 direction = spec['direction']192 if direction['bucket'] not in ('positive','negative'):193 raise ValueError('方向桶需要 positive 或 negative')194 self.direction = (direction['signal'], direction['bucket'] == 'positive')195196 def consume(self, decoded):197 """消费一条已匹配记录。除零、溢出或缺值直接使本文件失败。"""198 contribution, keep = self.contribution(decoded)199 self.add(contribution, keep)200201 def contribution(self, decoded):202 """只求当前帧贡献和方向选择，不修改累计状态。203204 参数: decoded为DBC物理数值字典。返回: (float64贡献, bool入选)。205 异常: 类型、单位换算、公式或方向值错误；未入桶也必须先求值。206 """207 contribution = _float(self.evaluate(decoded))208 if self.factor != 1:209 contribution = _affine(contribution, self.factor, 0, 'float64')210 keep = True211 if self.direction is not None:212 key, positive = self.direction213 value = decoded[key]214 _number_type(value)215 keep = value > 0 if positive else value < 0216 return contribution, keep217218def prepare_metric(spec):219 """创建一个计算项的求值/累计状态。参数为简单声明字典；拒绝冲突别名和未知指令。"""220 signals, parameters = spec['signals'], spec.get('parameters', {})221 if signals.keys() & parameters.keys():222 raise ValueError('参数与信号名称冲突')223 natural = _syntax(spec['expr'], signals, parameters)224 return _Metric(spec, _bind(spec.get('program', natural), signals))225226227def _prepare_program_metric(spec):228 """绑定内部已检查的指令，不重新解析原公式。229230 参数: spec来自公共 fast 编译器，或本会话已经检查的低层声明。231 返回: 每文件独立的求值/累计状态。异常: 指令、类型或数值声明错误。232 仅内部调用；普通字典入口仍通过 prepare_metric 检查文本，即使带有 program。233 """234 signals = spec['signals']235 if signals.keys() & spec.get('parameters', {}).keys():236 raise ValueError('参数与信号名称冲突')237 return _Metric(spec, _bind(spec['program'], signals))238239240def _check_program_spec(spec):241 """为低层会话检查一次文本及指令，并返回带指令的新声明。242243 参数: 原计算项字典，不修改它。返回: 浅复制的内部声明。244 异常: 与 prepare_metric 相同；非法文本不能借合法 program 绕过检查。245 """246 signals, parameters = spec['signals'], spec.get('parameters', {})247 if signals.keys() & parameters.keys():248 raise ValueError('参数与信号名称冲突')249 natural = _syntax(spec['expr'], signals, parameters)250 checked = dict(spec, program=spec.get('program', natural))251 _prepare_program_metric(checked)252 return checked253