chronaxis.execution.record_exports
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""只为显式导出接同次读取的帧或样本；复用既有writer和结果发布。23会话范围在真实EOF后确定；片段由窗口消费者直接交付，不复制片段表。4不再次读原日志或解码，默认精简导出不产生逐帧旁表。5"""6from ..core.models import Interval7from .record_signal_export import RecordSignalExport8from .legacy_frame_samples import LegacyFrameSamples9from ..results.frame_rebuild import FrameRebuildRead10from ..results.artifact_exports import ArtifactExportStore11from ..results.frame_exports import export_can_frames12from ..results.errors import ResultClosedError13from ..results.budget import _reserve_owned_ram,_release_owned_ram,_cleanup_owner141516class RecordExports:17 """拥有导出索引、在线信号写出与当前帧游标；原帧仓由来源游标所有。"""18 def __init__(self,plan,budget,path,cursors):19 """建立仅显式请求的索引。参数: 计划/预算/新路径/借用来源。返回: owner。异常: 原构造错。"""20 self._plan,self._budget,self._cursors=plan,budget,cursors21 self._held_cost=self._fixed_cost=022 self._exports=self._reader=self._export_owner=self._signal=self._rebuild=None23 self._closed=self._failed=self._finished=False24 self._option=plan.outputs.can_frames25 try:26 _reserve_owned_ram(self,65536,'_fixed_cost')27 self._require_clock()28 try:29 self._exports=ArtifactExportStore(budget,path,plan.dataset.session_id,plan.metadata.get('device_label','pending'),integrity=plan.diagnostics.integrity)30 except BaseException as error:31 self._exports=getattr(error,'artifact_export_store_cleanup_owner',None);raise32 if plan.outputs.signal_csv is not None:33 try:34 self._signal=RecordSignalExport(plan,budget,cursors,self._exports,path)35 except BaseException as error:36 self._signal=getattr(error,'record_signal_export_cleanup_owner',None);raise37 except BaseException as error:38 _cleanup_owner(self,error,'record_exports_cleanup_owner','记录流原帧导出');raise3940 def _require_clock(self):41 """多来源片段须有可比时钟；单来源不强求映射。参数: 无。返回: None。异常: 缺同步依据。"""42 plan=self._plan43 options=[]44 if self._option is not None:45 options.append((self._option,{self._option.source}))46 if plan.outputs.signal_csv is not None:47 option=plan.outputs.signal_csv48 options.append((option,{plan.bindings.resolve(n).source for n in option.spec.signals}))49 for option,sources in options:50 if option.scope!='segments':continue51 for window in (w for w in plan.windows if option.window is None or w.window_id==option.window):52 checked_sources=set(sources)53 rule=next(r for r in plan.rules if r.rule_id==window.event)54 names=set(rule.dependencies)55 choice=dict(plan.intervals.resolved)[window.window_id]56 if choice.driver:names.add(choice.driver)57 checked_sources.update(plan.bindings.resolve(n).source for n in names)58 if not names:checked_sources.update(s.name for s in plan.sources)59 evidenced={item.source for item in plan.bindings.synchronization}60 keys=set()61 for source in plan.bindings.sources:62 if source.name not in checked_sources:continue63 clock=source.timebase64 if source.name in evidenced or clock==plan.dataset.timebase or clock.time_type.value=='absolute' and plan.dataset.timebase.time_type.value=='absolute':65 keys.add(('dataset',))66 else:67 keys.add(('utc',) if clock.time_type.value=='absolute' else ('relative',clock.timebase_id,clock.origin))68 if len(keys)>1:raise ValueError('导出与片段缺少共同时间基准')6970 def _export(self,actual,segment_id=None):71 """借同次封口原帧写一份所需输出。参数: 实际范围/父片段。返回: None。异常: 原消费/发布错。"""72 option=self._option73 cursor=next(c for c in self._cursors if c.source_name==option.source)74 store=cursor._raw_store75 if store is None or store.closed or store.state!='sealed':76 raise ValueError('原帧导出缺少同次封口输入')77 interval=actual or Interval(self._plan.dataset.domain.start_ns,self._plan.dataset.domain.start_ns)78 try:79 if option.resampling is None:80 self._reader=store.iter_frames()81 else:82 context=cursor._dbc_context83 if context is None:raise ValueError('报文重采样缺同次编码定义')84 if cursor.legacy_frame_storage:85 source=next(s for s in self._plan.sources if s.name==option.source)86 try:87 self._reader=LegacyFrameSamples(store,cursor._message_store,context,self._budget,88 read_identity=cursor._read_identity,session_id=self._plan.dataset.session_id,channel=source.config.channel,trace=False)89 except BaseException as error:90 self._reader=getattr(error,'legacy_frame_samples_cleanup_owner',None);raise91 else:92 self._reader=store.iter_samples()93 try:94 self._rebuild=FrameRebuildRead(self._reader,option.resampling,interval,context.pairs(),self._budget,95 self._exports.path/f'rebuild-{self._exports.count:08d}',backend=self._plan.execution.backend,96 periods=context.periods,integrity=self._plan.diagnostics.integrity,trace=False)97 except BaseException as error:98 self._rebuild=getattr(error,'frame_rebuild_cleanup_owner',None);raise99 self._reader=None100 files=[]101 export_can_frames(self._rebuild or self._reader,option.spec,interval,self._exports.next_path(),self._budget,102 source=option.source,device_label=self._plan.metadata.get('device_label','pending'),compact=True,103 integrity=self._plan.diagnostics.integrity,file_consumer=files.extend)104 if self._rebuild is not None:self._rebuild.close();self._rebuild=None105 if self._reader is not None:self._reader.close();self._reader=None106 self._exports.append('can_frames',option.scope,actual,segment_id=segment_id,completed_files=files)107 except BaseException as error:108 self._export_owner=getattr(error,'export_cleanup_owner',None)109 self._failed=True;error.record_exports_cleanup_owner=self;raise110111 def session(self):112 """EOF后裁实际会话范围，空输入保持空输出。参数: 无。返回: None。异常: 原导出错。"""113 if self._signal is not None:self._signal.finish()114 if self._option is None or self._option.scope!='session': return115 cursor=next(c for c in self._cursors if c.source_name==self._option.source)116 actual=None117 if cursor.observed_start is not None and cursor.observed_end is not None:118 domain=self._plan.dataset.domain119 start,end=max(domain.start_ns,cursor.observed_start),min(domain.end_ns,cursor.observed_end)120 if start<=end: actual=Interval(start,end)121 self._export(actual)122123 def segment(self,segment):124 """直接接最终片段，不保存第二份片段历史。参数: CompactSegment。返回: None。异常: 原导出错。"""125 if self._signal is not None:self._signal.segment(segment)126 if self._option is not None and self._option.scope=='segments' and (self._option.window is None or segment.window_id==self._option.window):127 self._export(segment.actual,segment.segment_id)128129 def seal(self):130 """全部请求完成后封索引。参数: 无。返回: None。异常: 原封口错。"""131 self._exports.seal();self._finished=True132133 def take_exports(self,receiver):134 """唯一移交已完成结果；失败前责任仍可达。参数: 同预算父owner。返回: None。异常: 状态错。"""135 if not self._finished or self._closed or receiver._budget is not self._budget or receiver._artifact_exports is not None:136 raise ResultClosedError('原帧输出尚不可移交')137 owner=self._exports138 try:139 receiver._artifact_exports=owner140 finally:141 if receiver._artifact_exports is owner: self._exports=None142143 @property144 def closed(self):145 """返回实际关闭状态。参数: 无。返回: bool。异常: 无。"""146 return self._closed147148 def close(self):149 """先关当前输出及索引，再释放自有状态；不关闭借用原帧。参数: 无。返回: None。异常: 原关闭错。"""150 if self._closed:return151 for slot in ('_signal','_export_owner','_rebuild','_reader','_exports'):152 owner=getattr(self,slot)153 if owner is not None: owner.close();setattr(self,slot,None)154 self._plan=self._cursors=self._option=None155 _release_owned_ram(self,self._fixed_cost,'_fixed_cost');self._closed=True156