chronaxis.planning.backend_selection
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""依据整个计划的静态类别选择数值后端，不读取输入或试跑计算。23auto 只覆盖已测量的简单积分类别。其他组合沿用 NumPy；显式选择始终优先。4选择结果供校验、说明和运行共用，不保存逐条执行记录，也不在失败后重试。5"""67from __future__ import annotations89from dataclasses import dataclass, replace10from typing import TYPE_CHECKING1112if TYPE_CHECKING:13 from ..core.expression import StaticExpression14 from .plan import AnalysisPlan151617BACKENDS = ('reference', 'numpy', 'auto')18_SIMPLE = frozenset(('name', 'const', 'add', 'sub', 'mul', 'div', 'neg', 'abs',19 'unit_convert', 'to_float'))20_RAW = frozenset(('can_blf', 'can_asc', 'asc_or_blf', 'raw_can_csv'))212223@dataclass(frozen=True)24class BackendSelection:25 """保存一次静态选择：请求值、实际后端及可读原因，不代表已运行。"""2627 requested: str # 原配置或调用覆盖。28 selected: str # reference/numpy；非法 typed 声明为 unavailable。29 reason: str # 解释选择依据，不承诺具体提速比例。303132def simple_numeric_expression(expression: StaticExpression | None) -> bool:33 """检查首批实测的简单数值表达式。参数: 静态表达式。返回: bool。异常: 非法 typed 结构。3435 仅遍历配置树；短路、查表、宿主函数和时序运算留在通用类别。36 """37 if expression is None or not expression.dependencies:38 return False39 from ..core.types import SignalDType40 if expression.root.dtype not in (SignalDType.INT64, SignalDType.UINT64, SignalDType.FLOAT64):41 return False42 pending = [expression.root]43 while pending:44 node = pending.pop()45 if node.operator not in _SIMPLE:46 return False47 pending.extend(node.args)48 return True495051def select_backend(plan: AnalysisPlan) -> BackendSelection:52 """为整个计划选择一次后端。参数: AnalysisPlan。返回: BackendSelection。5354 异常: 非法 typed 结构的属性错误；未知后端作为 unavailable 交给静态诊断。55 不读取 DBC、日志或路径元数据，不修改原计划、身份或用户配置。56 """57 requested = plan.execution.backend58 if requested in ('reference', 'numpy'):59 return BackendSelection(requested, requested, '使用显式指定的数值后端')60 if requested != 'auto':61 return BackendSelection(requested, 'unavailable', '不支持的数值后端')62 conservative = BackendSelection('auto', 'numpy', '该计划不属于已测量的简单积分类别，沿用 NumPy')63 if (plan.extensions is not None or plan.transforms or plan.aggregations or plan.sequences64 or plan.rules or plan.windows or plan.message_calculations or not plan.calculations):65 return conservative66 choices = dict(plan.intervals.resolved)67 used = set()68 for calculation in plan.calculations:69 if (calculation.operation != 'integral' or calculation.scope != 'session'70 or calculation.method != 'zoh' or calculation.condition is not None71 or not simple_numeric_expression(calculation.signal)72 or calculation.calculation_id not in choices73 or choices[calculation.calculation_id].mode != 'actual'):74 return conservative75 used.update(calculation.signal.dependencies)76 definitions = tuple(item.definition for item in plan.signals if item.definition.name in used)77 if {item.name for item in definitions} != used:78 return conservative79 sources = {item.source for item in definitions}80 if len(sources) != 1:81 return conservative82 source = next((item for item in plan.sources if item.name in sources), None)83 if source is None:84 return conservative85 if source.source_type in _RAW and len(definitions) <= 2:86 from ..sources.can_spec import can_field_from_key87 fields = tuple(can_field_from_key(item.field) for item in definitions)88 if len({(f.channel, f.frame_id, f.is_extended, f.is_fd) for f in fields}) == 1:89 return BackendSelection('auto', 'reference', '同一报文定位的简单实际时间积分采用 reference')90 if source.source_type == 'signal_csv' and len(definitions) == 1:91 return BackendSelection('auto', 'reference', '单信号 CSV 实际时间积分采用 reference')92 return conservative939495def execution_plan(plan: AnalysisPlan) -> AnalysisPlan:96 """取得实际执行视图。参数: 原计划。返回: 原计划或仅后端变化的视图。9798 异常: 未支持后端为 ValueError；不修改 raw_config/identity，不重载来源。99 """100 selected = select_backend(plan)101 if selected.selected == 'unavailable':102 raise ValueError('未支持的执行backend：' + str(selected.requested))103 return plan if selected.selected == plan.execution.backend else replace(104 plan, execution=replace(plan.execution, backend=selected.selected))105