chronaxis.presets
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""校验领域预设v1的角色契约，展开为可序列化的类型化公共声明。23输入PresetSpec与当前SignalDefinition tuple，输出PlanFragment；不读取来源或执行分析。4先严格校验参数，再显式优先解析角色/检查转换后契约，最后生成简单别名表达式和节点。5候选语义名只属于本模块的版本化契约，不是通用核心领域目录或行业标准。6声明会交给AnalysisPlan正常编译；不使用eval、不推断方向/位置/阈值或补造真实数据。7输入/输出均不可变；JSON序列化返回独立副本。执行、资源预算和报文导出由执行层负责。8"""910# 算法与公开声明模型分开，避免AnalysisPlan/构建器循环。11from .core.binding import SignalDefinition12from .core.expression import DerivedDefinition13from .core.types import SignalDType as D14from .core.windows import WindowSpec15from .planning.checks import (PlanValidationError, choice_at, duration_at, json_value,16 array_at, number_at, object_at, text_at)17from .planning.preset_models import (RoleContract, RoleBinding, PresetSpec, PresetRule,18 PresetCalculation, PlanFragment, CHARGE_ROLES, FAULT_ROLES, COMMAND_ROLES)19from .planning.sequences import SequencePlan20from .timebase.conversion import INT64_MIN, INT64_MAX212223def _parameters(spec, signals=()):24 """按kind检查工程参数，保持原数值并将时长转换为整数ns。2526 参数: spec为PresetSpec；参数值来自独立JSON快照。27 返回: 新dict，仅内部使用已校验参数，不作为公开无类型片段。28 异常: PlanValidationError为缺项、未知参数、bool伪数值、非有限或错误范围。29 """30 # 将不可变快照还原为局部原生对象，严格拒绝未声明参数。31 path = '$.preset.parameters'32 if spec.preset_type == 'charge_session':33 raw = object_at(json_value(spec.parameters), path, ('mode_value', 'current_min', 'minimum_duration'),34 ('compute_energy', 'compute_charge', 'minimum_coverage', 'partial_policy'))35 result = {'mode_value': number_at(raw['mode_value'], path + '.mode_value', integer=True),36 'current_min': number_at(raw['current_min'], path + '.current_min'),37 'minimum_duration': duration_at(raw['minimum_duration'], path + '.minimum_duration'),38 'compute_energy': choice_at(raw.get('compute_energy', False), (False, True), path + '.compute_energy'),39 'compute_charge': choice_at(raw.get('compute_charge', False), (False, True), path + '.compute_charge'),40 'minimum_coverage': number_at(raw.get('minimum_coverage', 0), path + '.minimum_coverage', minimum=0, maximum=1),41 'partial_policy': choice_at(raw['partial_policy'], ('report', 'strict'), path + '.partial_policy') if 'partial_policy' in raw else None}42 elif spec.preset_type == 'fault_context':43 raw = object_at(json_value(spec.parameters), path, (), ('pre', 'post', 'emit_initial'))44 result = {'pre': duration_at(raw.get('pre', '0s'), path + '.pre'), 'post': duration_at(raw.get('post', '0s'), path + '.post'),45 'emit_initial': choice_at(raw.get('emit_initial', False), (False, True), path + '.emit_initial')}46 if result['pre'] == result['post'] == 0:47 raise PlanValidationError(path, 'fault_context至少一个pre/post为正')48 elif spec.preset_type == 'threshold_excursion':49 # v1明确高阈值进入、低阈值退出；拒绝重叠，不能在阈值相等时猜优先级。50 raw = object_at(json_value(spec.parameters), path, ('enter_threshold', 'exit_threshold'),51 ('unit', 'enter_delay', 'exit_delay'))52 if 'unit' in raw:53 unit = text_at(raw['unit'], path + '.unit')54 else:55 alias = spec.role_bindings.get('signal')56 candidates = [item for item in signals if item.name == alias]57 if len(candidates) != 1:58 raise PlanValidationError('$.preset.role_bindings.signal', '请指定唯一的已绑定信号以继承单位')59 unit = candidates[0].unit60 result = dict(unit=unit,61 enter_threshold=number_at(raw['enter_threshold'], path + '.enter_threshold'),62 exit_threshold=number_at(raw['exit_threshold'], path + '.exit_threshold'),63 enter_delay=duration_at(raw.get('enter_delay', '0ns'), path + '.enter_delay'),64 exit_delay=duration_at(raw.get('exit_delay', '0ns'), path + '.exit_delay'))65 if result['enter_threshold'] <= result['exit_threshold']:66 raise PlanValidationError(path, '进入阈值必须大于退出阈值；v1高于进入、低于退出，均含等号')67 else:68 raw = object_at(json_value(spec.parameters), path, ('max_delay',), ('min_delay', 'emit_initial', 'max_pending'))69 result = dict(min_delay=duration_at(raw.get('min_delay', '0s'), path + '.min_delay'),70 max_delay=duration_at(raw['max_delay'], path + '.max_delay'),71 emit_initial=choice_at(raw.get('emit_initial', False), (False, True), path + '.emit_initial'),72 max_pending=number_at(raw.get('max_pending', 65536), path + '.max_pending', integer=True, minimum=1))73 if result['min_delay'] > result['max_delay']:74 raise PlanValidationError(path, 'min_delay不能大于max_delay')75 return result767778def _mismatches(definition, contract, *, generic_unit=False):79 """列出转换后声明未满足的角色字段，不能用自由metadata放宽质量。8081 参数: definition为SignalDefinition；contract为本v1角色；generic_unit允许明确同名无转换的自定义单位。82 返回: tuple[str, ...]，空表示字段契约通过，不证明真实数据正确。83 异常: 无主动异常；输入类型由公共解析入口证明。84 """85 # 逐字段诊断方便缺项/歧义时列出候选不满足原因。86 errors = []87 if definition.dtype not in contract.dtypes:88 errors.append('dtype须为' + '/'.join(item.value for item in contract.dtypes))89 identity_unit = (generic_unit and definition.input_unit == definition.unit90 and definition.scale == 1 and definition.offset == 0 and definition.direction_factor == 1)91 if definition.unit != contract.unit or definition.unit_status != 'verified' and not identity_unit:92 errors.append(f'目标unit须为{contract.unit}且关系verified')93 if contract.sign_convention is not None and definition.sign_convention != contract.sign_convention:94 errors.append('sign_convention须为' + contract.sign_convention)95 if contract.measurement_location is not None and definition.measurement_location != contract.measurement_location:96 errors.append('measurement_location须为' + contract.measurement_location)97 # DBC默认有效期在运行启动时核实；声明阶段不为角色校验额外加载DBC。98 if (definition.valid_for_ns is not None and definition.valid_for_ns <= 099 or definition.sample_policy != 'backward_hold' or definition.identity()[-1] != contract.forbidden_quality):100 errors.append('正valid_for/backward_hold/固定基础质量策略未满足')101 return tuple(errors)102103104def resolve_preset_roles(spec: PresetSpec, signals: tuple[SignalDefinition, ...]) -> tuple[RoleBinding, ...]:105 """显式优先解析本次实际角色，否则按版本化语义契约寻找唯一合格候选。106107 参数: spec为公开不可变预设；signals为当前计划SignalDefinition tuple。108 返回: RoleBinding tuple，保留原定义与explicit/semantic_unique路径。109 异常: PlanValidationError为类型、重复别名、未知角色、缺项、候选歧义或契约冲突。110 边界: 显式别名无需semantic_id，仍检查类型/单位/方向/位置/质量；不选第一个候选。111 """112 if not isinstance(spec, PresetSpec) or not isinstance(signals, tuple) or any(not isinstance(item, SignalDefinition) for item in signals):113 raise PlanValidationError('$.preset', '需要PresetSpec与SignalDefinition tuple')114 parameters = _parameters(spec, signals)115 if spec.preset_type == 'threshold_excursion':116 # 任意物理量显式绑定；未填单位则继承该信号，不猜温度等领域名称。117 if 'signal' not in spec.role_bindings:118 raise PlanValidationError('$.preset.role_bindings.signal', 'threshold_excursion必须显式指定signal角色')119 from .planning.preset_models import RoleContract120 contracts = (RoleContract('signal', (D.INT64, D.UINT64, D.FLOAT64), parameters['unit'], 'threshold.signal'),)121 else:122 contracts = {'charge_session': CHARGE_ROLES, 'fault_context': FAULT_ROLES,123 'command_response': COMMAND_ROLES}[spec.preset_type]124 catalog = {item.name: item for item in signals}125 if len(catalog) != len(signals):126 raise PlanValidationError('$.signals', 'signal别名重复')127 unknown = set(spec.role_bindings) - {item.role for item in contracts}128 if unknown:129 raise PlanValidationError('$.preset.role_bindings', '未知角色：' + ', '.join(sorted(unknown)))130 result = []131 for contract in contracts:132 # voltage只在请求能量或明确提供该可选绑定时检查，不强迫无能量任务配置电压。133 if contract.role == 'voltage' and not parameters['compute_energy'] and contract.role not in spec.role_bindings:134 continue135 path = '$.preset.role_bindings.' + contract.role136 if contract.role in spec.role_bindings:137 alias = spec.role_bindings[contract.role]138 if alias not in catalog:139 raise PlanValidationError(path, f'未知别名：{alias}；可用：{", ".join(catalog)}')140 definition = catalog[alias]141 errors = _mismatches(definition, contract, generic_unit=spec.preset_type == 'threshold_excursion')142 if errors:143 raise PlanValidationError(path, f'{alias}契约不满足：{"; ".join(errors)}')144 result.append(RoleBinding(contract, definition, 'explicit'))145 else:146 # 只取本模块公开候选标识/版本，所有合格候选统计完才决定。147 candidates = [item for item in signals if item.semantic_id == contract.semantic_id]148 valid = [item for item in candidates if item.semantic_contract_version == contract.version and not _mismatches(item, contract)]149 if len(valid) != 1:150 details = '; '.join(item.name + ': ' + ', '.join(_mismatches(item, contract) +151 (() if item.semantic_contract_version == contract.version else ('semantic_contract_version须为1',))) for item in candidates)152 reason = '歧义候选：' + ', '.join(item.name for item in valid) if valid else '缺少符合契约的候选'153 raise PlanValidationError(path, reason + f'（{contract.semantic_id} v1）；' + details)154 result.append(RoleBinding(contract, valid[0], 'semantic_unique'))155 return tuple(result)156157158def build_preset(spec: PresetSpec, signals: tuple[SignalDefinition, ...]) -> PlanFragment:159 """校验v1预设并展开公共声明，不执行I/O或数值分析。160161 参数: spec为PresetSpec，保存id/preset_type/version/显式角色及参数；signals为当前类型化信号tuple。162 返回: PlanFragment，生成请求的功率/Wh/Ah、报警、阈值迟滞/逐段时长或响应配对。163 异常: PlanValidationError为参数/角色契约、模式编码范围及生成ID与信号冲突。164 流程: 检查角色→生成简单别名DSL→构造类型化节点→拒绝当前可见ID冲突。165 用户其他节点/多个预设的冲突由AnalysisPlan合入时继续检查。166 """167 resolved = resolve_preset_roles(spec, signals)168 roles = {item.contract.role: item.definition for item in resolved}169 parameters = _parameters(spec, signals)170 if spec.preset_type == 'charge_session':171 mode, current = roles['mode'], roles['current']172 minimum, maximum = (INT64_MIN, INT64_MAX) if mode.dtype is D.INT64 else (0, 2**64-1)173 code = parameters['mode_value']174 if not minimum <= code <= maximum or mode.valid_range is not None and not mode.valid_range[0] <= code <= mode.valid_range[1]:175 raise PlanValidationError('$.preset.parameters.mode_value', '模式编码超出已声明dtype/valid_range')176 # 只拼接已验证别名与原生有限数值字面量，不能注入属性或任意函数。177 condition = f'{mode.name} == {code!r} and {current.name} > {parameters["current_min"]!r}'178 rule = PresetRule(spec.id + '_activity', 'condition_duration', condition, parameters['minimum_duration'], 'onset')179 derived, calculations = [], []180 for requested, suffix, signal, unit in (181 (parameters['compute_energy'], 'energy', spec.id + '_power', 'Wh'),182 (parameters['compute_charge'], 'charge', current.name, 'Ah')):183 if requested:184 calculations.append(PresetCalculation(spec.id + '_' + suffix, 'event:' + rule.id, signal, unit,185 parameters['minimum_coverage'], parameters['partial_policy']))186 if parameters['compute_energy']:187 derived.append(DerivedDefinition(spec.id + '_power', roles['voltage'].name + ' * ' + current.name, 'W'))188 fragment = PlanFragment(tuple(derived), (rule,), (), tuple(calculations))189 elif spec.preset_type == 'fault_context':190 rule = PresetRule(spec.id + '_alarm', 'rising_edge', roles['alarm'].name, emit_initial=parameters['emit_initial'])191 window = WindowSpec(spec.id + '_context', rule.id, 'point', parameters['pre'], parameters['post'])192 fragment = PlanFragment((), (rule,), (window,), ())193 elif spec.preset_type == 'threshold_excursion':194 # 普通迟滞规则负责全部状态，逐段时长复用原duration核；总计由同一次状态推进保存。195 alias = roles['signal'].name196 rule = PresetRule(spec.id + '_activity', 'hysteresis', None,197 enter_condition=f'{alias} >= {parameters["enter_threshold"]!r}',198 exit_condition=f'{alias} <= {parameters["exit_threshold"]!r}',199 enter_delay_ns=parameters['enter_delay'], exit_delay_ns=parameters['exit_delay'])200 calculation = PresetCalculation(spec.id + '_duration', 'event:' + rule.id, None, 's',201 operation='duration', condition='True')202 fragment = PlanFragment(rules=(rule,), calculations=(calculation,))203 else:204 # 布尔角色变为两条普通rising_edge规则，配对仍走公开SequencePlan，没有隐藏计算器。205 request = PresetRule(spec.id + '_request', 'rising_edge', roles['request'].name, emit_initial=parameters['emit_initial'])206 response = PresetRule(spec.id + '_response', 'rising_edge', roles['response'].name, emit_initial=parameters['emit_initial'])207 sequence = SequencePlan(spec.id + '_pair', request.id, response.id,208 parameters['min_delay'], parameters['max_delay'], parameters['max_pending'])209 fragment = PlanFragment(rules=(request, response), sequences=(sequence,))210 # 不自动改生成名称，当前signal碰撞在公共构建器立即拒绝。211 generated = [item.name for item in fragment.derived] + [item.id for item in fragment.rules] + [item.window_id for item in fragment.windows] + [item.id for item in fragment.calculations]212 generated.extend(item.sequence_id for item in fragment.sequences)213 conflicts = set(generated) & {item.name for item in signals}214 if conflicts:215 raise PlanValidationError('$.preset.name', '生成ID冲突：' + ', '.join(sorted(conflicts)))216 return fragment217218219def expand_presets(config, signals):220 """在独立配置副本中合入预设片段，保留用户节点并拒绝全部生成ID冲突。221222 参数: config为dict，是调用方复制后的配置映射；本函数原地合入生成节点。223 signals为tuple[SignalDefinition, ...]，提供当前信号定义和预设角色匹配依据。224 返回: (PresetSpec tuple, 各预设RoleBinding tuple)，供计划保留原声明及角色追溯。225 异常: PlanValidationError为预设/用户结构或生成ID与用户/其他预设重名。226 副作用: 仅修改传入的局部derived/rules/windows/calculations/sequences；不修改raw_config快照或来源。227 """228 raw_derived = config.get('derived', {})229 if type(raw_derived) is not dict:230 raise PlanValidationError('$.derived', 'derived必须为别名对象')231 # 全部当前用户名称及节点ID都保留，生成项不得隐藏覆盖。232 occupied = {item.name for item in signals} | set(raw_derived)233 arrays = {}234 for key in ('rules', 'windows', 'calculations', 'sequences'):235 arrays[key] = array_at(config.get(key, []), '$.' + key)236 for index, row in enumerate(arrays[key]):237 if type(row) is dict and 'id' in row:238 raise PlanValidationError(f'$.{key}[{index}].id', '名称字段已统一为name，请替换id', 'unknown_field')239 if type(row) is not dict or 'name' not in row:240 raise PlanValidationError(f'$.{key}[{index}]', '节点必须包含name')241 occupied.add(text_at(row['name'], f'$.{key}[{index}].name', identifier=True))242 specs, bindings, preset_ids = [], [], set()243 for index, raw in enumerate(array_at(config.get('presets', []), '$.presets')):244 path = f'$.presets[{index}]'245 try:246 spec = PresetSpec.from_dict(raw)247 if spec.id in preset_ids:248 raise PlanValidationError('$.preset.name', 'preset id重复：' + spec.id)249 fragment = build_preset(spec, signals)250 bound = resolve_preset_roles(spec, signals)251 # to_dict承诺可直接合入用户配置；内部展开再转换到同一解析器形式。252 from .planning.config_groups import normalize_configuration253 serialized = normalize_configuration({'sources': [], 'signals': [], **fragment.to_dict()})254 generated = set(serialized['derived']) | {item['name'] for key in arrays for item in serialized.get(key, ())}255 if generated & occupied:256 raise PlanValidationError('$.preset.name', '生成ID冲突：' + ', '.join(sorted(generated & occupied)))257 except PlanValidationError as error:258 # 外层配置位置替换统一公共spec路径，不丢角色/参数诊断。259 raise PlanValidationError(error.path.replace('$.preset', path, 1), error.message, error.code) from error260 # 校验整个片段后再合入，失败不发布半个公开计划。261 raw_derived.update(serialized['derived'])262 for key in arrays:263 arrays[key].extend(serialized.get(key, ()))264 occupied.update(generated)265 preset_ids.add(spec.id)266 specs.append(spec)267 bindings.append(bound)268 config['derived'] = raw_derived269 config.update(arrays)270 return tuple(specs), tuple(bindings)271