chronaxis.results.frame_rebuild
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""按同次解码值离线生成网格报文，并与未选/未知原帧稳定合并。23输入FrameSample含精确原帧、物理值、实际复用选择器和已知报文事实；本模块4不再读原日志/DBC或调用解码。先用有界磁盘排序按完整身份分组，每组只持相邻5样本，再将生成/透传/未生成说明按时间有界排序。连续量用独立reference或6NumPy内核；基础帧等距取左，只更新允许位。未知报文默认透传，可显式最近邻。7所有暂存、格式对象和输入游标由本owner关闭，失败不发布成品。8"""910from dataclasses import dataclass, replace11from fractions import Fraction12from itertools import groupby13import json14from math import isfinite15from pathlib import Path1617from ..core.export_interpolation import linear_reference,linear_numpy18from ..core.models import Interval19from ..execution.history import _metadata_bytes20from ..legacy.records import frame_to_dict,frame_from_dict21from ..timebase.conversion import require_int64_ns,checked_add_ns22from .budget import _reserve_owned_ram,_release_owned_ram,_cleanup_owner23from .codec import _pack,_unpack24from .errors import ResultClosedError25from .frame_exports import ExportCANFrame,SkippedCANFrame26from .frame_patch import FrameRebuildRule,DBCFramePatch27from .frame_sort import FrameSort282930@dataclass(frozen=True)31class FrameSample:32 """保同次已解码事实；layout用原始选择器整数，非物理缩放值。"""3334 record: ExportCANFrame35 values: tuple[tuple[str,int | float | bool | None],...] = ()36 layout: tuple[tuple[str,int],...] = ()37 known: bool = False38 valid: bool = True3940 def __post_init__(self):41 """核冻结值与定位结构。参数: 无。返回: None。异常: 类型/重复字段/布局非法。42 """43 if type(self.record) is not ExportCANFrame or type(self.known) is not bool or type(self.valid) is not bool:44 raise TypeError('重建样本需要真实导出帧和明确布尔事实')45 for pairs,types in ((self.values,(int,float,bool,type(None))),(self.layout,(int,))):46 if type(pairs) is not tuple or len(pairs)>4096:47 raise ValueError('重建样本字段超过有限能力')48 names=set()49 for pair in pairs:50 if type(pair) is not tuple or len(pair)!=2 or type(pair[0]) is not str or not pair[0] or type(pair[1]) not in types or pair[0] in names:51 raise ValueError('重建样本字段重复或类型非法')52 names.add(pair[0])535455@dataclass(frozen=True)56class FrameGridSpec:57 """统一整数网格及明确的插值/最近邻上限，不按文件或批次重新起算。"""5859 period_ns: int | None # None为各已选报文的DBC周期，运行前一次解析。60 origin_ns: int61 max_gap_ns: int62 unknown_policy: str = 'passthrough'63 max_distance_ns: int | None = None6465 def __post_init__(self):66 """核网格与未知报文策略。参数: 无。返回: None。异常: 非法时间/策略/距离。67 """68 for name in ('period_ns','origin_ns','max_gap_ns'):69 if name=='period_ns' and self.period_ns is None:continue70 require_int64_ns(getattr(self,name),name)71 if self.period_ns is not None and self.period_ns<=0 or self.max_gap_ns<=0 or self.unknown_policy not in ('passthrough','nearest'):72 raise ValueError('需要正网格/间隔和显式未知策略')73 if self.max_distance_ns is not None:74 require_int64_ns(self.max_distance_ns)75 if self.max_distance_ns<0:76 raise ValueError('最近邻距离不能为负')77 if self.period_ns is None and self.unknown_policy=='nearest':78 raise ValueError('未知报文没有DBC周期；最近邻网格必须给固定period')79 if self.unknown_policy=='nearest' and self.max_distance_ns is None:80 raise ValueError('未知报文最近邻必须声明最大距离')818283def _frame_packet(record):84 """编码固定原帧及定位结构。参数: record。返回: JSON dict。异常: 原模型非法。85 """86 frame=frame_to_dict(record.frame)87 if not record.provenance:88 # writer实际用output_time_ns；默认暂存不再保存它的float秒副本。89 frame.pop('timestamp')90 packet={'sequence':record.sequence,'time_ns':record.time_ns,'frame':frame,'frame_origin':record.frame_origin}91 if record.output_time_ns!=record.time_ns:92 packet['output_time_ns']=record.output_time_ns93 if record.provenance:94 packet['provenance']=record.provenance95 if not record.remote_dlc_known:96 packet['remote_dlc_known']=False97 return packet9899100def _restore_frame(packet):101 """恢复本次认证暂存的固定原帧。参数: packet。返回: ExportCANFrame。异常: 非法模型。102 """103 output=packet.get('output_time_ns',packet['time_ns'])104 frame=packet['frame']105 if 'timestamp' not in frame:106 frame=dict(frame,timestamp=output/10**9)107 return ExportCANFrame(packet['sequence'],packet['time_ns'],output,108 frame_from_dict(frame),tuple(tuple(p) for p in packet.get('provenance',())),packet['frame_origin'],109 remote_dlc_known=packet.get('remote_dlc_known',True))110111112def _sample_packet(sample):113 """保存无效NaN等诊断而不输出非法JSON。参数: sample。返回: 固定dict。异常: 编码错。114 """115 return {'record':_frame_packet(sample.record),'values':_pack(sample.values),'layout':sample.layout,116 'known':sample.known,'valid':sample.valid}117118119def _restore_sample(packet):120 """恢复已认证样本，不加载DBC。参数: packet。返回: FrameSample。异常: 结构错。121 """122 return FrameSample(_restore_frame(packet['record']),_unpack(packet['values']),123 tuple(tuple(p) for p in packet['layout']),packet['known'],packet['valid'])124125126def _facts(sample):127 """比较同刻计算事实，排除来源序号但保实际载荷/布局/质量。参数: sample。返回: 有限JSON。异常: 编码错。128 """129 return json.dumps((frame_to_dict(sample.record.frame),sample.record.output_time_ns,130 _pack(sample.values),sample.layout,sample.known,sample.valid),sort_keys=True,allow_nan=False)131132133class FrameRebuildRead:134 """拥有输入、两份排序仓和有限DBC位域对象；只在完整准备后开放输出。"""135136 def __init__(self,records,spec,interval,rules,budget,work_dir,*,backend='numpy',check_cancel=None,periods=None,integrity=True,trace=True):137 """静态核配置并预约资源。参数: records、网格/范围、(rule,已加载Message)tuple、预算/新目录、后端/取消；periods为同次初始化已解析DBC周期。138 返回: 可关闭重建owner。异常: 配置/预算/I/O错，清理失败附frame_rebuild_cleanup_owner。139 """140 if type(spec) is not FrameGridSpec or type(interval) is not Interval or backend not in ('reference','numpy'):141 raise ValueError('报文重建需要网格、范围及明确后端')142 if type(rules) is not tuple or len(rules)>4096 or any(type(p) is not tuple or len(p)!=2 or type(p[0]) is not FrameRebuildRule for p in rules):143 raise ValueError('重建规则需要有限(rule,Message)tuple')144 identities=[(r.channel,r.frame_id,r.is_extended,r.is_fd) for r,_ in rules]145 if len(set(identities))!=len(identities):146 raise ValueError('不能对同一报文身份声明两个重建规则')147 self._budget=budget; self._held_cost=self._fixed_cost=self._group_cost=self._row_cost=0148 self._closed=False; self.ready=False; self.reader=None; self.input_sort=self.output_sort=None149 self.patchers=[]; self.pending_patcher=None; self.cursor=None150 self.spec=spec; self.interval=interval; self.backend=backend; self.check_cancel=check_cancel151 self._export_metadata=None152 self.trace=trace153 self.input_count=self.generated=self.passthrough=self.skipped=0154 self.grid=0155 self._periods=()156 self._group_period=spec.period_ns157 try:158 _reserve_owned_ram(self,2*1048576+_metadata_bytes(tuple(rule for rule,_ in rules))*16,'_fixed_cost')159 self.reader=iter(records)160 if spec.period_ns is None:161 from ..legacy.dbc_cycles import cycle_time_ns162 self._periods=(tuple(cycle_time_ns(message.cycle_time) for _,message in rules) if periods is None else periods)163 if type(self._periods) is not tuple or len(self._periods)!=len(rules) or any(type(p) is not int or not 0<p<(1<<63) for p in self._periods):164 raise ValueError('DBC逐报文周期与已选规则不符')165 elif periods is not None:166 raise ValueError('固定网格不能再附DBC周期')167 root=Path(work_dir)168 root.mkdir(parents=True,exist_ok=False)169 self.input_sort=FrameSort(root/'groups',budget,check_cancel=check_cancel,integrity=integrity)170 self.output_sort=FrameSort(root/'timeline',budget,check_cancel=check_cancel,integrity=integrity)171 for rule,message in rules:172 try:173 self.pending_patcher=DBCFramePatch(rule,message,budget)174 except BaseException as error:175 self.pending_patcher=getattr(error,'frame_patch_cleanup_owner',None)176 raise177 self.patchers.append(self.pending_patcher); self.pending_patcher=None178 except BaseException as error:179 # 构造失败也接管尚未赋值成功的真实排序对象。180 if self.input_sort is None:181 self.input_sort=getattr(error,'frame_sort_cleanup_owner',None)182 elif self.output_sort is None:183 self.output_sort=getattr(error,'frame_sort_cleanup_owner',None)184 _cleanup_owner(self,error,'frame_rebuild_cleanup_owner','报文重建构造')185 raise186187 def _cancel(self):188 """检查已关闭及用户取消。参数: 无。返回: None。异常: 关闭/取消。189 """190 if self._closed:191 raise ResultClosedError('报文重建已关闭')192 if self.check_cancel is not None:193 self.check_cancel()194195 def _output(self,item,identity=()):196 """保存真实/生成/未生成结果到时间排序。参数: item和有限身份。返回: None。异常: 排序/配额错。197 """198 if type(item) is SkippedCANFrame:199 packet={'skip':True,'time_ns':item.time_ns,'reason':item.reason,'details':item.details}200 key=(item.time_ns,2,0,str(identity)); self.skipped+=1201 else:202 packet={'skip':False,'record':_frame_packet(item)}203 key=(item.time_ns,int(item.frame_origin=='generated'),item.sequence,str(identity))204 if item.frame_origin=='generated':205 self.generated+=1206 else:207 self.passthrough+=1208 self.output_sort.append(key,packet)209210 def prepare(self):211 """单次消费同次样本，按身份重建并封口时间序列。参数: 无。返回: self。异常: 输入/解码事实/排序/预算/取消。212 """213 self._cancel()214 if self.ready:215 return self216 try:217 last=sequence=None218 for sample in self.reader:219 self._cancel()220 if type(sample) is not FrameSample:221 raise TypeError('报文重建需要同次FrameSample流')222 record=sample.record223 _reserve_owned_ram(self,1048576+_metadata_bytes((sample.values,sample.layout,record.provenance))*32,'_row_cost')224 try:225 if last is not None and (record.time_ns<last or record.sequence<=sequence):226 raise ValueError('原帧时间或原序回退，不能借离线排序掩盖输入错误')227 last=record.time_ns; sequence=record.sequence; self.input_count+=1228 if not self.interval.start_ns<=record.time_ns<self.interval.end_ns:229 continue230 frame=record.frame231 if frame.is_rx is None:232 raise ValueError('原始报文重建需要收发方向is_rx')233 matched=[i for i,p in enumerate(self.patchers) if p.rule.matches(frame)]234 if len(matched)>1:235 raise ValueError('同一实际帧匹配多个重建规则，请收紧帧类型约束')236 rule_index=matched[0] if matched else None237 rebuild=not(frame.is_error_frame or frame.is_remote_frame) and (238 rule_index is not None or not sample.known and self.spec.unknown_policy=='nearest')239 if not rebuild:240 self._output(record); continue241 if rule_index is not None and (not sample.known or frame.dlc!=len(frame.data)242 or len(frame.data)!=self.patchers[rule_index].message.length):243 raise ValueError('已选报文缺真实DBC解码身份或长度异常')244 if rule_index is not None:245 self.patchers[rule_index].accept_frame(frame)246 # 收发方向/实际帧型/通道分组，布局切换留在组内作为明确禁止跨越的边界。247 identity=(rule_index if rule_index is not None else -1,json.dumps(frame.channel),frame.arbitration_id,248 int(frame.is_extended_id),int(frame.is_fd),int(frame.is_rx))249 self.input_sort.append(identity+(record.time_ns,record.sequence),_sample_packet(sample))250 finally:251 _release_owned_ram(self,self._row_cost,'_row_cost')252 self._close_input(); self.input_sort.seal()253 for identity,rows in groupby(self.input_sort.rows(),key=lambda row:row[0][:-2]):254 patcher=self.patchers[identity[0]] if identity[0]>=0 else None255 size=0 if patcher is None else len(patcher.rule.continuous)256 _reserve_owned_ram(self,8*1048576+size*16384,'_group_cost')257 try:258 self._group((_restore_sample(row[2]) for row in rows),patcher,identity)259 finally:260 _release_owned_ram(self,self._group_cost,'_group_cost')261 self.output_sort.seal()262 self._export_metadata={'period_ns':self.spec.period_ns,'origin_ns':self.spec.origin_ns,263 'max_gap_ns':self.spec.max_gap_ns,'unknown_policy':self.spec.unknown_policy,264 'max_distance_ns':self.spec.max_distance_ns,'backend':self.backend,265 'quantization':'nearest_even-v1','quantization_input':'decimal_endpoints_integer_time',266 'range_ns':(self.interval.start_ns,self.interval.end_ns),'input_frames':self.input_count,267 'generated':self.generated,'passthrough':self.passthrough,'skipped':self.skipped,268 'checksum_recomputed':False,'rules':tuple({'channel':p.rule.channel,'frame_id':p.rule.frame_id,269 **({} if p.rule.is_extended is None else {'is_extended':p.rule.is_extended}),270 **({} if p.rule.is_fd is None else {'is_fd':p.rule.is_fd}),271 'continuous':p.rule.continuous,'protected':p.rule.protected,272 'same_time_policy':p.rule.same_time_policy,273 **({'period_ns':self._periods[i]} if self.spec.period_ns is None else {})}274 for i,p in enumerate(self.patchers))}275 if self.spec.period_ns is None:276 del self._export_metadata['period_ns']277 self._export_metadata['period_mode']='dbc'278 self.ready=True279 return self280 except BaseException as error:281 _cleanup_owner(self,error,'frame_rebuild_cleanup_owner','报文重建')282 raise283284 def _group(self,samples,patcher,identity):285 """保相邻两个同刻组并处理全部网格。参数: 单身份样本流、patcher或None、identity。返回: None。异常: 原重建错。286 """287 spec=self.spec288 self._group_period=spec.period_ns if spec.period_ns is not None else self._periods[identity[0]]289 period=self._group_period290 self.grid=spec.origin_ns+((self.interval.start_ns-spec.origin_ns+period-1)//period)*period291 left=pending=None; left_bad=pending_bad=False292 for sample in samples:293 self._cancel()294 if pending is not None and sample.record.time_ns==pending.record.time_ns:295 if patcher is not None and patcher.rule.same_time_policy=='last_by_sequence':296 if sample.record.sequence>pending.record.sequence:297 pending=sample298 else:299 pending_bad=pending_bad or _facts(sample)!=_facts(pending)300 continue301 if pending is not None:302 self._segment(left,pending,left_bad,pending_bad,patcher,identity)303 left,left_bad=pending,pending_bad304 pending=sample; pending_bad=False305 if pending is not None:306 self._segment(left,pending,left_bad,pending_bad,patcher,identity)307 self._segment(pending,None,pending_bad,False,patcher,identity)308309 def _reason(self,left,right,left_bad,right_bad,patcher):310 """确定整个两点间隔是否可插值。参数: 两端及冲突、patcher。返回: 原因或None。异常: 非法DBC布局。311 """312 if left_bad or right_bad:313 return 'same_time_conflict'314 if left is None:315 return 'missing_left'316 if right is None:317 return 'missing_right'318 if left.layout!=right.layout:319 return 'layout_changed'320 if left.record.output_time_ns-left.record.time_ns!=right.record.output_time_ns-right.record.time_ns:321 return 'time_mapping_changed'322 if not left.valid or not right.valid:323 return 'invalid_neighbor'324 names=patcher.selected(left.layout)325 if not names:326 return 'inactive_layout'327 for sample in (left,right):328 values=dict(sample.values)329 if any(type(values.get(n)) not in (int,float) or not isfinite(values[n]) for n in names):330 return 'invalid_neighbor'331 if right.record.time_ns-left.record.time_ns>self.spec.max_gap_ns:332 return 'gap_too_large'333 return None334335 def _skip(self,time_ns,reason,identity):336 """记录整个目标帧未生成，不补任一物理字段。参数: time_ns/reason/identity。返回: None。异常: 输出错。337 """338 self._output(SkippedCANFrame(time_ns,reason,json.dumps({'identity':identity},ensure_ascii=False) if self.trace else ''),identity)339340 def _emit(self,time_ns,base,left,right,patcher,values,identity,mode):341 """从同一基础帧生成目标并保完整两端引用。参数: 目标/基础/两端、位域器和值、身份/模式。返回: None。异常: 量化/输出错。342 """343 frame=base.record.frame; quantization=()344 if values is not None:345 # 数组结果是物理估计；离散码值的恰半边界另用原十进制值及整数时间证明。346 # 这是编码舍入，不重算状态、不读取原载荷，也不将参考内核当NumPy实现。347 a=dict(left.values); b=dict(right.values)348 ratio=Fraction(time_ns-left.record.time_ns,right.record.time_ns-left.record.time_ns)349 exact={name:Fraction(str(a[name]))+(Fraction(str(b[name]))-Fraction(str(a[name])))*ratio for name in values}350 frame,quantization=patcher.patch(frame,values,base.layout,exact_values=exact,trace=self.trace)351 output_ns=checked_add_ns(base.record.output_time_ns,time_ns-base.record.time_ns)352 provenance=()353 if self.trace:354 details={'mode':mode,'base_sequence':base.record.sequence,'base_ref':dict(base.record.provenance),355 'left_sequence':None if left is None else left.record.sequence,356 'right_sequence':None if right is None else right.record.sequence,357 'left_ref':None if left is None else dict(left.record.provenance),358 'right_ref':None if right is None else dict(right.record.provenance),359 'layout':base.layout,'quantization_rule':'nearest_even-v1','quantization':quantization,360 'checksum_recomputed':False}361 provenance=(('reconstruction',json.dumps(details,ensure_ascii=False,allow_nan=False,separators=(',',':'))),)362 result=ExportCANFrame(base.record.sequence,time_ns,output_ns,replace(frame,timestamp=output_ns/10**9),provenance,'generated')363 self._output(result,identity)364365 def _nearest(self,time_ns,left,right,left_bad,right_bad,identity):366 """未知报文整帧最近邻，等距选左且不越最大距离。参数: 时间/两端及冲突/身份。返回: None。异常: 输出错。367 """368 choices=[s for s,bad in ((left,left_bad),(right,right_bad)) if s is not None and not bad and s.valid]369 if not choices:370 self._skip(time_ns,'same_time_conflict' if left_bad or right_bad else 'no_neighbor',identity); return371 base=min(choices,key=lambda s:(abs(s.record.time_ns-time_ns),s.record.time_ns))372 if abs(base.record.time_ns-time_ns)>self.spec.max_distance_ns:373 self._skip(time_ns,'neighbor_too_far',identity); return374 self._emit(time_ns,base,left,right,None,None,identity,'nearest_frame')375376 def _segment(self,left,right,left_bad,right_bad,patcher,identity):377 """每块最多256网格点计算连续量，实点另外保原字节。参数: 两端/冲突、patcher、identity。返回: None。异常: 原计算/输出错。378 """379 stop=self.interval.end_ns if right is None else min(right.record.time_ns,self.interval.end_ns)380 reason=None if patcher is None else self._reason(left,right,left_bad,right_bad,patcher)381 while self.grid<stop:382 self._cancel()383 count=min(256,(stop-self.grid+self._group_period-1)//self._group_period)384 times=tuple(self.grid+i*self._group_period for i in range(count))385 arrays={}386 if patcher is not None and reason is None:387 names=patcher.selected(left.layout); a=dict(left.values); b=dict(right.values)388 function=linear_reference if self.backend=='reference' else linear_numpy389 arrays={n:function(times,left.record.time_ns,right.record.time_ns,a[n],b[n]) for n in names}390 for index,time in enumerate(times):391 if patcher is None:392 self._nearest(time,left,right,left_bad,right_bad,identity)393 elif reason is not None:394 self._skip(time,reason,identity)395 else:396 base=left if time-left.record.time_ns<=right.record.time_ns-time else right397 self._emit(time,base,left,right,patcher,{n:float(values[index]) for n,values in arrays.items()},identity,'linear')398 self.grid+=count*self._group_period399 if right is not None and self.grid==right.record.time_ns and self.grid<self.interval.end_ns:400 if patcher is None:401 self._nearest(self.grid,right,right,right_bad,right_bad,identity)402 elif right_bad:403 self._skip(self.grid,'same_time_conflict',identity)404 else:405 names=patcher.selected(right.layout); values=dict(right.values)406 if not right.valid or not names or any(type(values.get(n)) not in (int,float) or not isfinite(values[n]) for n in names):407 self._skip(self.grid,'invalid_update',identity)408 else:409 self._emit(self.grid,right,right,right,patcher,None,identity,'observed')410 self.grid+=self._group_period411412 def __iter__(self):413 """准备完成后开放一个时间游标。参数: 无。返回: self。异常: 准备或重复游标错。414 """415 self.prepare()416 if self.cursor is None:417 self.cursor=self.output_sort.rows()418 return self419420 @property421 def export_metadata(self):422 """返回本次已完成重建的有限清单事实。参数: 无。返回: dict或None。异常: 无。423 """424 return self._export_metadata425426 def __next__(self):427 """恢复下一生成/透传帧或缺失说明。参数: 无。返回: ExportCANFrame/SkippedCANFrame。异常: EOF/校验错。428 """429 if self.cursor is None:430 iter(self)431 self._cancel(); packet=next(self.cursor)[2]432 if packet['skip']:433 return SkippedCANFrame(packet['time_ns'],packet['reason'],packet['details'])434 return _restore_frame(packet['record'])435436 def _close_input(self):437 """真正关闭源样本游标。参数: 无。返回: None。异常: 原关闭错保责任。438 """439 if self.reader is not None:440 close=getattr(self.reader,'close',None)441 if close is not None:442 close()443 self.reader=None444445 def close(self):446 """关闭真实所有权后释放费用，保诊断暂存。参数: 无。返回: None。异常: 原关闭/释放错可重试。447 """448 if self._closed:449 return450 self._close_input()451 if self.cursor is not None:452 self.cursor.close(); self.cursor=None453 for owner in (self.input_sort,self.output_sort,self.pending_patcher,*self.patchers):454 if owner is not None:455 owner.close()456 self.patchers.clear(); self.pending_patcher=None457 _release_owned_ram(self,self._row_cost,'_row_cost')458 _release_owned_ram(self,self._group_cost,'_group_cost')459 _release_owned_ram(self,self._fixed_cost,'_fixed_cost'); self._closed=True460