chronaxis.results.blf_writer
源码基线:8904c2f744951857659abd40aaa3535d57faf53c
1"""独占持有一份BLF输出文件,委托既有python-can格式写入,不重写BLF编码。23写入器与底层文件属于同一owner;文件头构造、压缩或最终头更新失败时关闭实际句柄。4第三方stop至多调用一次,关闭重试仅处理尚未关闭的文件,不重复压缩或重写完成头。5"""6from pathlib import Path7from .budget import _cleanup_owner8910class BLFWrite:11 """管理一个新的BLF文件及其python-can writer,不接收来源路径或DBC。"""1213 def __init__(self,path):14 """独占创建文件并安装writer。参数: 新.blf路径。返回: owner。异常: 原创建/文件头错误。1516 传入文件对象由python-can的stop关闭;本对象只在第三方未完成关闭时收尾同一句柄。17 构造失败仍保留blf_write_cleanup_owner,不借重新打开文件修补失败。18 """19 self.path=Path(path)20 self._file=self._writer=None21 self._closed=self._stop_attempted=self._stop_succeeded=False22 try:23 if self.path.suffix.lower()!='.blf':raise ValueError('BLF写入需要.blf路径')24 from ..can_dependencies import import_can_dependency25 can=import_can_dependency('can')26 self._file=self.path.open('xb')27 self._writer=can.BLFWriter(self._file)28 except BaseException as error:29 _cleanup_owner(self,error,'blf_write_cleanup_owner','BLF写入器构造')30 raise3132 @property33 def closed(self):34 """查询实际关闭状态。3536 参数: 无。返回: bool,不表示失败文件可发布。异常: 无。37 """38 return self._closed3940 def write(self,message):41 """写入当前原帧一次。4243 参数: message为当前can.Message。返回: None。异常: 原写入错误或已结束时RuntimeError。44 不重试写入,避免重复添加帧。45 """46 if self._closed or self._stop_attempted:raise RuntimeError('BLF写入器已结束')47 self._writer.on_message_received(message)4849 def finish(self):50 """完成压缩与最终文件头。5152 参数: 无。返回: None,仅真实stop成功可正常返回。53 异常: 原收尾错误;已经失败的最终写入不能重试,返回RuntimeError。54 """55 if self._closed and self._stop_succeeded:return56 if self._stop_attempted:raise RuntimeError('BLF最终写入已经尝试,失败不能重试')57 self.close()58 if not self._stop_succeeded:raise RuntimeError('BLF最终写入未完成')5960 def close(self):61 """关闭同一实际文件,第三方stop最多一次。6263 参数: 无。返回: None,重复关闭安全。64 异常: 保留原收尾错误;实际文件仍未关闭时附可重试owner,其他错误加入notes。65 """66 if self._closed:return67 primary=None68 if self._writer is not None and not self._stop_attempted:69 self._stop_attempted=True70 try:71 self._writer.stop()72 self._stop_succeeded=True73 except BaseException as error:74 primary=error75 try:76 if self._file is not None and not self._file.closed:self._file.close()77 except BaseException as error:78 if primary is None:primary=error79 else:primary.add_note('BLF实际文件关闭附加失败: '+type(error).__name__)80 if self._file is None or self._file.closed:81 self._file=self._writer=None82 self._closed=True83 if primary is not None:84 if not self._closed:setattr(primary,'blf_write_cleanup_owner',self)85 raise primary86