chronaxis.execution.record_event_durations
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""按需计算每段事件的持续时长，只为周期模式保存必要驱动时间。23实际模式直接使用事件范围。固定/DBC需要知道事件两端落在哪对报文中，因而4保存每条所选驱动的一个整数时间；按索引二分定位，只回读相关记录对。该文件5没有数值、原帧或另一时间轴，未请求此类统计时不建立；关闭自动删除。6"""78import heapq9from ..core.models import Interval10from ..core.calculations import CalculationDefinition11from ..core.numeric import ScalarValue12from ..core.types import SignalDType as D,QualityFlag as Q13from ..results.errors import ResultClosedError14from ..results.spool import _reserve_owned_ram,_release_owned_ram,_cleanup_owner15from .record_time_calculation import RecordTimeCalculation16from .history import _metadata_bytes17from ..planning.record_execution import is_event_duration181920class RecordEventDurations:21 """拥有必要时点文件、尚缺右端的事件及一个短生命周期数值查询。"""2223 def __init__(self,plan,budget,work_dir,steps,consumer,*,indexes=None):24 """预付有限索引状态，不为actual创建文件。参数: 计划/预算/目录/周期/回调；indexes借用共享必要时点，不在此重复保存。返回: owner。异常: 原构造错误。"""25 self._plan,self._budget,self._path,self._steps,self._consumer=plan,budget,work_dir,steps,consumer26 self._requests=tuple(item for item in plan.calculations if is_event_duration(item))27 self._choices=dict(plan.intervals.resolved)28 self._held_cost=self._fixed_cost=self._pending_cost=029 self._indexes=indexes30 self._queue=[]31 self._serial=032 self._calculation=None33 self._closed=self._failed=False34 try:35 _reserve_owned_ram(self,131072+32768*len(self._requests)+32*_metadata_bytes(self._requests),'_fixed_cost')36 except BaseException as error:37 _cleanup_owner(self,error,'record_event_duration_cleanup_owner','事件时长构造')38 raise3940 def _open(self):41 """拒绝关闭/失败实例。参数: 无。返回: None。异常: ResultClosedError。"""42 if self._closed or self._failed:43 raise ResultClosedError('事件时长已关闭或失败')4445 def observe_group(self,group):46 """父已更新共享时点索引，尝试完成等待中的事件。参数: 完整组。返回: None。异常: 查询原错。"""47 self._open()48 self._flush_ready()4950 def observe_event(self,event):51 """只为匹配事件建立必要查询，缺右端时暂存端点和ID。参数: 事件。返回: None。异常: 预算/求值原错。"""52 self._open()53 for request in self._requests:54 if request.scope!='event:'+event.rule.rule_id:55 continue56 end=event.start_ns if event.end_ns is None else event.end_ns57 # 快捷时长与一般事件计算使用同一选择范围；确认等于结束时保留空区间。58 rule=next(item for item in self._plan.rules if item.rule_id==event.rule.rule_id)59 start=event.confirmed_ns if rule.selection_origin=='confirmed' else event.start_ns60 cost=8192+16*_metadata_bytes(event.event_id)61 _reserve_owned_ram(self,cost,'_pending_cost')62 heapq.heappush(self._queue,(end,self._serial,request,event.event_id,start,cost,63 bool(event.dependency_quality & Q.ESTIMATED)))64 self._serial+=165 self._flush_ready()6667 def _calculate(self,request,event_id,start_ns,end_ns,estimated=False):68 """只借取事件两侧及区间内的驱动点，复用原两后端时长核。6970 参数: 请求/父ID/真实端点。返回: None，结果交回调。异常: 构造/数值/发布原错。71 """72 step=self._steps[request.calculation_id]73 spec=CalculationDefinition(request.calculation_id,'duration',Interval(start_ns,end_ns),request.partial_policy,74 request.unit,'event',request.minimum_coverage,config_version=self._plan.config_version)75 try:76 self._calculation=RecordTimeCalculation(self._budget,spec,'1',D.BOOL,step,77 backend=self._plan.execution.backend,batch_intervals=64)78 except BaseException as error:79 self._calculation=getattr(error,'record_time_cleanup_owner',None)80 raise81 value=ScalarValue(True,D.BOOL,True,Q.NONE)82 if step is None:83 if start_ns<end_ns:84 self._calculation.consume(((start_ns,value),(end_ns,value)))85 else:86 driver=self._choices[request.calculation_id].driver87 count=self._indexes.count(driver)88 if count:89 first=self._indexes.lower(driver,start_ns)90 if first==count or first>0 and self._indexes.at(driver,first)>start_ns:91 first-=192 last=min(count-1,self._indexes.lower(driver,end_ns))93 for index in range(first,last+1):94 self._calculation.consume(((self._indexes.at(driver,index),value),))95 result=self._calculation.finish()96 if estimated:97 from .transform_partial import partial_result98 result=partial_result(result,request.partial_policy)99 self._consumer(request,event_id,result)100 self._calculation.close();self._calculation=None101102 def _flush_ready(self,final=False):103 """保留仍缺右端的请求，其余按必要范围计算；EOF不补记录。104105 参数: final只在真实来源EOF后使用。返回: None。异常: 原查询错，费用留到close。106 """107 pending=[]108 try:109 while self._queue:110 item=heapq.heappop(self._queue)111 end,serial,request,event_id,start,cost,estimated=item112 driver=self._choices[request.calculation_id].driver113 last=None if self._indexes is None or driver is None else self._indexes.last(driver)114 if not final and self._steps[request.calculation_id] is not None and (last is None or last<end):115 pending.append(item)116 continue117 self._calculate(request,event_id,start,end,estimated)118 item=None119 _release_owned_ram(self,cost,'_pending_cost')120 except BaseException:121 self._failed=True122 raise123 finally:124 for item in pending:125 heapq.heappush(self._queue,item)126127 def finish(self):128 """真实EOF后结算未完成范围，不假造周期尾段。参数: 无。返回: None。异常: 原计算错。"""129 self._open()130 self._flush_ready(final=True)131132 @property133 def closed(self):134 """返回关闭状态。参数: 无。返回: bool。异常: 无。"""135 return self._closed136137 def close(self):138 """只关闭自己的临时文件，成功后才撤销责任。参数: 无。返回: None。异常: 原关闭错可重试。"""139 if self._closed:140 return141 if self._calculation is not None:142 self._calculation.close();self._calculation=None143 self._queue.clear()144 self._plan=self._requests=self._consumer=self._choices=None145 _release_owned_ram(self,self._pending_cost,'_pending_cost')146 _release_owned_ram(self,self._fixed_cost,'_fixed_cost')147 self._closed=True148