第3章 CAN数据采集实战

第3章：CAN数据采集实战

说实话，CAN总线在车载领域太常见了。我最早接触CAN是在做BMS项目的时候，那时候被一堆CAN报文搞得头大。后来慢慢摸透了，发现其实没那么玄乎。今天咱们就用Python的python-can库，手把手把CAN数据采集这件事搞定。

3.1 python-can库的安装与基础

先装库，这个简单。我个人习惯用pip直接装：

pip install python-can

装完之后，咱们先看看能不能用。我建议你直接打开Python交互环境，试试导入：

import can
print(can.__version__)

能打印出版本号，说明装好了。嗯，这里要注意，不同操作系统下底层驱动不一样。Windows下用PCAN或者Kvaser，Linux下用SocketCAN，macOS下...说实话我不太推荐在macOS上做CAN开发，坑比较多。

3.2 CAN设备初始化与配置

初始化CAN设备，说白了就是告诉Python你要用哪个硬件、哪个通道、多少波特率。我做过一个项目，因为波特率配错了，排查了整整两天...后来发现是配置文件和代码里写的不一致。

来看看标准配置流程：

import can

# 配置总线参数
bus = can.interface.Bus(
    bustype='socketcan',      # 接口类型
    channel='can0',           # 通道名
    bitrate=500000,           # 波特率 500kbps
    receive_own_messages=False
)

# 检查是否初始化成功
if bus.state == can.BusState.ACTIVE:
    print("CAN总线初始化成功！")
else:
    print("初始化失败，请检查硬件连接")

这里有几个参数我得解释一下：

参数	说明	常见取值
bustype	硬件接口类型	socketcan, pcan, kvaser, vector
channel	通道名称	can0, can1, PCAN_USBBUS1
bitrate	波特率	125000, 250000, 500000, 1000000

注意：波特率必须和车上ECU的配置一致。我曾经遇到过一台车用250kbps，另一台用500kbps，混着用直接收不到数据。

3.3 发送CAN报文

发送报文其实就三步：构造报文、发送、检查结果。你想想看，这跟发微信差不多，写好内容、点发送、看有没有发送成功。

# 构造一个标准帧
msg = can.Message(
    arbitration_id=0x123,     # CAN ID
    data=[0x11, 0x22, 0x33], # 数据
    is_extended_id=False,     # 标准帧
    dlc=3                     # 数据长度
)

# 发送
try:
    bus.send(msg)
    print(f"发送成功: ID=0x{msg.arbitration_id:x}, 数据={msg.data.hex()}")
except can.CanError:
    print("发送失败！检查总线是否正常")

扩展帧和标准帧的区别，说白了就是ID长度不同。标准帧11位ID，扩展帧29位ID。我建议你平时用标准帧就够了，除非你明确知道车上用的是扩展帧。

3.4 接收CAN报文

接收报文有两种方式：阻塞式和非阻塞式。我个人更喜欢非阻塞式，因为不会卡住主程序。

# 非阻塞接收
msg = bus.recv(timeout=0.1)  # 100ms超时
if msg:
    print(f"收到报文: ID=0x{msg.arbitration_id:x}")
    print(f"数据: {msg.data.hex()}")
    print(f"时间戳: {msg.timestamp}")

# 循环接收
while True:
    msg = bus.recv(timeout=0.5)
    if msg:
        # 处理报文
        process_message(msg)

小技巧：实际项目中，我一般把接收放在独立线程里。这样主线程可以干别的事，比如UI刷新、数据存储。

3.5 解析CAN ID与数据

这才是重头戏。CAN报文本身只是一串字节，你得知道每个字节代表什么意思。我记得第一次解析车速信号时，拿着DBC文件看了半天，才搞明白怎么把原始数据转成物理值。

常见的解析方式：

def parse_vehicle_speed(data):
    """
    解析车速信号
    假设：车速在byte0-byte1，分辨率0.01 km/h，偏移量0
    """
    raw = (data[0] << 8) | data[1]
    speed = raw * 0.01
    return speed

def parse_engine_rpm(data):
    """
    解析发动机转速
    假设：转速在byte2-byte3，分辨率0.125 rpm，偏移量0
    """
    raw = (data[2] << 8) | data[3]
    rpm = raw * 0.125
    return rpm

# 使用示例
msg = bus.recv(timeout=1)
if msg:
    if msg.arbitration_id == 0x100:  # 车速报文ID
        speed = parse_vehicle_speed(msg.data)
        print(f"当前车速: {speed:.2f} km/h")
    elif msg.arbitration_id == 0x200:  # 转速报文ID
        rpm = parse_engine_rpm(msg.data)
        print(f"发动机转速: {rpm:.1f} rpm")

3.6 实战：完整的采集脚本

把上面这些串起来，就是一个能用的采集工具了。我写了个简单的，你直接拿去改改就能用：

import can
import time
import csv
from datetime import datetime

class CANCollector:
    def __init__(self, channel='can0', bitrate=500000):
        self.bus = can.interface.Bus(
            bustype='socketcan',
            channel=channel,
            bitrate=bitrate
        )
        self.running = False
        self.data_buffer = []
        
    def start_collection(self, duration=10):
        """开始采集，duration秒"""
        self.running = True
        start_time = time.time()
        
        print(f"开始采集，持续{duration}秒...")
        
        while self.running and (time.time() - start_time) < duration:
            msg = self.bus.recv(timeout=0.1)
            if msg:
                record = {
                    'timestamp': datetime.now().isoformat(),
                    'can_id': hex(msg.arbitration_id),
                    'dlc': msg.dlc,
                    'data': msg.data.hex(),
                    'is_extended': msg.is_extended_id
                }
                self.data_buffer.append(record)
                print(f"[{record['timestamp']}] ID={record['can_id']} 数据={record['data']}")
        
        self.running = False
        print(f"采集完成，共收到{len(self.data_buffer)}条报文")
    
    def save_to_csv(self, filename='can_data.csv'):
        """保存到CSV文件"""
        if not self.data_buffer:
            print("没有数据可保存")
            return
        
        with open(filename, 'w', newline='') as f:
            writer = csv.DictWriter(f, fieldnames=[
                'timestamp', 'can_id', 'dlc', 'data', 'is_extended'
            ])
            writer.writeheader()
            writer.writerows(self.data_buffer)
        
        print(f"数据已保存到 {filename}")

# 使用示例
if __name__ == '__main__':
    collector = CANCollector(channel='can0', bitrate=500000)
    collector.start_collection(duration=30)
    collector.save_to_csv('test_data.csv')

    核心要点：
    初始化时确认硬件类型和波特率匹配
接收报文用非阻塞方式，加超时防止卡死
解析数据时注意字节序和分辨率
生产环境一定要加异常处理和日志

3.7 避坑指南

做CAN采集这几年，我踩过的坑真不少。给你列几个最常见的：

波特率不匹配：我曾经在测试车上配了250kbps，结果实车是500kbps，啥都收不到。排查了半小时才发现。
终端电阻：CAN总线两端需要120欧姆终端电阻。少一个，长距离传输就会丢包。
ID过滤：车上报文很多，每秒几百条。不加过滤的话，你的程序会被撑爆。用can的硬件过滤功能。
时间戳精度：Python的time.time()精度不够。需要高精度时间戳的话，用硬件时间戳或者time.perf_counter()。

好了，CAN数据采集这块就讲这么多。你回去拿个CAN卡连上车试试，跑一遍上面的代码，很快就能上手。下一章咱们聊CAN FD，那个带宽更大，适合大数据量传输的场景。

📚 课程目录