chronaxis.planning.plan
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""把分析配置转换成经过静态检查的分析计划。23使用入口：4 AnalysisPlan.from_dict(config, base_dir=...)接收一个配置字典。5 load_analysis_plan(path)读取一个JSON/YAML配置文件，再调用from_dict检查内容。6输出都是AnalysisPlan，其中保存来源、信号、表达式、规则、窗口和指标，7以及计划采用的计算方式、资源预算和输出格式。89处理顺序是：检查字段与版本 → 复制配置 → 检查来源和信号声明10→ 检查表达式及相互引用 → 保存计划。静态检查是根据配置声明判断11是否合法，不读取待分析的日志或CSV；省略CAN信号元数据时按需读DBC，不执行计算。1213from_dict的相对路径以base_dir为基准，省略时使用当前工作目录；14load_analysis_plan以各定义的声明文件所在目录为基准，一层imports先展开。路径只转换为绝对定位，不读取目标。15配置字典和列表会复制并冻结，调用者之后修改原配置不会改变计划快照。1617当前计划支持的来源由parse_sources检查；预设先校验角色并合入公共节点；尚未接入的旧profile、18旧完整积分兼容和未接入原始输出格式会被拒绝。message_calculations仅显式19current_message-v1逐消息计算；原帧和重建帧统一由can_frames声明。20execution中的预算与outputs中的格式21只是经过检查的声明；实际执行由AnalysisRunner落实，文件发布由AnalysisResult.write负责。22配置通过不表示本次来源已读取、实际结果或输出已经验证。23identity用于区分配置和来源定位声明，不证明磁盘上的输入内容已经核实。2425本模块依赖核心声明模型和planning中的字段检查、来源解析及节点解析。26from_dict不读分析数据，按需读取DBC元数据；load_analysis_plan还读取配置及一层公共定义并及时关闭句柄。27"""2829# 创建不可变模型并计算声明哈希；语法解析交共享配置入口。30from dataclasses import dataclass31from hashlib import sha25632from pathlib import Path33from collections.abc import Mapping34# 核心模型提供本计划的名称、来源关系和时间范围检查。35from ..core.declarations import DeclarationBindings, DeclaredSource36from ..core.expression import DerivedDefinition, StaticExpression37from .config_groups import public_errors38from .diagnostic_collection import diagnose_plan, current_collection, usable_node, collect_node_error39from ..core.models import Interval40from ..core.types import TimeType41# 固定CAN配置和profile标识只含静态模型，不导入reader、writer或旧依赖。42from ..sources.can_spec import LegacyCANSpec43from ..legacy.records import PROFILE_ID44# 字段、来源和计算节点分别交给对应解析模块。45from ..configuration import load_configuration, ConfigurationError46from ..configuration.diagnostics import locate_error47from .checks import (PlanValidationError, configuration_error, array_at, canonical, choice_at, coordinate_at, json_value,48 metadata_at, model_at, number_at, object_at, require_json_value, text_at, timezone_at)49from .inputs import parse_sources, parse_signals50from .aggregations import AggregationPlan, parse_aggregations51from .transforms import TemporalTransformPlan, build_transforms52from .lookups import parse_lookups53from .models import DatasetPlan, DerivedPlan, ExecutionPlan, MessageCalculationPlan, CalculationPlan, OutputPlan, RulePlan, SignalPlan, SourcePlan, WindowPlan54from .message_calculations import parse_message_calculations55from .legacy_execution import require_can_decode_scope56from .nodes import expression_at, parse_calculations, parse_rules, parse_windows57from .preset_models import PresetSpec, RoleBinding58from .sequences import SequencePlan, parse_sequences59from .intervals import IntervalPlan, DiagnosticPlan606162@dataclass(frozen=True)63class AnalysisPlan:64 """保存一个会话的分析配置，以及静态检查得到的对象。6566 通过from_dict或load_analysis_plan构建，依次保存输入来源、信号、派生计算、67 条件规则、窗口与指标。实例字段不能重新赋值；原配置及自由元数据68 保存为独立的只读快照，不保留调用者可变字典或列表的引用。6970 bindings只描述“哪个信号来自哪里”，derived_expressions是已检查的71 表达式结构；二者不表示真实数据已经读取或能够直接运行。72 execution与outputs保存执行及输出声明，创建本类不会启动计算或发布文件。7374 identity由配置和来源定位声明计算，用于区分计划；不能据此断言CSV内容、75 实际列、内存使用或分析结果已核实。各次构建的绑定和声明相互独立。76 """7778 config_version: int # 当前固定为4；拒绝旧配置，实际能力由已接线节点约束。79 dataset: DatasetPlan # 分析对象、会话和请求时间范围。80 sources: tuple[SourcePlan, ...] # 已解析的来源配置，保持用户声明顺序。81 signals: tuple[SignalPlan, ...] # 原字段与信号名称、类型、单位及有效期声明。82 derived: tuple[DerivedPlan, ...] # 由已有信号计算的新信号，例如功率及其元数据。83 rules: tuple[RulePlan, ...] # 已经检查的布尔条件和时序规则。84 windows: tuple[WindowPlan, ...] # 已检查的活动或时间点窗口，以及所引用的规则。85 calculations: tuple[CalculationPlan, ...] # 统计指标及其范围、计算口径、单位和缺失数据策略。86 presets: tuple[PresetSpec, ...] # 原预设v1声明，冻结并保留生成参数。87 preset_bindings: tuple[tuple[RoleBinding, ...], ...] # 本次解析角色的原定义与契约，不继承旧来源。88 execution: ExecutionPlan # 计算方式、分批大小和资源预算声明，尚未实际执行。89 outputs: OutputPlan # 结果表及清单的格式声明，尚未写出文件。90 bindings: DeclarationBindings # 名称与来源之间的声明关系，不含实际读取的信号值。91 derived_expressions: tuple[StaticExpression, ...] # 全部派生表达式的检查结果，已拒绝循环依赖。92 metadata: Mapping # 调用者附加的只读元数据，不改变计算规则。93 raw_config: Mapping # 原配置的独立只读副本。94 base_dir: Path # 相对来源路径的解释基准，保存为绝对路径。95 identity: str # 配置及来源定位声明的SHA-256哈希，不是文件内容哈希。96 message_calculations: tuple[MessageCalculationPlan, ...] = () # 显式当前消息计算；旧位置参数及缺省行为保留。97 sequences: tuple[SequencePlan, ...] = () # 可选点事件FIFO配对；旧计划不增加结果表。98 intervals: 'IntervalPlan | None' = None # 当前记录流的有限间隔声明。99 diagnostics: 'DiagnosticPlan | None' = None # 分别声明追溯、完整性检查与现场留存。100 aggregations: tuple[AggregationPlan, ...] = () # 仅配置后消费必要结果及总体证据。101 transforms: tuple[TemporalTransformPlan, ...] = () # 依赖在前的离散分析序列。102103 def __post_init__(self):104 """拒绝直接构造或replace得到的旧版本计划，不只检查字典入口。"""105 from ..core.unit_conversion import require_config_version106 require_config_version(self.config_version)107108 @property109 def extensions(self):110 """读取计划使用的宿主注册。参数: 无。返回: ExtensionRegistry或None。异常: 无。"""111 return self.bindings.extensions112113 @property114 def lookups(self):115 """读取本配置声明的标定表。参数: 无。返回: LookupTablePlan 元组。异常: 无。"""116 return self.bindings.lookups117118 @classmethod119 @public_errors120 @diagnose_plan121 def from_dict(cls, config: dict, *, base_dir: Path | str | None = None, extensions=None) -> 'AnalysisPlan':122 """复制配置字典，按唯一支持的版本4检查静态要求；执行能力由所选后端决定。123124 参数:125 extensions: 仅宿主 Python 可提供的不可变 ExtensionRegistry。省略时只识别内置操作。126 config (dict): 使用Python原生字典、列表和标量的分析声明；可由YAML加载取得。127 必填config_version、dataset、sources、signals；derived、rules、windows、128 sequences、calculations、message_calculations、presets、execution、outputs、129 intervals、diagnostics和metadata按需提供。版本4可用lookups声明标定表。130 字段、名称、表达式和引用都按对应规则检查。131 base_dir (Path | str | None): 相对来源路径的解释基准；默认None，132 使用当前工作目录。传入Path或文本时转换为绝对路径。133 返回:134 AnalysisPlan: 保存经过静态检查的来源、信号及计算节点。135 raw_config和metadata是独立的只读快照；创建计划不读取分析数据。136 异常:137 PlanValidationError: 字段或版本不合法，名称冲突、表达式或引用错误，138 输入模型不合规，或配置请求尚未接入的能力；错误保留配置位置。139 OSError: 获取当前工作目录或转换路径时发生系统错误，可原样传播。140 处理流程:141 1. 检查顶层字段、复制JSON结构，并核对当前配置结构版本。142 2. 解析来源、会话范围和信号声明，建立本计划的名称与来源关系。143 3. 检查全部派生表达式、规则、窗口、指标及执行和输出选项。144 4. 计算声明哈希并返回计划；此阶段不打开来源文件或执行分析。145 """146 # 先检查顶层必填项和允许项；附加信息只能放在metadata中。147 if type(config) is dict and 'imports' in config:148 raise PlanValidationError('$.imports','from_dict不读取公共定义文件','unresolved_imports',149 hint='使用load_analysis_plan加载配置文件，或传入config resolve展开后的字典')150 collection = current_collection()151 if type(config) is not dict or any(key not in config for key in ('config_version','dataset','sources','signals')):152 collection.finish()153 root = config154 # 复制并冻结配置，拒绝非JSON值、非有限数和非文本键。155 # 先保留原JSON类型及循环诊断，再冻结，避免tuple被隐式转为合法数组。156 require_json_value(root, '$')157 frozen = metadata_at(root, '$')158 # 用副本继续解析，避免调用者之后修改原配置影响本次检查。159 root = json_value(frozen)160 from .config_groups import normalize_configuration161 with collection.stage('$', check=False) as ready:162 root = normalize_configuration(root)163 # 当前唯一结构采用记录间隔和按需输出默认值。164 version = None165 with collection.stage('$.config_version') as ready:166 if ready:167 version = choice_at(root['config_version'], (4,), '$.config_version')168 lookups = ()169 if 'lookups' in root:170 with collection.stage('$.lookups', requires=version is not None) as ready:171 if ready:172 lookups = parse_lookups(root['lookups'], version)173 # 只接受Path或文本路径；这里不读取目标或解析符号链接。174 if base_dir is not None and not isinstance(base_dir, (Path, str)):175 # 拒绝其他对象，避免配置触发自定义路径转换代码。176 raise PlanValidationError('$', 'base_dir必须为Path或文本')177 # 省略base_dir时使用当前工作目录，否则将基准转为绝对路径。178 base = Path.cwd() if base_dir is None else Path(base_dir).absolute()179 # 解析来源配置，保存定位和格式声明；此时仍不打开CSV。180 sources = ()181 with collection.stage('$.sources', requires=True, check=False) as ready:182 if ready:183 sources = parse_sources(root['sources'], base)184 # 以首个来源声明的时钟作为分析基准，多源关系在依赖检查时核对。185 timebase = sources[0].config.timebase if sources else None186 # dataset只接受规定的单对象、单会话和时间范围字段。187 dataset = None188 with collection.stage('$.dataset', requires=bool(sources), check=True) as ready:189 if ready:190 raw_dataset = object_at(root['dataset'], '$.dataset', ('name',), ('entity', 'session', 'domain', 'metadata'))191 # 检查数据集名称，避免把路径或表达式当作标识。192 dataset_id = text_at(raw_dataset['name'], '$.dataset.name', identifier=True)193 # 普通单任务沿用数据集名；对象与会话仍由本次计划实例隔离，不共享状态。194 entity = text_at(raw_dataset.get('entity', dataset_id), '$.dataset.entity', identifier=True)195 session = text_at(raw_dataset.get('session', dataset_id), '$.dataset.session', identifier=True)196 # 省略domain是真正的无时间筛选声明，不制造假端点或实际观测范围。197 domain_explicit = 'domain' in raw_dataset198 domain = None199 if domain_explicit:200 # 只有用户显式请求时间域才解析坐标；显式null仍是配置错误。201 raw_domain = object_at(raw_dataset['domain'], '$.dataset.domain', ('start', 'end'))202 tz = timezone_at(sources[0].timezone_text, '$.dataset.domain.timezone')203 start = coordinate_at(raw_domain['start'], '$.dataset.domain.start', absolute=timebase.time_type is TimeType.ABSOLUTE, tz=tz)204 end = coordinate_at(raw_domain['end'], '$.dataset.domain.end', absolute=timebase.time_type is TimeType.ABSOLUTE, tz=tz)205 domain = model_at(Interval, '$.dataset.domain', start, end)206 # 保存对象、会话、范围和只读元数据，供后续引用。207 dataset = DatasetPlan(dataset_id, entity, session, domain, timebase,208 metadata_at(raw_dataset.get('metadata', {}), '$.dataset.metadata'), domain_explicit)209 # 检查每个信号的原字段、类型、单位转换、有效期和来源定位。210 signals = ()211 with collection.stage('$.signals', requires=bool(sources) and version is not None, check=False) as ready:212 if ready:213 from .dbc_defaults import requested_enum_codes214 signals = parse_signals(root['signals'], sources, config_version=version,215 enum_codes=requested_enum_codes(root))216 # 延迟导入避免公开预设直接导入时产生模块循环；只对局部副本展开，无来源I/O。217 from ..presets import expand_presets218 presets, preset_bindings = (), ()219 with collection.stage('$.presets', requires=not root.get('presets') or collection.usable('$.signals')) as ready:220 if ready:221 presets, preset_bindings = expand_presets(root, tuple(item.definition for item in signals))222 # 派生项按名称组织；类型由表达式推导，不单独指定有效期。223 raw_derived = root.get('derived', {})224 # 派生项必须是名称到配置的字典，不能由数组位置猜名称。225 if type(raw_derived) is not dict:226 raw_derived = {}227 # 保持用户声明顺序，收集已经检查的派生项。228 derived = []229 # 逐项检查字段与元数据；收集完成后再检查整个表达式依赖图。230 for name, item in raw_derived.items():231 # 记录当前派生项的配置位置，供错误信息定位。232 path = f'$.derived.{name}'233 # 先检查名称是否合法；表达式别名不能包含属性访问语法。234 if not usable_node(path, {'name': name}):235 continue236 try:237 text_at(name, path, identifier=True)238 # 只接受规定的表达式、单位和可选语义字段，拒绝未知字段。239 item = object_at(item, path, ('expr',), ('unit', 'semantic_id', 'semantic_contract_version', 'sign_convention', 'measurement_location', 'metadata'))240 # 建立派生声明，由核心检查表达式文本和单位是否有效。241 unit = text_at(item['unit'], path + '.unit') if 'unit' in item else None242 definition = model_at(DerivedDefinition, path, name, item['expr'], unit)243 # 保存可选语义信息；名称相同不能证明测量含义一致。244 annotations = tuple(text_at(item[key], path + '.' + key) if key in item else None for key in ('semantic_id', 'semantic_contract_version', 'sign_convention', 'measurement_location'))245 # 保存只读元数据；表达式不能通过对象属性访问这些信息。246 derived.append(DerivedPlan(definition, *annotations, metadata_at(item.get('metadata', {}), path + '.metadata')))247 except PlanValidationError as error:248 collect_node_error(error, path, {'name': name})249 # 将来源名称、时钟、定位声明和偏移组合为检查用对象。250 declared_sources = tuple(DeclaredSource(item.name, item.config.timebase, item.input_identity, item.config.offset_ns) for item in sources)251 # 在当前会话内统一检查来源、信号和派生名称，建立声明关系。252 bindings = None253 with collection.stage('$.signals/derived', requires=dataset is not None and version is not None, check=True) as ready:254 if ready:255 bindings = model_at(DeclarationBindings, '$.signals/derived', dataset.session_id, timebase, dataset.domain, declared_sources,256 tuple(item.definition for item in signals), tuple(item.definition.name for item in derived),257 config_version=version, lookups=lookups, extensions=extensions)258 # 检查所有派生定义，包括未被引用的项，拒绝未知名称或循环依赖。259 definitions = tuple(item.definition for item in derived)260 transforms = ()261 if 'transforms' in root:262 with collection.stage('$.transforms', requires=bindings is not None) as ready:263 if ready:264 bindings, transforms = build_transforms(root['transforms'],bindings,definitions,265 timezone=timezone_at(sources[0].timezone_text,'$.transforms.origin'))266 # 保存已经检查的表达式结构，不把配置文本当作Python代码执行。267 expressions_list = []268 collection.checking_derived = True269 for item in definitions:270 previous_count = len(expressions_list)271 with collection.stage(f'$.derived.{item.name}', requires=bindings is not None) as ready:272 if ready:273 expressions_list.append(expression_at(item.name, f'$.derived.{item.name}.expr', bindings, definitions, unit=item.unit))274 if len(expressions_list) == previous_count:275 collection.invalid_names.add(item.name)276 expressions = tuple(expressions_list)277 collection.checking_derived = False278 if not collection.errors:279 # 运行侧保留完整绑定身份；逐项隔离检查只用于定位错误，不能流入执行器。280 expressions = tuple(expression_at(item.name, f'$.derived.{item.name}.expr', bindings, definitions, unit=item.unit) for item in definitions)281 # 坏派生项已单列；无关规则不因编译器扫描整个派生集合而重复收到同一个错误。282 definitions = tuple(item for item in definitions if item.name not in collection.invalid_names)283 # 逐项检查执行选项及默认值；资源预算由实际执行器落实。284 execution = None285 with collection.stage('$.execution', requires=True, check=True) as ready:286 if ready:287 raw_execution = object_at(root.get('execution', {}), '$.execution', (), ('backend', 'batch_rows', 'memory_budget_bytes', 'result_buffer_rows', 'partial_policy', 'on_budget_exceeded', 'spool_quota_bytes', 'process_count'))288 # 只允许'numpy'或'reference'，不为未知实现名称自动回退。289 backend = choice_at(raw_execution.get('backend', 'auto'), ('numpy', 'reference', 'auto'), '$.execution.backend')290 # 分批行数、内存预算和结果缓冲行数必须为正整数，不能是布尔值。291 limits = tuple(number_at(raw_execution.get(key, default), '$.execution.' + key, integer=True, minimum=1) for key, default in (('batch_rows', 65536), ('memory_budget_bytes', 268435456), ('result_buffer_rows', 4096)))292 # 检查数据覆盖不完整时采用'report'还是'strict'策略。293 partial = choice_at(raw_execution.get('partial_policy', 'report'), ('report', 'strict'), '$.execution.partial_policy')294 # 超出资源预算的处理只接受'error'，不允许静默丢数据。295 choice_at(raw_execution.get('on_budget_exceeded', 'error'), ('error',), '$.execution.on_budget_exceeded')296 # 磁盘配额若给出则必须为正整数；None表示没有配置该配额。297 quota = number_at(raw_execution['spool_quota_bytes'], '$.execution.spool_quota_bytes', integer=True, minimum=1) if raw_execution.get('spool_quota_bytes') is not None else None298 # 保存检查后的执行选项，包含本次实际采用的默认值。299 process_count = number_at(raw_execution.get('process_count', 1), '$.execution.process_count', integer=True, minimum=1)300 execution = ExecutionPlan(backend, *limits, partial, quota, process_count=process_count)301 # 规则、窗口和指标共用标识集合，避免同一计划出现重复ID。302 ids = {t.name for t in transforms}303 # 先检查条件和规则，供窗口按当前规则名称建立引用。304 rules = ()305 with collection.stage('$.rules', requires=bindings is not None, check=False) as ready:306 if ready:307 rules = parse_rules(root.get('rules', []), bindings, definitions, ids)308 # 检查活动合并或固定窗口的边界与引用关系。309 windows = ()310 with collection.stage('$.windows', requires=bindings is not None, check=False) as ready:311 if ready:312 windows = parse_windows(root.get('windows', []), rules, ids)313 # 检查指标所用值、条件、统计范围、联合来源依赖和单位。314 calculations = ()315 with collection.stage('$.calculations', requires=bindings is not None and execution is not None, check=False) as ready:316 if ready:317 calculations = parse_calculations(root.get('calculations', []), bindings, definitions, rules, windows, execution, ids)318 # 新消息模式在同一ID空间解析；不向原配置副本补默认，旧plan身份recipe保持。319 message_calculations = ()320 with collection.stage('$.message_calculations', requires=bindings is not None, check=False) as ready:321 if ready:322 message_calculations = parse_message_calculations(root.get('message_calculations', []), sources, bindings, ids)323 sequences = ()324 with collection.stage('$.sequences', requires=bindings is not None, check=False) as ready:325 if ready:326 sequences = parse_sequences(root.get('sequences', []), rules, bindings, ids)327 aggregations = ()328 if 'aggregations' in root:329 with collection.stage('$.aggregations', requires=execution is not None and version is not None) as ready:330 if ready:331 aggregations = parse_aggregations(root['aggregations'],version,calculations,rules,windows,sequences,execution,ids)332 # 检查结果输出配置，此处仅保存格式，不创建输出文件。333 outputs = None334 with collection.stage('$.outputs', requires=bindings is not None, check=True) as ready:335 if ready:336 raw_outputs = object_at(root.get('outputs', {}), '$.outputs', (), ('tables', 'manifest', 'contributions','signal_csv','can_frames'))337 # 结果表只接受CSV或JSONL，清单只接受JSON。338 outputs = OutputPlan(choice_at(raw_outputs.get('tables', 'readable_csv'), ('readable_csv', 'csv', 'jsonl'), '$.outputs.tables'), choice_at(raw_outputs.get('manifest', 'json'), ('json',), '$.outputs.manifest'),339 choice_at(raw_outputs.get('contributions', 'none'), ('none', 'full'), '$.outputs.contributions'))340 # 新导出默认关闭；解析只核引用和时间政策，不创建原始读取器或加载DBC。341 from dataclasses import replace342 from .export_options import parse_signal_output,parse_frame_output343 outputs=replace(outputs,signal_csv=parse_signal_output(raw_outputs.get('signal_csv'),signals,windows,sources=sources),344 can_frames=parse_frame_output(raw_outputs.get('can_frames'),sources,windows))345 # 设备标签是可选的客户文本；在静态检查时拒绝空白和非文本，避免运行到导出才失败。346 metadata = None347 with collection.stage('$.metadata', check=False) as ready:348 if ready:349 metadata = metadata_at(root.get('metadata', {}), '$.metadata')350 if 'device_label' in metadata:351 text_at(metadata['device_label'], '$.metadata.device_label')352 # 在构造完整计划前拒绝任何结构/依赖错误，不输出部分可执行对象。353 collection.finish()354 from dataclasses import replace355 # 对配置副本和来源定位声明计算哈希，不读取文件内容。356 identity_parts = {'plan_version': 'config4-static-v1', 'config': frozen, 'sources': [item.input_identity for item in sources]}357 if extensions is not None:358 identity_parts['extensions'] = extensions.identity359 identity = sha256(canonical(identity_parts).encode('utf-8')).hexdigest()360 # 返回完整计划；原配置和附加元数据均保存为独立的只读副本。361 plan = cls(version, dataset, sources, signals, tuple(derived), rules, windows, calculations, presets, preset_bindings, execution, outputs,362 bindings, expressions, metadata, frozen, base, identity,363 message_calculations=message_calculations, sequences=sequences, aggregations=aggregations,364 transforms=transforms)365 # 在全部消费者已确定后解析有限间隔选择，DBC实际周期留运行初始化。366 from .intervals import parse_intervals, parse_diagnostics367 plan = replace(plan, intervals=parse_intervals(root.get('intervals', {}), plan),368 diagnostics=parse_diagnostics(root.get('diagnostics', {})))369 # 省略范围由同一次读取的实际观测确定，不提前扫描或补造时间端点。370 # 来源策略依赖最终输出与全部节点；完整计划确定后再拒绝不支持的requested组合。371 require_can_decode_scope(plan)372 return plan373374375376def load_analysis_plan(path: Path | str, *, extensions=None) -> AnalysisPlan:377 """读取JSON/YAML配置及一层公共定义，按各自声明目录解释相对来源路径。378379 参数:380 extensions: 宿主可选的不可变注册；文件中不能声明或导入代码。381 path (Path | str): 配置文件路径；不接受目录扫描或任意路径对象。382 配置按UTF-8读取，允许文件开头带BOM。383 返回:384 AnalysisPlan: 经过from_dict静态检查的计划。只读取主配置和显式公共定义文件，385 不打开其中声明的日志或CSV；省略信号元数据时按需读取DBC。386 异常:387 PlanValidationError: 路径类型错误、文件读取或编码错误、JSON/YAML语法错误、388 重复键、非有限常量或配置声明错误。加载错误带配置文件定位，389 通过原异常链保留更详细的原因。390 副作用:391 打开指定配置文件，正常或异常退出时均关闭句柄；不写文件。392 """393 # 与from_dict一致，只接受Path或文本配置路径。394 if not isinstance(path, (Path, str)):395 # 类型错误直接定位到配置路径参数。396 raise PlanValidationError('$', '配置路径必须为Path或文本')397 # 转为绝对路径，用于打开指定配置及关联错误位置。398 target = Path(path).absolute()399 # 将文件读取、编码、语法和配置检查错误统一关联到该文件。400 try:401 document = load_configuration(target)402 from ..configuration.imports import expand_document_imports403 document = expand_document_imports(document)404 # 配置文件的词法父目录继续作为相对来源定位基准，不读取来源。405 return AnalysisPlan.from_dict(document.data, base_dir=target.parent, extensions=extensions)406 except ConfigurationError as error:407 raise configuration_error(error) from error408 except PlanValidationError as error:409 raise locate_error(error,document)410 # 包装读取与配置检查异常，并保留原异常链。411 except (OSError, UnicodeError, ValueError, TypeError) as error:412 # 已有配置错误保留错误码，其他加载错误使用'load_error'。413 raise PlanValidationError('$',str(error),'load_error',source=str(target)) from error414