Raspberry Pi电池电压在Node+Express本地Web应用的高效展示方案咨询

当前通过JSON文件中转数据的方案存在磁盘IO开销大、实时性不足的问题，以下是几种更高效的实现方式，适配你的Node+Express环境：

1. Python与Node直接TCP通信，跳过文件IO #

让Python采集数据后直接通过TCP发送给Node服务，数据存在内存中，前端通过API获取，彻底避免磁盘读写开销。

Python采集端代码 #

import socket
import time
from datetime import datetime
import json

# 替换为实际的GPIO电压读取逻辑
def read_battery_voltage():
    return round(3.7 + (time.time() % 1)*0.3, 2)

node_host = 'localhost'
node_port = 8888

while True:
    voltage = read_battery_voltage()
    timestamp = datetime.now().isoformat()
    data = json.dumps({"timestamp": timestamp, "voltage": voltage})
    
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.connect((node_host, node_port))
        s.sendall(data.encode('utf-8'))
        s.close()
    except Exception as e:
        print(f"发送失败: {e}")
    time.sleep(1)  # 调整采集间隔

Node+Express服务端代码 #

const express = require('express');
const net = require('net');
const app = express();
const port = 3000;

// 内存存储最新电压数据，可扩展为数组保存历史记录
let latestBatteryData = { timestamp: '', voltage: 0 };

// 启动TCP服务接收Python数据
const tcpServer = net.createServer((socket) => {
    socket.on('data', (data) => {
        try {
            latestBatteryData = JSON.parse(data.toString());
            console.log('更新数据:', latestBatteryData);
        } catch (err) {
            console.log('数据解析失败:', err);
        }
    });
});
tcpServer.listen(8888, 'localhost', () => {
    console.log('TCP监听端口8888');
});

// 前端API接口
app.get('/api/battery', (req, res) => {
    res.json(latestBatteryData);
});

// 托管前端静态文件
app.use(express.static('public'));

app.listen(port, () => {
    console.log(`Web服务运行在http://localhost:${port}`);
});

前端请求代码 #

// 定时拉取最新数据
setInterval(async () => {
    const res = await fetch('/api/battery');
    const data = await res.json();
    document.getElementById('voltage').textContent = `${data.voltage}V`;
    document.getElementById('time').textContent = data.timestamp;
}, 1000);

2. 用SSE实现实时推送，减少前端轮询开销 #

通过Server-Sent Events让Node主动向前端推送新数据，无需前端频繁发起请求，实时性更高，资源消耗更低。

Node端新增SSE接口 #

app.get('/api/battery-sse', (req, res) => {
    res.writeHead(200, {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive'
    });

    // 先发送当前最新数据
    res.write(`data: ${JSON.stringify(latestBatteryData)}\n\n`);

    // 定时推送更新
    const pushInterval = setInterval(() => {
        res.write(`data: ${JSON.stringify(latestBatteryData)}\n\n`);
    }, 1000);

    // 客户端断开时清理资源
    req.on('close', () => {
        clearInterval(pushInterval);
        res.end();
    });
});

前端SSE连接代码 #

const eventSource = new EventSource('/api/battery-sse');
eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    document.getElementById('voltage').textContent = `${data.voltage}V`;
    document.getElementById('time').textContent = data.timestamp;
};

eventSource.onerror = (error) => {
    console.error('SSE连接异常:', error);
    eventSource.close();
};

3. Python作为Node子进程运行，简化通信 #

直接在Node中启动Python采集脚本，通过标准输出接收数据，无需额外的网络服务，适合轻量场景。

Node端代码 #

const express = require('express');
const { spawn } = require('child_process');
const app = express();
const port = 3000;

let latestBatteryData = { timestamp: '', voltage: 0 };

// 启动Python采集脚本
const pythonProc = spawn('python3', ['battery_reader.py']);

pythonProc.stdout.on('data', (data) => {
    try {
        latestBatteryData = JSON.parse(data.toString().trim());
    } catch (err) {
        console.log('解析Python输出失败:', err);
    }
});

pythonProc.stderr.on('data', (data) => {
    console.error('Python脚本错误:', data.toString());
});

app.get('/api/battery', (req, res) => {
    res.json(latestBatteryData);
});
app.use(express.static('public'));
app.listen(port, () => {
    console.log(`Web服务运行在http://localhost:${port}`);
});

Python脚本（battery_reader.py） #

import sys
import time
from datetime import datetime
import json

def read_battery_voltage():
    return round(3.7 + (time.time() % 1)*0.3, 2)

while True:
    voltage = read_battery_voltage()
    timestamp = datetime.now().isoformat()
    print(json.dumps({"timestamp": timestamp, "voltage": voltage}))
    sys.stdout.flush()  # 强制输出避免缓冲
    time.sleep(1)

方案选择建议 #

• 单设备轻量场景：优先选方案3（子进程），实现最简单，无额外依赖；
• 追求实时性：选方案2（SSE），避免前端轮询浪费资源；
• 未来有扩展需求（多采集节点/多客户端）：可引入MQTT消息队列，实现更灵活的分布式数据传输。

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