chronaxis.results.signal_grid
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""从有序信号历史生成离线线性网格，每个别名只保留相邻两组真实样本。23目标网格origin+k*period贯穿文件和批次，限制在明确半开导出范围。先收齐同刻4更新，按原绑定的last_by_sequence或冲突政策确定候选；缺两侧、无效更新或5间隔过大均产出空值及原因。其他状态信号由原始CSV路径保留，不进入本计算。6所有保留样本及有限网格块计费，失败/取消由上层文件owner接管关闭。7"""89from dataclasses import dataclass10from math import isnan11from ..core.binding import SignalDefinition12from ..core.models import Interval13from ..core.types import SignalDType14from ..core.validation import require_identifier15from ..core.export_interpolation import linear_reference, linear_numpy16from ..timebase.conversion import require_int64_ns17from ..execution.record_sample import RecordSample18from ..execution.history_records import HistorySample, estimate_record_bytes19from .budget import _reserve_owned_ram, _release_owned_ram, _cleanup_owner20from .errors import ResultClosedError212223@dataclass(frozen=True)24class SignalGridSpec:25 """显式连续信号及统一网格；不从DBC名字、dtype或消息周期推断类别/间隔。"""2627 continuous: tuple[str,...]28 period_ns: int | None # None仅用于尚未解析的DBC声明；运行网格必须已解析。29 origin_ns: int30 max_gap_ns: int3132 def __post_init__(self):33 """核静态网格。参数: 无。返回: None。异常: 类型/数量/时间或名称非法。34 """35 if type(self.continuous) is not tuple or not 1 <= len(self.continuous) <= 4096:36 raise ValueError('连续信号需要有限非空tuple')37 for name in self.continuous:38 require_identifier(name,'signal_grid.continuous')39 if len(set(self.continuous))!=len(self.continuous):40 raise ValueError('连续信号重复')41 for key in ('period_ns','origin_ns','max_gap_ns'):42 value=getattr(self,key)43 if key=='period_ns' and value is None:continue44 if type(value) is not int:45 raise TypeError('网格时间需要原生int纳秒')46 require_int64_ns(value,key)47 if self.period_ns is not None and self.period_ns<=0 or self.max_gap_ns<=0:48 raise ValueError('网格周期和最大间隔必须为正')495051@dataclass(frozen=True)52class GridRow:53 """一个目标点的估计或缺失；左右引用是实际样本，不能伪造为目标点观测。"""5455 time_ns: int56 name: str57 unit: str58 value: int | float | None59 status: str60 reason: str61 quality: int62 left: HistorySample | None63 right: HistorySample | None646566def _same_fact(left,right):67 """沿既有同刻事实比较，包括无效NaN。参数: 两HistorySample。返回: bool。异常: 模型属性错误。68 """69 a,b=left.value,right.value70 return (a.valid==b.valid and a.quality==b.quality and type(a.value) is type(b.value)71 and (a.value==b.value or type(a.value) is float and isnan(a.value) and isnan(b.value)))727374class SignalGrid:75 """拥有每信号两个有界样本与单调目标游标；push/finish产出必须在下一次push前耗尽。"""7677 def __init__(self,spec,definitions,interval,budget,*,backend='numpy',check_cancel=None):78 """核声明并预付有限状态。参数: spec/definitions/interval/budget、backend和取消回调。7980 返回: owner。异常: 声明/预算错；持续释放失败附signal_grid_cleanup_owner。81 """82 if type(spec) is not SignalGridSpec or spec.period_ns is None or type(interval) is not Interval or backend not in ('reference','numpy'):83 raise TypeError('线性导出需要明确网格、范围和后端')84 if type(definitions) is not tuple or any(type(d) is not SignalDefinition for d in definitions):85 raise TypeError('线性导出需要原绑定定义tuple')86 # 类型声明必须在读取之前核对，空文件也不能让状态/枚举误入连续插值。87 for name in spec.continuous:88 matches=tuple(d for d in definitions if d.name==name)89 if len(matches)!=1 or matches[0].dtype not in (SignalDType.FLOAT64,SignalDType.INT64,SignalDType.UINT64):90 raise ValueError('连续信号需要唯一数值绑定: '+name)91 self.spec,self.interval,self.backend,self._check_cancel=spec,interval,backend,check_cancel92 self._budget=budget93 self._held_cost=self._fixed_cost=self._sample_cost=094 self._closed=self._finished=False95 self.states=None96 try:97 _reserve_owned_ram(self,262144+len(spec.continuous)*4096,'_fixed_cost')98 # ceil除法保证负时间、非零origin及跨文件都落在同一网格，不重置到首个样本。99 first=spec.origin_ns+(-((spec.origin_ns-interval.start_ns)//spec.period_ns))*spec.period_ns100 self.states={name:{'definition':next(d for d in definitions if d.name==name),'next':first,101 'left':None,'pending':None,'left_conflict':False,'pending_conflict':False,102 'left_cost':0,'pending_cost':0} for name in spec.continuous}103 except BaseException as error:104 _cleanup_owner(self,error,'signal_grid_cleanup_owner','线性网格构造')105 raise106107 def _row(self,state,time_ns,value,status,reason,left,right):108 """构造单个有限结果。参数: state/目标/值/状态/原因/两端。返回: GridRow。异常: 无。109 """110 definition=state['definition']111 quality=(0 if left is None else int(left.value.quality)) | (0 if right is None else int(right.value.quality))112 return GridRow(time_ns,definition.name,definition.unit,value,status,reason,quality,left,right)113114 def _until(self,state,limit_ns,right,conflict):115 """输出limit_ns之前的完整有限块。参数: state、独占limit_ns、右端/冲突。返回: GridRow迭代。异常: 数值/取消错误。116 """117 left=state['left']; end=min(limit_ns,self.interval.end_ns)118 if left is None and right is None:119 reason='no_observation'120 elif left is None:121 reason='missing_left'122 elif right is None:123 reason='missing_right'124 elif state['left_conflict'] or conflict:125 reason='same_time_conflict'126 elif not left.value.valid or not right.value.valid:127 reason='invalid_neighbor'128 elif right.time_ns-left.time_ns>self.spec.max_gap_ns:129 reason='gap_too_large'130 else:131 reason=''132 while state['next']<end:133 if self._check_cancel is not None:134 self._check_cancel()135 # 每块最多256个点，内存费用已在构造预付；很长范围不会建立整段网格数组。136 count=min(256,(end-1-state['next'])//self.spec.period_ns+1)137 times=tuple(state['next']+i*self.spec.period_ns for i in range(count))138 if reason:139 values=(None,)*count140 else:141 kernel=linear_reference if self.backend=='reference' else linear_numpy142 values=kernel(times,left.time_ns,right.time_ns,left.value.value,right.value.value)143 for time,value in zip(times,values):144 yield self._row(state,time,None if value is None else float(value),145 'missing' if reason else 'estimated',reason,left,right)146 state['next']=time+self.spec.period_ns147148 def _resolve(self,state):149 """结算已收齐的同刻更新。参数: state。返回: GridRow迭代。异常: 数值/预算/取消错误。150 """151 right=state['pending']152 if right is None:153 return154 conflict=state['pending_conflict']155 yield from self._until(state,right.time_ns,right,conflict)156 # 命中真实点直接使用该点，即使前一端缺失/无效/间隔太大也不阻止真实观测。157 if state['next']==right.time_ns and state['next']<self.interval.end_ns:158 reason='same_time_conflict' if conflict else '' if right.value.valid else 'invalid_update'159 yield self._row(state,right.time_ns,None if reason else right.value.value,160 'missing' if reason else 'observed',reason,right,right)161 state['next']+=self.spec.period_ns162 _release_owned_ram(self,state['left_cost'],'_sample_cost')163 state.update(left=right,left_cost=state['pending_cost'],left_conflict=conflict,164 pending=None,pending_cost=0,pending_conflict=False)165166 def push(self,sample):167 """处理一个连续信号更新，不跨范围借样本。参数: sample历史记录。返回: 已确定网格行迭代。异常: 声明/顺序/资源错。168 """169 if self._closed or self._finished:170 raise ResultClosedError('网格已关闭或完成')171 if type(sample) not in (HistorySample,RecordSample) or sample.name not in self.states:172 raise TypeError('网格需要所选连续信号历史')173 state=self.states[sample.name]; definition=state['definition']174 if (sample.value.dtype is not definition.dtype or sample.unit!=definition.unit175 or sample.source_ref.source_name!=definition.source or sample.source_ref.field!=definition.field):176 raise ValueError('网格样本与原绑定不一致')177 if not self.interval.start_ns<=sample.time_ns<self.interval.end_ns:178 return179 pending=state['pending']180 if pending is not None and sample.time_ns<pending.time_ns:181 raise ValueError('网格输入同信号时间回退')182 if pending is not None and sample.time_ns==pending.time_ns:183 # 默认冲突产生明确缺值；last_by_sequence只按稳定原序号选定，不按到达次序。184 if sample.sequence==pending.sequence:185 raise ValueError('网格同刻原序号重复')186 conflict=state['pending_conflict'] or not _same_fact(pending,sample) and definition.same_time_policy=='error'187 if sample.sequence<pending.sequence:188 state['pending_conflict']=conflict189 return190 cost=estimate_record_bytes(sample)191 _reserve_owned_ram(self,cost,'_sample_cost')192 old=state['pending_cost']193 state.update(pending=sample,pending_cost=cost,pending_conflict=conflict)194 _release_owned_ram(self,old,'_sample_cost')195 else:196 yield from self._resolve(state)197 cost=estimate_record_bytes(sample)198 _reserve_owned_ram(self,cost,'_sample_cost')199 state.update(pending=sample,pending_cost=cost,pending_conflict=False)200201 def finish(self):202 """完整EOF后输出剩余真实点及缺后侧网格。参数: 无。返回: GridRow迭代。异常: 资源/取消错。203 """204 if self._closed or self._finished:205 raise ResultClosedError('网格不可再次完成')206 for state in self.states.values():207 yield from self._resolve(state)208 yield from self._until(state,self.interval.end_ns,None,False)209 self._finished=True210211 def close(self):212 """释放保留样本与有限状态，可重试。参数: 无。返回: None。异常: 原预算释放错误。213 """214 if self._closed:215 return216 self.states=None217 _release_owned_ram(self,self._sample_cost,'_sample_cost')218 _release_owned_ram(self,self._fixed_cost,'_fixed_cost')219 self._closed=True220