chronaxis.execution.record_rule_timeline
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用最新必要信号在线推进一条规则，不保存输入历史或第二时间轴。23只暂留一个完整条件组，等下一组或真实EOF确定其是否为末点。实际模式调度4必要TTL和规则截止；记录周期模式只在选定驱动更新时增加等待量。输出直接5交调用方结果消费者，输入/共享状态和结果仓不归本对象所有。6"""78from dataclasses import replace9from ..core.temporal import ConditionTick, TemporalRule10from ..core.types import Truth as T, QualityFlag as Q11from ..results.errors import ResultClosedError12from ..results.spool import _reserve_owned_ram, _release_owned_ram, _cleanup_owner13from .driver_expression import DriverExpression14from .temporal_buffer import BudgetedTemporalDetector15from .history import _metadata_bytes16from .record_alarm_duration import RecordAlarmDuration17from .record_delta_condition import RecordDeltaCondition181920class RecordRuleTimeline:21 """拥有规则表达式、检测器和一个待定条件组；原始组由父会话原子提交。"""2223 def __init__(self,plan,request,state,budget,step_ns,driver,observation_ends,consumer,step_consumer=None):24 """建立有限声明，不读来源，也不把请求域当作已观测范围。2526 参数: plan/request同计划规则、state共享最新值、budget共享预算、step_ns27 为None或周期，driver为所选别名，observation_ends借用EOF终点，consumer接事件。28 返回: 新owner。异常: 原构造错误附record_rule_cleanup_owner。29 """30 self._plan,self._request,self._state,self._budget=plan,request,state,budget31 self._step,self._driver,self._ends,self._consumer=step_ns,driver,observation_ends,consumer32 self._step_consumer=step_consumer33 self._held_cost=self._fixed_cost=034 self._expressions=[]35 self._detector=self._pending_owner=self._batch=None36 self._alarm=self._delta=None37 self._delta_emitted=False38 # 常量大小的业务概况，不保留条件轨迹；关闭仍由原固定预算负责。39 self._seen_true=self._seen_false=self._seen_unknown=False40 # 仅对象计数请求需要发现缺口的原因，并忽略EOF人为封口的UNKNOWN状态。41 self._count_quality = Q.NONE42 self._track_count = any(c.count_target == request.rule_id for c in plan.calculations43 if c.operation in ('point_event_count', 'activity_count')) or any(44 w.event == request.rule_id and any(c.operation == 'segment_count' and c.count_target == w.window_id45 for c in plan.calculations) for w in plan.windows)46 self._event_count=047 if plan.aggregations and not self._track_count:48 from .result_aggregations import discovery_rules49 self._track_count=any(request.rule_id in discovery_rules(plan,agg) for agg in plan.aggregations)50 self._required_aliases=set()51 self._pending=None52 self._next_expiry=self._last_driver=None53 self._start=None54 self._end=plan.dataset.domain.end_ns55 self._full_domain=not plan.dataset.domain_explicit56 self._closed=self._closing=self._failed=self._finished=False57 try:58 _reserve_owned_ram(self,131584+32*_metadata_bytes(request),'_fixed_cost')59 for expression in request.expressions:60 try:61 owner=DriverExpression(expression,state,None,backend=plan.execution.backend,62 derived=tuple(item.definition for item in plan.derived),observation_ends=observation_ends)63 except BaseException as error:64 self._pending_owner=getattr(error,'driver_expression_cleanup_owner',None)65 raise66 self._expressions.append(owner)67 if not owner._has_where:68 self._required_aliases.update(expression.dependencies)69 if request.rule_type=='window_delta':70 try:71 self._delta=RecordDeltaCondition(budget,request.window_delta,72 request.condition.root.dtype,plan.execution.backend,step_ns)73 except BaseException as error:74 self._delta=getattr(error,'record_delta_cleanup_owner',None)75 raise76 from ..planning.transforms import physical_dependencies77 names=physical_dependencies(plan,set(request.dependencies)|({driver} if driver else set()))78 self.sources=frozenset(state.definitions[name].source for name in names)79 if not self.sources:80 self.sources=frozenset(source.name for source in plan.sources)81 if request.stability is not None and request.stability.track_duration:82 try:83 self._alarm=RecordAlarmDuration(budget,step_ns,plan.execution.backend,plan.dataset.domain)84 except BaseException as error:85 self._alarm=getattr(error,'record_alarm_cleanup_owner',None)86 raise87 except BaseException as error:88 _cleanup_owner(self,error,'record_rule_cleanup_owner','在线规则构造')89 raise9091 def extend_observation(self, end_ns):92 """更新单遍读取已见的末端；最后条件组仍等待后继或EOF。参数: 实际末端。返回: None。异常: 回退。"""93 if not self._full_domain or self._finished:94 return95 self._end=end_ns96 if self._detector is not None:97 self._detector.extend_observed_end(end_ns)98 if self._alarm is not None:99 self._alarm._window=self._plan.dataset.domain100101 def _open(self):102 """拒绝失效实例。参数: 无。返回: None。异常: ResultClosedError。"""103 if self._closed or self._closing or self._failed:104 raise ResultClosedError('在线规则已关闭或失败')105106 def begin(self,observed_start):107 """用实际观察起点建立检测器，空来源由父跳过。108109 参数: observed_start实际首端。返回: None。异常: 构造/范围错误附可清理owner。110 """111 self._open()112 if self._start is not None:113 raise ValueError('在线规则不能重复开始')114 start=max(observed_start,self._plan.dataset.domain.start_ns)115 if start>self._end:116 self._finished=True117 return118 request=self._request119 rule=TemporalRule(request.rule_id,self._plan.dataset.dataset_id,self._plan.dataset.session_id,120 request.temporal_rule_type,request.minimum_duration_ns or 0,request.emit_initial,request.window_duration_ns or 0,121 duration_comparison=request.duration_comparison)122 try:123 self._detector=BudgetedTemporalDetector(self._budget,rule,start,self._end,124 backend=self._plan.execution.backend,125 stability=replace(request.stability,track_duration=False) if self._alarm is not None else request.stability,126 start_end=request.start_end,127 record_clock=self._step is not None)128 self._start=start129 except BaseException as error:130 self._detector=getattr(error,'temporal_detector_cleanup_owner',None)131 self._failed=True132 error.record_rule_cleanup_owner=self133 raise134135 def _snapshot(self,time_ns,actual,terminal=False):136 """收齐进出条件的点/右侧值，保一个必要组。参数: 实际时间/更新/EOF。返回: None。异常: 求值原错。"""137 ticks=[]138 deadlines=[]139 selected=self._state.get(self._driver) if self._driver else None140 is_driver=actual and selected is not None and selected.time_ns==time_ns141 for expression in self._expressions:142 evidence=self._required_aliases if expression._has_where else None143 point,_=expression.evaluate_at(time_ns,continuous=False,unobserved=evidence)144 right,deadline=expression.evaluate_at(time_ns,unobserved=evidence if not terminal else None)145 if self._delta is not None:146 point,right=self._delta.update(time_ns,point,right,deadline,driver=is_driver)147 deadline=self._delta.next_deadline148 point_state=T.TRUE if point.valid and point.value else T.FALSE if point.valid else T.UNKNOWN149 right_state=T.TRUE if right.valid and right.value else T.FALSE if right.valid else T.UNKNOWN150 point_quality=Q.NONE if point.valid else point.quality151 quality=Q.NONE if right.valid else right.quality152 observed_end=any(self._ends.get(source)==time_ns for source in self.sources)153 if observed_end and expression._has_where:154 observed_end=expression.active_end_at(time_ns)155 updated=self._delta is not None or (actual and expression.actual_update_at(time_ns))156 actual_unknown=updated and not point.valid157 if actual_unknown and expression._has_where and self._delta is None:158 actual_unknown=expression.actual_unknown_at(time_ns,point)159 ticks.append(ConditionTick(time_ns,T.UNKNOWN if terminal else right_state,160 quality|Q.MISSING if terminal else quality,actual_update=updated,161 observation_end=observed_end,terminal=terminal,point_state=point_state,point_quality=point_quality,162 actual_unknown=actual_unknown,diagnostic_quality=point.quality|right.quality|163 (Q.ESTIMATED if point.partial or right.partial and not terminal else Q.NONE)))164 if deadline is not None:165 deadlines.append(deadline)166 advance=None167 is_driver=False168 if self._step is not None:169 selected=self._state.get(self._driver)170 is_driver=actual and selected is not None and selected.time_ns==time_ns171 advance=self._step if is_driver and self._last_driver is not None else 0172 if is_driver:173 self._last_driver=time_ns174 self._pending=(tuple(ticks),advance,is_driver)175 self._next_expiry=min(deadlines) if deadlines else None176177 def _commit(self):178 """交付一个完整条件组和增量事件，消费失败不返回前缀。参数: 无。返回: None。异常: 原错。"""179 ticks,advance,driver=self._pending180 self._batch=self._detector.push((ticks[0],),None if len(ticks)==1 else (ticks[1],),181 advances=None if advance is None else (advance,))182 events=self._batch.events183 if self._delta is not None and self._request.retrigger=='once':184 if self._delta_emitted:185 events=()186 elif events:187 self._delta_emitted=True188 # 真实末点可以说明当时条件，但EOF人为补出的UNKNOWN不算数据缺口。189 tick=ticks[0]190 truths=(tick.point_state,) if tick.terminal else (tick.state,tick.point_state)191 self._seen_true |= T.TRUE in truths192 if self._track_count:193 for item in ticks:194 self._count_quality |= item.diagnostic_quality & Q.ESTIMATED195 for item in ticks:196 if item.point_state == T.UNKNOWN:197 self._count_quality |= item.point_quality198 if not item.terminal and item.state == T.UNKNOWN:199 self._count_quality |= item.quality200 self._seen_false |= T.FALSE in truths201 self._seen_unknown |= any(T.UNKNOWN in ((item.point_state,) if item.terminal202 else (item.state,item.point_state)) for item in ticks)203 self._event_count += len(events)204 primary=None205 try:206 if self._alarm is not None:207 state=T.UNKNOWN if ticks[0].terminal else self._detector._detector.state208 self._alarm.consume(ticks[0].time_ns,state)209 if self._step_consumer is not None:210 self._step_consumer(self._request.rule_id,ticks[0],events)211 for event in events:212 self._consumer(event)213 except BaseException as error:214 primary=error215 raise216 finally:217 event=ticks=None218 try:219 self._batch.close()220 self._batch=None221 except BaseException as cleanup:222 if primary is None:223 raise224 primary.add_note('规则输出批清理失败：'+type(cleanup).__name__)225 self._pending=None226227 def _drain_before(self,time_ns):228 """只推进已经由后续观测证明存在的区间。参数: 下个实际边界。返回: None。异常: 原推进错。"""229 if self._pending is None and self._detector._detector.last_time is None and self._start<time_ns:230 self._snapshot(self._start,False)231 while True:232 if self._pending is not None:233 if self._pending[0][0].time_ns>=time_ns:234 return235 self._commit()236 next_times=[value for value in (self._next_expiry,self._detector.next_deadline_ns) if value is not None]237 if not next_times or min(next_times)>=time_ns:238 return239 self._snapshot(min(next_times),False)240241 def before_group(self,time_ns):242 """父提交新组前处理旧状态上的TTL和实际等待。参数: 新组时间。返回: None。异常: 原错。"""243 self._open()244 if self._finished or self._start is None or time_ns<self._start:245 return246 try:247 if time_ns>self._end:248 self.finish(self._end)249 else:250 self._drain_before(time_ns)251 except BaseException as error:252 self._failed=True253 error.record_rule_cleanup_owner=self254 raise255256 def after_group(self,time_ns):257 """父原子提交后保存新条件组；实际终点在同刻更新之后结算。258259 参数: 新组时间。返回: None。异常: 原求值/事件错误，失败只能关闭。260 """261 self._open()262 if self._finished or self._start is None or time_ns<self._start:263 return264 try:265 terminal=time_ns==self._end and not self._full_domain266 self._snapshot(time_ns,True,terminal=terminal)267 if terminal:268 self._commit()269 self._batch=self._detector.finish()270 self._batch.close();self._batch=None271 self._finished=True272 except BaseException as error:273 self._failed=True274 error.record_rule_cleanup_owner=self275 raise276277 def clock_group(self,group):278 """待定旧条件已经交付后，接收报警累计所需的完整驱动对。279280 参数: 新的真实同刻组。返回: None。异常: 累计顺序/预算错误。281 不改变事件坐标，也不把请求域外样本算入报警；只借它们确定裁剪分母。282 """283 if self._alarm is not None and self._step is not None and any(s.name==self._driver for s in group.samples):284 self._alarm.driver(group.time_ns)285286 def finish(self,observed_end):287 """所有必要来源EOF后按真实末端封口，不外推到请求结束。288289 参数: observed_end必要来源联合末端。返回: None。异常: 原错，未完成不发布结果。290 """291 self._open()292 if self._finished or self._start is None:293 return294 try:295 end=min(observed_end,self._end)296 if end<self._start:297 self._finished=True298 return299 self._detector.set_observed_end(end)300 self._end=end301 self._drain_before(end)302 if self._pending is not None and self._pending[0][0].time_ns==end:303 ticks,advance,driver=self._pending304 self._pending=(tuple(replace(t,terminal=True,state=T.UNKNOWN,quality=t.quality|Q.MISSING) for t in ticks),advance,driver)305 else:306 self._snapshot(end,False,terminal=True)307 self._commit()308 self._batch=self._detector.finish()309 self._batch.close();self._batch=None310 self._finished=True311 except BaseException as error:312 self._failed=True313 error.record_rule_cleanup_owner=self314 raise315316 def outcome(self):317 """冻结规则结果原因，不读取输入。参数: 无。返回: RuleOutcome。异常: 未完成或已关闭。318319 有事件时始终保留计数；缺信号和UNKNOWN不被解释成条件为假。320 """321 self._open()322 if not self._finished and self._start is not None:323 raise ValueError('规则未结束，不能报告结果概况')324 from ..results.analysis_outcome import RuleOutcome325 # 无观测只报告实际求值需要且一直没有观测的别名。静态绑定仍校验两个分支。326 missing=tuple(name for name in self._request.dependencies if self._state.get(name) is None327 and (self._start is None or name in self._required_aliases))328 # 已产生的事件与未观测/未知事实并存；无事件时才用数据充分性解释原因。329 if self._event_count:330 status='events_found'331 elif self._start is None or missing:332 status='no_observation'333 elif self._seen_unknown:334 status='insufficient_data'335 elif self._request.rule_type=='condition_duration' and self._seen_true:336 status='duration_not_met'337 elif not self._seen_true:338 status='condition_not_met'339 else:340 status='no_event'341 return RuleOutcome(self._request.rule_id,status,self._event_count,self._seen_unknown,missing)342343 def count_evidence(self, domain):344 """判定规则实际观测是否覆盖整个目标会话，保留发现是否完整的证据。345346 参数: domain为最终会话域。返回: 已观测、至少部分可判定、完整、质量位。347 异常: 规则尚未结束或已关闭时沿outcome检查传播。348 """349 outcome = self.outcome()350 observed = self._start is not None and outcome.status != 'no_observation'351 known = self._seen_true or self._seen_false352 # 已确认的末点可以产生点事件；显式请求域超出来源实际首末时仍有缺口。353 covers_request = (observed and domain is not None354 and self._start <= domain.start_ns and self._end >= domain.end_ns)355 complete = covers_request and not self._seen_unknown and not outcome.unobserved_signals and not self._count_quality & Q.ESTIMATED356 quality = self._count_quality357 if not covers_request or outcome.unobserved_signals:358 quality |= Q.MISSING359 return observed, known, complete, quality360361 @property362 def duration_result(self):363 """借出明确请求的最终累计；空来源返回无观测。参数: 无。返回: tuple或None。异常: 未完成错误。"""364 if self._alarm is None:365 return None366 if not self._finished and self._start is not None:367 raise ValueError('报警尚未完成')368 return self._alarm.finish()369370 @property371 def closed(self):372 """返回资源关闭状态。参数: 无。返回: bool。异常: 无。"""373 return self._closed374375 def close(self):376 """释放自有批、表达式和检测器，保留失败槽供重试。参数: 无。返回: None。异常: 原关闭错。"""377 if self._closed:378 return379 self._closing=True380 for slot in ('_batch','_pending_owner','_detector','_alarm','_delta'):381 owner=getattr(self,slot)382 if owner is not None:383 owner.close()384 setattr(self,slot,None)385 for index,owner in enumerate(self._expressions):386 if owner is not None:387 owner.close();self._expressions[index]=None388 self._pending=self._consumer=self._step_consumer=self._state=self._plan=self._ends=self._required_aliases=None389 _release_owned_ram(self,self._fixed_cost,'_fixed_cost')390 self._closed=True391