chronaxis.analysis_cli
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""把公共AnalysisRunner的检查、解释和实际分析接到CLI，提供文本和JSON报告。23输入为解析后的命令参数，load_analysis_plan只读取指定JSON/YAML并按配置目录解释来源路径。4validate和explain直接调用同一AnalysisRunner接口，不读取CSV、DBC或旧日志，不建立5运行目录。序列化逐项选择公开报告字段，不遍历任意对象或另做一套配置校验。6报告中的预算和有效期是声明，execution_ready不证明实际输入已核实。7run先检查输出根与独占工作路径，再交给AnalysisRunner运行，在同一结果上write并关闭。8只有真实关闭成功才打印成功摘要；失败保留原原因链和现存产物，不假报回滚。910本模块仅在实际计划命令时加载；CLI帮助与参数错误仍只需要标准库。11"""1213import json14from pathlib import Path15import sys1617# 这些接口只有计划命令实际执行时才加载；取得AnalysisRunner类不会创建运行会话。18from . import AnalysisRunner, load_analysis_plan19from .planning.checks import PlanValidationError20from .results.errors import AnalysisError212223def validation_json(report) -> dict:24 """保留公共校验报告的全部固定字段和未验证警告。2526 参数: report为AnalysisRunner.validate返回的ValidationReport。27 返回: 原生JSON字典；元组和只读映射转换为列表和dict，不改变字段含义。28 异常: 接口不符的编程错误原样传播，不靠反射补缺失属性。29 """30 # 显式白名单避免漏掉能力限制，也避免把内部状态或任意元数据输出到控制台。31 return dict(valid=report.valid, errors=list(report.errors), warnings=list(report.warnings),32 capability_status=dict(report.capability_status), execution_ready=report.execution_ready)333435def explanation_json(report) -> dict:36 """将依赖、节点后端、历史需求和预算声明完整交给CLI使用者。3738 参数: report为AnalysisRunner.explain返回的ExplainReport。39 返回: 固定JSON字典，节点保持原顺序，plan_identity保持原配置身份。40 异常: 接口不符的编程错误原样传播；不推算实测峰值或来源观测范围。41 """42 # 各节点只含原公开报告字段；state_driver与数值后端沿原报告区别显示。43 nodes = [dict(node_id=node.node_id, node_type=node.node_type, dependencies=list(node.dependencies),44 unit=node.unit, backend=node.backend) for node in report.nodes]45 requirements = report.input_requirements46 scopes = ([] if requirements is None else [dict(source=item.source, input_check=item.input_check,47 time_scope=item.time_scope, required_aliases=list(item.required_aliases),48 decode_frame_ids=list(item.decode_frame_ids), reasons=list(item.reasons)) for item in requirements.sources])49 if report.intervals is not None and requirements is not None:50 for item,scope in zip(requirements.sources,scopes):51 scope['active']=item.active52 output = dict(source_execution=[dict(item) for item in report.source_execution],53 input_requirements=scopes, validation=validation_json(report.validation), plan_identity=report.plan_identity,54 source_names=list(report.source_names), nodes=nodes,55 history_ns=dict(report.history_ns), budget=dict(report.budget),56 requested_backend=report.requested_backend, selected_backend=report.selected_backend,57 backend_reason=report.backend_reason)58 # 解释已解析的间隔声明，不推测DBC周期或输出第二时间轴。59 if report.intervals is not None:60 choices = {}61 for name, choice in report.intervals.resolved:62 choices[name] = dict(mode=choice.mode, step_ns=choice.step_ns, driver=choice.driver)63 output['intervals'] = choices64 if report.diagnostics is not None:65 output['diagnostics'] = dict(trace=report.diagnostics.trace,66 integrity=report.diagnostics.integrity, retain_work=report.diagnostics.retain_work)67 return output686970def run_static(arguments) -> int:71 """只加载配置并输出共享AnalysisRunner报告，不触碰来源或创建分析产物。7273 参数: arguments含command=validate或explain、plan路径和可选backend覆盖。74 返回: 0表示静态有效且当前有执行路径，2表示报告中存在能力错误。75 异常: 配置加载、依赖、JSON输出或取消错误向主CLI传播，由统一边界处理。76 副作用: 只读取计划JSON、写stdout；来源路径按load_analysis_plan解释，不按CLI另猜。77 """78 # 只构建静态计划和无状态AnalysisRunner；这些步骤不验证来源文件是否存在或可读。79 plan = load_analysis_plan(arguments.plan)80 runner = AnalysisRunner(backend=arguments.backend)81 # 两种命令使用相同有效计划，覆盖后端不会伪造新的配置或输入身份。82 if arguments.command == 'validate':83 validation = runner.validate(plan, process_count=arguments.process_count)84 payload = validation_json(validation)85 else:86 report = runner.explain(plan, process_count=arguments.process_count)87 validation = report.validation88 payload = explanation_json(report)89 # 人读输出复用同一计划及报告；机器格式保持明确字段，不重新解析输入。90 if arguments.format=='json':91 print(json.dumps(payload, ensure_ascii=False, allow_nan=False))92 else:93 from .configuration.explanation import validation_text, explanation_text94 text=(validation_text(plan,validation) if arguments.command=='validate'95 else explanation_text(plan,report,backend_override=arguments.backend))96 print(text)97 return 0 if validation.valid and validation.execution_ready and not validation.errors else 29899100def _new_output(arguments) -> Path:101 """在扫描来源前检查可复用输出根和冲突工作路径。102103 参数: arguments包含output和可选work_dir，由argparse限定为路径文本。104 返回: 绝对输出Path；不创建目录，工作目录实际分配仍归AnalysisRunner。105 异常: FileExistsError或原lstat错误为文件失败；PlanValidationError为路径冲突。106 """107 output = Path(arguments.output).absolute()108 # 输出根可复用，工作目录仍须独占；两者都保留原来的存在性检查。109 candidates = (output,) if arguments.work_dir is None else (output, Path(arguments.work_dir).absolute())110 if arguments.work_dir is not None:111 work = Path(arguments.work_dir).absolute().resolve()112 destination = output.resolve()113 # AnalysisRunner会先建立work；若它位于output内部，随后发布必因output已存在而失败。114 if work == destination or destination in work.parents:115 raise PlanValidationError('$.cli.work_dir', '工作目录不能等于输出目录或位于输出目录内部')116 for path in candidates:117 try:118 # 工作路径必须新建；输出允许已有目录，悬空链接不允许。119 # 最终输出安装的竞态仍由原write检查，运行会话也保留原工作目录检查。120 path.lstat()121 except FileNotFoundError:122 continue123 if path == output and path.is_dir() and not path.is_symlink():124 continue125 raise FileExistsError('工作目录须新建，输出目标须为目录：' + str(path))126 return output127128129def _cause_chain(error) -> tuple:130 """取得有限且不重复的真实cause链，供诊断和致命异常判断共用。131132 参数: error为实际异常。返回: 最多16个原异常对象，保持外层到根因顺序。133 异常: 无主动异常；不访问任意业务对象，不解析日志或重建错误。134 """135 chain, seen = [], set()136 current = error137 # 防御外部异常的cause循环；不因诊断无限遍历而阻塞清理或退出。138 while current is not None and id(current) not in seen and len(chain) < 16:139 seen.add(id(current))140 chain.append(current)141 current = current.__cause__142 return tuple(chain)143144145def run_analysis(arguments) -> int:146 """复用真实分析和发布，成功关闭所有结果资源后才报告completed。147148 参数: arguments含plan、output、可选backend及work_dir；输出目录可复用。149 返回: 成功0；已知分析/发布/关闭失败1，并说明输出可能已经发布。150 异常: 配置/路径错误交主CLI分类；取消原样传播；MemoryError及已知编程错误151 的直接或AnalysisError原因链继续抛出，不伪装成普通业务错误。152 副作用: AnalysisRunner建立专用暂存并读来源，write在输出根新增本次文件；不删除已有产物。153 """154 plan = load_analysis_plan(arguments.plan)155 output = _new_output(arguments)156 runner = AnalysisRunner(backend=arguments.backend)157 manifest = None158 try:159 # 结果拥有其真实仓和资源；write不重新运行来源，退出上下文完成实际close。160 with runner.run(plan, work_dir=arguments.work_dir, process_count=arguments.process_count) as result:161 manifest = result.write(output)162 summary = dict(status=manifest.status, run_id=manifest.run_id,163 plan_identity=result.metadata.plan_hash, backend=result.metadata.backend,164 output=str(manifest.output_dir), manifest=str(manifest.manifest_path),165 row_counts=dict(manifest.row_counts), device_label=result.metadata.device_label)166 if result.metadata.analysis_outcome is not None:167 summary['analysis_outcome']=result.metadata.analysis_outcome.to_dict()168 if result.table_format == 'readable_csv':169 from .results.readable_format import outcome_messages170 summary['result_files'] = [path.name for path in manifest.output_files]171 summary['result_messages'] = outcome_messages(result.metadata.analysis_outcome)172 except (AnalysisError, OSError) as error:173 chain = _cause_chain(error)174 # AnalysisRunner可能包装构造内存/编程错误；保留该公开异常及原cause，不归类普通失败。175 if any(isinstance(item, (MemoryError, TypeError, AttributeError, AssertionError, NameError)) for item in chain):176 raise177 causes = ' <- '.join(type(item).__name__ + ': ' + str(item).replace('\n', ' ') for item in chain)178 completed = manifest or next((getattr(item, 'completed_manifest') for item in chain179 if getattr(item, 'completed_manifest', None) is not None), None)180 if completed is not None:181 print('结果已发布，清单：' + str(completed.manifest_path) + '；收尾失败：' + causes, file=sys.stderr)182 return 1183 # 目录可能已安装后才发生close错误，非零退出不证明输出被撤销。184 print('分析或发布失败（输出可能已发布，请核对目标目录 ' + str(output) + '）：' + causes,185 file=sys.stderr)186 return 1187 # 本行在with真实退出之后，不能提前把analysis_complete当成发布和关闭成功。188 print(json.dumps(summary, ensure_ascii=False, allow_nan=False))189 return 0190