chronaxis.execution.record_signal_export
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""会话信号CSV随同次合并流在线写出；只暂留最后一个所需时刻组。23最后组在真实EOF范围确定后才写，保半开范围而不保存整段信号历史。4线性网格仍使用原SignalGrid；未请求的别名不进入本owner。5"""6from dataclasses import replace7from ..core.models import Interval8from ..sources.can_spec import can_field_from_key9from ..legacy.records import field_to_dict10from ..results.signal_exports import SignalCSVWriter,export_signal_csv11from .record_export_samples import RecordExportSamples12from ..results.budget import _reserve_owned_ram,_release_owned_ram,_cleanup_owner13from .history import _metadata_bytes141516class RecordSignalExport:17 """拥有一个在线writer和待定末组，借用最终导出索引。"""18 def __init__(self,plan,budget,cursors,exports,path):19 """预约必要别名状态，不读取文件。参数: 计划/预算/游标/结果索引。返回: owner。异常: 构造原错。"""20 self._plan,self._budget,self._cursors,self._exports=plan,budget,cursors,exports21 self._held_cost=self._fixed_cost=self._pending_cost=022 self._writer=self._history=None;self._pending=()23 self._closed=False24 self._option=plan.outputs.signal_csv25 self._definitions=tuple(s.definition for s in plan.signals if s.definition.name in self._option.spec.signals)26 self._names=set(self._option.spec.signals)27 self._sources={d.source for d in self._definitions}28 try:29 _reserve_owned_ram(self,65536+32*_metadata_bytes(self._definitions),'_fixed_cost')30 plan.bindings.require_synchronized(tuple(self._names))31 if self._option.period_driver is not None:32 definition=next(d for d in self._definitions if d.name==self._option.period_driver)33 cursor=next(c for c in cursors if c.source_name==definition.source)34 locator=field_to_dict(can_field_from_key(definition.field))35 period=next(f['cycle_ns'] for f in cursor.metadata['fields'] if f['locator']==locator)36 self._option=replace(self._option,period_driver=None,spec=replace(self._option.spec,37 resampling=replace(self._option.spec.resampling,period_ns=period)))38 if self._option.scope=='segments':39 try:40 self._history=RecordExportSamples(budget,path,self._definitions,plan.dataset.session_id)41 except BaseException as error:42 self._history=getattr(error,'record_export_samples_cleanup_owner',None);raise43 except BaseException as error:44 _cleanup_owner(self,error,'record_signal_export_cleanup_owner','在线信号导出');raise4546 def update_observation(self, plan):47 """接收同次读取已见范围，在线writer不会预造末端样本。参数: 内部计划。返回: None。异常: 无。"""48 self._plan=plan49 if self._writer is not None:50 self._writer.interval=Interval(self._writer.interval.start_ns,plan.dataset.domain.end_ns)51 if self._writer.worker is not None:52 self._writer.worker.interval=self._writer.interval5354 def _bounds(self):55 """只查所选来源实际首末，不扫描输入。参数: 无。返回: 实际范围或None。异常: 无。"""56 starts=[c.observed_start for c in self._cursors if c.source_name in self._sources and c.observed_start is not None]57 ends=[c.observed_end for c in self._cursors if c.source_name in self._sources and c.observed_end is not None]58 if not starts or not ends:return None59 domain=self._plan.dataset.domain60 start,end=max(min(starts),domain.start_ns),min(max(ends),domain.end_ns)61 return Interval(start,end) if start<=end else None6263 def _begin(self,final=False):64 """第一次需要写出时确定真实起点，末端暂以请求上限约束。参数: 是否EOF。返回: None。异常: 构造原错。"""65 if self._writer is not None:return66 actual=self._bounds();domain=self._plan.dataset.domain67 interval=(actual or Interval(domain.start_ns,domain.start_ns)) if final else Interval(68 domain.start_ns if actual is None else actual.start_ns,domain.end_ns)69 try:70 self._writer=SignalCSVWriter(self._option.spec,interval,self._exports.next_path(),self._budget,71 device_label=self._plan.metadata.get('device_label','pending'),definitions=self._definitions,72 backend=self._plan.execution.backend,compact=True,integrity=self._plan.diagnostics.integrity)73 except BaseException as error:74 self._writer=getattr(error,'export_cleanup_owner',None);raise7576 def _flush(self):77 """交出上一实际组并释放待定费用。参数: 无。返回: None。异常: 原写入/清理错。"""78 for sample in self._pending:self._writer.push(sample)79 self._pending=()80 _release_owned_ram(self,self._pending_cost,'_pending_cost')8182 def observe(self,group):83 """仅保所选样本的一个完整时刻；不在最新值状态里重复保存导出值。参数: 真实组。返回: None。异常: 原错。"""84 samples=tuple(sample for sample in group.samples if sample.name in self._names)85 if not samples:return86 if self._history is not None:87 for sample in samples:self._history.append(sample)88 return89 self._begin()90 self._flush()91 _reserve_owned_ram(self,4096+sum(4096+_metadata_bytes(s)*16 for s in samples),'_pending_cost')92 self._pending=samples9394 def finish(self):95 """EOF收紧末端，再交末组和网格，登记唯一输出。参数: 无。返回: None。异常: 原发布错。"""96 if self._option.scope!='session':return97 self._begin(final=True)98 actual=self._bounds()99 end=self._writer.interval.start_ns if actual is None else actual.end_ns100 self._writer.observed_end(end)101 self._flush()102 self._writer.complete()103 self._exports.append('signal_csv','session',actual,completed_files=self._writer.files)104 self._writer=None105106 def segment(self,segment):107 """片段确定后只读必要样本，复用同一导出器。参数: CompactSegment。返回: None。异常: 原导出错。"""108 if self._option.scope!='segments' or self._option.window!=segment.window_id:return109 actual=segment.actual110 interval=actual or Interval(self._plan.dataset.domain.start_ns,self._plan.dataset.domain.start_ns)111 try:112 files=[]113 export_signal_csv(self._history.rows(interval),self._option.spec,interval,self._exports.next_path(),114 self._budget,device_label=self._plan.metadata.get('device_label','pending'),definitions=self._definitions,115 backend=self._plan.execution.backend,compact=True,integrity=self._plan.diagnostics.integrity,116 file_consumer=files.extend)117 self._exports.append('signal_csv','segments',actual,segment_id=segment.segment_id,completed_files=files)118 except BaseException as error:119 self._writer=getattr(error,'export_cleanup_owner',None);raise120121 @property122 def closed(self):123 """返回资源关闭状态。参数: 无。返回: bool。异常: 无。"""124 return self._closed125126 def close(self):127 """关闭尚未发布writer及末组，保留原输入和借用索引。参数: 无。返回: None。异常: 原关闭错。"""128 if self._closed:return129 if self._writer is not None:self._writer.close();self._writer=None130 if self._history is not None:self._history.close();self._history=None131 self._pending=();self._plan=self._cursors=self._exports=self._definitions=None132 _release_owned_ram(self,self._pending_cost,'_pending_cost')133 _release_owned_ram(self,self._fixed_cost,'_fixed_cost');self._closed=True134