如何用Python导出汽车OBD实时数据至.txt/.csv文件

1. 如何将从汽车获取的OBD实时数据导出至.txt、.csv或其他合适的文件中？
2. 使用watch方法还是通过响应来实现数据导出是否可行？

当前使用的Python代码如下：

import obd #import required libraries
import time
import serial

connection = obd.Async( fast= False) #auto-connects to USB or RF port


#RPM
def new_rpm(r):
    print("RPM:",r.value)
    print("	")
    
#SPEED
def new_speed(r):
    print("SPEED:",r.value)
    print("	")
    
    

#Is it ok to do that with the watch or with the response ?    
connection.watch(obd.commands.RPM, callback=new_rpm)
connection.watch(obd.commands.SPEED, callback=new_speed)


connection.start()

#the callback now will be Enabled for new values
time.sleep(60)
connection.stop()

一、实现数据导出到文件 #

1. 导出到TXT文件 #

直接在回调函数中以追加模式写入文件，搭配时间戳能清晰记录数据采集时间：

import obd
import time
from datetime import datetime

connection = obd.Async(fast=False)

# 定义TXT文件路径
txt_file_path = "obd_data.txt"

def new_rpm(r):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    data_line = f"{timestamp} | RPM: {r.value}\n"
    # 追加写入，避免覆盖原有内容
    with open(txt_file_path, "a", encoding="utf-8") as f:
        f.write(data_line)
    print("RPM:", r.value)

def new_speed(r):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    data_line = f"{timestamp} | SPEED: {r.value}\n"
    with open(txt_file_path, "a", encoding="utf-8") as f:
        f.write(data_line)
    print("SPEED:", r.value)

connection.watch(obd.commands.RPM, callback=new_rpm)
connection.watch(obd.commands.SPEED, callback=new_speed)

connection.start()
time.sleep(60)
connection.stop()

2. 导出到CSV文件 #

CSV适合结构化数据，方便后续用Excel或数据分析工具处理，用Python内置csv模块实现：

import obd
import time
import csv
from datetime import datetime

connection = obd.Async(fast=False)

csv_file_path = "obd_data.csv"
# 初始化CSV并写入表头
with open(csv_file_path, "w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerow(["Timestamp", "Metric", "Value", "Unit"])

def write_to_csv(metric, value):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    # 提取数值单位（如果存在）
    unit = value.unit if hasattr(value, "unit") else ""
    with open(csv_file_path, "a", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([timestamp, metric, value.magnitude, unit])

def new_rpm(r):
    write_to_csv("RPM", r.value)
    print("RPM:", r.value)

def new_speed(r):
    write_to_csv("SPEED", r.value)
    print("SPEED:", r.value)

connection.watch(obd.commands.RPM, callback=new_rpm)
connection.watch(obd.commands.SPEED, callback=new_speed)

connection.start()
time.sleep(60)
connection.stop()

二、watch方法 vs 直接响应获取 #

两种方式都可行，适配不同场景：

• watch方法：适合实时监控场景，新数据返回时自动触发回调，无需主动轮询，代码结构简洁，完全能用来实现数据导出，匹配你当前的代码逻辑，适合长时间持续采集。
• 直接响应获取：如果是按需定时采集（比如每隔固定时间主动请求一次），可以用connection.query()方法，示例：

import obd
import time
import csv
from datetime import datetime

connection = obd.OBD(fast=False)
csv_file_path = "obd_data_polling.csv"

with open(csv_file_path, "w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerow(["Timestamp", "Metric", "Value", "Unit"])

def write_to_csv(metric, value):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    unit = value.unit if hasattr(value, "unit") else ""
    with open(csv_file_path, "a", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([timestamp, metric, value.magnitude, unit])

# 主动轮询采集60秒
start_time = time.time()
while time.time() - start_time < 60:
    rpm_response = connection.query(obd.commands.RPM)
    if not rpm_response.is_null():
        write_to_csv("RPM", rpm_response.value)
        print("RPM:", rpm_response.value)
    
    speed_response = connection.query(obd.commands.SPEED)
    if not speed_response.is_null():
        write_to_csv("SPEED", speed_response.value)
        print("SPEED:", speed_response.value)
    
    time.sleep(1) # 每隔1秒请求一次

总结：实时持续采集选watch+回调更高效；按需定时采集用主动query更灵活，两种方式都能完成数据导出需求。

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