基于Flask实现树莓派OBD数据上传与程序远程管控方案咨询

整体架构 #

树莓派端部署Flask服务，负责三个核心功能：

• 远程启停OBD数据采集程序
• 监听CSV文件生成事件，提供数据展示接口
• 渲染Web管控界面，实现可视化操作与数据展示

树莓派端代码实现 #

1. Flask服务核心代码（app.py） #

from flask import Flask, render_template, jsonify, request
import subprocess
import os
import csv
from threading import Thread
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

app = Flask(__name__)

# 配置参数，根据实际路径修改
COLLECTOR_PATH = "/home/pi/obd_collector.py"
CSV_DIR = "/home/pi/obd_data"
PID_FILE = "/tmp/obd_collector.pid"
latest_csv_data = []

# 创建CSV存储目录
os.makedirs(CSV_DIR, exist_ok=True)

class CSVFileHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory and event.src_path.endswith(".csv"):
            global latest_csv_data
            time.sleep(1)  # 等待文件写入完成
            with open(event.src_path, 'r', encoding='utf-8') as f:
                reader = csv.reader(f)
                latest_csv_data = list(reader)

# 启动文件监听线程
def start_watchdog():
    event_handler = CSVFileHandler()
    observer = Observer()
    observer.schedule(event_handler, CSV_DIR, recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

Thread(target=start_watchdog, daemon=True).start()

def get_collector_pid():
    if os.path.exists(PID_FILE):
        with open(PID_FILE, 'r') as f:
            pid = f.read().strip()
        if pid.isdigit() and os.path.exists(f"/proc/{pid}"):
            return int(pid)
    return None

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/api/start_collector', methods=['POST'])
def start_collector():
    if get_collector_pid() is not None:
        return jsonify({"status": "error", "msg": "采集程序已在运行"})
    # 启动采集程序，后台运行并记录PID
    proc = subprocess.Popen(['python3', COLLECTOR_PATH], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    with open(PID_FILE, 'w') as f:
        f.write(str(proc.pid))
    return jsonify({"status": "success", "msg": "采集程序已启动"})

@app.route('/api/stop_collector', methods=['POST'])
def stop_collector():
    pid = get_collector_pid()
    if pid is None:
        return jsonify({"status": "error", "msg": "采集程序未运行"})
    try:
        os.kill(pid, 9)
        os.remove(PID_FILE)
        return jsonify({"status": "success", "msg": "采集程序已停止"})
    except Exception as e:
        return jsonify({"status": "error", "msg": f"停止失败: {str(e)}"})

@app.route('/api/get_latest_data')
def get_latest_data():
    return jsonify({"data": latest_csv_data})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=False)

2. OBD数据采集程序示例（obd_collector.py） #

替换为你实际的OBD采集逻辑，这里模拟生成CSV文件：

import csv
import time
import os
from datetime import datetime

CSV_DIR = "/home/pi/obd_data"
os.makedirs(CSV_DIR, exist_ok=True)

# 模拟OBD数据采集
def collect_obd_data():
    while True:
        # 生成文件名（按时间戳）
        filename = f"{CSV_DIR}/obd_data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        with open(filename, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            # 写入表头
            writer.writerow(["时间", "车速(km/h)", "发动机转速(rpm)", "冷却液温度(℃)"])
            # 模拟采集10条数据
            for i in range(10):
                current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                speed = 30 + i*2
                rpm = 1500 + i*100
                temp = 80 + i
                writer.writerow([current_time, speed, rpm, temp])
                time.sleep(1)
        # 每次采集完成后休眠5秒，可根据实际需求调整
        time.sleep(5)

if __name__ == '__main__':
    try:
        collect_obd_data()
    except KeyboardInterrupt:
        print("采集程序已停止")

3. Web前端模板（templates/index.html） #

在Flask项目根目录创建templates文件夹，放入该文件：

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>OBD数据管控平台</title>
    <style>
        .container { max-width: 1200px; margin: 20px auto; padding: 0 20px; }
        .control-btn { padding: 10px 20px; margin-right: 10px; font-size: 16px; cursor: pointer; }
        .start-btn { background-color: #4CAF50; color: white; border: none; }
        .stop-btn { background-color: #f44336; color: white; border: none; }
        .status { margin: 10px 0; font-size: 18px; }
        table { width: 100%; border-collapse: collapse; margin-top: 20px; }
        th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
        th { background-color: #f2f2f2; }
    </style>
</head>
<body>
    <div class="container">
        <h1>OBD数据采集管控</h1>
        <div class="status" id="status">当前状态: 未运行</div>
        <button class="control-btn start-btn" onclick="startCollector()">启动采集</button>
        <button class="control-btn stop-btn" onclick="stopCollector()">停止采集</button>
        <div id="data-table-container">
            <h2>最新采集数据</h2>
            <table id="data-table">
                <!-- 数据将通过JS动态填充 -->
            </table>
        </div>
    </div>

    <script>
        // 定时刷新数据，每3秒一次
        setInterval(fetchLatestData, 3000);

        async function startCollector() {
            const res = await fetch('/api/start_collector', { method: 'POST' });
            const data = await res.json();
            alert(data.msg);
            updateStatus();
        }

        async function stopCollector() {
            const res = await fetch('/api/stop_collector', { method: 'POST' });
            const data = await res.json();
            alert(data.msg);
            updateStatus();
        }

        async function updateStatus() {
            const statusElement = document.getElementById('status');
            // 简化状态更新，实际可新增/api/get_status接口做精准判断
            statusElement.textContent = "当前状态: " + (statusElement.textContent.includes("未运行") ? "运行中" : "未运行");
        }

        async function fetchLatestData() {
            const res = await fetch('/api/get_latest_data');
            const data = await res.json();
            const table = document.getElementById('data-table');
            table.innerHTML = '';
            if (data.data.length === 0) {
                table.innerHTML = '<tr><td colspan="4">暂无数据</td></tr>';
                return;
            }
            // 生成表头
            const headerRow = document.createElement('tr');
            data.data[0].forEach(cell => {
                const th = document.createElement('th');
                th.textContent = cell;
                headerRow.appendChild(th);
            });
            table.appendChild(headerRow);
            // 生成数据行
            for (let i = 1; i < data.data.length; i++) {
                const row = document.createElement('tr');
                data.data[i].forEach(cell => {
                    const td = document.createElement('td');
                    td.textContent = cell;
                    row.appendChild(td);
                });
                table.appendChild(row);
            }
        }

        // 页面加载时初始化数据
        window.onload = fetchLatestData;
    </script>
</body>
</html>

部署与配置步骤 #

1. 安装依赖
   在树莓派终端执行：

pip3 install flask watchdog psutil

2. 调整原采集程序的开机自启
   如果原采集程序通过systemd或rc.local设置了开机自启，建议关闭该配置，改为通过Flask服务控制启停。
3. 设置Flask服务开机自启
   创建systemd服务文件/etc/systemd/system/obd-web.service：

[Unit]
Description=OBD Data Web Service
After=network.target

[Service]
User=pi
WorkingDirectory=/home/pi
ExecStart=/usr/bin/python3 /home/pi/app.py
Restart=always

[Install]
WantedBy=multi-user.target

然后执行：

sudo systemctl daemon-reload
sudo systemctl enable obd-web.service
sudo systemctl start obd-web.service

4. 访问Web界面
   在同一局域网内的设备浏览器中输入http://<树莓派IP>:5000即可访问管控界面。

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