chronaxis.fast.worker

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""在本工作进程内执行事先验证的参数，只上传本文件的少量结果。"""2from hashlib import sha2563import os4from pathlib import Path5from time import perf_counter6from .run import run_prepared7from .._batch_worker import _diagnostics8from ..dbc_loading import load_dbc_file910def _run_file(packet, directory, cancel):11    """在已持有该文件的工作进程直接分析并发布。返回任务状态及可选的小型汇总行。"""12    start = perf_counter()13    spec = packet['spec']14    facts = dict(job_id=packet['job_id'],plan_path=packet['path'],input_identity=str(spec['path']))15    try:16        if cancel.is_set():17            raise InterruptedError('文件启动前收到取消请求')18        database = None19        if packet['total']:20            # 求和要求同一份解码声明。核对本进程实际加载的 DBC，变动则该成员失败。21            import cantools22            dbc = Path(spec['dbc'])23            before = _dbc_digest(dbc)24            database = load_dbc_file(dbc, cantools)25            if before != spec['_dbc_digest'] or _dbc_digest(dbc) != before:26                raise ValueError('求总量期间模板的 DBC 内容发生改变')27        directory.mkdir(exist_ok=False)28        value = run_prepared(spec,directory/'output',engine='base_hpc_v1',29                             _return_rows=packet['total'], cancel_event=cancel,database=database,30                             _programs_checked=True)31        manifest, rows = value if packet['total'] else (value, None)32        return dict(**facts,status='completed',output_dir=manifest.parent,manifest_path=manifest,33                              run_id=directory.name,elapsed_seconds=perf_counter()-start), rows34    except InterruptedError as exc:35        return dict(**facts,status='cancelled',error_type='InterruptedError',36                              error_message=str(exc),elapsed_seconds=perf_counter()-start), None37    except Exception as exc:38        return dict(**facts,status='failed',error_type=type(exc).__name__,39                              error_message=str(exc),elapsed_seconds=perf_counter()-start), None404142def _dbc_digest(path):43    """读取解码声明的摘要；只在请求跨文件求和时使用。"""44    with Path(path).open('rb') as source:45        return sha256(source.read()).hexdigest()464748def file_response(packet, directory, cancel):49    """仅传已冻结的结果字段；接收方使用与通用批量相同的结果模型校验。"""50    try:51        result, rows = _run_file(packet, directory, cancel)52    except BaseException as exc:53        result = dict(job_id=packet['job_id'],status='failed',plan_path=packet.get('path'),54                      error_type=type(exc).__name__,error_message=str(exc),error_chain=_diagnostics(exc))55        rows = None56    result['worker_pid'] = os.getpid()57    response = ('fast_result', result, False)58    return response + (rows,) if packet['total'] else response59