chronaxis.results.threshold_metadata
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""保存一次阈值报警预设在实际观测范围内的时长汇总。23输入是同次迟滞状态机累计的整数纳秒，不再次读取信号或推算缺口内状态。4有未知时间时只报告已知报警累计，完整总时长为空；无观测不返回正常零。5快照只随有限预设数增长，由结果清单预算承担，重复发布不会重新计算。6"""78from dataclasses import dataclass9from math import isfinite10from ..core.models import Interval11from ..core.validation import require_identifier121314@dataclass(frozen=True)15class ThresholdDurationMetadata:16 """冻结报警、已知与未知时长，完整总数只在整个观测范围已知时给出。"""1718 preset_id: str # 本次计划的预设ID。19 rule_id: str # 实际生成的迟滞规则ID，可关联事件表。20 observed_start_ns: int | None # 实际时钟组起点，无观测为None。21 observed_end_ns: int | None # 实际时钟组末点，不向请求末端外推。22 known_active_duration_ns: int # 已知报警区间的准确整数纳秒和。23 known_duration_ns: int # 已知报警或未报警的总纳秒。2425 def __post_init__(self):26 """检查固定计数和观测范围的一致性。2728 参数: 无。返回: None。异常: TypeError/ValueError为身份、范围或计数矛盾。29 """30 require_identifier(self.preset_id, 'threshold.preset_id')31 require_identifier(self.rule_id, 'threshold.rule_id')32 if (self.observed_start_ns is None) != (self.observed_end_ns is None):33 raise ValueError('阈值汇总观测端点必须同时存在或同时为空')34 if self.observed_start_ns is not None:35 if type(self.observed_start_ns) is not int or type(self.observed_end_ns) is not int:36 raise TypeError('阈值汇总端点须原生整数纳秒')37 Interval(self.observed_start_ns, self.observed_end_ns)38 if any(type(value) is not int for value in (self.known_active_duration_ns, self.known_duration_ns)):39 raise TypeError('阈值汇总时长须原生整数纳秒')40 if not 0 <= self.known_active_duration_ns <= self.known_duration_ns <= self.observed_duration_ns:41 raise ValueError('报警/已知时长不得超过实际观测范围')4243 @property44 def observed_duration_ns(self):45 """取得实际半开范围长度。参数: 无。返回: int纳秒。异常: 无。"""46 return 0 if self.observed_start_ns is None else self.observed_end_ns - self.observed_start_ns4748 @property49 def unknown_duration_ns(self):50 """取得未确定报警状态的时长。参数: 无。返回: int纳秒。异常: 无。"""51 return self.observed_duration_ns - self.known_duration_ns5253 @property54 def total_duration_ns(self):55 """完整总时长不能由未知区间推测。参数: 无。返回: int或None。异常: 无。"""56 return self.known_active_duration_ns if self.observed_duration_ns > 0 and self.unknown_duration_ns == 0 else None5758 @property59 def status(self):60 """区分完整、部分、全未知和无观测。参数: 无。返回: 状态文本。异常: 无。"""61 if self.observed_duration_ns == 0:62 return 'no_observation'63 if self.known_duration_ns == 0:64 return 'indeterminate'65 return 'partial' if self.unknown_duration_ns else 'complete'6667 def to_dict(self):68 """将纳秒事实及便于阅读的秒数写入清单。6970 参数: 无。返回: 独立dict。异常: 无；调用方在复制前预付元数据费用。71 """72 return dict(preset_id=self.preset_id, rule_id=self.rule_id,73 observed_start_ns=self.observed_start_ns, observed_end_ns=self.observed_end_ns,74 observed_duration_ns=self.observed_duration_ns, known_duration_ns=self.known_duration_ns,75 known_active_duration_ns=self.known_active_duration_ns, unknown_duration_ns=self.unknown_duration_ns,76 total_duration_ns=self.total_duration_ns,77 total_duration_s=None if self.total_duration_ns is None else self.total_duration_ns / 10**9,78 known_active_duration_s=self.known_active_duration_ns / 10**9, status=self.status)798081@dataclass(frozen=True)82class SummaryThresholdDuration:83 """新合同只保存一个累计值和单位；partial说明它只是已知部分。"""8485 preset_id: str86 rule_id: str87 value: float | None88 status: str89 unit: str = 's'9091 def __post_init__(self):92 """检查唯一单位和未知总量。参数: 无。返回: None。异常: 非法身份/数值/状态。"""93 require_identifier(self.preset_id,'preset_id')94 require_identifier(self.rule_id,'rule_id')95 if self.value is not None and (type(self.value) not in (int,float) or not isfinite(self.value) or self.value<0):96 raise ValueError('报警时长必须为非负有限值或未知')97 if self.unit!='s' or self.status not in ('no_observation','indeterminate','partial','complete'):98 raise ValueError('报警汇总单位或状态无效')99 if (self.status in ('no_observation','indeterminate')) != (self.value is None):100 raise ValueError('无观测/全未知为空，完整或部分累计必须有可核实值')101102 def to_dict(self):103 """发布一种数值单位，不附重复ns/s或时间范围。参数: 无。返回: dict。异常: 无。"""104 return dict(preset_id=self.preset_id,rule_id=self.rule_id,unit=self.unit,105 value=self.value,status=self.status)106