chronaxis.batch.models
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""保存独立批量任务及其完成、失败和取消事实。23输入是已构建计划或计划文件路径；输出保留任务、计划、会话和实际输入身份。4模型不读取来源、不建立进程，也不将多个任务的数值结果相加。清理失败时只保留5可显式重试的最高资源所有者，错误本身转成文本，避免长期保存异常栈和输入数组。6"""78from dataclasses import dataclass, field9from pathlib import Path10from typing import Protocol1112from ..core.validation import require_identifier13from ..planning.plan import AnalysisPlan141516class Closable(Protocol):17 """表示已经存在、由调用者负责关闭的资源，不创建占位执行对象。"""1819 def close(self) -> None:20 """关闭资源。参数: 无。返回: None。异常: 原关闭错误，可恢复后重试。"""21 ...222324class CleanupHandle:25 """只持有未关闭的最高所有者；关闭成功后撤销引用，重复关闭无副作用。"""2627 def __init__(self, owner: Closable):28 """接管真实恢复对象。参数: owner有close方法。返回: 新句柄。异常: 无。"""29 self._owner = owner3031 @property32 def closed(self) -> bool:33 """读取恢复是否完成。参数: 无。返回: bool。异常: 无。"""34 return self._owner is None3536 def close(self) -> None:37 """重试关闭，不在失败时丢所有者。参数: 无。返回: None。异常: 原关闭错误。"""38 if self._owner is not None:39 self._owner.close()40 self._owner = None414243@dataclass(frozen=True)44class BatchJob:45 """声明一个独立分析；同会话连续文件应包含在本任务的一个计划中。"""4647 job_id: str # 批量范围内唯一的简单标识，不用它拼接任意相对路径。48 plan: AnalysisPlan | Path # 冻结计划或文件路径；路径到真正启动时才加载。4950 def __post_init__(self):51 """检查声明而不读文件。参数: 无。返回: None。异常: 名称或计划类型错误。"""52 require_identifier(self.job_id, 'job_id')53 if not isinstance(self.plan, (AnalysisPlan, Path)):54 raise TypeError('BatchJob.plan需要AnalysisPlan或Path')555657@dataclass(frozen=True)58class BatchPlan:59 """保存批量配置；只关联独立计划路径，不能把一个会话拆成多个时间任务。"""6061 jobs: tuple[BatchJob, ...] # 声明顺序就是最终清单顺序。62 process_count: int = 1 # 工作进程上限，默认1；各任务仍有独立预算。63 fail_fast: bool = False # 默认继续其他独立任务。64 backend: str | None = None # 可选统一覆盖，不修改原计划文件。6566 def __post_init__(self):67 """检查冻结声明。参数: 无。返回: None。异常: 类型、范围或重复任务错误；无I/O。"""68 if type(self.jobs) is not tuple or any(not isinstance(j, BatchJob) for j in self.jobs):69 raise TypeError('jobs需要BatchJob tuple')70 if len({j.job_id.casefold() for j in self.jobs}) != len(self.jobs):71 raise ValueError('批量job_id不能重复或仅大小写不同')72 if type(self.process_count) is not int or self.process_count < 1:73 raise ValueError('process_count需要正原生int')74 if type(self.fail_fast) is not bool:75 raise TypeError('fail_fast需要bool')76 if self.backend is not None and (type(self.backend) is not str or self.backend not in ('reference', 'numpy', 'auto')):77 raise ValueError('backend需要auto、reference、numpy或null')7879 @classmethod80 def from_dict(cls, config, *, base_dir=None):81 """检查批量配置；jobs只关联路径，template读取模板并展开文件集合。8283 参数: config原生dict；base_dir为计划路径相对基准，默认当前目录。84 返回: 冻结BatchPlan。异常: PlanValidationError定位字段；不读取业务报文，模板可读取DBC元数据。85 """86 from ..planning.checks import object_at, array_at, text_at, model_at, PlanValidationError, number_at, choice_at87 from ..planning.diagnostic_collection import DiagnosticCollection88 from ..planning.checks import require_json_value89 require_json_value(config, '$')90 diagnostics = DiagnosticCollection(config, 'batch-v1.schema.json')91 names = set()92 if type(config) is dict and type(config.get('jobs')) is list:93 for index, job in enumerate(config['jobs']):94 path = f'$.jobs[{index}]'95 if type(job) is not dict or not diagnostics.usable(path):96 continue97 name = job.get('name', f'job_{index+1}').casefold()98 if name in names:99 diagnostics.add(PlanValidationError(path+'.name', '任务名称不能重复或仅大小写不同',100 'duplicate_id', hint='为每个独立任务指定唯一name'))101 names.add(name)102 # 批量自身结构一次报告；jobs子计划仍在工作进程启动时才加载。103 diagnostics.finish()104 if type(config) is dict and ('workers' in config or 'max_workers' in config):105 raise PlanValidationError('$.process_count', '请使用process_count设置分析工作进程数；workers/max_workers不再接受')106 raw = object_at(config, '$', ('batch_version',), ('jobs', 'template', 'inputs', 'source', 'process_count', 'fail_fast', 'backend'))107 choice_at(raw['batch_version'],(1,),'$.batch_version')108 base, jobs = Path.cwd() if base_dir is None else Path(base_dir).absolute(), []109 if 'template' in raw:110 if 'jobs' in raw or 'inputs' not in raw:111 raise PlanValidationError('$','template需要inputs且不能同时提供jobs')112 from .templates import template_jobs113 jobs=template_jobs(raw,base)114 elif 'jobs' not in raw or 'inputs' in raw or 'source' in raw:115 raise PlanValidationError('$','需要jobs，或template与inputs')116 names=set()117 for index, value in enumerate(array_at(raw.get('jobs', []), '$.jobs')):118 path = '$.jobs[' + str(index) + ']'119 item = object_at(value, path, ('plan',), ('name',))120 job_id = text_at(item.get('name',f'job_{index+1}'), path + '.name', identifier=True)121 if job_id.casefold() in names:122 raise PlanValidationError(path+'.name','任务名称不能重复或仅大小写不同','duplicate_id',123 hint='为每个独立任务指定唯一name')124 names.add(job_id.casefold())125 target = Path(text_at(item['plan'], path + '.plan'))126 if not target.is_absolute():127 target = base / target128 jobs.append(BatchJob(job_id, target.absolute()))129 return model_at(cls, '$', tuple(jobs),130 number_at(raw.get('process_count',1),'$.process_count',integer=True,minimum=1),131 choice_at(raw.get('fail_fast',False),(False,True),'$.fail_fast'),132 choice_at(raw.get('backend'),(None,'reference','numpy','auto'),'$.backend'))133134135@dataclass(frozen=True)136class BatchJobResult:137 """保存一个任务的最终状态；发布后关闭失败仍为failed，并保留已发布路径。"""138139 job_id: str # 原任务身份。140 status: str # completed、failed或cancelled。141 plan_path: str | None = None # 路径计划的原绝对位置；内存计划无路径。142 plan_identity: str | None = None # 配置身份，不冒充实际输入内容身份。143 dataset_id: str | None = None # 无法加载计划时三项会话字段均为空。144 entity_id: str | None = None145 session_id: str | None = None146 input_identity: str | None = None # 只有实际分析返回后才记录。147 run_id: str | None = None # 同计划重跑仍有独立运行身份。148 output_dir: Path | None = None # 实际发布成功的目录，不预填未生成输出。149 manifest_path: Path | None = None # 本次任务实际完成清单，不猜固定文件名。150 error_type: str | None = None # 原异常类型或未启动取消原因。151 error_message: str | None = None # 原错误说明，不用成功值覆盖失败。152 error_chain: tuple[str, ...] = () # cause链和清理notes，保留错误来源。153 elapsed_seconds: float = 0.0 # 单任务墙钟秒，含发布及关闭，不宣称CPU耗时。154 memory_budget_bytes: int | None = None # 该任务的新分析层预算；不是整进程RSS。155 worker_pid: int | None = None # 实际执行任务的工作进程PID；未启动项为空。156 recovery: CleanupHandle | None = field(default=None, repr=False, compare=False)157158 def __post_init__(self):159 """拒绝伪成功状态。参数: 无。返回: None。异常: 非法状态或成功缺少事实。"""160 if self.status not in ('completed', 'failed', 'cancelled'):161 raise ValueError('批量任务状态无效')162 if self.status == 'completed' and (self.output_dir is None or self.run_id is None):163 raise ValueError('completed需要实际发布目录及运行身份')164165 def to_dict(self) -> dict:166 """生成可保存的事实。参数: 无。返回: JSON对象。异常: 无；不序列化资源对象。"""167 names = ('job_id', 'status', 'plan_path', 'plan_identity', 'dataset_id', 'entity_id',168 'session_id', 'input_identity', 'run_id', 'error_type', 'error_message',169 'elapsed_seconds', 'memory_budget_bytes', 'worker_pid')170 result = {name: getattr(self, name) for name in names}171 result.update(output_dir=None if self.output_dir is None else str(self.output_dir),172 manifest_path=None if self.manifest_path is None else str(self.manifest_path),173 error_chain=list(self.error_chain),174 cleanup_pending=self.recovery is not None and not self.recovery.closed)175 return result176177178@dataclass(frozen=True)179class BatchReport:180 """按原任务顺序汇总结果；每个任务数值表仍在各自目录，不跨口径归约。"""181182 output_dir: Path # 可复用的输出根，任务子目录和报告按本批时间区分。183 jobs: tuple[BatchJobResult, ...] # 最终结果顺序不受工作进程完成先后影响。184 process_count: int # 本次配置的工作进程上限。185 fail_fast: bool # 是否在观察到失败后停止派发。186 cancel_requested: bool # 用户取消或KeyboardInterrupt是否曾被观察到。187 device_label: str # 调用方明确提供；缺省pending，不猜测机器。188 worker_process_count: int = 0 # 本次实际启动的工作进程数，空批次为0。189 infrastructure_error: str | None = None # 调度或汇总基础设施错误，与任务业务失败分开。190 report_path: Path | None = None # 成功发布后返回的实际报告路径。191 _publication: object = field(default=None, repr=False, compare=False) # 持有本批名称，关闭释放。192193 @property194 def status(self) -> str:195 """汇总失败优先状态。参数: 无。返回: failed/cancelled/completed。异常: 无。"""196 if self.infrastructure_error is not None or any(job.status == 'failed' for job in self.jobs):197 return 'failed'198 return 'cancelled' if any(job.status == 'cancelled' for job in self.jobs) else 'completed'199200 def to_dict(self) -> dict:201 """输出身份和计数。参数: 无。返回: JSON对象。异常: 无。不合并指标值。"""202 return dict(batch_version=1, status=self.status, output_dir=str(self.output_dir),203 process_count=self.process_count, executor='processes', worker_process_count=self.worker_process_count, fail_fast=self.fail_fast,204 cancel_requested=self.cancel_requested, device_label=self.device_label,205 infrastructure_error=self.infrastructure_error,206 budget_scope='per_task_new_analysis_layer',207 counts={s: sum(j.status == s for j in self.jobs)208 for s in ('completed', 'failed', 'cancelled')},209 jobs=[job.to_dict() for job in self.jobs])210211 def close(self) -> None:212 """重试所有未完清理。参数: 无。返回: None。异常: 首个关闭错误，其他项仍尝试。"""213 first = None214 if self._publication is not None:215 try:216 self._publication.close()217 except BaseException as error:218 first = error219 for job in self.jobs:220 if job.recovery is not None:221 try:222 job.recovery.close()223 except BaseException as error:224 # 一项失败不能阻止其他独立任务释放资源；原句柄仍由本报告保留。225 if first is None:226 first = error227 else:228 first.add_note('其他批量任务清理失败：' + str(error))229 if first is not None:230 raise first231