chronaxis.planning.transforms
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""解析 v4 时序变换的名称图、时间网格、输入表达式及输出单位。23每个离散序列作为独立的分析来源声明，不向真实 reader 增加字段。先按表达式4引用拓扑校验所有声明（包括不用的声明），再给消费者提供同一套最终绑定。5本模块不扫描数据、不生成网格点，也不创建历史。6"""78import ast9from dataclasses import dataclass, field, replace10from hashlib import sha25611from ..core.binding import SignalDefinition12from ..core.calculations import CalculationDefinition, CalculationConversion, calculation_conversion13from ..core.declarations import DeclaredSource14from ..core.expression import StaticExpression, validate_expression15from ..core.models import Interval16from ..core.types import SignalDType as D, TimeType17from .checks import (PlanValidationError, array_at, canonical, choice_at, coordinate_at, duration_at,18 model_at, number_at, object_at, text_at)1920ROLLING = ('rolling_time_mean', 'rolling_minimum', 'rolling_maximum', 'rolling_integral')21RATES = ('adjacent_rate', 'window_rate', 'sample_fit_slope')22MAX_TRANSFORMS = 64 # 防止用户定义无界依赖链和运行回调栈。232425@dataclass(frozen=True)26class TemporalTransformPlan:27 """按拓扑序保存一个已验证的离散变换，预算与运行状态在任务启动时创建。"""2829 name: str30 operation: str31 expression: StaticExpression32 output: SignalDefinition33 conversion: CalculationConversion34 physical: tuple[str, ...]35 window_ns: int | None36 step_ns: int | None37 origin_ns: int38 max_gap_ns: int | None39 partial_policy: str40 minimum_coverage: float41 host_spec: object = field(default=None,repr=False,compare=False) # 仅运行宿主注册的实际网格操作。424344def _names(text, path):45 """提取引用边。参数: 表达式文本及位置。返回: 名称集合。异常: 带位置的语法错误。"""46 try:47 tree = ast.parse(text, mode='eval')48 except (SyntaxError, TypeError) as error:49 raise PlanValidationError(path, '变换输入须为合法表达式') from error50 return {n.id for n in ast.walk(tree) if isinstance(n, ast.Name)}515253def build_transforms(raw, bindings, derived, *, timezone=None):54 """解析全图。参数: 配置项、物理声明、派生定义及时间区。返回: 最终声明和变换元组。异常: 配置错误。"""55 if bindings.config_version != 4:56 raise PlanValidationError('$.transforms', 'transforms需要config_version=4')57 rows, edges, plans = {}, {}, {}58 original = {s.name for s in bindings.signals}59 formulas = {d.name: d for d in derived}60 for d in derived:61 edges[d.name] = _names(d.expr, '$.derived.'+d.name)62 entries=array_at(raw,'$.transforms')63 if len(entries)>MAX_TRANSFORMS:64 raise PlanValidationError('$.transforms',f'最多支持{MAX_TRANSFORMS}个时序变换')65 for i, item in enumerate(entries):66 path = f'$.transforms[{i}]'67 base = object_at(item, path, ('name','operation','expr'), ('window','step','origin','max_gap','unit',68 'minimum_coverage','partial_policy'))69 name = text_at(base['name'], path+'.name', identifier=True)70 if name in original or name in formulas or name in rows:71 raise PlanValidationError(path+'.name', '时序、派生与原信号名称必须唯一')72 registry = getattr(bindings,'extensions',None)73 registered = registry.transforms if registry is not None else {}74 op = choice_at(base['operation'], ROLLING+RATES+tuple(registered), path+'.operation')75 params = ('max_gap',) if op == 'adjacent_rate' else ('window','step')76 extra = ('unit',) if op == 'adjacent_rate' else ('unit','origin','minimum_coverage','partial_policy')77 item = object_at(base,path,('name','operation','expr',*params),extra)78 rows[name] = path,item79 edges[name] = _names(text_at(item['expr'],path+'.expr'),path+'.expr')8081 ordered, done, visiting = [], set(), set()82 def visit(name):83 """按依赖先后检查循环。参数: 原简单名称。返回: 无。异常: 包括派生/变换混合环。"""84 if name in visiting:85 raise PlanValidationError('$.transforms', '派生或时序变换循环依赖：'+name)86 if name in done or name not in edges:87 return88 visiting.add(name)89 for dep in sorted(edges[name]):90 visit(dep)91 visiting.remove(name)92 done.add(name)93 ordered.append(name)9495 for name in edges:96 visit(name)97 available = []98 source_names = {s.name for s in bindings.sources}99 for name in ordered:100 if name in formulas:101 available.append(formulas[name])102 continue103 path, item = rows[name]104 op = item['operation']105 local = replace(bindings,derived_names=tuple(d.name for d in available))106 host_spec = registered.get(op)107 expr = model_at(validate_expression,path+'.expr',item['expr'],local,derived=tuple(available),108 unit=host_spec.input.unit if host_spec is not None else None)109 if host_spec is not None and expr.root.dtype is not host_spec.input.dtype:110 raise PlanValidationError(path+'.expr','宿主时序变换需要与声明端口一致的输入dtype')111 if expr.root.dtype not in (D.FLOAT64,D.INT64,D.UINT64):112 raise PlanValidationError(path+'.expr','时序数值变换需要数值输入')113 physical = tuple(dict.fromkeys(alias for dep in expr.dependencies114 for alias in (plans[dep].physical if dep in plans else (dep,))))115 model_at(bindings.require_synchronized,path+'.expr',physical)116 grid = op != 'adjacent_rate'117 window = duration_at(item['window'],path+'.window',positive=True) if grid else None118 if host_spec is not None and window > host_spec.history_ns:119 raise PlanValidationError(path+'.window','时序窗口超过宿主注册的history_ns')120 step = duration_at(item['step'],path+'.step',positive=True) if grid else None121 gap = None if grid else duration_at(item['max_gap'],path+'.max_gap',positive=True)122 origin = coordinate_at(item['origin'],path+'.origin',absolute=bindings.timebase.time_type is TimeType.ABSOLUTE,123 tz=timezone) if 'origin' in item else 0124 policy = choice_at(item.get('partial_policy','strict'),('strict','report'),path+'.partial_policy')125 coverage = number_at(item.get('minimum_coverage',0),path+'.minimum_coverage',minimum=0,maximum=1)126 unit = host_spec.output.unit if host_spec is not None else expr.root.unit127 if op in RATES:128 from ..core.rate_transforms import rate_unit129 unit = model_at(rate_unit,path+'.unit',unit)130 # 平均值/极值保留输入尺度，积分从规范输入推导面积；最后统一兼容目标。131 calculation = 'integral' if op=='rolling_integral' else 'sample_minimum'132 target = item.get('unit', None if calculation=='integral' else unit)133 if calculation == 'integral':134 from ..core.unit_registry import resolve_unit135 child = expr.root136 while child.operator=='unit_convert' and not child.unit_explicit:137 child = child.args[0]138 original_unit = resolve_unit(child.unit)139 if original_unit is not None and original_unit.offset:140 raise PlanValidationError(path+'.expr','绝对摄氏度滚动积分须先显式转换到K')141 spec = CalculationDefinition(name,calculation,Interval(0,0),output_unit=target,config_version=4)142 conversion = model_at(calculation_conversion,path+'.unit',spec,unit)143 dtype = conversion.dtype(expr.root.dtype) if op in ('rolling_minimum','rolling_maximum') else D.FLOAT64144 if host_spec is not None:145 dtype = conversion.dtype(host_spec.output.dtype)146 source = '_temporal_'+name147 while source in source_names:148 source = '_'+source149 source_names.add(source)150 identity = sha256(canonical((item,registry.identity) if host_spec is not None else item).encode('utf-8')).hexdigest()151 declared = DeclaredSource(source,bindings.timebase,'temporal:'+identity)152 output = SignalDefinition(name,source,'temporal.'+name,dtype,conversion.unit,step or gap,config_version=4)153 bindings = replace(bindings,sources=bindings.sources+(declared,),signals=bindings.signals+(output,),154 temporal_inputs=bindings.temporal_inputs+((name,physical),))155 plans[name] = TemporalTransformPlan(name,op,expr,output,conversion,physical,window,step,origin,gap,policy,coverage,host_spec)156 # 重编译到最终绑定；执行时按原声明复核，不把临时拓扑身份交给 reader。157 return bindings, tuple(replace(p,expression=model_at(validate_expression,rows[p.name][0]+'.expr',158 p.expression.text,bindings,derived=derived,unit=p.host_spec.input.unit if p.host_spec is not None else None))159 for p in plans.values())160161162def active_transforms(plan, aliases):163 """从消费者逆向收集必要变换。参数: 计划、输出别名集合。返回: 拓扑序元组。异常: 已建计划错误。"""164 needed = set(aliases)165 selected = []166 for item in reversed(plan.transforms):167 if item.name in needed:168 selected.append(item)169 needed.update(item.expression.dependencies)170 return tuple(reversed(selected))171172173def physical_dependencies(plan, aliases):174 """展开原采集依赖。参数: 计划和表达式输入别名。返回: 保序去重原别名。异常: 已建计划错误。"""175 transforms = {p.name:p.physical for p in plan.transforms}176 return tuple(dict.fromkeys(n for a in aliases for n in transforms.get(a,(a,))))177