Windows环境下用Python实现DLT日志转txt/csv及自动化采集检索

前置依赖准备 #

• 提前执行adb devices确认USB连接的ECU设备可被正常识别，记录设备序列号
• 本地完成DLT Viewer安装，记录程序可执行文件的绝对路径
• Python环境安装所需依赖：pip install psutil python-dlt pandas

各功能模块实现 #

1. 后台启动DLT Viewer并连接ECU #

通过Windows进程启动参数隐藏DLT Viewer窗口，后台自动完成ADB通道的ECU连接，无需人工操作界面。

import subprocess
import psutil
import time
import os

# 全局配置项，按需修改
DLT_PROGRAM_PATH = r"C:\Program Files (x86)\dlt-viewer\dlt-viewer.exe"
ECU_ADB_SERIAL = ""  # 填入adb devices返回的目标ECU序列号
dlt_main_process = None

def init_dlt_connect(run_background: bool = True) -> bool:
    """初始化DLT连接，run_background为True时后台隐藏运行"""
    global dlt_main_process
    startup_config = subprocess.STARTUPINFO()
    if run_background:
        startup_config.dwFlags |= subprocess.STARTF_USESHOWWINDOW
        startup_config.wShowWindow = 0  # 隐藏窗口参数
    dlt_main_process = subprocess.Popen(
        [DLT_PROGRAM_PATH, "--adb-device", ECU_ADB_SERIAL, "--auto-connect-ecu"],
        startupinfo=startup_config
    )
    time.sleep(3)  # 等待连接握手完成
    return psutil.pid_exists(dlt_main_process.pid)

2. 日志采集启动API #

触发后自动拉取全量实时日志流，支持写入txt或csv格式文件到预设路径。

# 日志存储根目录
LOG_ROOT_DIR = r"D:\test_dlt_logs"
active_log_path = None
log_capture_process = None

def start_capture(save_format: str = "txt", custom_filename: str = None) -> bool:
    """
    启动全量日志采集
    :param save_format: 支持txt、csv两种存储格式
    :param custom_filename: 自定义日志文件名，默认按时间戳自动生成
    """
    global active_log_path, log_capture_process
    os.makedirs(LOG_ROOT_DIR, exist_ok=True)
    if not custom_filename:
        custom_filename = f"dlt_full_log_{time.strftime('%Y%m%d_%H%M%S')}"
    active_log_path = os.path.join(LOG_ROOT_DIR, f"{custom_filename}.{save_format}")

    startup_config = subprocess.STARTUPINFO()
    startup_config.dwFlags |= subprocess.STARTF_USESHOWWINDOW
    startup_config.wShowWindow = 0
    # 调用DLT实时导出能力
    capture_cmd = [
        DLT_PROGRAM_PATH, "--live-export", active_log_path,
        "--export-format", save_format, "--all-messages"
    ]
    log_capture_process = subprocess.Popen(capture_cmd, startupinfo=startup_config)
    return True

3. 日志采集停止API #

触发后安全终止采集进程，等待缓冲区日志写入磁盘后返回存储结果，避免日志截断丢失。

def stop_capture() -> dict:
    """停止日志采集，返回文件存储状态"""
    global log_capture_process, active_log_path
    if log_capture_process and psutil.pid_exists(log_capture_process.pid):
        log_capture_process.terminate()
        try:
            log_capture_process.wait(timeout=5)  # 等待写入完成
        except subprocess.TimeoutExpired:
            log_capture_process.kill()
    # 校验文件完整性
    save_ok = os.path.exists(active_log_path) and os.path.getsize(active_log_path) > 1024
    return {"is_saved": save_ok, "file_path": active_log_path}

4. 自定义字段检索校验功能 #

支持按传入的payload关键词、App ID、CT ID（上下文ID）匹配日志，返回字段存在性判定结果和命中条数。

from dlt import DLTReader

def verify_log_field(log_file_path: str, target_payload: str = None, target_appid: str = None, target_ctid: str = None) -> dict:
    """
    检索日志中指定字段是否存在
    所有传入的匹配条件需同时满足才算命中
    """
    hit_count = 0
    with open(log_file_path, "rb") as f:
        for msg in DLTReader(f):
            current_payload = str(msg.payload)
            current_appid = msg.apid
            current_ctid = msg.ctid
            is_match = True
            if target_payload and target_payload not in current_payload:
                is_match = False
            if target_appid and target_appid != current_appid:
                is_match = False
            if target_ctid and target_ctid != current_ctid:
                is_match = False
            if is_match:
                hit_count += 1
    return {
        "field_exists": hit_count > 0,
        "hit_number": hit_count
    }

踩坑提示：如果当前安装的DLT Viewer版本不支持上述命令行参数，可以直接替换为DLT官方命令行工具dlt-receive实现日志拉取，不需要依赖GUI程序，后台运行稳定性更高。采集过程中建议加进程存活巡检，若采集进程意外退出可自动拉起，避免日志漏采。

内容的提问来源于stack exchange，提问作者Alice