Python中如何在树莓派上使用Flask框架实时读取串口信息？

硬件用的是树莓派+二维码扫描器，程序上用的是 python、flask、pyserial，这套方案现在遇到的问题就是，二维码数据有更新的时候，怎样去实时将数据显示在 html 页面中？
各位高手帮忙提点意见，最好有现成案例
Python中如何在树莓派上使用Flask框架实时读取串口信息？

17 回复

实时更新这一般是前段通过 js 的 ajax 动态更新的，和 flask 关系不大吧。


在树莓派上通过Flask实时读取串口，核心是结合pyserial进行串口通信，并使用Flask的Streaming或Server-Sent Events (SSE)技术。下面是一个使用Flask-SSE扩展的完整示例，它更简洁高效。

首先，安装必要的库：

pip install flask pyserial flask-sse

然后，创建应用文件（例如app.py）：

from flask import Flask, render_template
from flask_sse import sse
import serial
import threading
import time

app = Flask(__name__)
app.config["REDIS_URL"] = "redis://localhost"  # SSE需要Redis，需先安装运行Redis
app.register_blueprint(sse, url_prefix='/stream')

# 配置串口参数，根据你的设备调整（如/dev/ttyUSB0或/dev/ttyAMA0）
SERIAL_PORT = '/dev/ttyUSB0'
BAUD_RATE = 9600

def read_serial():
    """后台线程函数，持续读取串口并发送SSE事件"""
    try:
        ser = serial.Serial(SERIAL_PORT, BAUD_RATE, timeout=1)
        print(f"开始监听串口 {SERIAL_PORT}...")
        while True:
            if ser.in_waiting > 0:
                line = ser.readline().decode('utf-8', errors='ignore').strip()
                if line:
                    with app.app_context():
                        sse.publish({"message": line}, type='serial_data')
            time.sleep(0.01)  # 短暂延迟避免CPU过高占用
    except Exception as e:
        print(f"串口读取错误: {e}")

@app.route('/')
def index():
    """提供前端页面"""
    return render_template('index.html')

if __name__ == '__main__':
    # 启动后台串口读取线程
    thread = threading.Thread(target=read_serial, daemon=True)
    thread.start()
    app.run(host='0.0.0.0', port=5000, debug=True)

接着，创建templates/index.html前端页面：

<!DOCTYPE html>
<html>
<head>
    <title>串口实时数据</title>
</head>
<body>
    <h1>串口实时数据流</h1>
    <div id="data" style="white-space: pre-wrap; font-family: monospace;"></div>

    <script>
        const eventSource = new EventSource('/stream');
        const dataDiv = document.getElementById('data');

        eventSource.addEventListener('serial_data', function(event) {
            const data = JSON.parse(event.data);
            dataDiv.textContent += data.message + '\\n';
            // 自动滚动到底部
            dataDiv.scrollTop = dataDiv.scrollHeight;
        });

        eventSource.onerror = function(err) {
            console.error("SSE连接错误:", err);
            eventSource.close();
        };
    </script>
</body>
</html>

关键点说明：

  1. 串口配置：SERIAL_PORT和BAUD_RATE需匹配你的硬件设置（如树莓派GPIO串口通常是/dev/ttyAMA0或/dev/ttyS0）。
  2. SSE与Redis：Flask-SSE依赖Redis作为消息代理，需先安装Redis（sudo apt install redis-server）并确保服务运行。
  3. 后台线程：串口读取在独立线程中运行，避免阻塞Flask主线程。
  4. 前端：使用EventSource API订阅SSE流，实时更新数据。

运行步骤：

  1. 确保Redis已启动：sudo systemctl start redis
  2. 运行Python应用：python app.py
  3. 浏览器访问 http://树莓派IP:5000

如果不想用Redis，可以用Flask-SocketIO实现WebSocket方案，但SSE更轻量且适合单向数据流。注意串口权限问题，可能需要将用户加入dialout组（sudo usermod -aG dialout $USER并重新登录）。

总结：用Flask-SSE加后台线程最直接。

长连接 websocket

websocket 正解

做过类似的
https://github.com/but0n/Avem_HUD

用 pyserial 读取数据
flask 返回 json
js ajax 获取数据
canvas 渲染数据

有案例代码可以参考吗？

请问有类似的代码可以参考吗？

我网上找不到 flask+pyserial 的案例，哪位大神提供一下给我吧

轮询的话直接在 Flask 的请求里面调 pyserial 就能做，不需要案例，用 WebSocket 的话可能麻烦点。

4 楼啊

二维码扫描器淘宝买的么

买了

好像很贵 = =

400 元左右

请问 .route(’/api/status’) def api():…后为什么不用 render_template(‘xxx.html’)，我看 flask 的教程上是每个 app.route 都要引入 html 模板啊，难道可以和 def index():共同 index.html 模版？

还有想问你不用 websocket 为什么可以保持数据实时更新？

你好,参照你的项目给了我很大的帮助;最近在做一个扫描器的项目,遇到了一个问题就是 python 和 html 之间的通信建立不起来,代码如下:

#红外传感器
import RPi.GPIO as GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)
GPIO.setup(18, GPIO.IN, pull_up_down=GPIO.PUD_UP)
#16 进制库
import binascii

#FLASK 框架
from flask import Flask, url_for, render_template, jsonify

#实例化串口
import serial
import time
ser = serial.Serial(’/dev/ttyUSB0’, 9600, timeout = None) #一直等待

app = Flask(name)

.route(’/’)
def index():
return render_template(‘index.html’)

.route("/json/")
def scan(channel):
def order_list():
a = ‘7e 00 08 01 00 02 01 ab cd’ #触发扫描器的指令
a_list = []
for i in a.split():
a_list.append(binascii.a2b_hex(i)) #转为 16 进制
return a_list
ser.write(order_list()) #触发扫描
scandata = ser.readline() #读取数据
data = {“zhipinming”: “brian”,“age”: scandata,“has_car”: False}
return jsonify(data=data)

#18 引脚的中断和边缘检测,发生改变时调用 scan 函数
#红外感应一次,扫描器工作一次
GPIO.add_event_detect(18, GPIO.FALLING, callback=scan)
if name == ‘main’:
app.run()


运行后提示错误:
Traceback (most recent call last):
File “/home/pi/Desktop/tgpscan/scan.py”, line 35, in scan
return jsonify(data=data)
File “/usr/lib/python2.7/dist-packages/flask/json.py”, line 234, in jsonify
if current_app.config[‘JSONIFY_PRETTYPRINT_REGULAR’] <br> File “/usr/lib/python2.7/dist-packages/werkzeug/local.py”, line 338, in getattr
return getattr(self._get_current_object(), name)
File “/usr/lib/python2.7/dist-packages/werkzeug/local.py”, line 297, in _get_current_object
return self.__local()
File “/usr/lib/python2.7/dist-packages/flask/globals.py”, line 34, in _find_app
raise RuntimeError(‘working outside of application context’)
RuntimeError: working outside of application context

回到顶部