使用python的pywin32库实现CANape工程自动化案例

CANape Python 自动化录数据 & MF4 数据分析 — 教学大纲与脚本合集

目标：用 Python 控制 CANape 做自动化测量与录制（生成 MF4），并用 Python 对 MF4 做批量离线分析与流程化处理。

目录

1. 概述与先决条件

• 平台：Windows（CANape COM API 依赖 Windows COM）
• CANape：已安装并有合适 License（如果要录制 MF4 需有 Logging License）
• Python：3.8+ 推荐 3.10/3.11
• 必须 Python 库：pywin32, asammdf（或 mdfreader）、pandas, matplotlib, numpy 等
• 硬件：支持的 Vector/ETAS/PCAN 接口，网线或 CAN 总线连接
• 权限：运行 Python 脚本的 Windows 账户需要有启动 CANape 与访问硬件的权限

2. CANape Automation 概览（COM 接口）

CANape 提供 COM Automation 接口（通常 ProgID 类似 CANape.Application 或 Vector.CANape.Application）。通过 win32com.client.Dispatch 可以在 Python 中控制 CANape：打开工程、连接硬件、启动测量、管理 DAQ lists、开始/停止录制、导出数据等。

注意：不同 CANape 版本 ProgID 可能不同，建议用 win32com.client.Dispatch("CANape.Application") 或查看注册表/Help->About 找到正确 ProgID。

3. 环境搭建与依赖安装

# 在 Windows 上创建虚拟环境并安装依赖
python –m venv venv
venv\\Scripts\\activate
pip install pywin32 asammdf pandas matplotlib numpy

• pywin32 用于 COM Automation。
• asammdf 是开源库用于读取/处理 MF4（MDF4）文件（也可以选 mdfreader）。

4. 常见使用场景与案例一览

• 基础录制：加载工程 -> 选择 DAQ list -> Start Measurement -> Start Logging -> Stop -> 保存 MF4
• 条件触发录制：当某个信号超限或事件发生，自动启动短时录制
• 定时/调度录制：按计划（cron-like）批量采集，例如夜间连续多次采集
• 批量标定+验证：自动写入多个参数组合，运行工况并记录 MF4 以验证性能
• 批量离线分析：读取 MF4、做统计、绘图并生成报告

5. 详解案例与 Python 模板

以下代码示例均可直接复制到 .py 文件运行（需要在 Windows + CANape 环境）。代码注释里给出了可调整参数。

案例 A：基础 — 连接 CANape、加载工程、开始/停止测量并录 MF4

# canape_basic_record.py
import time
import os
import win32com.client as win32

CANAPE_PROGID = 'CANape.Application' # 视你的 CANape 安装而定
PROJECT_FILE = r'C:\\Path\\To\\Your\\Project.cape' # 改为你的工程文件路径
OUTPUT_DIR = r'C:\\AI_Project\\CanapeRecordings'

os.makedirs(OUTPUT_DIR, exist_ok=True)

app = win32.Dispatch(CANAPE_PROGID)
# 如果 CANape 未启动，这行会启动 CANape
app.Visible = True

# 打开项目（如果已经打开可跳过）
app.OpenProject(PROJECT_FILE)

# 等待项目加载
time.sleep(2)

# 连接硬件/ECU（通常会有 Connect/Start）
# 这里的对象名和方法需根据你 CANape 版本调整
measurement = app.Measurement

# Start measurement (开始测量)
measurement.Start()
print('Measurement started')

# Start logging: 指定输出文件名
timestamp = time.strftime('%Y%m%d_%H%M%S')
outfile = os.path.join(OUTPUT_DIR, f'record_{timestamp}.mf4')

# Logging API 可能在不同版本下有不同对象，尝试常见方法
try:
recorder = app.Recorder
recorder.FileName = outfile
recorder.Start()
print('Recording started ->', outfile)
except Exception as e:
print('Recorder API not direct-accessible:', e)
# 备选：通过 Measurement.StartRecording(…) 或者使用 Commands

