chronaxis.execution.result_aggregations
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在最终结果行产生时按具名引用分组。只保存归约状态和精确分位数所需的样本。"""23from fractions import Fraction4from math import isfinite5from ..core.quantiles import ExactQuantileBuffer6from ..core.types import SignalDType as D, QualityFlag as Q7from ..planning.aggregations import aggregation_unit8from ..results.aggregation_row import SummaryAggregationRow9from ..results.summary import SummaryCalculationRow, SummaryEventRow, SummarySegmentRow, SummarySequenceRow10from ..results.budget import _reserve_owned_ram, _release_owned_ram, _cleanup_owner111213def discovery_rules(plan, spec):14 """定位总体发现所依赖的原规则。参数: 主计划与聚合。返回: 规则ID元组。异常: 引用未校验。"""15 if spec.source_kind=='rule':16 return (spec.source_id,)17 if spec.source_kind=='sequence':18 pair=next(item for item in plan.sequences if item.sequence_id==spec.source_id)19 return tuple(dict.fromkeys((pair.start_rule,pair.response_rule)))20 if spec.source_kind=='calculation':21 calc=next(item for item in plan.calculations if item.calculation_id==spec.source_id)22 if calc.scope=='session':23 return ()24 scope,parent=calc.scope.split(':')25 if scope=='event':26 return (parent,)27 else:28 parent=spec.source_id29 window=next(w for w in plan.windows if w.window_id==parent)30 return (window.event,)313233class _Group:34 """一个实际键的有限状态。二阶和在精确有理数域累计，输出边界检查数值范围。"""3536 def __init__(self):37 """创建空候选和零计数。参数: 无。返回: 新组。异常: 分配错误传播。"""38 self.total=self.used=self.excluded=self.bad=self.partials=039 self.matched=self.timeout=self.insufficient=040 self.sum=self.squares=Fraction(0)41 self.quality=Q.NONE42 self.quantile=None43 self.integer=True4445 def accept(self,spec,row,budget,backend,dtype,needs_coverage):46 """从一条源行提取候选、分类及实际数值。参数: 计划、行、预算、后端、原dtype。返回: None。异常: 预算或模型错误。"""47 self.total+=148 if spec.source_kind=='sequence':49 self.matched+=row.status=='matched'50 self.timeout+=row.status=='no_response'51 self.insufficient+=row.status=='indeterminate'52 if spec.operation=='count':53 # 计已观测对象身份。已确认请求的响应未知不使该请求本身消失。54 self.used+=155 return56 if spec.source_kind=='sequence':57 if row.status=='no_response':58 self.excluded+=159 return60 if row.status=='indeterminate':61 self.bad+=162 self.quality|=row.quality63 return64 if row.quality & Q.ESTIMATED:65 self.bad+=166 self.quality|=row.quality67 return68 value=row.delay_ns69 else:70 if row.unit!=spec.input_unit:71 raise ValueError('指标结果行单位与同一来源计划不符')72 if row.status=='complete' and row.value is None and row.reason=='empty_selection':73 self.excluded+=174 return75 if (row.value is None or row.status=='indeterminate' or getattr(row,'coverage_status',None)=='indeterminate'76 or needs_coverage and not hasattr(row,'coverage_status')77 or row.status=='partial' and not spec.include_partial):78 self.bad+=179 self.quality|=row.quality or Q.MISSING80 return81 self.partials+=row.status=='partial'82 value=row.value83 self.quality|=row.quality84 self.used+=185 self.integer &= type(value) is int86 if spec.operation in ('quantile','median'):87 if self.quantile is None:88 self.quantile=ExactQuantileBuffer('sample_'+spec.operation,spec.q,dtype,backend,budget)89 if dtype is D.FLOAT64 and type(value) is int and int(float(value))!=value:90 self.quality|=Q.PRECISION_LOSS91 self.bad+=192 self.used-=193 self.partials-=getattr(row,'status','')=='partial'94 return95 if backend=='numpy':96 import numpy as np97 from ..core.numeric_numpy import DTYPES98 self.quantile.append(np.asarray([value],dtype=DTYPES[dtype]),np.ones(1,dtype=bool))99 else:100 self.quantile.append((value,))101 else:102 exact=Fraction(value)103 self.sum+=exact104 if spec.operation=='variance':105 self.squares+=exact*exact106107 def value(self,spec,factor,shift):108 """最终一次换算，不平均行覆盖率或块分位数。参数: 计划及已核单位系数。返回: 数值、原因、质量。异常: 原预算/排序错误。"""109 if spec.operation=='count':110 return self.used,'ok',Q.NONE111 if not self.used:112 return None,'empty_population',Q.NONE113 if spec.operation=='variance' and self.used<=spec.ddof:114 return None,'insufficient_count',Q.NONE115 if spec.operation in ('median','quantile'):116 value,quality=self.quantile.finish()117 if value is None:118 return None,'numeric_error',quality119 number=Fraction(value)*factor+shift120 integral=type(value) is int121 elif spec.operation=='variance':122 number=(self.squares-self.sum*self.sum/self.used)/(self.used-spec.ddof)*factor123 integral=False124 elif spec.operation=='mean':125 number=self.sum/self.used*factor+shift126 integral=False127 else:128 number=self.sum*factor+shift*self.used129 integral=self.integer130 if integral and isinstance(number,Fraction) and number.denominator==1:131 if -(1<<63)<=number.numerator<1<<64:132 return number.numerator,'ok',Q.NONE133 return None,'numeric_error',Q.OVERFLOW134 try:135 result=float(number)136 except OverflowError:137 return None,'numeric_error',Q.OVERFLOW138 if (not isfinite(result) or result==0 and number139 or self.integer and abs(number)>=1<<53 and Fraction(result)!=number):140 return None,'numeric_error',Q.PRECISION_LOSS141 return result,'ok',Q.NONE142143144class ResultAggregations:145 """订阅同一次运行的必要源行，分组先预约，异常由拥有者统一关闭。"""146147 def __init__(self,plan,budget):148 """预留有限声明、已知空组和结果元数据。参数: 已校验计划和共享账本。返回: owner。异常: 配置/预算错误含清理句柄。"""149 self._plan,self._budget=plan,budget150 self._held_cost=self._group_cost=0151 self._groups={}152 self._subscribers={}153 self._types={}154 self._scales={}155 self._coverage=set()156 self._seen=set()157 self._closed=False158 try:159 _reserve_owned_ram(self,8192+8192*len(plan.aggregations),'_group_cost')160 for spec in plan.aggregations:161 self._subscribers.setdefault((spec.source_kind,spec.source_id),[]).append(spec)162 self._scales[spec]=aggregation_unit(spec.operation,spec.input_unit,spec.unit)[1:]163 self._types[spec]=self._input_dtype(spec)164 if spec.source_kind=='calculation' and any(c.calculation_id==spec.source_id and165 c.minimum_coverage>0 for c in plan.calculations):166 self._coverage.add(spec)167 except BaseException as exc:168 _cleanup_owner(self,exc,'aggregation_cleanup_owner','结果聚合')169 raise170171 def _input_dtype(self,spec):172 """确定分位数排序列的原类型。参数: 已校验聚合。返回: dtype。异常: 引用不符。"""173 if spec.source_kind=='sequence':174 return D.INT64175 if spec.source_kind!='calculation' or spec.operation not in ('median','quantile'):176 return D.FLOAT64177 calc=next(c for c in self._plan.calculations if c.calculation_id==spec.source_id)178 if self._plan.extensions is not None and calc.operation in self._plan.extensions.reductions:179 from ..core.calculations import CalculationDefinition,calculation_conversion180 from ..core.models import Interval181 value=self._plan.extensions.reductions[calc.operation].output182 unit=CalculationDefinition('stat','sample_minimum',Interval(0,0),output_unit=calc.unit,config_version=4)183 return calculation_conversion(unit,value.unit).dtype(value.dtype)184 if calc.operation.endswith('_count'):185 return D.UINT64186 if calc.operation in ('sample_minimum','sample_maximum','min_time','max_time'):187 from ..core.calculations import CalculationDefinition,calculation_conversion188 from ..core.models import Interval189 unit=CalculationDefinition('stat',calc.operation,Interval(0,0),output_unit=calc.unit,config_version=4)190 return calculation_conversion(unit,calc.signal.root.unit).dtype(calc.signal.root.dtype)191 return D.FLOAT64192193 def _group(self,spec,row):194 """按稳定字段取得或创建组。参数: 计划、实际源行或已知固定空组。返回: 状态。异常: 预算或键不合法。"""195 dataset=self._plan.dataset196 values={'dataset_id':row.dataset_id if row is not None else dataset.dataset_id,197 'session_id':row.session_id if row is not None else dataset.session_id,198 'entity_id':row.entity_id if row is not None else dataset.entity,199 spec.source_kind+'_id':spec.source_id}200 if 'parent_id' in spec.group_by:201 values['parent_id']=row.parent_id202 key=tuple(values[k] for k in spec.group_by)203 identity=(spec,key)204 if identity not in self._groups:205 _reserve_owned_ram(self,32768+16*sum(len(k) for k in key),'_group_cost')206 self._groups[identity]=_Group()207 return self._groups[identity]208209 def accept(self,row):210 """只分派实际受订阅的结果行。参数: 本次产出的行。返回: None。异常: 预算/输入矛盾。"""211 if isinstance(row,SummaryCalculationRow):212 key=('calculation',row.calculation_id)213 elif isinstance(row,SummarySequenceRow):214 key=('sequence',row.rule_id)215 elif isinstance(row,SummaryEventRow):216 key=('rule',row.rule_id)217 elif isinstance(row,SummarySegmentRow):218 key=('window',row.window_id)219 else:220 return221 for spec in self._subscribers.get(key,()):222 self._seen.add(spec)223 self._group(spec,row).accept(spec,row,self._budget,self._plan.execution.backend,self._types[spec],spec in self._coverage)224225 def rows(self,rules,sequence):226 """在原规则完成后交付总体判定及组结果。参数: 原规则owner、首序号。返回: 逐组行。异常: 未完成或预算错误。"""227 by_id={owner._request.rule_id:owner for owner in rules}228 ds=self._plan.dataset229 for spec in self._plan.aggregations:230 if spec not in self._seen and 'parent_id' not in spec.group_by:231 self._group(spec,None)232 for (spec,key),group in self._groups.items():233 evidence=[by_id[name].count_evidence(ds.domain) for name in discovery_rules(self._plan,spec)]234 complete=all(e[2] for e in evidence)235 observed=all(e[0] for e in evidence)236 determinate=all(e[1] for e in evidence)237 quality=group.quality238 for _,_,_,flags in evidence:239 quality|=flags240 partial=not complete or group.bad>0 or group.partials>0241 number,reason,flag=group.value(spec,*self._scales[spec])242 quality|=flag243 if not group.used and (not observed or not determinate or group.bad):244 number,status,reason=None,'indeterminate','no_observation' if not observed else 'insufficient_data'245 elif reason=='numeric_error' or reason=='insufficient_count':246 status='indeterminate'247 elif partial:248 status,reason='partial','partial_population'249 if spec.partial_policy=='strict':250 number=None251 else:252 status='empty' if number is None else 'complete'253 identity=dict(zip(spec.group_by,key))254 yield SummaryAggregationRow(sequence,identity.get('dataset_id',ds.dataset_id),identity.get('entity_id',ds.entity),255 identity.get('session_id',ds.session_id),spec.name,spec.source_kind,256 spec.source_id,spec.operation,spec.group_by,key,number,spec.unit,status,reason,group.total,257 group.used,group.excluded,group.bad,group.partials,group.matched,group.timeout,group.insufficient,258 0,complete,quality)259 sequence+=1260261 def close(self):262 """释放精确分位数和自己的分组费用，可重试。参数: 无。返回: None。异常: 原账本错误传播。"""263 if self._closed:264 return265 for group in self._groups.values():266 if group.quantile is not None:267 group.quantile.close()268 self._groups.clear()269 _release_owned_ram(self,self._group_cost,'_group_cost')270 self._closed=True271