chronaxis.results.summary
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""定义按需执行的精简指标行，不要求不存在的完整输入证明。23只保留结果身份、业务数值、单位及完成/质量状态。事件或片段通过parent_id关联，4不逐行重复配置、信号定义、时间对照或贡献；公开结果只接受当前扁平模型。5本模型只校验结果字段，不读取输入或执行指标计算。6"""78from ..core.moments import MOMENTS9from ..core.quantiles import SAMPLE_QUANTILES10from dataclasses import dataclass11import json12from math import isfinite1314from ..core.types import QualityFlag as Q15from ..core.counts_v4 import COUNT_OPERATIONS, OBJECT_COUNTS16from ..core.validation import require_identifier, require_text17from ..timebase.conversion import require_int64_ns181920@dataclass(frozen=True)21class SummaryCalculationRow:22 """一条精简指标；未知用None，不能把缺失转换为正常零。"""2324 # 固定身份只用于关联本次结果，不代替原始输入哈希或逐条来源。25 sequence: int26 dataset_id: str27 entity_id: str28 session_id: str29 calculation_id: str30 # 算子及父范围给出业务含义；session没有父对象。31 operation: str32 scope: str33 parent_id: str | None34 # 数值与质量必须一起读取，None不等于有效零。35 value: int | float | None36 unit: str37 status: str38 reason: str39 quality: Q4041 def __post_init__(self):42 """核对身份、数值类型及状态，不补造来源或时间信息。4344 参数: 无。返回: None。异常: TypeError/ValueError为字段或组合错误。45 """46 if type(self.sequence) is not int or self.sequence < 0:47 raise ValueError('精简指标序号需要非负原生int')48 for name in ('dataset_id', 'entity_id', 'session_id', 'calculation_id'):49 require_identifier(getattr(self, name), 'summary.'+name)50 if type(self) is not SummaryHostedCalculationRow and self.operation not in ('integral', 'duration', 'min_time', 'max_time', 'time_weighted_mean',51 'sample_minimum', 'sample_maximum', 'sample_mean', 'legacy_sum', *COUNT_OPERATIONS, *MOMENTS, *SAMPLE_QUANTILES):52 raise ValueError('精简指标不支持该算子')53 # 即使外部重新读取结果行，计数也必须是无量纲定宽整数，且对象归属整次会话。54 if self.operation in COUNT_OPERATIONS and (self.unit != '1' or self.value is not None55 and (type(self.value) is not int or not 0 <= self.value < 1 << 64)):56 raise ValueError('计数结果需要单位1及可用的uint64整数或None')57 if self.operation in OBJECT_COUNTS and self.scope != 'session':58 raise ValueError('对象计数结果必须属于session')59 if self.scope not in ('session', 'event', 'segment'):60 raise ValueError('精简指标scope无效')61 if self.scope == 'session':62 if self.parent_id is not None:63 raise ValueError('会话指标不能声明父事件或片段')64 else:65 require_text(self.parent_id, 'summary.parent_id')66 if self.value is not None and (type(self.value) not in (int, float)67 or type(self.value) is float and not isfinite(self.value)):68 raise ValueError('精简指标值需要有限数值或None')69 require_text(self.unit, 'summary.unit')70 require_text(self.reason, 'summary.reason')71 if self.status not in ('complete', 'partial', 'indeterminate'):72 raise ValueError('精简指标status无效')73 if self.status == 'indeterminate' and self.value is not None:74 raise ValueError('无法确定的指标不能含可用数值')75 if not isinstance(self.quality, Q):76 raise TypeError('精简指标质量需要QualityFlag')777879SUMMARY_FIELDS = ('sequence', 'dataset_id', 'entity_id', 'session_id', 'calculation_id', 'operation',80 'scope', 'parent_id', 'value', 'unit', 'status', 'reason', 'quality')818283@dataclass(frozen=True)84class SummaryHostedCalculationRow(SummaryCalculationRow):85 """宿主归约产生的纯数据行；读取版本和标识不允许动态执行插件。"""8687 extension_version: str88 coverage_status: str | None = None8990 def __post_init__(self):91 """验证固定行协议和清晰的宿主身份。参数: 无。返回: None。异常: 身份或覆盖状态错误。"""92 super().__post_init__()93 require_identifier(self.operation,'summary.hosted_operation')94 require_text(self.extension_version,'summary.extension_version')95 if self.coverage_status not in (None,'sufficient','indeterminate'):96 raise ValueError('宿主归约覆盖判定无效')979899@dataclass(frozen=True)100class SummaryCoverageCalculationRow(SummaryCalculationRow):101 """仅明确设置覆盖门槛时补一个判定，不重复输出时长、比例或输入明细。"""102103 coverage_status: str104105 def __post_init__(self):106 """保留数值政策与覆盖门槛两个不同事实。参数: 无。返回: None。异常: 非法判定。"""107 super().__post_init__()108 if self.coverage_status not in ('sufficient','indeterminate'):109 raise ValueError('覆盖门槛判定无效')110111112def _require_row_identity(row):113 """核精简事件和配对的公共身份。参数: 冻结行。返回: None。异常: 类型或字段错误。"""114 if type(row.sequence) is not int or row.sequence < 0:115 raise ValueError('精简结果序号需要非负原生int')116 for name in ('dataset_id','entity_id','session_id','rule_id'):117 require_identifier(getattr(row,name),'summary.'+name)118 if not isinstance(row.quality,Q):119 raise TypeError('精简结果质量需要QualityFlag')120121122@dataclass(frozen=True)123class SummaryEventRow:124 """一条业务事件；开始、确认、结束各有不同含义，不含原帧或逐条来源副本。"""125126 sequence: int127 dataset_id: str128 entity_id: str129 session_id: str130 rule_id: str131 event_id: str132 start_ns: int133 confirmed_ns: int134 end_ns: int | None135 status: str136 trigger: str137 quality: Q138139 def __post_init__(self):140 """核实际业务端点，不能把未知结束填为零。参数: 无。返回: None。异常: 合同错误。"""141 _require_row_identity(self)142 require_text(self.event_id,'event_id')143 start=require_int64_ns(self.start_ns,'start_ns')144 confirmed=require_int64_ns(self.confirmed_ns,'confirmed_ns')145 if confirmed < start:146 raise ValueError('确认不能早于开始')147 if self.status == 'point':148 if self.end_ns is not None or start != confirmed:149 raise ValueError('点事件没有活动结束时间')150 elif self.status in ('closed','interrupted','truncated','timeout'):151 end=require_int64_ns(self.end_ns,'end_ns')152 if not start < end or confirmed > end:153 raise ValueError('活动实际端点顺序无效')154 else:155 raise ValueError('精简事件状态无效')156 require_text(self.trigger,'trigger')157158159@dataclass(frozen=True)160class SummarySequenceRow:161 """一条配对结论，只关联既有事件ID并保存所选口径的延迟。"""162163 sequence: int164 dataset_id: str165 entity_id: str166 session_id: str167 rule_id: str168 result_id: str169 request_id: str170 response_id: str | None171 decided_ns: int172 delay_ns: int | None173 status: str174 quality: Q175176 def __post_init__(self):177 """核配对及未知结果，不逐行重复规则和事件。参数: 无。返回: None。异常: 合同错误。"""178 _require_row_identity(self)179 for name in ('result_id','request_id'):180 require_text(getattr(self,name),name)181 require_int64_ns(self.decided_ns,'decided_ns')182 if self.status == 'matched':183 require_text(self.response_id,'response_id')184 if require_int64_ns(self.delay_ns,'delay_ns') < 0:185 raise ValueError('配对延迟不能为负')186 elif self.status in ('no_response','indeterminate'):187 if self.response_id is not None or self.delay_ns is not None:188 raise ValueError('未匹配结果不能伪造响应和延迟')189 if self.status == 'indeterminate' and self.quality == Q.NONE:190 raise ValueError('未知配对必须说明质量原因')191 if self.status == 'no_response' and self.quality != Q.NONE:192 raise ValueError('缺失观测不能声明确定未响应')193 else:194 raise ValueError('精简配对状态无效')195196197@dataclass(frozen=True)198class SummarySegmentRow:199 """片段只输出实际边界、状态和父事件ID，不重复请求边界/原事件/来源。"""200 sequence: int201 dataset_id: str202 entity_id: str203 session_id: str204 window_id: str205 segment_id: str206 start_ns: int | None207 end_ns: int | None208 status: str209 parent_event_ids: tuple[str,...]210 quality: Q211212 def __post_init__(self):213 """检查实际范围和必需父关联。参数: 无。返回: None。异常: 字段/组合错误。"""214 if type(self.sequence) is not int or self.sequence<0:215 raise ValueError('片段sequence需非负原生int')216 for name in ('dataset_id','entity_id','session_id','window_id'):217 require_identifier(getattr(self,name),name)218 require_text(self.segment_id,'segment_id')219 if not isinstance(self.quality,Q) or type(self.parent_event_ids) is not tuple or not self.parent_event_ids:220 raise TypeError('片段需质量及非空父ID tuple')221 if len(set(self.parent_event_ids))!=len(self.parent_event_ids):222 raise ValueError('片段父ID不能重复')223 for parent in self.parent_event_ids:224 require_text(parent,'parent_event_id')225 if self.status=='empty':226 if self.start_ns is not None or self.end_ns is not None:227 raise ValueError('空片段不能伪造实际范围')228 elif self.status in ('complete','partial'):229 if require_int64_ns(self.start_ns)>=require_int64_ns(self.end_ns):230 raise ValueError('片段需要正实际范围')231 else:232 raise ValueError('片段状态错误')233234235SUMMARY_EVENT_FIELDS = ('sequence','dataset_id','entity_id','session_id','rule_id','event_id',236 'start_ns','confirmed_ns','end_ns','status','trigger','quality')237SUMMARY_SEQUENCE_FIELDS = ('sequence','dataset_id','entity_id','session_id','rule_id','result_id',238 'request_id','response_id','decided_ns','delay_ns','status','quality')239SUMMARY_MODELS = {SummaryCalculationRow: ('calculations',SUMMARY_FIELDS,7),240 SummaryEventRow: ('events',SUMMARY_EVENT_FIELDS,8), SummarySequenceRow: ('sequences',SUMMARY_SEQUENCE_FIELDS,9)}241SUMMARY_MODELS[SummarySegmentRow]=('segments',('sequence','dataset_id','entity_id','session_id','window_id',242 'segment_id','start_ns','end_ns','status','parent_event_ids','quality'),11)243@dataclass(frozen=True)244class SummaryContributionRow:245 """一条明确请求的逐消息贡献；仅定位计算，不复制原帧或输入值。"""246247 sequence: int248 dataset_id: str249 entity_id: str250 session_id: str251 calculation_id: str252 source: str253 original_sequence: int254 evaluation_index: int255 value: float | None256 reason: str257 quality: Q258259 def __post_init__(self):260 """核定位及贡献/跳过互斥。参数: 无。返回: None。异常: 类型或字段错误。"""261 for name in ('sequence','original_sequence','evaluation_index'):262 if type(getattr(self,name)) is not int or getattr(self,name)<0:263 raise ValueError('贡献定位需非负原生整数')264 for name in ('dataset_id','entity_id','session_id','calculation_id','source'):265 require_identifier(getattr(self,name),name)266 if not isinstance(self.quality,Q):267 raise TypeError('贡献质量需QualityFlag')268 if self.value is None:269 if self.reason not in ('missing_input','condition_false','evaluation_error','nonfinite_input'):270 raise ValueError('跳过贡献需要明确原因')271 elif type(self.value) is not float or not isfinite(self.value) or self.reason!='contributed':272 raise ValueError('有效贡献需要有限float及contributed状态')273274275SUMMARY_MODELS[SummaryContributionRow]=('contributions',('sequence','dataset_id','entity_id','session_id',276 'calculation_id','source','original_sequence','evaluation_index','value','reason','quality'),12)277@dataclass(frozen=True)278class SummaryInputRow:279 """按请求保存一次所需输入；字段原值一份，不保存双时间表示或原帧副本。"""280 sequence: int281 dataset_id: str282 entity_id: str283 session_id: str284 source: str285 original_sequence: int286 time_ns: int | None287 record_type: str288 locator: str289 values_json: str290 quality: Q291292 def __post_init__(self):293 """核定位与有限JSON标量，拒绝任意嵌套模型。参数: 无。返回: None。异常: 字段/格式错误。"""294 for name in ('sequence','original_sequence'):295 if type(getattr(self,name)) is not int or getattr(self,name)<0:raise ValueError('追溯序号需非负int')296 for name in ('dataset_id','entity_id','session_id','source'):require_identifier(getattr(self,name),name)297 if self.time_ns is not None:require_int64_ns(self.time_ns)298 if self.record_type not in ('decoded_message','signal_sample','signal_clock') or not isinstance(self.quality,Q):299 raise ValueError('追溯类型/质量无效')300 require_text(self.locator,'trace.locator')301 if type(self.values_json) is not str:raise TypeError('追溯值需JSON文本')302 values=json.loads(self.values_json)303 if type(values) is not dict:raise ValueError('追溯值需字段对象')304 for name,value in values.items():305 require_text(name,'trace.field')306 if type(value) is dict:307 if set(value)!= {'nonfinite'} or value['nonfinite'] not in ('nan','inf','-inf'):308 raise ValueError('追溯只允许明确非有限值标签')309 elif value is not None and (type(value) not in (str,int,float,bool)310 or type(value) is float and not isfinite(value)):311 raise ValueError('追溯字段需有限标量或明确非有限标签')312313314SUMMARY_MODELS[SummaryInputRow]=('input_trace',('sequence','dataset_id','entity_id','session_id',315 'source','original_sequence','time_ns','record_type','locator','values_json','quality'),13)316@dataclass(frozen=True)317class SummarySelectionRow:318 """一条真实更新的选择结果；保业务值及三值选择，不重复输入证明。"""319 sequence: int320 dataset_id: str321 entity_id: str322 session_id: str323 calculation_id: str324 scope: str325 parent_id: str | None326 time_ns: int327 value: int | float | bool | str | None328 dtype: str329 unit: str330 valid: bool331 selection: str332 quality: Q333334 def __post_init__(self):335 """核选择行身份和值类型。参数: 无。返回: None。异常: 类型/结构错误。"""336 if type(self.sequence) is not int or self.sequence<0:raise ValueError('选择序号需要非负int')337 for name in ('dataset_id','entity_id','session_id','calculation_id'):require_identifier(getattr(self,name),name)338 if self.scope not in ('session','event','segment'):raise ValueError('选择范围类型无效')339 if self.scope=='session':340 if self.parent_id is not None:raise ValueError('会话选择不能声明父对象')341 else:require_text(self.parent_id,'selection.parent_id')342 require_int64_ns(self.time_ns);require_text(self.unit,'selection.unit')343 kinds={'float64':float,'int64':int,'uint64':int,'bool':bool,'enum':str}344 if self.dtype not in kinds or type(self.valid) is not bool or not isinstance(self.quality,Q):345 raise TypeError('选择类型/有效性/质量错误')346 if self.selection not in ('true','unknown') or self.selection=='unknown' and self.quality==Q.NONE:347 raise ValueError('选择仅支持TRUE或有质量原因的UNKNOWN')348 if self.valid:349 if type(self.value) is not kinds[self.dtype] or type(self.value) is float and not isfinite(self.value):350 raise ValueError('选择有效值与dtype不符')351 if self.dtype=='int64' and not -(1<<63)<=self.value<(1<<63) or self.dtype=='uint64' and not 0<=self.value<(1<<64):352 raise ValueError('选择整数超出dtype')353 elif self.value is not None or self.quality==Q.NONE:354 raise ValueError('选择无效值必须为空并保留质量原因')355356357SUMMARY_MODELS[SummarySelectionRow]=('selections',('sequence','dataset_id','entity_id','session_id','calculation_id',358 'scope','parent_id','time_ns','value','dtype','unit','valid','selection','quality'),14)359from .aggregation_row import SummaryAggregationRow, AGGREGATION_FIELDS360SUMMARY_MODELS[SummaryAggregationRow] = ('aggregations',AGGREGATION_FIELDS,15)361SUMMARY_COLUMNS = {table: names for table,names,_ in SUMMARY_MODELS.values()}362SUMMARY_MODELS[SummaryCoverageCalculationRow] = ('calculations',SUMMARY_FIELDS+('coverage_status',),10)363SUMMARY_MODELS[SummaryHostedCalculationRow] = ('calculations',SUMMARY_FIELDS+('extension_version','coverage_status'),16)364365366def summary_fields(row):367 """转换必要平坦字段，CSV和JSONL共用，不附另一份row_json。368369 参数: SummaryCalculationRow。返回: 固定字段dict。异常: TypeError为错误根类型。370 """371 try:372 if type(row) not in SUMMARY_MODELS:373 raise TypeError('精简字段需要SummaryCalculationRow')374 result = {name: getattr(row, name) for name in SUMMARY_MODELS[type(row)][1]}375 result['quality'] = int(row.quality)376 return result377 finally:378 row = None379