chronaxis.execution.transform_stream
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在原始记录流及下游规则之间插入按需的时序变换。23每一层只持有自己的最新输入、滚动窗口，以及固定周期等待后帧确认时尚未4交付的有界原组。待同一时刻的原更新与变换值均齐全，再原子交给下层。5无全会话网格、输入重扫或对照时间列；预算、取消和真实EOF沿主执行器。6"""7from collections import deque8from dataclasses import replace9from fractions import Fraction10from ..core.numeric import ScalarValue11from ..core.types import QualityFlag as Q, SignalDType as D12from ..core.models import SourceRef13from ..results.budget import _reserve_owned_ram, _release_owned_ram, _cleanup_owner14from .driver_expression import DriverExpression15from .groups import DecodedTimeGroup16from .history_records import estimate_record_bytes17from .record_sample import RecordSample18from .state import LatestSignalState19from .transform_window import TransformSpan, TransformWindow, first_grid202122class TransformStage:23 """拥有一层的状态与前瞻，向唯一后继按实际时间递增地交付组。"""2425 def __init__(self, plan, item, defs, choice, period, ends, cursors, budget, emit):26 """参数: 计划/变换、有效定义、原时钟/周期、EOF/游标、预算及后继。返回: owner。异常: 预算/绑定错误。"""27 self.item,self.backend,self._budget,self.emit=item,plan.execution.backend,budget,emit28 self._held_cost=self._fixed_cost=self._queue_cost=self._work_cost=029 self.state=self.driver=self.history=None30 self.host = None31 self.host_finished = False32 self.waiting=deque()33 self.epoch=self.next_grid=self.last=None34 self.last_point=self.last_right=self.last_deadline=None35 self.last_update=False36 self.adjacent=self.clock_start=None37 self.adjacent_weight=Fraction(0)38 self.adjacent_known=True39 self.sequence=040 self.ends,self.cursors=ends,cursors41 self.period,self.clock=period,choice.driver42 self.sources={plan.bindings.resolve(n).source for n in item.physical}43 if not self.sources:44 self.sources=set(cursors)45 self.domain=plan.bindings.domain46 self.ref=SourceRef(plan.dataset.session_id,item.output.source,item.output.field)47 self.names=set(item.expression.dependencies)|({self.clock} if self.clock else set())48 self.closed=self.failed=False49 try:50 _reserve_owned_ram(self,65536+1024*len(self.names),'_fixed_cost')51 self.state=LatestSignalState(budget,tuple(d for d in defs if d.name in self.names),plan.dataset.session_id)52 # 时钟决定区间长度，数值始终按输入表达式的真实更新独立推进。53 self.driver=DriverExpression(item.expression,self.state,None,backend=self.backend,54 derived=tuple(d.definition for d in plan.derived),observation_ends=ends)55 self.history=TransformWindow(budget,item.window_ns or item.max_gap_ns,period)56 if item.host_spec is not None:57 from ..extension_runtime import HostedTransform58 self.host = HostedTransform(item.host_spec,self.backend,budget)59 except BaseException as error:60 if self.host is None and getattr(error,'extension_cleanup_owner',None) is not None:61 self.host = error.extension_cleanup_owner62 if self.driver is None:63 self.driver=getattr(error,'driver_expression_cleanup_owner',None)64 if self.state is None:65 self.state=getattr(error,'state_cleanup_owner',None)66 _cleanup_owner(self,error,'transform_cleanup_owner','时序变换')67 raise6869 def _ensure_active(self):70 """拒绝关闭或失败后的继续推进，避免重发未完成网格。参数: 无。返回: None。异常: 生命周期错误。"""71 from ..results.errors import ResultClosedError72 if self.closed or self.failed:73 raise ResultClosedError('时序变换已关闭或失败')7475 def bounds(self):76 """取得实际可观测起点和已知终点。参数: 无。返回: 起点/可选终点。异常: 游标错误。"""77 start=min((self.cursors[n].observed_start for n in self.sources78 if self.cursors[n].observed_start is not None),default=None)79 complete=all(self.cursors[n]._finished for n in self.sources)80 end=max((self.cursors[n].observed_end for n in self.sources if self.cursors[n].observed_end is not None),81 default=None) if complete else None82 if self.domain is not None:83 if start is not None:84 start=max(start,self.domain.start_ns)85 end=self.domain.end_ns if end is None else min(self.domain.end_ns,end)86 if complete and end is not None:87 self.ends[self.item.output.source]=end88 return start,end8990 def _invalid(self,quality):91 """构造明确未知的输出。参数: 原因。返回: ScalarValue。异常: 非法质量位。"""92 return ScalarValue(None,self.item.output.dtype,False,quality or Q.MISSING)9394 def _sample(self,t,value):95 """构造离散输出。参数: 网格/实际样本坐标和值。返回: RecordSample。异常: 模型错误。"""96 sample=RecordSample(self.item.name,t,self.sequence,value,self.item.output.unit,self.ref)97 self.sequence+=198 return sample99100 def _at_grid(self,t,point,update):101 """从原区间读取[t-W,t)及变化率端点。参数: 网格时刻、可选同刻点和更新。返回: ScalarValue。异常: 预算/数值错误。"""102 values = weights = fit = span = left_value = right_value = value = None103 try:104 op=self.item.operation105 values,weights,fit=[],[],[]106 valid=Fraction(0)107 width=Fraction(0)108 quality=Q.NONE109 left_value=right_value=None110 partial=False111 for span,left,right,ratio in self.history.slices(t):112 w=(right-left)*ratio113 if right_value is None:114 right_value=span.value if span.deadline is None or t<span.deadline else self._invalid(Q.STALE)115 left_value=span.value if span.deadline is None or left<span.deadline else self._invalid(Q.STALE)116 good=max(0,min(right,span.deadline)-left)*ratio if span.deadline is not None else w117 if span.value.valid and good:118 valid+=good119 partial |= span.value.partial120 weights.append(good)121 values.append(span.value.value)122 if not span.value.valid:123 quality|=span.value.quality124 elif good<w:125 quality|=Q.STALE126 width+=w127 if op=='sample_fit_slope' and span.update and span.start>=left and span.point.valid:128 fit.append((-width,span.point))129 partial |= span.point.partial130 # 斜率会使用右端新读数；即使此前整个窗口都有完整实测值，新读数也可能是估值。131 if point is not None and point.partial and (op=='window_rate' or op=='sample_fit_slope' and update):132 partial=True133 quality|=Q.ESTIMATED134 window=self.item.window_ns135 if width<window:136 quality|=Q.MISSING137 partial |= valid<window138 if (not valid or partial and self.item.partial_policy=='strict'139 or valid < Fraction(str(self.item.minimum_coverage))*window):140 return self._invalid(quality)141 if self.host is not None:142 from ..extension_runtime import advance_grid143 value = advance_grid(self.host,t,window,values,weights,self.item.conversion)144 elif op.startswith('rolling_'):145 from ..core.rolling_values import rolling_value146 value=rolling_value(op,self.backend,values,weights,self.item.conversion,self.item.expression.root.dtype)147 else:148 from ..core.rate_transforms import rate_difference, rate_fit149 right_value=point if point is not None else right_value150 if op=='window_rate':151 if width!=window or left_value is None or right_value is None:152 return self._invalid(Q.MISSING)153 value=rate_difference(self.backend,left_value,right_value,window)154 else:155 fit.reverse()156 if update and point is not None and point.valid:157 fit.append((0,point))158 value=rate_fit(self.backend,[p for p,_ in fit],[s for _,s in fit])159 value=self._convert_rate(value)160 return replace(value,partial=True) if value.valid and partial else value161 finally:162 # 网格费用由调用层释放，本层错误回溯不再持有窗口临时列表或最后一段。163 # 只清自身局部引用；历史与用户回调仍保有的数据不会被修改。164 values = weights = fit = span = left_value = right_value = point = value = None165166 def _convert_rate(self,value):167 """换算变化率目标单位。参数: 自然单位结果。返回: 新标量。异常: 转换错误。"""168 if self.item.conversion.factor==1:169 return value170 if self.backend=='reference':171 from ..core.affine import affine_reference172 return affine_reference(value,self.item.conversion.factor,0,D.FLOAT64)173 from ..core.affine_numpy import affine_numpy174 from .where_state import scalar_columns175 from .calculation_values import scalar_from_column176 return scalar_from_column(affine_numpy(scalar_columns({'v':value})['v'],self.item.conversion.factor,0,D.FLOAT64),0)177178 def _deliver(self,group,t,value=None):179 """同刻真实更新与离散输出一次交付。参数: 可选原组、实际时刻、可选结果。返回: 无。异常: 下游错误。"""180 if value is None:181 if group is not None:182 self.emit(group)183 else:184 samples=() if group is None else group.samples185 self.emit(DecodedTimeGroup(t,(*samples,self._sample(t,value))))186187 def _drain(self,group,point=None,update=False,*,through=None,unresolved=False):188 """按网格和原组时间归并。参数: 当前组/点、真实更新或EOF截止。返回: 无。异常: 下游/预算错误。"""189 self._ensure_active()190 t=group.time_ns if group is not None else through191 start,end=self.bounds()192 if self.next_grid is None and start is not None:193 self.next_grid=first_grid(start,self.item.origin_ns,self.item.step_ns)194 limit=t if end is None else min(t,end)195 while True:196 clock=group.time_ns if group is not None else None197 tick=self.next_grid if self.next_grid is not None and self.next_grid<=limit else None198 if clock is None and tick is None:199 break200 when=clock if tick is None else tick if clock is None else min(clock,tick)201 value=None202 if tick==when:203 failure = None204 try:205 # 申请也在保护域内，扣账后取消仍有本阶段的实际费用槽。206 _reserve_owned_ram(self,8192+1024*len(self.history.spans),'_work_cost')207 value=self._invalid(Q.MISSING) if unresolved else self._at_grid(when,point if when==t else None,208 update and when==t)209 except BaseException as error:210 failure = error211 self.failed = True212 error.transform_cleanup_owner = self213 raise214 finally:215 try:216 _release_owned_ram(self,self._work_cost,'_work_cost')217 except BaseException as cleanup:218 self.failed = True219 original = cleanup if failure is None else failure220 original.transform_cleanup_owner = self221 if failure is None:222 raise223 failure.add_note('时序变换网格工作区释放失败: ' + type(cleanup).__name__)224 self.next_grid+=self.item.step_ns225 self._deliver(group if clock==when else None,when,value)226 if clock==when:227 group=None228229 def _append_span(self,t,pair_width=None):230 """闭合上一条原始保持。参数: 后继实际时刻和所属驱动对跨度。返回: 无。异常: 预算/顺序错误。"""231 if self.last is not None and self.last<t:232 self.history.append(TransformSpan(self.last,t,self.last_right,self.last_deadline,self.last_update,233 self.last_point,pair_width))234235 def push(self,group):236 """消费上游完整组。参数: 原子组。返回: 无；向后继交付。异常: 原输入/预算/求值错误。"""237 self._ensure_active()238 t=group.time_ns239 if self.period is None:240 self._process(group,None)241 return242 drive=any(s.name==self.clock for s in group.samples)243 if self.clock_start is None:244 if drive:245 self.clock_start=t246 self._process(group,None,unresolved=True)247 return248 if drive:249 width=t-self.clock_start250 self._flush_waiting(width)251 self._process(group,width)252 self.clock_start=t253 elif self.ends.get(self.state.definitions[self.clock].source,t)>=t:254 # 等原驱动右端确定后，以原时间顺序回放这几组；每个真实值只求值一次。255 cost=1024+sum(estimate_record_bytes(s) for s in group.samples)256 _reserve_owned_ram(self,cost,'_queue_cost')257 self.waiting.append((group,cost))258 else:259 self._flush_waiting(None)260 self._process(group,None,unresolved=True)261262 def _flush_waiting(self,width):263 """以已确认的原对处理所有待定组。参数: 驱动实际跨度或EOF的None。返回: 无。异常: 交付/退款错误。"""264 while self.waiting:265 group,cost=self.waiting[0]266 self._process(group,width,unresolved=width is None)267 self.waiting.popleft()268 _release_owned_ram(self,cost,'_queue_cost')269270 def _process(self,group,width,*,unresolved=False):271 """按已知计量推进原更新和网格。参数: 完整组、原驱动跨度、计量未知标志。返回: 无。异常: 原错。"""272 t=group.time_ns273 start,end=self.bounds()274 if start is None or t<start:275 self.emit(group)276 return277 if self.item.operation!='adjacent_rate' and self.last is None and self.period is None and start<t:278 self.last=start279 self.last_right,self.last_deadline=self.driver.evaluate_at(start)280 self.last_point=self.last_right281 # 后帧可能刚揭示来源EOF；在覆盖旧状态之前，按真实末端重新确定左侧有效截止。282 if self.last is not None:283 self.last_right,self.last_deadline=self.driver.evaluate_at(self.last)284 selected=tuple(s for s in group.samples if s.name in self.names)285 if selected:286 self.state.apply(DecodedTimeGroup(t,selected))287 updated=self.driver.actual_update_at(t)288 point,_=self.driver.evaluate_at(t,continuous=False)289 if self.item.operation=='adjacent_rate':290 output=None291 if self.last is not None:292 elapsed=t-self.last293 if not unresolved:294 self.adjacent_weight+=Fraction(elapsed*self.period,width) if width else elapsed295 self.adjacent_known &= (not unresolved and self.last_right.valid and296 (self.last_deadline is None or self.last_deadline>=t))297 if updated and (end is None or t<=end):298 from ..core.rate_transforms import rate_difference299 ok=(self.adjacent is not None and t-self.adjacent[0]<=self.item.max_gap_ns300 and self.adjacent_known and self.adjacent_weight>0)301 output=(self._convert_rate(rate_difference(self.backend,self.adjacent[1],point,self.adjacent_weight))302 if ok else self._invalid(point.quality or Q.MISSING))303 if output.valid and (point.partial or self.adjacent[1].partial):304 output=replace(output,partial=True)305 self.adjacent,self.adjacent_known,self.adjacent_weight=(t,point),point.valid,Fraction(0)306 self.last=t307 self._deliver(group,t,output)308 return309 if not unresolved:310 self._append_span(t,width)311 self._drain(group,point,updated,unresolved=unresolved)312 self.last,self.last_point,self.last_update=t,point,updated313 self.last_right,self.last_deadline=self.driver.evaluate_at(t)314 self.history.trim()315316 def finish(self):317 """完成真正EOF后标记宿主状态正常结束。参数: 无。返回: None。异常: 最后网格失败仍通知取消。"""318 self._ensure_active()319 self._finish_grid()320 self.host_finished = True321322 def _finish_grid(self):323 """按真实来源EOF补最后可定网格，缺少后继原帧不外推周期。参数: 无。返回: 无。异常: 交付错误。"""324 start,end=self.bounds()325 self._flush_waiting(None)326 if self.item.operation=='adjacent_rate' or start is None or end is None:327 return328 if self.period is None:329 if self.last is None:330 self.last=start331 self.last_right,self.last_deadline=self.driver.evaluate_at(start)332 self.last_point=self.last_right333 self.last_right,self.last_deadline=self.driver.evaluate_at(self.last)334 self._append_span(end)335 self._drain(None,through=end,unresolved=self.period is not None)336337 def close(self):338 """清理等待组、真实区间和最新值。参数: 无。返回: 无。异常: 退款失败可重试。"""339 if self.closed:340 return341 for owner in (self.driver,self.state,self.history):342 if owner is not None:343 owner.close()344 if self.host is not None:345 self.host.close(cancelled=not self.host_finished)346 self.host = None347 self.waiting.clear()348 self.emit=self.last_point=self.last_right=None349 for field in ('_work_cost','_queue_cost','_fixed_cost'):350 _release_owned_ram(self,getattr(self,field),field)351 self.closed=True352