pyserial读取Flask中的串行端口（可能使用gevent）

from flask import Flask import flask import serial app = Flask(__name__) app.debug = True def event_barcode(): ser = serial.Serial() ser.port = 0 ser.baudrate = 9600 ser.bytesize = 8 ser.parity = serial.PARITY_NONE ser.stopbits = serial.STOPBITS_ONE ser.open() s = ser.read(7) yield 'data: %s\n\n' % s @app.route('/barcode') def barcode(): newresponse = flask.Response(event_barcode(), mimetype="text/event-stream") newresponse.headers.add('Access-Control-Allow-Origin', '*') return newresponse if __name__ == '__main__': app.run(port=8080, threaded=True)

<!DOCTYPE HTML> <html> <head> <meta http-equiv=Content-Type content="text/html; charset=utf-8"> <title>TEST</title> <script src="https://ajax.googleapis.com/ajax/libs/jquery/1.8.3/jquery.min.js" type="text/javascript" charset="utf-8"></script> <script> $(document).ready(function(){ if (!!window.EventSource) { console.log('SSE supported.'); var source = new EventSource('http://localhost:8080/barcode'); source.addEventListener('message', function(e) { console.log(e.data); }, false); source.addEventListener('open', function(e) { console.log('Connection was opened.'); }, false); source.addEventListener('error', function(e) { if (e.readyState == EventSource.CLOSED) { console.log('Connection was closed.'); } }, false); } else { console.log('SSE notsupported.'); } }); </script> </head> <body> </body> </html>

2条回答

网友

1楼 · 编辑于 2024-05-14 23:05:43

回答我自己的问题

似乎目前只有Firefox支持SSE的CORS->； article
在Janus Troelsen的帮助下我找到了连接打开并跨行发送多个条形码（请参见代码以下）
就性能而言，我似乎只能建立一个连接，但是那可能是因为我只有一个串行端口连接无法再打开串行端口。我想会有用的从烧瓶里拿出来的，但是有socketio和gevents的东西会用的更好，因为它更适合这份工作。有一个有趣的文章here。

对于代码：

import flask
import serial
from time import sleep

app = flask.Flask(__name__)
app.debug = True

def event_barcode():
    messageid = 0
    ser = serial.Serial()
    ser.port = 0
    ser.baudrate = 9600
    ser.bytesize = 8
    ser.parity = serial.PARITY_NONE
    ser.stopbits = serial.STOPBITS_ONE
    ser.timeout = 0
    try:
        ser.open()
    except serial.SerialException, e:
         yield 'event:error\n' + 'data:' + 'Serial port error({0}): {1}\n\n'.format(e.errno, e.strerror)
         messageid = messageid + 1
    str_list = []
    while True:
        sleep(0.01)
        nextchar = ser.read()
        if nextchar:
            str_list.append(nextchar)
        else:
            if len(str_list) > 0:
                yield 'id:' + str(messageid) + '\n' + 'data:' + ''.join(str_list) + '\n\n'
                messageid = messageid + 1
                str_list = []

@app.route('/barcode')
def barcode():
    newresponse = flask.Response(event_barcode(), mimetype="text/event-stream")
    newresponse.headers.add('Access-Control-Allow-Origin', '*')
    newresponse.headers.add('Cache-Control', 'no-cache')
    return newresponse

if __name__ == '__main__':
    app.run(port=8080, threaded=True)

因为我想支持多个浏览器，所以SSE现在不是我的选择。我会查一下websockets并尝试从中学习。

网友

2楼 · 编辑于 2024-05-14 23:05:43

以下是一些可能有帮助的建议（我一直想发布一些类似基于“django sse”的“flask sse”：

https://gist.github.com/3680055

https://gist.github.com/3687523

也有用-https://github.com/jkbr/chat/blob/master/app.py

“redissestream”类使用redis作为后端在线程之间进行通信（尽管gevent可以做到这一点？），并“侦听”redis发布事件。

虽然“PeriodicSseStream”不需要redis，但它不能在烧瓶线程之间进行通信，即使用来自另一个响应的信息；没有像redis这样的东西，单独的线程（流和为另一个用户服务的线程）就不能通信。

正如Janus所说，生成器只返回一个结果-它必须产生多个结果，在这种情况下，它必须被封装在一个循环中，在每次串行轮询之后都会无休止地产生结果；您还需要决定什么将限制轮询，它是受时间限制（定期轮询）还是其他限制（例如，如果读取串行端口已经花了一段时间）？

我不太了解sse的性能，也不知道它的支持程度（以及wrt跨域），但是如果考虑socket.io，您可以使用this来改进web套接字performance?

相关问题更多 >

编程相关推荐

热门问题

热门文章