chronaxis.execution.record_windows
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""从最终事件表生成窗口，不再复制一份事件阶段或重扫原输入。23每个窗口只保最后可能继续合并的父组及有限新父；记录周期借共享实际时刻索引。4边界扩展/合并继续使用既有两个独立窗口核，默认输出精简片段。5"""6from dataclasses import replace7from fractions import Fraction8from ..core.models import Interval9from ..core.temporal import Event,TemporalRule10from ..core.windows import WindowSpec11from ..core.windows_reference import build_segments_reference12from ..core.windows_numpy import build_segments_numpy13from ..results.summary import SummaryEventRow14from ..results.codec import decode_row15from ..core.types import QualityFlag as Q16from ..results.errors import ResultClosedError,ResultFormatError17from ..results.spool import _reserve_owned_ram,_release_owned_ram,_cleanup_owner18from .history import _metadata_bytes192021class RecordWindows:22 """拥有最终事件表读槽和有限待合并父组；共享时点索引、结果仓均为借用。"""2324 def __init__(self,plan,budget,indexes,steps,consumer):25 """预付声明，不读取输入或事件表。参数: 计划/预算/时点索引/周期/片段回调。返回: owner。异常: 原构造错。"""26 self._plan,self._budget,self._indexes,self._steps,self._consumer=plan,budget,indexes,steps,consumer27 self._held_cost=self._fixed_cost=self._parents_cost=self._reader_cost=self._workspace_cost=028 self._handle=None29 self._parents={}30 self._since={}31 self._observation={}32 self._specs={}33 self._rules={}34 self._closed=self._failed=False35 self._choices=dict(plan.intervals.resolved)36 self._coverage_needed={calculation.scope.split(":",1)[1] for calculation in plan.calculations37 if calculation.scope.startswith("segment:") and calculation.minimum_coverage>0}38 try:39 _reserve_owned_ram(self,131072+64*_metadata_bytes(plan.windows),'_fixed_cost')40 for request in plan.windows:41 self._parents[request.window_id]=[]42 self._since[request.window_id]=043 self._specs[request.window_id]=WindowSpec(request.window_id,request.event,request.anchor,44 request.pre_ns,request.post_ns,request.merge,request.exit_delay_ns)45 for rule in plan.rules:46 self._rules[rule.rule_id]=TemporalRule(rule.rule_id,plan.dataset.dataset_id,plan.dataset.session_id,47 rule.temporal_rule_type,rule.minimum_duration_ns or 0,rule.emit_initial,rule.window_duration_ns or 0,48 duration_comparison=rule.duration_comparison)49 self._build=build_segments_reference if plan.execution.backend=='reference' else build_segments_numpy50 except BaseException as error:51 _cleanup_owner(self,error,'record_windows_cleanup_owner','按需窗口构造')52 raise5354 def _bounds(self,window,event):55 """按所选周期定位实际边界；actual由原核直接扩展。参数: 窗口/事件。返回: 四元组或None。异常: 原查找错。"""56 step=self._steps[window.window_id]57 if step is None:58 return None59 driver=self._choices[window.window_id].driver60 if not self._indexes.count(driver):61 return event.start_ns,event.start_ns,True,True62 end=event.start_ns if event.end_ns is None else event.end_ns63 extension=window.exit_delay_ns if window.merge=='exit_delay' else window.post_ns64 left,left_clip=self._indexes.shift(driver,event.start_ns,-window.pre_ns,step)65 right,right_clip=self._indexes.shift(driver,end,extension,step)66 return left,right,left_clip,right_clip6768 def _coverage(self,window,segment,parents):69 """仅明确门槛请求才算窗口已取得部分比例，不保存各记录的第二时间。7071 参数: 窗口/结果/本组父事件。返回: 0—1或无法确定的None。异常: 原时间读取错。72 actual沿原请求秒数；周期模式按已选记录跨度与请求周期数比较。73 """74 if segment.actual is None:75 return 0.76 ids=set(segment.parent_event_ids)77 selected=tuple(parent for parent in parents if parent.event_id in ids)78 extension=window.exit_delay_ns if window.merge=='exit_delay' else window.post_ns79 step=self._steps[window.window_id]80 if step is None:81 left=min(parent.start_ns-window.pre_ns for parent in selected)82 right=max((parent.start_ns if parent.end_ns is None else parent.end_ns)+extension for parent in selected)83 return segment.actual.duration_ns()/(right-left)84 driver=self._choices[window.window_id].driver85 starts=[]86 ends=[]87 for parent in selected:88 start,outside=self._indexes.position(driver,parent.start_ns)89 end,outside_end=self._indexes.position(driver,parent.start_ns if parent.end_ns is None else parent.end_ns)90 if outside or outside_end:91 return None92 starts.append(start-Fraction(window.pre_ns,step))93 ends.append(end+Fraction(extension,step))94 actual_left,_=self._indexes.position(driver,segment.actual.start_ns)95 actual_right,_=self._indexes.position(driver,segment.actual.end_ns)96 length=max(ends)-min(starts)97 return 0. if length==0 else min(1.,max(0.,float((actual_right-actual_left)/length)))9899 def _process(self,window,final=False):100 """复用窗口核，非末批只保最后合并组，其他片段直接输出。101102 参数: 窗口/真实EOF标识。返回: None。异常: 预算/窗口/发布原错，责任保留到close。103 """104 name=window.window_id105 entries=self._parents[name]106 if not entries:107 return108 parents=bounds=segments=None109 primary=None110 try:111 # 入队费用是8192+32*事件元数据；工作副本仍收8192+16*元数据。112 # 从已核费用推导同一金额，不重新遍历长合并组中未变化的父事件。113 workspace = 65536 + sum(4096 + item[1] // 2 for item in entries)114 _reserve_owned_ram(self,workspace,'_workspace_cost')115 parents=tuple(item[0] for item in entries)116 bounds=None if self._steps[name] is None else tuple(item[2] for item in entries)117 segments=self._build(parents,self._specs[name],self._observation[name],bounds=bounds,compact=True)118 limit=len(segments) if final else max(0,len(segments)-1)119 for index in range(limit):120 segment=segments[index]121 if name in self._coverage_needed:122 segment=replace(segment,coverage=self._coverage(window,segment,parents))123 self._consumer(segment)124 segment=None125 keep=0126 if not final:127 first=segments[-1].parent_event_ids[0]128 keep=next(index for index,item in enumerate(entries) if item[0].event_id==first)129 else:130 keep=len(entries)131 cost=sum(item[1] for item in entries[:keep])132 del entries[:keep]133 parents=bounds=segments=None134 _release_owned_ram(self,cost,'_parents_cost')135 self._since[name]=0136 except BaseException as error:137 primary=error138 raise139 finally:140 parents=bounds=segments=None141 try:142 _release_owned_ram(self,self._workspace_cost,'_workspace_cost')143 except BaseException as cleanup:144 if primary is None:145 raise146 # 真实写出/计算错误优先；未退费用仍在本owner，交由父会话恢复。147 primary.add_note('窗口工作区清理附加失败：'+type(cleanup).__name__)148149 def finish(self,cursors,spool):150 """所有来源结束后只读一次最终事件表，并生成片段。151152 参数: 真正EOF的游标、同次最终结果仓。返回: None。异常: 缺范围/短读/损坏/预算/发布原错。153 未请求窗口时调用方不创建本对象，空事件不建立额外中间仓。154 """155 if self._closed or self._failed:156 raise ResultClosedError('窗口已关闭或失败')157 if not spool.counts.get('events',0):158 return159 row=event=payload=None160 try:161 definitions={item.definition.name:item.definition for item in self._plan.signals}162 rules={rule.rule_id:rule for rule in self._plan.rules}163 for window in self._plan.windows:164 names=set(rules[window.event].dependencies)165 driver=self._choices[window.window_id].driver166 if driver:167 names.add(driver)168 from ..planning.transforms import physical_dependencies169 names=physical_dependencies(self._plan,names)170 sources={definitions[name].source for name in names} if names else {source.name for source in self._plan.sources}171 starts=[c.observed_start for c in cursors if c.source_name in sources and c.observed_start is not None]172 ends=[c.observed_end for c in cursors if c.source_name in sources and c.observed_start is not None and c.observed_end is not None]173 if starts and ends:174 left=max(min(starts),self._plan.dataset.domain.start_ns)175 right=min(max(ends),self._plan.dataset.domain.end_ns)176 if left<=right:177 self._observation[window.window_id]=Interval(left,right)178 path=spool._seal_table('events')179 _reserve_owned_ram(self,spool.reader_cost_for('events'),'_reader_cost')180 self._handle=path.open('rb',buffering=8192)181 maximum=spool.row_limit('events')182 count=0183 while True:184 payload=self._handle.readline(maximum+1)185 if not payload:186 break187 if len(payload)>maximum or not payload.endswith(b'\n'):188 raise ResultFormatError('窗口事件输入短读或行超限')189 row=decode_row(payload,maximum)190 if type(row) is not SummaryEventRow:191 raise ResultFormatError('窗口需要同次精简事件表')192 count+=1193 event=Event(row.event_id,self._rules[row.rule_id],'point' if row.end_ns is None else 'activity',194 row.start_ns,row.confirmed_ns,row.end_ns,row.status,row.trigger,195 row.quality & ~Q.ESTIMATED if row.status in ('point','timeout','closed') else row.quality,row.quality)196 for window in self._plan.windows:197 if window.event!=row.rule_id:198 continue199 if window.window_id not in self._observation:200 raise ValueError('有事件却没有实际窗口观测范围')201 cost=8192+32*_metadata_bytes(event)202 _reserve_owned_ram(self,cost,'_parents_cost')203 bounds=self._bounds(window,event)204 self._parents[window.window_id].append((event,cost,bounds))205 self._since[window.window_id]+=1206 if self._since[window.window_id]>=min(self._plan.execution.batch_rows,64):207 self._process(window)208 row=event=payload=None209 if count!=spool.counts['events']:210 raise ResultFormatError('窗口事件表记录缺失')211 self._handle.close();self._handle=None212 _release_owned_ram(self,self._reader_cost,'_reader_cost')213 for window in self._plan.windows:214 self._process(window,final=True)215 except BaseException:216 self._failed=True217 raise218 finally:219 row=event=payload=None220221 @property222 def closed(self):223 """返回资源关闭状态。参数: 无。返回: bool。异常: 无。"""224 return self._closed225226 def close(self):227 """关闭读槽并释放剩余父组，不删除最终事件表。参数: 无。返回: None。异常: 原关闭错，可重试。"""228 if self._closed:229 return230 if self._handle is not None:231 self._handle.close();self._handle=None232 self._parents.clear()233 self._plan=self._consumer=self._indexes=None234 for slot in ('_parents_cost','_workspace_cost','_reader_cost','_fixed_cost'):235 _release_owned_ram(self,getattr(self,slot),slot)236 self._closed=True237