# 录制 30 秒做示例
time.sleep(30)

# 停止录制
try:
recorder.Stop()
print('Recording stopped')
except:
pass

# 停止测量
measurement.Stop()
print('Measurement stopped')

# 关闭项目
# app.CloseProject()

说明与调整：不同 CANape 版本 API 方法名可能差异（例如 Recorder 对象可能不存在）；如无直接方法，可以通过发送菜单命令或使用 app.ExecuteCommand("…" )（部分版本支持）或操作 GUI 对象。

案例 B：触发录制（按信号条件触发）并保存带元数据的 MF4

思路：持续读取信号（或订阅事件），当触发条件满足时，调用 Recorder 开始短时记录并保存文件名/元数据（例如测试工况、运行参数）。

# canape_trigger_record.py
import time
import os
import win32com.client as win32

app = win32.Dispatch('CANape.Application')
app.Visible = True
app.OpenProject(r'C:\\Path\\To\\Project.cape')

# 获取信号值的 API：可能是 app.Measurement.GetValue("SignalName")
SIG = 'Engine.Speed' # 示例 signal name
THRESH = 3000 # RPM
OUTDIR = r'C:\\AI_Project\\CanapeRecordings\\trigger'
os.makedirs(OUTDIR, exist_ok=True)

measurement = app.Measurement

def get_signal_value(signal_name):
try:
# API 风格可能为 Measurement.GetValue 或 VariableBrowser 等
return float(measurement.GetValue(signal_name))
except Exception as e:
print('GetValue error', e)
return None

# 轮询监听
while True:
v = get_signal_value(SIG)
if v and v > THRESH:
ts = time.strftime('%Y%m%d_%H%M%S')
outfile = os.path.join(OUTDIR, f'trigger_{ts}.mf4')
try:
app.Recorder.FileName = outfile
app.Recorder.Start()
print('Triggered recording ->', outfile)
time.sleep(10) # record 10s
app.Recorder.Stop()
print('Saved', outfile)
except Exception as e:
print('Recorder error', e)
# 避免重复触发
time.sleep(5)
time.sleep(0.1)

说明：如果 CANape 支持事件回调（CAPL/CANape Event API），可将触发器写在 CAPL，CAPL 调用 COM/外部接口触发 Python。但在纯 Python 轮询通常也能工作（代价是 CPU 与延迟）。

案例 C：排程/定时录制（夜间批量采集）

思路：读取一个 CSV 定时表（每行是开始时间/持续时长/备注），按表依次运行。

# canape_scheduler.py
import time
import csv
from datetime import datetime
import win32com.client as win32

app = win32.Dispatch('CANape.Application')
app.OpenProject(r'C:\\Path\\To\\Project.cape')

SCHEDULE_CSV = r'C:\\Path\\to\\schedule.csv' # columns: start_iso,duration_seconds,tag

with open(SCHEDULE_CSV, newline='') as f:
reader = csv.DictReader(f)
tasks = list(reader)

for t in tasks:
start = datetime.fromisoformat(t['start_iso'])
dur = int(t['duration_seconds'])
tag = t.get('tag','')
now = datetime.now()
wait = (start – now).total_seconds()
if wait > 0:
print('Waiting', wait, 'seconds until', start)
time.sleep(wait)
# Start measurement + recording
app.Measurement.Start()
fn = f"C:\\\\Records\\\\record_{start.strftime('%Y%m%d_%H%M%S')}_{tag}.mf4"
app.Recorder.FileName = fn
app.Recorder.Start()
print('Recording', fn)
time.sleep(dur)
app.Recorder.Stop()
app.Measurement.Stop()
print('Done', fn)

案例 D：批量标定写入（通过 CANape COM 自动写入标定参数）并记录验证数据

思路：从 CSV 中读取每一组标定参数（名称和值），调用 CANape API 写入（WriteValue/SetCalParam），然后运行一次短记录验证。示例给出伪代码，具体 API 名需你本地 CANape 对照。

# canape_batch_cal.py
import csv, time, os
import win32com.client as win32

