python开发:blf格式CAN数据转 asc格式文件工具
本文介绍了一个基于Python开发的BLF格式转ASC格式的CAN数据转换工具。该工具使用python-can库处理CAN日志文件,并采用tkinter构建GUI界面。开发环境需安装Python 3.6+和python-can库,推荐使用VS Code或PyCharm作为开发工具。项目采用面向对象设计,包含主应用类、文件转换类和界面组件。核心转换功能通过BLFReader和ASCWriter实现C
python开发:blf格式CAN数据转 asc格式文件工具
1. 概述
这个工具将使用 Python 语言开发,结合 python-can 库处理 CAN 日志文件,以及 tkinter 库创建图形用户界面,开发一个blf格式转asc格式工具。
2. 开发环境准备
2.1 安装 Python
首先需要安装 Python 3.6 或更高版本。你可以从 Python 官方网站 下载适合你操作系统的安装包。
2.2 安装依赖库
使用 pip 安装所需的库:
pip install python-can
2.3 开发工具
推荐使用 Visual Studio Code 或 PyCharm 作为开发环境,它们提供了良好的 Python 支持和代码编辑体验。
3. 项目结构设计
我们的项目将采用面向对象的方式设计,主要包含以下组件:
- 主应用类:负责管理整个应用的生命周期和界面
- 文件转换类:处理 BLF 到 ASC 的转换逻辑
- 界面组件:包括文件选择、进度显示、日志记录等功能模块
4. 核心功能实现
4.1 BLF 到 ASC 的转换功能
首先实现文件转换的核心功能:
from can.io import BLFReader, ASCWriter
def convert_blf_to_asc(blf_file_path, asc_file_path):
try:
# 打开 BLF 文件
with BLFReader(blf_file_path) as log_reader:
# 创建 ASC 文件写入器
with ASCWriter(asc_file_path) as log_writer:
# 遍历 BLF 文件中的每条消息并写入 ASC 文件
for msg in log_reader:
log_writer.on_message_received(msg)
print(f"成功将 {blf_file_path} 转换为 {asc_file_path}")
except FileNotFoundError:
print(f"错误:未找到文件 {blf_file_path}")
except Exception as e:
print(f"发生未知错误: {e}")
4.2 图形用户界面设计
使用 tkinter 创建用户界面:
import tkinter as tk
from tkinter import filedialog, ttk, messagebox
import threading
import time
from datetime import datetime
class BLFtoASCConverterApp:
def __init__(self, root):
self.root = root
self.root.title("CANoe BLF转ASC文件转换器")
self.root.geometry("600x450")
# 初始化UI组件
self._create_widgets()
# 转换状态变量
self.total_files = 0
self.processed_files = 0
self.successful_conversions = 0
self.failed_conversions = 0
self.is_converting = False
def _create_widgets(self):
# 创建主框架
main_frame = ttk.Frame(self.root, padding="20")
main_frame.pack(fill=tk.BOTH, expand=True)
# 文件选择部分
file_frame = ttk.LabelFrame(main_frame, text="文件选择", padding="10")
file_frame.pack(fill=tk.X, pady=(0, 10))
ttk.Label(file_frame, text="BLF文件:").grid(row=0, column=0, sticky=tk.W, pady=5)
self.file_path_var = tk.StringVar()
ttk.Entry(file_frame, textvariable=self.file_path_var, width=40).grid(row=0, column=1, padx=5, pady=5)
ttk.Button(file_frame, text="浏览...", command=self._browse_files).grid(row=0, column=2, padx=5, pady=5)
# 选项部分
options_frame = ttk.LabelFrame(main_frame, text="转换选项", padding="10")
options_frame.pack(fill=tk.X, pady=(0, 10))
self.overwrite_var = tk.BooleanVar(value=True)
ttk.Checkbutton(options_frame, text="覆盖已存在的ASC文件", variable=self.overwrite_var).grid(row=0, column=0, sticky=tk.W, pady=5)
self.verbose_var = tk.BooleanVar(value=True)
ttk.Checkbutton(options_frame, text="详细日志输出", variable=self.verbose_var).grid(row=0, column=1, sticky=tk.W, pady=5)
# 进度部分
progress_frame = ttk.LabelFrame(main_frame, text="转换进度", padding="10")
progress_frame.pack(fill=tk.X, pady=(0, 10))
self.progress_var = tk.DoubleVar()
self.progress_bar = ttk.Progressbar(progress_frame, variable=self.progress_var, length=100, mode='determinate')
self.progress_bar.pack(fill=tk.X, pady=5)
self.status_var = tk.StringVar(value="就绪")
ttk.Label(progress_frame, textvariable=self.status_var).pack(anchor=tk.W, pady=2)
# 统计信息
stats_frame = ttk.LabelFrame(main_frame, text="转换统计", padding="10")
stats_frame.pack(fill=tk.X, pady=(0, 10))
self.stats_var = tk.StringVar(value="待转换文件: 0 | 成功: 0 | 失败: 0")
ttk.Label(stats_frame, textvariable=self.stats_var).pack(anchor=tk.W, pady=2)
# 日志部分
log_frame = ttk.LabelFrame(main_frame, text="转换日志", padding="10")
log_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 10))
self.log_text = tk.Text(log_frame, wrap=tk.WORD, height=8)
self.log_text.pack(fill=tk.BOTH, expand=True, pady=5)
scrollbar = ttk.Scrollbar(self.log_text, command=self.log_text.yview)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
self.log_text.config(yscrollcommand=scrollbar.set)
# 按钮部分
button_frame = ttk.Frame(main_frame)
button_frame.pack(fill=tk.X, pady=(10, 0))
ttk.Button(button_frame, text="开始转换", command=self._start_conversion, width=15).pack(side=tk.LEFT, padx=5)
ttk.Button(button_frame, text="清空日志", command=self._clear_log, width=15).pack(side=tk.LEFT, padx=5)
ttk.Button(button_frame, text="退出", command=self.root.quit, width=15).pack(side=tk.RIGHT, padx=5)
# 其他方法实现(略)
4.3 实现文件选择功能
添加文件选择对话框功能:
def _browse_files(self):
files = filedialog.askopenfilenames(
title="选择BLF文件",
filetypes=[("BLF Files", "*.blf"), ("All Files", "*.*")]
)
if files:
file_names = [os.path.basename(file) for file in files]
self.file_path_var.set(f"{len(files)}个文件已选择")
self.selected_files = files
4.4 实现多线程转换
为避免转换过程中界面卡顿,使用多线程处理:
def _start_conversion(self):
if not hasattr(self, 'selected_files') or not self.selected_files:
messagebox.showwarning("警告", "请先选择要转换的BLF文件")
return
if self.is_converting:
messagebox.showinfo("提示", "正在进行转换,请等待当前任务完成")
return
# 初始化转换状态
self.total_files = len(self.selected_files)
self.processed_files = 0
self.successful_conversions = 0
self.failed_conversions = 0
self.is_converting = True
# 更新UI状态
self.progress_var.set(0)
self.status_var.set(f"准备转换 {self.total_files} 个文件...")
self.stats_var.set(f"待转换文件: {self.total_files} | 成功: 0 | 失败: 0")
self._clear_log()
self._log_message(f"开始转换: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
# 在单独的线程中执行转换
conversion_thread = threading.Thread(target=self._perform_conversion)
conversion_thread.daemon = True
conversion_thread.start()
4.5 实现详细的转换逻辑
添加完整的转换逻辑,包括错误处理和进度更新:
def _perform_conversion(self):
for file_path in self.selected_files:
if not self.is_converting:
break
blf_file = os.path.basename(file_path)
asc_file = os.path.splitext(file_path)[0] + ".asc"
self._log_message(f"\n正在处理: {blf_file}")
self.status_var.set(f"正在处理: {blf_file}")
# 检查文件是否存在
if os.path.exists(asc_file) and not self.overwrite_var.get():
self._log_message(f"跳过: {os.path.basename(asc_file)} 已存在")
self.failed_conversions += 1
else:
try:
start_time = time.time()
self._convert_blf_to_asc(file_path, asc_file)
elapsed_time = time.time() - start_time
self._log_message(f"成功: 转换完成,耗时 {elapsed_time:.2f} 秒")
self.successful_conversions += 1
except Exception as e:
self._log_message(f"错误: {str(e)}")
self.failed_conversions += 1
self.processed_files += 1
progress = (self.processed_files / self.total_files) * 100
self.progress_var.set(progress)
self.stats_var.set(f"待转换文件: {self.total_files - self.processed_files} | 成功: {self.successful_conversions} | 失败: {self.failed_conversions}")
# 转换完成
self.is_converting = False
self.status_var.set("转换完成")
self._log_message(f"\n转换完成: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
self._log_message(f"总文件数: {self.total_files}, 成功: {self.successful_conversions}, 失败: {self.failed_conversions}")
# 在主线程中显示完成消息
self.root.after(0, lambda: messagebox.showinfo("完成", f"转换完成!\n总文件数: {self.total_files}\n成功: {self.successful_conversions}\n失败: {self.failed_conversions}"))
def _convert_blf_to_asc(self, blf_file_path, asc_file_path):
try:
# 打开 BLF 文件
with BLFReader(blf_file_path) as log_reader:
# 创建 ASC 文件写入器
with ASCWriter(asc_file_path) as log_writer:
# 获取总消息数用于进度显示
if self.verbose_var.get():
total_messages = getattr(log_reader, 'number_of_messages', None)
if total_messages is not None:
self._log_message(f"总消息数: {total_messages}")
# 遍历 BLF 文件中的每条消息
messages_processed = 0
for msg in log_reader:
# 将消息写入 ASC 文件
log_writer.on_message_received(msg)
# 每处理1000条消息更新一次日志
messages_processed += 1
if messages_processed % 1000 == 0 and total_messages:
progress = (messages_processed / total_messages) * 100
self._log_message(f"已处理消息: {messages_processed}/{total_messages} ({progress:.1f}%)")
except FileNotFoundError:
raise Exception(f"未找到文件: {blf_file_path}")
except Exception as e:
raise Exception(f"转换过程中发生错误: {str(e)}")
4.6 实现日志记录功能
添加日志记录和界面更新方法:
def _log_message(self, message):
# 在主线程中更新日志
self.root.after(0, lambda: self._update_log_text(message))
def _update_log_text(self, message):
self.log_text.insert(tk.END, message + "\n")
self.log_text.see(tk.END)
self.conversion_log.append(message)
def _clear_log(self):
self.log_text.delete(1.0, tk.END)
5. 主程序入口
添加程序入口点:
if __name__ == "__main__":
root = tk.Tk()
app = BLFtoASCConverterApp(root)
root.mainloop()
6. 代码优化与扩展
6.1 添加中文支持
为确保中文正常显示,可在初始化时设置字体:
# 设置中文字体支持
self.style = ttk.Style()
self.style.configure("TLabel", font=("SimHei", 10))
self.style.configure("TButton", font=("SimHei", 10))
self.style.configure("TProgressbar", thickness=20)
7. 打包
使用 pyinstaller 将程序打包为可执行文件:
pip install pyinstaller
pyinstaller --onefile --windowed blf_to_asc_converter.py
可以生成一个独立的可执行文件,无需用户安装 Python 和依赖库。
8. 完整代码
```python
import can
from can.io import BLFReader, ASCWriter
import tkinter as tk
from tkinter import filedialog, ttk, messagebox
import os
import threading
import time
from datetime import datetime
class BLFtoASCConverterApp:
def __init__(self, root):
self.root = root
self.root.title("CANoe BLF转ASC文件转换器")
self.root.geometry("600x550")
self.root.resizable(True, True)
# 设置中文字体支持
self.style = ttk.Style()
self.style.configure("TLabel", font=("SimHei", 10))
self.style.configure("TButton", font=("SimHei", 10))
self.style.configure("TProgressbar", thickness=20)
# 创建界面元素
self._create_widgets()
# 转换状态变量
self.total_files = 0
self.processed_files = 0
self.successful_conversions = 0
self.failed_conversions = 0
self.conversion_log = []
self.is_converting = False
def _create_widgets(self):
# 创建主框架
main_frame = ttk.Frame(self.root, padding="20")
main_frame.pack(fill=tk.BOTH, expand=True)
# 文件选择部分
file_frame = ttk.LabelFrame(main_frame, text="文件选择", padding="10")
file_frame.pack(fill=tk.X, pady=(0, 10))
ttk.Label(file_frame, text="BLF文件:").grid(row=0, column=0, sticky=tk.W, pady=5)
self.file_path_var = tk.StringVar()
ttk.Entry(file_frame, textvariable=self.file_path_var, width=40).grid(row=0, column=1, padx=5, pady=5)
ttk.Button(file_frame, text="浏览...", command=self._browse_files).grid(row=0, column=2, padx=5, pady=5)
# 选项部分
options_frame = ttk.LabelFrame(main_frame, text="转换选项", padding="10")
options_frame.pack(fill=tk.X, pady=(0, 10))
self.overwrite_var = tk.BooleanVar(value=True)
ttk.Checkbutton(options_frame, text="覆盖已存在的ASC文件", variable=self.overwrite_var).grid(row=0, column=0, sticky=tk.W, pady=5)
self.verbose_var = tk.BooleanVar(value=True)
ttk.Checkbutton(options_frame, text="详细日志输出", variable=self.verbose_var).grid(row=0, column=1, sticky=tk.W, pady=5)
# 进度部分
progress_frame = ttk.LabelFrame(main_frame, text="转换进度", padding="10")
progress_frame.pack(fill=tk.X, pady=(0, 10))
self.progress_var = tk.DoubleVar()
self.progress_bar = ttk.Progressbar(progress_frame, variable=self.progress_var, length=100, mode='determinate')
self.progress_bar.pack(fill=tk.X, pady=5)
self.status_var = tk.StringVar(value="就绪")
ttk.Label(progress_frame, textvariable=self.status_var).pack(anchor=tk.W, pady=2)
# 统计信息
stats_frame = ttk.LabelFrame(main_frame, text="转换统计", padding="10")
stats_frame.pack(fill=tk.X, pady=(0, 10))
self.stats_var = tk.StringVar(value="待转换文件: 0 | 成功: 0 | 失败: 0")
ttk.Label(stats_frame, textvariable=self.stats_var).pack(anchor=tk.W, pady=2)
# 日志部分
log_frame = ttk.LabelFrame(main_frame, text="转换日志", padding="10")
log_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 10))
self.log_text = tk.Text(log_frame, wrap=tk.WORD, height=8, font=("SimHei", 9))
self.log_text.pack(fill=tk.BOTH, expand=True, pady=5)
scrollbar = ttk.Scrollbar(self.log_text, command=self.log_text.yview)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
self.log_text.config(yscrollcommand=scrollbar.set)
# 按钮部分
button_frame = ttk.Frame(main_frame)
button_frame.pack(fill=tk.X, pady=(10, 0))
ttk.Button(button_frame, text="开始转换", command=self._start_conversion, width=15).pack(side=tk.LEFT, padx=5)
ttk.Button(button_frame, text="清空日志", command=self._clear_log, width=15).pack(side=tk.LEFT, padx=5)
ttk.Button(button_frame, text="退出", command=self.root.quit, width=15).pack(side=tk.RIGHT, padx=5)
def _browse_files(self):
files = filedialog.askopenfilenames(
title="选择BLF文件",
filetypes=[("BLF Files", "*.blf"), ("All Files", "*.*")]
)
if files:
file_names = [os.path.basename(file) for file in files]
self.file_path_var.set(f"{len(files)}个文件已选择")
self.selected_files = files
def _start_conversion(self):
if not hasattr(self, 'selected_files') or not self.selected_files:
messagebox.showwarning("警告", "请先选择要转换的BLF文件")
return
if self.is_converting:
messagebox.showinfo("提示", "正在进行转换,请等待当前任务完成")
return
# 初始化转换状态
self.total_files = len(self.selected_files)
self.processed_files = 0
self.successful_conversions = 0
self.failed_conversions = 0
self.conversion_log = []
self.is_converting = True
# 更新UI状态
self.progress_var.set(0)
self.status_var.set(f"准备转换 {self.total_files} 个文件...")
self.stats_var.set(f"待转换文件: {self.total_files} | 成功: 0 | 失败: 0")
self._clear_log()
self._log_message(f"开始转换: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
# 在单独的线程中执行转换,避免阻塞UI
conversion_thread = threading.Thread(target=self._perform_conversion)
conversion_thread.daemon = True
conversion_thread.start()
def _perform_conversion(self):
for file_path in self.selected_files:
if not self.is_converting:
break
blf_file = os.path.basename(file_path)
asc_file = os.path.splitext(file_path)[0] + ".asc"
self._log_message(f"\n正在处理: {blf_file}")
self.status_var.set(f"正在处理: {blf_file}")
# 检查文件是否存在
if os.path.exists(asc_file) and not self.overwrite_var.get():
self._log_message(f"跳过: {os.path.basename(asc_file)} 已存在")
self.failed_conversions += 1
else:
try:
start_time = time.time()
self._convert_blf_to_asc(file_path, asc_file)
elapsed_time = time.time() - start_time
self._log_message(f"成功: 转换完成,耗时 {elapsed_time:.2f} 秒")
self.successful_conversions += 1
except Exception as e:
self._log_message(f"错误: {str(e)}")
self.failed_conversions += 1
self.processed_files += 1
progress = (self.processed_files / self.total_files) * 100
self.progress_var.set(progress)
self.stats_var.set(f"待转换文件: {self.total_files - self.processed_files} | 成功: {self.successful_conversions} | 失败: {self.failed_conversions}")
# 转换完成
self.is_converting = False
self.status_var.set("转换完成")
self._log_message(f"\n转换完成: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
self._log_message(f"总文件数: {self.total_files}, 成功: {self.successful_conversions}, 失败: {self.failed_conversions}")
# 在主线程中显示完成消息
self.root.after(0, lambda: messagebox.showinfo("完成", f"转换完成!\n总文件数: {self.total_files}\n成功: {self.successful_conversions}\n失败: {self.failed_conversions}"))
def _convert_blf_to_asc(self, blf_file_path, asc_file_path):
try:
# 打开.blf文件
with BLFReader(blf_file_path) as log_reader:
# 创建一个用于写入.asc文件的对象
with ASCWriter(asc_file_path) as log_writer:
# 获取总消息数用于进度显示
if self.verbose_var.get():
total_messages = getattr(log_reader, 'number_of_messages', None)
if total_messages is not None:
self._log_message(f"总消息数: {total_messages}")
# 遍历.blf文件中的每一条消息
messages_processed = 0
for msg in log_reader:
# 将消息写入.asc文件
log_writer.on_message_received(msg)
# 每处理1000条消息更新一次日志
messages_processed += 1
if messages_processed % 1000 == 0 and total_messages:
progress = (messages_processed / total_messages) * 100
self._log_message(f"已处理消息: {messages_processed}/{total_messages} ({progress:.1f}%)")
except FileNotFoundError:
raise Exception(f"未找到文件: {blf_file_path}")
except Exception as e:
raise Exception(f"转换过程中发生错误: {str(e)}")
def _log_message(self, message):
# 在主线程中更新日志
self.root.after(0, lambda: self._update_log_text(message))
def _update_log_text(self, message):
self.log_text.insert(tk.END, message + "\n")
self.log_text.see(tk.END)
self.conversion_log.append(message)
def _clear_log(self):
self.log_text.delete(1.0, tk.END)
if __name__ == "__main__":
root = tk.Tk()
app = BLFtoASCConverterApp(root)
root.mainloop()
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐


所有评论(0)