chronaxis.results.signal_exports
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""将指定信号的已解码历史写为原始采样及可选连续量网格CSV长表。23输入为同次读取生成的HistorySample/HistoryProgress流及显式导出范围，不打开日志4或DBC；输出original.csv及清单，启用线性时另写resampled.csv。异步信号分别5一行，状态/计数不补点；显式连续量交给独立有界网格owner。6值无效时留空并保质量原因和完整来源引用；原无效值以既有安全编码保存。7逐行缓冲和编码先付预算，输入正常/失败/取消均关闭，整个目录最终独占发布。8"""910import csv11from dataclasses import dataclass12from io import StringIO13import json1415from ..core.models import Interval16from ..core.validation import require_identifier, require_text17from ..execution.record_sample import RecordSample18from ..execution.history_records import HistorySample, HistoryProgress, estimate_record_bytes19from ..execution.history import _metadata_bytes20from .budget import _reserve_owned_ram, _release_owned_ram, _cleanup_owner21from .codec import _pack22from .export_files import ExportFiles23from .signal_grid import SignalGridSpec, SignalGrid242526@dataclass(frozen=True)27class SignalCSVExportSpec:28 """指定待导出的已绑定别名；字段含义和单位沿用原绑定，不由文件名推断。"""2930 signals: tuple[str,...] # 唯一非空别名，最多4096，保持配置顺序。31 resampling: SignalGridSpec | None = None # 默认原始采样，显式连续量才进入线性网格。3233 def __post_init__(self):34 """只核静态选择。参数: 无。返回: None。异常: 类型/数量/重复或非法别名。35 """36 if type(self.signals) is not tuple or not 1 <= len(self.signals) <= 4096:37 raise ValueError('信号CSV需要1到4096个显式别名')38 for name in self.signals:39 require_identifier(name,'signal_csv_export.signal')40 if len(set(self.signals)) != len(self.signals):41 raise ValueError('信号CSV别名重复')42 if self.resampling is not None and (type(self.resampling) is not SignalGridSpec43 or any(name not in self.signals for name in self.resampling.continuous)):44 raise ValueError('重采样必须是所选信号的明确子集')454647_COLUMNS = ('time_ns','source','signal','field','value','unit','valid','quality','status',48 'sequence','source_ref_json','original_value_json')495051def _csv_bytes(values):52 """编码一个有界CSV行。参数: values有限标量列表。返回: UTF8 bytes。异常: CSV/Unicode/内存错。53 """54 # newline交由CSV处理，引号、逗号和换行按标准转义，不拼接未经转义的来源文字。55 stream = StringIO(newline='')56 csv.writer(stream,lineterminator='\n').writerow(values)57 return stream.getvalue().encode('utf-8')585960def _sample_row(sample, compact=False):61 """提取原样本可读列及完整证据。参数: sample已校验HistorySample，compact关闭原文/来源列。返回: 行tuple。异常: 安全编码错。62 """63 value = sample.value64 if compact:65 return (sample.time_ns,sample.name,value.value if value.valid else '',sample.unit,int(value.quality))66 return (sample.time_ns,sample.source_ref.source_name,sample.name,sample.source_ref.field,67 value.value if value.valid else '',sample.unit,value.valid,int(value.quality),68 'observed' if value.valid else 'invalid',sample.sequence,69 json.dumps(_pack(sample.source_ref),ensure_ascii=False,allow_nan=False,separators=(',',':')),70 json.dumps(_pack(sample.original_value),ensure_ascii=False,allow_nan=False,separators=(',',':'))71 if sample.original_present else '')727374def _reference_json(sample):75 """编码实际端点引用。参数: sample或None。返回: JSON文本或空字符串。异常: 安全编码错。76 """77 return '' if sample is None else json.dumps(_pack(sample.source_ref),ensure_ascii=False,78 allow_nan=False,separators=(',',':'))798081def _write_grid(owner, rows, counts, compact=False):82 """有界编码网格行及两端引用。参数: owner/rows/固定计数字典及compact引用开关。返回: None。异常: 编码/预算/I/O/取消错。83 """84 owner.select_file('resampled.csv')85 for row in rows:86 _reserve_owned_ram(owner,8192+_metadata_bytes(row)*16,'_row_cost')87 try:88 # 左右时间和引用均来自实际采样，目标时间不伪装成原行记录。89 left,right=row.left,row.right90 values=(row.time_ns,row.name,'' if row.value is None else row.value,row.unit,91 row.status,row.reason,row.quality)92 if not compact:93 values+=('' if left is None else left.time_ns,'' if right is None else right.time_ns,94 _reference_json(left),_reference_json(right))95 owner.append(_csv_bytes(values))96 counts[row.status]+=197 finally:98 _release_owned_ram(owner,owner._row_cost,'_row_cost')99100101class SignalCSVWriter(ExportFiles):102 """复用原写文件和线性网格，支持逐样本推进而不强制暂存整份输入。"""103104 def __init__(self,spec,interval,destination,budget,*,device_label,check_cancel=None,105 definitions=(),backend='numpy',compact=False,integrity=True,timestamped=False):106 """预付输出状态并建立所需表。参数: 导出规格/范围/目标/预算与后端。返回: owner。异常: 原构造错。"""107 if type(compact) is not bool:108 raise TypeError('compact必须为bool')109 if type(spec) is not SignalCSVExportSpec or type(interval) is not Interval:110 raise TypeError('信号导出需要选择及明确Interval')111 require_text(device_label, 'device_label')112 if (backend not in ('reference','numpy')113 or check_cancel is not None and not callable(check_cancel)):114 raise ValueError('导出需要有效后端及可选取消回调')115 super().__init__(destination,budget,integrity=integrity,timestamped=timestamped)116 self.spec,self.interval,self.definitions=spec,interval,definitions117 self.device_label,self.backend,self.compact=device_label,backend,compact118 self._check_cancel=check_cancel119 self._count=self._invalid=0120 self._grid_counts={'observed':0,'estimated':0,'missing':0}121 self._selected=None122 try:123 _reserve_owned_ram(self,4096+sum(len(n.encode('utf-8'))*16+256 for n in spec.signals),'_fixed_cost')124 self._selected=set(spec.signals)125 if spec.resampling is not None:126 try:127 self.worker=SignalGrid(spec.resampling,definitions,interval,budget,128 backend=backend,check_cancel=check_cancel)129 except BaseException as error:130 self.worker=getattr(error,'signal_grid_cleanup_owner',None);raise131 self.open_file('resampled.csv')132 self.append(_csv_bytes(('time_ns','signal','value','unit','status','reason','quality')+133 (() if compact else ('left_time_ns','right_time_ns','left_ref_json','right_ref_json'))))134 if not compact or spec.resampling is None or self._selected-set(spec.resampling.continuous):135 self.select_file('original.csv')136 self.append(_csv_bytes(('time_ns','signal','value','unit','quality') if compact else _COLUMNS))137 except BaseException as error:138 self._failed = True139 _cleanup_owner(self,error,'export_cleanup_owner','信号CSV构造');raise140141 def push(self,sample):142 """写一个已验证样本或推进原网格。参数: HistorySample/Progress。返回: None。异常: 原输入/写入错。"""143 if self._closed:144 raise ValueError('信号导出已关闭')145 if self._check_cancel is not None: self._check_cancel()146 if type(sample) is HistoryProgress: return147 if type(sample) not in (HistorySample,RecordSample):148 raise TypeError('信号导出只接受已解码历史记录')149 if sample.name not in self._selected or not self.interval.start_ns<=sample.time_ns<self.interval.end_ns:150 return151 if self.worker is not None and sample.name in self.spec.resampling.continuous:152 _write_grid(self,self.worker.push(sample),self._grid_counts,self.compact)153 return154 _reserve_owned_ram(self,estimate_record_bytes(sample)*2,'_row_cost')155 try:156 self.select_file('original.csv')157 self.append(_csv_bytes(_sample_row(sample,self.compact)))158 finally:159 _release_owned_ram(self,self._row_cost,'_row_cost')160 self._count+=1;self._invalid+=not sample.value.valid161162 def observed_end(self,end_ns):163 """在线调用方在末组交付前收紧EOF；不能用它改已经写出的记录。164165 参数: 实际末端，须在初始范围中。返回: None。异常: 越界。166 """167 if not self.interval.start_ns<=end_ns<=self.interval.end_ns:168 raise ValueError('导出末端超出声明范围')169 self.interval=Interval(self.interval.start_ns,end_ns)170 if self.worker is not None: self.worker.interval=self.interval171172 def complete(self):173 """结束网格、写一次必要清单并原子发布。参数: 无。返回: 输出Path。异常: 原完成错。"""174 interval,spec=self.interval,self.spec175 metadata={'export_type':'signal_csv','device_label':self.device_label,'signals':spec.signals,176 'range_ns':(interval.start_ns,interval.end_ns),'resampling':False,177 'rows':self._count,'observed':self._count-self._invalid,'invalid':self._invalid}178 if self.worker is not None:179 _write_grid(self,self.worker.finish(),self._grid_counts,self.compact)180 grid=spec.resampling181 metadata.update(resampling={'mode':'linear','continuous':grid.continuous,182 'period_ns':grid.period_ns,'origin_ns':grid.origin_ns,'max_gap_ns':grid.max_gap_ns,183 'causal':False,'numeric':'float64_estimate','ordering':'per_signal_monotonic'},184 resampled_counts=self._grid_counts,backend=self.backend,185 bindings=tuple({'name':d.name,'source':d.source,'field':d.field,'unit':d.unit,186 'dtype':d.dtype.value,'same_time_policy':d.same_time_policy}187 for d in self.definitions if d.name in grid.continuous))188 if self.compact: metadata['row_format']='compact-v2'189 return self.finish(metadata)190191192def export_signal_csv(records,spec,interval,destination,budget,*,device_label,check_cancel=None,193 definitions=(),backend='numpy',compact=False,integrity=True,file_consumer=None):194 """消费既有记录流并复用在线写出器，保原公开调用与关闭责任。195196 参数: 可关闭记录流、规格/范围/目标/预算；设备、取消、绑定、后端、精简开关。197 integrity只在请求时计算文件摘要；file_consumer接收写入时取得的有限文件身份，供内部登记复用。198 返回: 公开ExportManifest含实际路径；内部file_consumer模式返回Path。异常: 原格式/输入/预算/取消错；清理失败附真实owner。199 """200 if file_consumer is not None and not callable(file_consumer):201 raise TypeError('file_consumer须为回调')202 owner=SignalCSVWriter(spec,interval,destination,budget,device_label=device_label,203 check_cancel=check_cancel,definitions=definitions,backend=backend,compact=compact,integrity=integrity,204 timestamped=file_consumer is None)205 try:206 owner.reader=iter(records)207 for sample in owner.reader: owner.push(sample)208 output=owner.complete()209 if file_consumer is not None: file_consumer(tuple(owner.files))210 return output211 except BaseException as error:212 owner._failed = True213 _cleanup_owner(owner,error,'export_cleanup_owner','信号CSV导出');raise214