app = win32.Dispatch('CANape.Application')
app.OpenProject(r'C:\\Path\\To\\Project.cape')

CSV = r'C:\\Path\\to\\cal_jobs.csv' # columns: param_name,value
OUTDIR = r'C:\\Records\\cal_verification'

with open(CSV) as f:
reader = csv.DictReader(f)
for i,row in enumerate(reader):
name = row['param_name']
val = float(row['value'])
print('Writing', name, val)
# 伪 API，视 CANape 版本替换
try:
app.Calibration.WriteValue(name, val) # 视版本而定
except Exception as e:
print('WriteValue failed, trying alternative', e)
# 可尝试通过 VariableBrowser 获取变量，用 SetValue 等
# 短录 5s 做验证
app.Measurement.Start()
fn = os.path.join(OUTDIR, f'cal_{i}_{name}_{val}.mf4')
app.Recorder.FileName = fn
app.Recorder.Start()
time.sleep(5)
app.Recorder.Stop()
app.Measurement.Stop()
print('Saved verification', fn)

案例 E：故障重放（回放 CAN 数据并同步录制）

• 使用 CANape 自带的 Replay 功能或 Virtual CAN channel，把 MF4/ASC/BLF 回放到总线上，同时记录 ECU 响应为新 MF4。可以通过 COM 自动化启动 Replay 并启动 Recorder。

# canape_replay_and_record.py
import win32com.client as win32
import time

app = win32com.client.Dispatch('CANape.Application')
app.OpenProject(r'C:\\Path\\To\\ReplayProject.cape')

# 假定项目中配置了 Replay object 名为 'Replay1'
replay = app.Replay # 可能需要具体对象名
replay.LoadFile(r'C:\\Samples\\events.blf')
replay.Start()

app.Measurement.Start()
app.Recorder.FileName = r'C:\\Records\\replay_response.mf4'
app.Recorder.Start()

# 等待回放完成
while replay.IsPlaying:
time.sleep(1)

app.Recorder.Stop()
app.Measurement.Stop()
print('Replay done and recorded')

6. MF4 离线分析：使用 asammdf / mdfreader / pandas

安装：

pip install asammdf matplotlib pandas numpy

读取并显示信号

# analyse_mf4_basic.py
from asammdf import MDF
import matplotlib.pyplot as plt

mf4 = MDF('C:\\AI_Project\\CanapeRecordings\\record_20251124_120000.mf4')
# 列出信号名
print(mf4.channels_db.keys())

# 读取信号
sig = mf4.get('Engine.Speed') # 返回 signal 对象 (time, samples)
# 绘图
plt.plot(sig.timestamps, sig.samples)
plt.xlabel('time (s)')
plt.ylabel('rpm')
plt.title('Engine Speed')
plt.show()

批量统计与导出 CSV

# batch_analyse.py
from asammdf import MDF
import pandas as pd
import os

INDIR = r'C:\\AI_Project\\CanapeRecordings'
OUTDIR = r'C:\\AI_Project\\CanapeReports'
os.makedirs(OUTDIR, exist_ok=True)

for fn in os.listdir(INDIR):
if fn.lower().endswith('.mf4'):
path = os.path.join(INDIR, fn)
m = MDF(path)
# 假设要提取 3个信号
signals = ['Engine.Speed','Vehicle.Speed','Throttle.Position']
df = pd.DataFrame()
for s in signals:
try:
ch = m.get(s)
df[s+'_t'] = ch.timestamps
df[s] = ch.samples
except Exception as e:
print('skip', s, 'error', e)
# 简单统计
stats = df.describe()
stats.to_csv(os.path.join(OUTDIR, fn + '.stats.csv'))
df.to_csv(os.path.join(OUTDIR, fn + '.signals.csv'))
print('Processed', fn)

高级：同步不同采样率信号、重采样

# resample_and_sync.py
from asammdf import MDF
import numpy as np
import pandas as pd

