chronaxis.fast.run
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""准备和执行单会话同帧累计，仅在全部输入、累计和资源关闭成功后发布结果。"""23import csv4import json5from pathlib import Path6from tempfile import TemporaryDirectory78from ..results.publication import PublicationNames, output_root91011def run_prepared(spec: dict, output: str | Path, *, engine: str, database: object | None = None,12 cancel_event: object | None = None, _return_rows: bool = False,13 work_dir: str | Path | None = None, _programs_checked: bool = False) -> Path | tuple[Path, tuple]:14 """执行冻结会话并最后发布清单。1516 参数: spec编译声明，output结果目录，engine执行器标识，database可复用DBC，17 cancel_event可选取消，_return_rows控制独立batch同时取得未舍入行，work_dir暂存父目录。18 _programs_checked仅供紧接公共编译器的内部调用；默认False仍检查全部原始公式。19 返回: 清单Path；_return_rows=True时返回(Path, 行tuple)。异常: 执行/取消/发布错误。20 """21 table_format = spec.get('table_format','jsonl')22 if table_format not in ('jsonl','csv','readable_csv'):23 raise ValueError('fast仅支持readable_csv、csv或jsonl汇总')24 root = output_root(output)25 from .session import run_session26 execution = {}27 rows,warnings = run_session(spec,database,cancel_event=cancel_event,work_dir=work_dir,28 statistics=execution,_programs_checked=_programs_checked)29 if cancel_event is not None and cancel_event.is_set():30 raise InterruptedError('发布结果前取消')31 table = 'results.jsonl' if table_format == 'jsonl' else 'results.csv'32 manifest = 'run_manifest.json'33 names = PublicationNames(root, (table,manifest))34 try:35 with TemporaryDirectory(dir=root,prefix='.fast-') as private:36 temp = Path(private)37 with (temp/table).open('x',encoding='utf-8',newline='') as out:38 if table_format == 'readable_csv':39 writer = csv.writer(out)40 writer.writerow(['计算项','结果','单位','状态','说明'])41 reasons = {'no_target_frames':'没有目标报文','no_target_in_domain':'分析范围内没有目标报文',42 'empty_file':'文件无观测数据','no_selected_direction':'有效目标帧均未进入该方向桶'}43 for row in rows:44 writer.writerow([row['name'],row['value'],row['unit'],45 '完整' if row['status']=='complete' else '无法确定',46 reasons.get(row.get('reason'),'')])47 elif table_format == 'csv':48 writer = csv.DictWriter(out,fieldnames=['name','value','unit','status','reason'])49 writer.writeheader()50 writer.writerows(rows)51 else:52 for row in rows:53 out.write(json.dumps(row,ensure_ascii=False,allow_nan=False)+'\n')54 record = dict(engine=engine,status='completed',inputs=[item['path'] for item in spec.get('inputs', [dict(path=spec['path'])])],55 execution=execution,warnings=warnings,56 outputs=[{'name':names.names[table]}])57 (temp/manifest).write_text(json.dumps(record,ensure_ascii=False,indent=2),encoding='utf8')58 names.publish(temp,manifest)59 path = root/names.names[manifest]60 return (path,tuple(rows)) if _return_rows else path61 finally:62 names.close()636465def run(config: str | Path, output: str | Path, *, process_count: int | None = None,66 work_dir: str | Path | None = None, cancel_event: object | None = None) -> Path:67 """运行一份fast单会话配置，所有分卷成功后发布汇总。6869 参数: config为YAML/JSON配置路径；output为结果目录；process_count为可选正整数70 工作进程上限覆盖，None沿用配置默认1；work_dir为可选暂存父目录；cancel_event71 为可选带is_set方法的取消事件。覆盖不修改配置，1在当前进程执行。72 返回: pathlib.Path，本次成功清单。异常: 配置/文件/资源/数值错误及InterruptedError。73 """74 from .config import compile_config75 from .session import effective_spec76 spec = effective_spec(compile_config(config,_with_database=True),process_count)77 return run_prepared(spec,output,engine='base_hpc_v1',database=spec.pop('_database'),78 work_dir=work_dir,cancel_event=cancel_event,_programs_checked=True)79