chronaxis.execution.frame_capture
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""为显式报文导出保存同次记录和新编码所需DBC上下文。23默认关闭时不创建这些对象。TextFrameStore只保存本次文本读取已交付的原帧、4物理值和定位，可为多个窗口有限回放；不重新读原文件。FrameDBCContext负责5新编码的DBC身份/字段声明，可借给新文本解码器共用一次加载。旧BLF链自身的6DBC加载和解码不改变；为新增编码另加载同一哈希的定义属于新输出的显式成本。7"""89from hashlib import file_digest10from pathlib import Path1112from ..dbc_loading import load_dbc_file1314from ..legacy.records import LegacyCANField15from ..legacy.binding import dbc_candidates, select_message16from ..results.budget import _reserve_owned_ram,_release_owned_ram,_cleanup_owner17from ..results.errors import ResultClosedError18from ..results.frame_exports import as_export_frame19from ..results.frame_patch import DBCFramePatch20from ..results.frame_rebuild import FrameSample,_sample_packet,_restore_sample21from ..results.frame_sort import FrameSort22from .history import _metadata_bytes232425class FrameDBCContext:26 """拥有供新输出使用的DBC定义、摘要和额外请求字段，借用者不负责释放。"""2728 def __init__(self,path,rules,budget,*,cycle_periods=False,integrity=True):29 """加载并核身份，生成连续量/选择器请求。参数: path、规则tuple、budget。返回: owner。异常: DBC/预算/字段/I/O错。30 """31 if type(integrity) is not bool: raise TypeError("integrity需要bool")32 self.path=Path(path); self.rules=rules; self._budget=budget33 self._held_cost=self._fixed_cost=0; self.closed=self._closed=False34 self.database=self._handle=self._patcher=None; self.sha256=None; self.extra_fields=();self.periods=None35 self._candidates=(); self._selected=[]; self._identities=[]; self._cycle_periods=cycle_periods36 self._by_frame_id={}; self._cycle_values={}37 try:38 size=self.path.stat().st_size39 _reserve_owned_ram(self,1048576+size*64+_metadata_bytes(rules)*32,'_fixed_cost')40 before_stat=self.path.stat()41 before=self._hash() if integrity else None42 from ..can_dependencies import import_can_dependency43 cantools = import_can_dependency('cantools')44 self.database=load_dbc_file(str(self.path), cantools)45 # 导出的未知报文策略需要判断DBC是否认识该帧；一次建索引，不逐帧扫描DBC。46 for message in self.database.messages:47 self._by_frame_id.setdefault(message.frame_id,[]).append(message)48 after_stat=self.path.stat()49 changed=(before_stat.st_size,before_stat.st_mtime_ns,before_stat.st_ino)!=(after_stat.st_size,after_stat.st_mtime_ns,after_stat.st_ino)50 if changed or integrity and self._hash()!=before:51 raise ValueError('新增编码加载DBC期间内容发生变化')52 self.sha256=before; self.version=cantools.__version__53 fields=[]; candidates=[]54 for rule in rules:55 choices=dbc_candidates(self.database,rule.frame_id,(*rule.continuous,*rule.protected),rule.is_extended,rule.is_fd)56 if not choices:57 raise ValueError('DBC缺少重建报文或指定信号: '+str(rule.frame_id))58 candidates.append(choices)59 # 候选定义均须可重建；最终选择等实际帧到达，不按DBC顺序取第一个。60 for message in choices:61 if cycle_periods and id(message) not in self._cycle_values:62 # 声明DBC网格时，在打开日志前拒绝缺失或非法周期；只读取一次。63 from ..legacy.dbc_cycles import cycle_time_ns64 self._cycle_values[id(message)]=cycle_time_ns(message.cycle_time)65 try:66 self._patcher=DBCFramePatch(rule,message,budget)67 except BaseException as error:68 self._patcher=getattr(error,'frame_patch_cleanup_owner',None)69 raise70 self._patcher.close(); self._patcher=None71 names=tuple(dict.fromkeys((*rule.continuous,*(s.name for s in message.signals if s.is_multiplexer))))72 fields.extend(LegacyCANField(rule.channel,name,rule.frame_id,message.is_extended_frame,rule.is_fd) for name in names)73 self.extra_fields=tuple(dict.fromkeys(fields))74 self._candidates=tuple(candidates)75 self._selected=[None]*len(rules); self._identities=[None]*len(rules)76 except BaseException as error:77 _cleanup_owner(self,error,'frame_dbc_cleanup_owner','导出DBC上下文')78 raise7980 def _hash(self):81 """核同一文件的完整内容身份。参数: 无。返回: SHA256文本。异常: 打开/读取/真实关闭错。82 """83 self._handle=self.path.open('rb')84 digest=file_digest(self._handle,'sha256').hexdigest()85 self._handle.close(); self._handle=None86 return digest8788 def require_open(self):89 """禁止使用已释放DBC。参数: 无。返回: None。异常: ResultClosedError。90 """91 if self.closed:92 raise ResultClosedError('导出DBC上下文已关闭')9394 def pairs(self):95 """交付规则与同次定义的有限借用。参数: 无。返回: tuple。异常: 已关闭或身份缺失。96 """97 self.require_open()98 pairs=[]99 for index,rule in enumerate(self.rules):100 message=self._selected[index]101 if message is None:102 choices=self._candidates[index]103 # 从未收到该报文时无需构造一个猜测的多候选输出规则。104 if len(choices)!=1:105 continue106 message=choices[0]107 pairs.append((rule,message))108 if self._cycle_periods:109 self.periods=tuple(self._cycle_values[id(message)] for _,message in pairs)110 return tuple(pairs)111112 def observe(self,frame):113 """用正在读取的实际帧确定输出定义，禁止一条自动规则混用两种身份。114115 参数: frame为原始帧。返回: 匹配的DBC Message或None。异常: DBC或实际身份歧义。116 副作用: 每条规则只保存一个已识别定义和身份，不保留额外帧或扫描文件。117 """118 self.require_open()119 if frame.is_remote_frame or frame.is_error_frame:120 return None121 selected=[]122 for index,rule in enumerate(self.rules):123 if not rule.matches(frame):124 continue125 message=select_message(self._candidates[index],frame)126 if message is None:127 continue128 identity=(frame.is_extended_id,frame.is_fd)129 prior=self._identities[index]130 if prior is not None and prior!=identity:131 differing='is_extended' if prior[0]!=identity[0] else 'is_fd'132 raise ValueError('报文重建规则匹配到多种实际帧，请补充'+differing)133 self._identities[index]=identity; self._selected[index]=message134 selected.append(message)135 if len(selected)>1:136 raise ValueError('同一实际报文匹配多个重建规则，请收紧帧类型约束')137 return selected[0] if selected else None138139 def is_known_frame(self,frame):140 """只判断未知报文输出策略，不要求未选报文的信号或定义唯一。141142 参数: frame为原始帧。返回: bool。异常: 已关闭时ResultClosedError。143 副作用: 无，不解码或记录帧内容。144 """145 self.require_open()146 return any(message.is_extended_frame==frame.is_extended_id147 and (not frame.is_fd or message.is_fd)148 for message in self._by_frame_id.get(frame.arbitration_id,()))149150 def close(self):151 """在借用者关闭后释放定义和实际句柄。参数: 无。返回: None。异常: 原关闭/释放错可恢复。152 """153 if self.closed:154 return155 if self._handle is not None:156 self._handle.close(); self._handle=None157 if self._patcher is not None:158 self._patcher.close(); self._patcher=None159 self.database=None; self.extra_fields=();self.periods=None160 self._candidates=(); self._selected=[]; self._identities=[]; self._by_frame_id={}; self._cycle_values={}161 _release_owned_ram(self,self._fixed_cost,'_fixed_cost'); self.closed=self._closed=True162163164class TextFrameStore:165 """保存连续原序的同次文本CAN事实；封口后允许多个窗口逐个回放。"""166167 def __init__(self,path,budget,source,session_id,read_identity,*,database_context=None,channel=None,integrity=True,trace=True):168 """建立有限排序暂存及真实读取身份。参数: 新路径、budget、源/会话/read身份，可借用重建DBC及声明通道。返回: owner。异常: I/O/预算错。169 """170 self.path=Path(path); self._budget=budget; self._held_cost=self._fixed_cost=self._row_cost=0171 self.source_name=source; self.session_id=session_id; self.read_identity=read_identity172 self.database_context=database_context; self.channel=channel; self.trace=trace173 self.sort=None; self.count=0; self.last_ns=None; self.state='building'; self.closed=self._closed=False174 try:175 _reserve_owned_ram(self,131072,'_fixed_cost')176 try:177 self.sort=FrameSort(self.path,budget,integrity=integrity)178 except BaseException as error:179 self.sort=getattr(error,'frame_sort_cleanup_owner',None)180 raise181 except BaseException as error:182 _cleanup_owner(self,error,'text_frame_store_cleanup_owner','文本原帧暂存构造')183 raise184185 def append(self,record):186 """保原序原时间，不让排序掩盖回退。参数: DecodedCANRecord同次记录。返回: None。异常: 时序/状态/预算/I/O错。187 """188 if self.closed or self.state!='building' or record.sequence!=self.count:189 raise ValueError('文本原帧仓状态或连续序号不符')190 if self.last_ns is not None and record.raw.time_ns<self.last_ns:191 raise ValueError('文本原帧时间回退')192 _reserve_owned_ram(self,1048576+_metadata_bytes((record.values,record.frame_values,record.frame_layout))*32,'_row_cost')193 try:194 known=record.status!='unknown'195 if self.database_context is not None:196 self.database_context.require_open()197 frame=record.raw.frame198 # 实际类型只在同次读取中匹配，输出不使用按ID覆盖的DBC字典。199 message=self.database_context.observe(frame)200 known=frame.channel==self.channel and self.database_context.is_known_frame(frame)201 sample=FrameSample(as_export_frame(record,compact=not self.trace),record.frame_values,record.frame_layout,known)202 self.sort.append((record.sequence,),_sample_packet(sample))203 self.count+=1; self.last_ns=record.raw.time_ns204 finally:205 _release_owned_ram(self,self._row_cost,'_row_cost')206207 def seal(self):208 """完整真实EOF后封口。参数: 无。返回: None。异常: 状态/配额/I/O错。209 """210 if self.closed or self.state!='building':211 raise ResultClosedError('文本原帧仓不能重复封口')212 self.sort.seal(); self.state='sealed'213214 def iter_samples(self):215 """打开一个有界完整回放，提前关闭不声称完整核验。参数: 无。返回: FrameSample生成器。异常: 仓状态/篡改/I/O错。216 """217 if self.closed or self.state!='sealed':218 raise ResultClosedError('文本原帧仓不可读')219 cursor=self.sort._read(self.sort.path/f'{self.sort.level}-0.jsonl')220 try:221 for _,sequence,packet in cursor:222 sample=_restore_sample(packet)223 if sample.record.sequence!=sequence:224 raise ValueError('文本原帧暂存序号不一致')225 yield sample226 finally:227 cursor.close()228229 def iter_frames(self):230 """只交付原帧及定位，不重新解码。参数: 无。返回: ExportCANFrame生成器。异常: 原回放错。231 """232 cursor=self.iter_samples()233 try:234 for sample in cursor:235 yield sample.record236 finally:237 cursor.close()238239 def close(self):240 """关闭全部真实回放FD和排序仓，保诊断文件。参数: 无。返回: None。异常: 清理错可重试。241 """242 if self.closed:243 return244 if self.sort is not None:245 self.sort.close()246 _release_owned_ram(self,self._row_cost,'_row_cost')247 _release_owned_ram(self,self._fixed_cost,'_fixed_cost'); self.closed=self._closed=True248