m = MDF('sample.mf4')
ch1 = m.get('Engine.Speed')
ch2 = m.get('CAN_bus_signal')
# 将两信号重采样到相同时间轴
t0 = max(ch1.timestamps[0], ch2.timestamps[0])
t1 = min(ch1.timestamps[–1], ch2.timestamps[–1])
fs = 100 # target freq
new_t = np.arange(t0, t1, 1.0/fs)

from numpy import interp
v1 = interp(new_t, ch1.timestamps, ch1.samples)
v2 = interp(new_t, ch2.timestamps, ch2.samples)

import matplotlib.pyplot as plt
plt.plot(new_t, v1)
plt.plot(new_t, v2)
plt.show()

7. 常见问题与故障排查清单

• COM Dispatch 失败：确认 CANape 已安装并存在 COM 接口；检查 ProgID（可用 regedit 查找）
• Recorder 对象不存在：不同版本 API 不同，查看 CANape 的 Automation 文档或使用 dir(app) 调试
• 权限/License 问题：没有 Logging License 会导致无法录制 MF4
• 信号名称不匹配：确保使用的 signal name 与项目中 Variable Browser 一致（大小写/命名空间）
• 硬件连接失败：检查驱动、权限、通道号与波特率
• 并发冲突：若同一硬件被其它软件占用，CANape 无法访问

8. 截图建议与 PPT / 培训材料大纲

• 幻灯片结构（建议 12 页）

• 标题 + 目标
• CANape 概述与应用场景
• 环境与依赖（图示 CANape + 硬件）
• Python + COM 自动化简介（代码截图）
• 案例 A：基础录制（步骤截图：Project->Start Measurement->Recorder）
• 案例 B：触发录制（示意图 + 代码）
• 案例 C：排程录制（流程图）
• 案例 D：批量标定与验证（CSV -> 写入 -> 记录）
• MF4 分析工具链（asammdf 演示截图）
• 实操检查清单（步骤）
• 常见错误与解决方案（表格）
• Q&A / 后续资源

• 截图建议：

  • Project 打开界面（Show Project tree）
  • Measurement/Recorder 配置页截图（突出 FileName 与 Start/Stop）
  • Variable Browser 展示 signal 名称
  • asammdf 示例绘图（Matplotlib plot）

9. 附录：常用 COM 接口函数映射（示例）

具体接口可能随 CANape 版本差异较大，以下为常见名词映射供调试参考：

• app.OpenProject(path) — 打开工程
• app.CloseProject() — 关闭工程
• app.Measurement.Start() / .Stop() — 启动/停止 Measurement
• app.Recorder.Start() / .Stop() — 启动/停止 Recording（若可用）
• app.Recorder.FileName = '…mf4' — 设置输出文件
• app.VariableBrowser.GetValue('SignalName') 或 app.Measurement.GetValue('SignalName') — 读取信号
• app.Calibration.WriteValue('ParamName', value) — 写入标定参数（示例）

CANape的工作中的案例

对信号名加密，就是给客户的CANape工具是加密后的信号 客户看不到信号名。

import os
import sys
import glob
import time
import shutil
import pyautogui
import subprocess
import win32com.client

class CANapeController:
def __init__(self):
self.MT_project = sys.argv[1]
self.workspace = sys.argv[2]
self.project_path = os.path.join(self.MT_project, 'fsvsvsv')
self.canape_project = os.path.join(self.project_path, 'asdvsavd', 'Canape')
self.canape_database = os.path.join(self.project_path, 'adsssav', 'database', 'a2l')
self.command = r'C:\\sdad\\sdaasd\\sadas\\sdadas.bat'
self.a2l_file = 'sadd.a2l'
self.whitelist_file = os.path.join(self.workspace, 'builtfsfa', 'mt_a2l_encryption', 'whitelist.txt')
try:
self.canape = win32com.client.DispatchEx("CANape.Application")
print('CANape dispatched.')
except Exception as e:
print(f"Error dispatching CANape: {e}")

