chronaxis.planning.process_execution

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""静态检查单会话文件读取并行的支持范围，不打开日志或DBC。23进程数与文件归并分别检查，明确拒绝未接入的来源及嵌套并行。4只声明执行上限，不能凭校验成功证明文件内部有序或实际使用了多个进程。5"""67# 只由独立批量工作进程设置，不读取环境变量或根据CPU数量隐式改变用户配置。8IN_BATCH_WORKER = False91011def process_execution_errors(plan):12    """核有效计划的进程与归并声明。参数: plan。返回: 错误tuple。异常: 无输入IO。13    """14    count = plan.execution.process_count15    if type(count) is not int or count < 1:16        return ('execution.process_count需要大于等于1的整数，不接受bool',)17    errors = []18    merged = False19    for source in plan.sources:20        merge = getattr(source.config, 'file_merge', 'concatenate')21        if merge not in ('concatenate', 'time_order'):22            errors.append(source.name + '的file_merge需要concatenate或time_order')23        merged |= merge == 'time_order'24    if (count > 1 or merged) and (len(plan.sources) != 1 or plan.sources[0].source_type not in ('can_blf', 'can_asc', 'asc_or_blf')):25        errors.append('文件并行读取或时间归并当前需要一个ASC/BLF来源；其他来源或多个来源请使用process_count: 1及默认文件接续')26    if count > 1 or merged:27        if IN_BATCH_WORKER:28            errors.append('独立批量工作进程内不再建立文件读取进程池；请直接运行此连续分析计划')29        if any(getattr(source.config,'decode_scope','requested')=='all' for source in plan.sources):30            errors.append('并行读取及时间归并当前需要reader.decode_scope: requested')31    return tuple(errors)323334def source_execution_facts(plan):35    """解释文件组织与读取进程上限。参数: 有效计划。返回: 有限dict元组。异常: 无；不读输入。36    """37    from ..sources.can_spec import LegacyCANSpec38    from ..sources.text_can_spec import CANTextSpec39    result=[]40    if type(plan.execution.process_count) is not int or plan.execution.process_count<1:41        return ()42    for source in plan.sources:43        config=source.config44        if source.source_type not in ('can_blf','can_asc','asc_or_blf'):45            continue46        if not isinstance(config,(LegacyCANSpec,CANTextSpec)):47            continue48        if source.source_type=='can_blf' and type(config) is not LegacyCANSpec:49            continue50        if source.source_type!='can_blf' and type(config) is not CANTextSpec:51            continue52        files=config.input_paths if source.source_type=='can_blf' else config.inputs53        requested=plan.execution.process_count54        merge=config.file_merge55        # 接续只有多文件且有效上限大于1才建读取进程；时间归并沿独立工作槽。56        effective=min(requested,len(files))57        limit=effective if effective>1 or merge=='time_order' else 058        result.append(dict(source=source.name,file_count=len(files),file_merge=merge,59            read_process_limit=limit,continuous_session=True,60            temporary_merge_files=merge=='time_order'))61    return tuple(result)62