chronaxis.fast.config
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""把一个CAN来源编译成fast会话，共用Chronaxis的单位、类型与来源位置规则。23会话保存完整输入序列、合并方式及资源额度，工作进程接收其中一个文件的声明。4"""5from dataclasses import asdict6from hashlib import sha2567from pathlib import Path89from ..configuration import ConfigurationError, load_configuration10from ..configuration.diagnostics import locate_error11from ..configuration.imports import expand_document_imports12from ..legacy.binding import dbc_candidates13from ..planning.checks import PlanValidationError, configuration_error14from ..planning.dbc_defaults import load_dbc, scoped_dbc_cache15from ..planning.plan import AnalysisPlan16from ..sources.asc import ASCSpec17from ..sources.can_spec import LegacyCANSpec, can_field_from_key18from .compute import encode_program, _prepare_program_metric192021def _unsupported(path, message):22 """构造可由已有配置诊断和 CLI 消费的错误。"""23 return PlanValidationError(path, message, 'unsupported_fast', hint='修改配置或使用通用 chronaxis run')242526def _capabilities(raw):27 """仅检查显式请求，避免把通用计划的默认设置误判为用户要求。"""28 errors = []29 for key in ('derived', 'lookups', 'rules', 'windows', 'sequences', 'calculations', 'presets',30 'aggregations', 'transforms', 'intervals', 'preset_bindings'):31 if raw.get(key):32 errors.append(_unsupported('$.' + key, 'fast 只执行当前帧数值公式'))33 for key in ('execution', 'diagnostics'):34 group = raw.get(key)35 if isinstance(group, dict):36 for field, value in group.items():37 if key == 'execution' and field in ('process_count', 'memory_budget_bytes', 'spool_quota_bytes'):38 continue39 if key == 'diagnostics' and value is False:40 continue41 errors.append(_unsupported(f'$.{key}.{field}', 'fast 不执行该资源或诊断策略'))42 outputs = raw.get('outputs')43 if isinstance(outputs, dict):44 for key in ('signal_csv', 'can_frames'):45 if outputs.get(key) is not None:46 errors.append(_unsupported('$.outputs.' + key, 'fast 只发布指标汇总'))47 if outputs.get('contributions', 'none') != 'none':48 errors.append(_unsupported('$.outputs.contributions', 'fast 不生成逐帧明细'))49 sources = raw.get('sources')50 if isinstance(sources, list):51 if len(sources) != 1:52 errors.append(_unsupported('$.sources', '当前 fast 任务须有一个来源'))53 for i, src in enumerate(sources):54 if not isinstance(src, dict):55 continue56 if 'legacy_profile' in src:57 errors.append(_unsupported(f'$.sources[{i}].legacy_profile', '显式锁定的 reader 与 fast 冲突'))58 reader = src.get('reader')59 if isinstance(reader, dict):60 for key, value in reader.items():61 if (key,value) not in (('input_check','auto'), ('decode_scope','requested'), ('decode_policy','strict')):62 errors.append(_unsupported(f'$.sources[{i}].reader.{key}', 'fast 仅支持按需读取与严格长度检查'))63 metrics = raw.get('message_calculations')64 if not metrics:65 errors.append(_unsupported('$.message_calculations', '至少需要一项逐消息累计'))66 elif isinstance(metrics, list):67 for i, item in enumerate(metrics):68 if isinstance(item, dict) and 'profile' in item:69 errors.append(_unsupported(f'$.message_calculations[{i}].profile', '显式锁定的解码器与 fast 冲突'))70 return errors717273def _source(plan):74 """展开同一来源的文件和时间解释，不打开日志。7576 参数: 静态AnalysisPlan。返回: 可序列化会话声明。异常: 不支持来源或解码策略。77 """78 source, = plan.sources79 cfg = source.config80 if source.source_type not in ('can_blf', 'can_asc', 'asc_or_blf'):81 raise _unsupported('$.sources[0]', 'fast只读取ASC/BLF')82 if cfg.decode_policy != 'strict':83 raise _unsupported('$.sources[0].reader', 'fast要求严格帧长')84 inputs = []85 if isinstance(cfg, LegacyCANSpec):86 inputs = [dict(path=str(path), offset_ns=cfg.offset_ns) for path in cfg.input_paths]87 else:88 for item in cfg.inputs:89 if isinstance(item, ASCSpec):90 inputs.append(dict(path=str(item.path), asc_options={k:v for k,v in asdict(item).items() if k != 'path'}))91 else:92 inputs.append(dict(path=str(item.path), offset_ns=item.offset_ns))93 if not inputs:94 raise _unsupported('$.sources[0].paths', 'fast需要至少一个输入文件')95 domain = plan.dataset.domain96 return dict(**inputs[0], inputs=inputs, dbc=str(cfg.dbc_path), channel=cfg.channel + 1,97 domain_ns=[domain.start_ns, domain.end_ns] if domain else None,98 table_format=plan.outputs.tables, file_merge=cfg.file_merge,99 process_count=plan.execution.process_count,100 memory_budget_bytes=plan.execution.memory_budget_bytes,101 spool_quota_bytes=plan.execution.spool_quota_bytes)102103104def _flag(selector, fields, name):105 """合并同一公式中每个信号和选择器的身份限制；显式矛盾不可合并。"""106 flags = {getattr(v,name) for v in (selector,*fields)} - {None}107 if len(flags) > 1:108 raise ValueError('同一帧公式的 ' + name + ' 互相冲突')109 return next(iter(flags), None)110111112def _metric(plan, item, database):113 """静态验证本次请求的信号及原单位，将已检查的表达式树编码为简单数值指令。"""114 required = set(item.expr.dependencies)115 if item.disjunction:116 required.add(item.disjunction.signal)117 signals, fields = {}, {}118 for name in sorted(required):119 definition = plan.bindings.resolve(name)120 if definition.dtype.value not in ('int64','uint64','float64'):121 raise ValueError(name + ' 需要数值类型')122 if definition.valid_range or definition.invalid_values:123 raise ValueError(name + ' 声明了 fast 尚不支持的有效范围或无效值')124 if definition.unit_status != 'verified':125 raise ValueError(name + ' 未验证单位或换算')126 field = can_field_from_key(definition.field)127 fields[name] = field128 signals[name] = dict(signal=field.signal, dtype=definition.dtype.value,129 scale=definition.scale, offset=definition.offset)130 if definition.input_dtype is not None:131 signals[name]['input_dtype'] = definition.input_dtype.value132 selector = item.update_message133 extended = _flag(selector, fields.values(), 'is_extended')134 fd = _flag(selector, fields.values(), 'is_fd')135 messages = dbc_candidates(database, selector.frame_id, [v.signal for v in fields.values()], extended, fd)136 if len(messages) != 1:137 raise ValueError('DBC 候选不唯一，请补充扩展帧或 FD 标志')138 message, = messages139 if message.is_container or message.is_multiplexed():140 raise ValueError('fast 首版不支持 container 或 multiplex 报文')141 for name, field in fields.items():142 unit = message.get_signal_by_name(field.signal).unit143 if unit and unit != plan.bindings.resolve(name).input_unit:144 raise ValueError(name + ' 的 DBC 单位与声明的输入单位不符')145 direction = item.disjunction146 spec = dict(name=item.calculation_id,frame_id=selector.frame_id,is_extended=message.is_extended_frame,147 is_fd=fd,expr=item.expr.text,unit=item.unit,signals=signals,148 parameters={p.name:p.value for p in item.parameters},149 direction=dict(signal=fields[direction.signal].signal,bucket=direction.bucket) if direction else None,150 output_scale=1,program=encode_program(item.expr.root))151 # 公共编译器已检查原式；只绑定已编码指令，避免第二套文本语法重新解释同一公式。152 _prepare_program_metric(spec)153 return spec, message154155156def _compile_plan(plan, with_database):157 """编译公共静态计划的单会话声明，保留全部分卷；仅加载本计划的DBC元数据。"""158 import cantools159 errors = []160 result = _source(plan)161 try:162 before = sha256(Path(result['dbc']).read_bytes()).hexdigest()163 database = load_dbc(result['dbc'])164 if sha256(Path(result['dbc']).read_bytes()).hexdigest() != before:165 raise ValueError('编译期间DBC内容变化')166 result['_dbc_digest'] = before167 except (OSError, ValueError, cantools.database.errors.Error) as exc:168 raise PlanValidationError('$.sources[0].dbc_by_channel', str(exc)) from exc169 metrics = []170 bindings = {}171 for i, item in enumerate(plan.message_calculations):172 try:173 metric, message = _metric(plan,item,database)174 # 同一实际帧只解码一次。不同公式若各自选中不同定义，不能等到 worker 才拒绝。175 fd = metric['is_fd']176 for actual_fd in ((False, True) if fd is None else (fd,)):177 key = (metric['frame_id'], metric['is_extended'], actual_fd)178 decoder = None if actual_fd and not message.is_fd else message179 if bindings.setdefault(key,decoder) is not decoder:180 raise ValueError(f'ID {key[0]:#x} 在同一实际帧型中对应多个 DBC 定义')181 metrics.append(metric)182 except (TypeError, ValueError, OverflowError) as exc:183 errors.append(_unsupported(f'$.message_calculations[{i}]', str(exc)))184 if errors:185 raise PlanValidationError.from_diagnostics(errors)186 result['metrics'] = metrics187 if with_database:188 result['_database'] = database189 return result190191192def compile_template_plan(plan, *, template_path=None, _with_database=False):193 """为 batch.template_jobs 展开的一个真实文件编译 fast 任务。194195 只使用原模板中显式声明的选项判断能力冲突，不把通用缺省值解释为用户要求。196 模板绑定错误带模板路径及当前文件说明。此步骤不执行分析或打开原始日志。197 """198 from ..planning.checks import json_value199 raw = json_value(plan.raw_config)200 try:201 errors = _capabilities(raw)202 if errors:203 raise PlanValidationError.from_diagnostics(errors)204 return _compile_plan(plan, _with_database)205 except PlanValidationError as exc:206 if template_path is not None:207 source = raw['sources'][0]208 input_path = source.get('path', source.get('paths', ''))209 for diagnostic in exc.diagnostics:210 diagnostic.source = str(template_path)211 diagnostic.hint = f'模板输入 {input_path}。' + (diagnostic.hint or '')212 diagnostic.refresh_text()213 exc.source = str(template_path)214 exc.refresh_text()215 raise216217218@scoped_dbc_cache()219def compile_config(path, *, _with_database=False):220 """读取配置和导入定义，验证DBC与fast子集。默认返回可序列化会话声明。221222 参数 path 是 JSON/YAML 文件路径；_with_database 仅给同进程立即执行时复用数据库。223 错误带配置路径。此入口只准备计算，不打开日志或创建输出目录。224 """225 target = Path(path).absolute()226 try:227 doc = expand_document_imports(load_configuration(target))228 except ConfigurationError as exc:229 raise configuration_error(exc) from exc230 try:231 if not isinstance(doc.data,dict):232 raise _unsupported('$', '需要单任务分析配置')233 errors = _capabilities(doc.data)234 if errors:235 raise PlanValidationError.from_diagnostics(errors)236 plan = AnalysisPlan.from_dict(doc.data,base_dir=target.parent)237 return _compile_plan(plan, _with_database)238 except PlanValidationError as exc:239 raise locate_error(exc,doc)240