def boot_CANape_project(self):
if not os.path.exists(self.canape_project):
print(f"Project file does not exist: {self.canape_project}")
return None
try:
self.canape.Open1(self.canape_project, 1, 100000000, True)
print('CANape initialized.')
# self.canape.Open2(project_path, 1, 100000000, True, True, 1)
time.sleep(20)
except Exception as e:
print(f"Error initializing CANape: {e}")
return None

def save_Cnax(self):
try:
save_cns = os.path.join(self.project_path, 'save_cnax.cns')
with open(save_cns, 'w') as cns_file:
# 0: configuration is not actually saved (only reset the modify flag) 1: configuration is saved to file (default)
cns_file.write('SaveConfiguration(1,"sadd.cnax");\\n')
self.canape.RunScript(save_cns)
print("saved successfully!!")
except Exception as e:
print(f"Error saving project: {e}")

def run_command_in_cmd(self, command):

pyautogui.hotkey('win', 'r')
time.sleep(1)
pyautogui.write('cmd')
pyautogui.press('enter')
time.sleep(1)

pyautogui.write(command)
time.sleep(5)
pyautogui.press('enter')
time.sleep(300)
pyautogui.write('exit')
time.sleep(1)
pyautogui.press('enter')

def a2l_update_encryption(self):
try:
full_command = f'"{self.command}" "{self.canape_database}" "{self.a2l_file}" –CNAX "{self.whitelist_file}"'
print("Executing command:", full_command)
self.run_command_in_cmd(full_command)

except Exception as e:
print(f"An error occurred: {e}")

def replace_a2l(self):
time.sleep(5)
generated_file = os.path.join(self.canape_database, 'sadd.a2l')
if os.path.exists(generated_file):
original_a2l_file = os.path.join(self.canape_database, self.a2l_file)
if os.path.exists(original_a2l_file):
os.remove(original_a2l_file)
print(f"Deleted original file: {original_a2l_file}")
os.rename(generated_file, original_a2l_file)
print(f"Renamed {generated_file} to {original_a2l_file}")
txt_files = glob.glob(os.path.join(self.canape_database, '*.txt'))
for txt_file in txt_files:
os.remove(txt_file)
print(f"Deleted text file: {txt_file}")

def load_cnax(self):
time.sleep(10)
self.canape_cnax = os.path.join(self.canape_project, 'sadd.cnax')
self.canape.LoadCNAFile(self.canape_cnax)
print("loaded successfully!!")

def quit_canape(self):
try:
time.sleep(30)
self.canape.Quit()
print("CANape exited successfully.")
except Exception as e:
print(f"Error quitting CANape: {e}")

def delete_file(self, path, suffix):
files = glob.glob(os.path.join(path, suffix))
for file in files:
os.remove(file)
print(f"Deleted text file: {file}")

def delete_dir(self, dir_path):
if os.path.exists(dir_path):
try:
shutil.rmtree(dir_path)
print(f"The directory '{dir_path}' has been successfully deleted.")
except Exception as e:
print(f"Error deleting directory: {e}")
else:
print(f"The directory '{dir_path}' does not exist.")

def delete_confidential_files(self):
cna_files = os.path.join(self.canape_project, 'DASY_BASE_01.cna')
if os.path.exists(cna_files):
os.remove(cna_files)
addon_dirs = os.path.join(self.canape_project, 'addon')
delete_tools = os.path.join(self.project_path, 'tools')
exec_files = os.path.join(self.project_path, 'executables')
self.delete_dir(addon_dirs)
self.delete_dir(delete_tools)
self.delete_file(self.project_path, "*.txt")
self.delete_file(self.project_path, "*.cns")
self.delete_file(exec_files, "*.elf")
self.delete_file(exec_files, "*.map")

if __name__ == "__main__":

canape_controller = CANapeController()
canape_controller.a2l_update_encryption()
canape_controller.boot_CANape_project()
canape_controller.save_Cnax()
canape_controller.replace_a2l()
canape_controller.load_cnax()
canape_controller.save_Cnax()
canape_controller.quit_canape()
canape_controller.delete_confidential_files()