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()    
Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