chronaxis.planning.aggregations
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""校验 v4 结果表聚合的来源、字段、分组、总体政策与输出单位。"""23from dataclasses import dataclass4from fractions import Fraction5from .checks import PlanValidationError, array_at, object_at, choice_at6from .nodes import register_id7from ..core.unit_registry import resolve_unit, output_label, conversion8from ..core.moments import moment_source_unit9from ..core.quantiles import quantile_probability101112@dataclass(frozen=True)13class AggregationPlan:14 """一次已编译的归约，只消费同一分析中的一个具名结果来源。"""1516 name: str17 source_kind: str18 source_id: str19 operation: str20 field: str | None21 group_by: tuple[str, ...]22 input_unit: str23 unit: str24 include_partial: bool25 partial_policy: str26 q: int | float | None27 ddof: int282930def aggregation_unit(operation, source_unit, target):31 """核实已输出量的尺度，方差用平方温差，返回可复算的输出换算。3233 参数: operation 为聚合操作，source_unit 为引用结果的单位，target 可省略。34 返回: 输出单位、倍率、偏移；偏移在求和时按参与数累计。异常: 单位不可比或未知变换。35 """36 if operation == 'count':37 if target not in (None,'1'):38 raise ValueError('count 的单位只能为 1')39 return '1', 1, 040 source = resolve_unit(source_unit)41 if source is None:42 if operation != 'variance' and target in (None,source_unit):43 return source_unit, 1, 044 raise ValueError('未知单位不能换算或求方差')45 base = moment_source_unit('sample_variance',source) if operation=='variance' else source46 # 方差保留原尺度的平方，默认输出文本必须能再次解析。47 spelling = target or (f'({source_unit})^2' if operation=='variance' else source_unit)48 if operation == 'variance' and source.temperature_role == 'absolute' and target is None:49 spelling = '(delta_degC)^2' if source.offset else '(delta_K)^2'50 dst = resolve_unit(target) if target is not None else base51 if dst is None:52 raise ValueError('聚合输出单位未登记')53 scale, shift = conversion(base,dst)54 factor = scale.factor if scale.pi_power == 0 else Fraction(scale.to_float())55 return output_label(dst,spelling), factor, shift565758def parse_aggregations(raw, version, calculations, rules, windows, sequences, execution, ids):59 """把声明列表解析为不可变归约计划，拒绝选择行或任意字段访问。6061 参数: raw 为数组，version/各来源目录/execution/ids 均来自同一主计划。62 返回: AggregationPlan 元组，顺序与配置一致。异常: PlanValidationError 定位错误条目。63 """64 if version != 4:65 raise PlanValidationError('$.aggregations','aggregations 仅在配置版本4中可用')66 catalogs = {'calculation':{r.calculation_id:r for r in calculations},'sequence':{r.sequence_id:r for r in sequences},67 'rule':{r.rule_id:r for r in rules},'window':{r.window_id:r for r in windows}}68 plans = []69 for i,value in enumerate(array_at(raw,'$.aggregations')):70 path = f'$.aggregations[{i}]'71 item = object_at(value,path,('name','source','operation'),72 ('field','group_by','unit','partial_policy','include_partial','q','ddof'))73 name = register_id(item['name'],path+'.name',ids)74 source = object_at(item['source'],path+'.source',(),tuple(catalogs))75 if len(source) != 1:76 raise PlanValidationError(path+'.source','需要一个具名结果对象')77 kind,source_id = next(iter(source.items()))78 if type(source_id) is not str or source_id not in catalogs[kind]:79 raise PlanValidationError(path+'.source.'+kind,'引用的结果对象不存在')80 source_plan = catalogs[kind][source_id]81 if kind=='calculation' and source_plan.operation=='select':82 raise PlanValidationError(path+'.source','select 行不是标量结果')83 op = choice_at(item['operation'],('count','sum','mean','variance','quantile','median')84 if kind in ('calculation','sequence') else ('count',),path+'.operation')85 field = None if op=='count' else 'delay_ns' if kind=='sequence' else 'value'86 if 'field' in item and (field is None or item['field'] != field):87 raise PlanValidationError(path+'.field','此操作不支持所选字段')88 keys = tuple(array_at(item.get('group_by',[]),path+'.group_by'))89 allowed = {'dataset_id','entity_id','session_id',kind+'_id'}90 if kind=='calculation' and source_plan.scope!='session':91 allowed.add('parent_id')92 if any(type(key) is not str or key not in allowed for key in keys) or len(set(keys))!=len(keys):93 raise PlanValidationError(path+'.group_by','分组键不适用于此来源或重复')94 include = item.get('include_partial',False)95 if type(include) is not bool or include and (kind!='calculation' or op=='count'):96 raise PlanValidationError(path+'.include_partial','仅数值指标允许纳入 partial 数值')97 policy=choice_at(item.get('partial_policy',execution.partial_policy),('report','strict'),path+'.partial_policy')98 q=item.get('q')99 if op=='quantile':100 try:101 quantile_probability('sample_quantile',q)102 except ValueError as error:103 raise PlanValidationError(path+'.q',str(error)) from error104 elif 'q' in item:105 raise PlanValidationError(path+'.q','只有 quantile 接受 q')106 ddof=item.get('ddof',0)107 if 'ddof' in item and (op!='variance' or type(ddof) is not int or ddof not in (0,1)):108 raise PlanValidationError(path+'.ddof','只有 variance 接受 ddof=0 或 1')109 input_unit = '1' if op=='count' else 'ns' if kind=='sequence' else source_plan.unit110 try:111 unit,_,_=aggregation_unit(op,input_unit,item.get('unit'))112 except (TypeError,ValueError) as error:113 raise PlanValidationError(path+'.unit',str(error)) from error114 plans.append(AggregationPlan(name,kind,source_id,op,field,keys,input_unit,unit,include,policy,q,ddof))115 return tuple(plans)116