Flask SocketIO實(shí)現(xiàn)動態(tài)繪圖的示例詳解
Flask-SocketIO 是基于 Flask 的一個(gè)擴(kuò)展，用于簡化在 Flask 應(yīng)用中集成 WebSocket 功能。WebSocket 是一種在客戶端和服務(wù)器之間實(shí)現(xiàn)實(shí)時(shí)雙向通信的協(xié)議，常用于實(shí)現(xiàn)實(shí)時(shí)性要求較高的應(yīng)用，如聊天應(yīng)用、實(shí)時(shí)通知等，使得開發(fā)者可以更輕松地構(gòu)建實(shí)時(shí)性要求較高的應(yīng)用。通過定義事件處理函數(shù)，可以實(shí)現(xiàn)雙向?qū)崟r(shí)通信，為應(yīng)用提供更加豐富和實(shí)時(shí)的用戶體驗(yàn)。
前端參數(shù)拼接
Flask 提供了針對WebSocket的支持插件flask_socketio直接通過pip命令安裝即可導(dǎo)入使用，同時(shí)前端也需要引入SocketIO.js庫文件。
如下代碼通過ECharts圖表庫和WebSocket技術(shù)實(shí)現(xiàn)了一個(gè)實(shí)時(shí)監(jiān)控主機(jī)CPU負(fù)載的動態(tài)折線圖。通過WebSocket連接到Flask應(yīng)用中的Socket.IO命名空間，前端通過實(shí)時(shí)接收后端傳來的CPU負(fù)載數(shù)據(jù)，動態(tài)更新折線圖，展示1分鐘、5分鐘和15分鐘的CPU負(fù)載趨勢。同時(shí)，通過控制臺打印實(shí)時(shí)數(shù)據(jù)，實(shí)現(xiàn)了方便的調(diào)試和監(jiān)控功能。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<script type="text/javascript" src="https://www.lyshark.com/javascript/jquery/3.5.1/jquery.min.js"></script>
<script type="text/javascript" src="https://www.lyshark.com/javascript/socket.io/socket.io.min.js"></script>
<script type="text/javascript" src="https://www.lyshark.com/javascript/echarts/5.3.0/echarts.min.js"></script>
</head>
<body>
<div id="Linechart" style="height:500px;width:1200px;border:1px solid #673ab7;padding:10px;"></div>
<!-- 執(zhí)行繪圖函數(shù)-->
<script type="text/javascript" charset="UTF-8">
var display = function(time,x,y,z)
{
var echo = echarts.init(document.getElementById("Linechart"));
var option = {
title: {
left: 'left',
text: 'CPU 利用表動態(tài)監(jiān)控',
},
// 調(diào)節(jié)大小
grid: {
left: '3%',
right: '4%',
bottom: '3%',
containLabel: true
},
// tooltip 鼠標(biāo)放上去之后會自動出現(xiàn)坐標(biāo)
tooltip: {
trigger: 'axis',
axisPointer: {
type: 'cross',
label: {
backgroundColor: '#6a7985'
}
}
},
legend: {
data: ['1分鐘負(fù)載', '5分鐘負(fù)載', '15分鐘負(fù)載']
},
xAxis: {
type: 'category',
// data: ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
data: time
},
yAxis: {
type: 'value'
},
series:
[
{
name: "1分鐘負(fù)載",
stack: "總量",
//data: [10, 25, 99, 87, 54, 66, 2],
data: x,
type: 'line',
areaStyle: {}
},
{
name: "5分鐘負(fù)載",
stack: "總量",
//data: [89, 57, 85, 44, 25, 4, 54],
data: y,
type: 'line',
areaStyle: {}
},
{
name: "15分鐘負(fù)載",
stack: "總量",
//data: [1, 43, 2, 12, 5, 4, 7],
data: z,
type: 'line',
areaStyle: {}
}
]
};
echo.setOption(option,true);
}
</script>
<!-- 負(fù)責(zé)對參數(shù)的解析,填充數(shù)據(jù) -->
<script type="text/javascript" charset="UTF-8">
var time =["","","","","","","","","","","","","","","","","","","","","","","","","","","","","","","","","","",""];
var cpu_load1 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0];
var cpu_load5 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0];
var cpu_load15 = [0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0];
var update_function = function(recv)
{
time.push(recv.datetime);
cpu_load1.push(parseFloat(recv.load1));
cpu_load5.push(parseFloat(recv.load5));
cpu_load15.push(parseFloat(recv.load15));
if(time.length >=10)
{
time.shift();
cpu_load1.shift();
cpu_load5.shift();
cpu_load15.shift();
console.log("時(shí)間數(shù)組: " + time);
console.log("1分鐘: " + cpu_load1);
console.log("5分鐘: " + cpu_load5);
console.log("15分鐘: " + cpu_load15);
// 調(diào)用繪圖函數(shù)
display(time,cpu_load1,cpu_load5,cpu_load15);
}
};
</script>
<!-- 負(fù)責(zé)接收目標(biāo)主機(jī)的CPU負(fù)載情況 -->
<script type="text/javascript" charset="UTF-8">
$(document).ready(function()
{
namespace = '/Socket';
var socket = io.connect("http://" + document.domain + ":" + location.port + namespace);
socket.emit("message",{"data":"hello lyshark"}); // 初始化完成后,發(fā)送一條消息.
socket.on('response', function(recv) {
console.log("時(shí)間: " + recv.datetime);
console.log("1分鐘: " + recv.load1);
console.log("5分鐘: " + recv.load5);
console.log("15分鐘: " + recv.load15);
// 調(diào)用函數(shù)完成數(shù)據(jù)填充
update_function(recv);
});
});
</script>
</body>
</html>
后臺代碼使用Flask和Flask-SocketIO搭建了一個(gè)實(shí)時(shí)監(jiān)控主機(jī)CPU負(fù)載的WebSocket應(yīng)用，并將數(shù)據(jù)通過socketio.emit函數(shù)將數(shù)據(jù)推送給前端展示。
關(guān)鍵點(diǎn)概括如下：
Flask和SocketIO集成：
使用Flask框架創(chuàng)建了一個(gè)Web應(yīng)用，并通過Flask-SocketIO集成了WebSocket功能，實(shí)現(xiàn)了實(shí)時(shí)雙向通信。
消息接收與實(shí)時(shí)推送：
定義了socket事件處理函數(shù)，用于接收前端通過WebSocket發(fā)送的消息。在無限循環(huán)中，通過socketio.sleep方法設(shè)置每2秒推送一次實(shí)時(shí)的CPU負(fù)載數(shù)據(jù)給前端。
前端連接和斷開事件：
定義了connect和disconnect事件處理函數(shù)，分別在WebSocket連接建立和斷開時(shí)觸發(fā)。在控制臺打印相應(yīng)信息，用于監(jiān)控連接狀態(tài)。
實(shí)時(shí)數(shù)據(jù)推送：
使用socketio.emit方法實(shí)時(shí)將CPU負(fù)載數(shù)據(jù)推送給前端，以更新折線圖。推送的數(shù)據(jù)包括當(dāng)前時(shí)間、1分鐘負(fù)載、5分鐘負(fù)載和15分鐘負(fù)載。
前端頁面渲染：
通過Flask的render_template方法渲染了一個(gè)HTML頁面，用于展示實(shí)時(shí)更新的CPU負(fù)載折線圖。
調(diào)試信息輸出：
在每個(gè)事件處理函數(shù)中使用print語句輸出調(diào)試信息，方便監(jiān)測WebSocket連接和消息的傳遞過程。
總體來說，該應(yīng)用實(shí)現(xiàn)了一個(gè)簡單而實(shí)用的實(shí)時(shí)監(jiān)控系統(tǒng)，通過WebSocket技術(shù)實(shí)時(shí)推送主機(jī)CPU負(fù)載數(shù)據(jù)至前端，為用戶提供了實(shí)時(shí)可視化的監(jiān)控體驗(yàn)。
from flask import Flask,render_template,request
from flask_socketio import SocketIO
import time,psutil
async_mode = None
app = Flask(__name__)
app.config['SECRET_KEY'] = "lyshark"
socketio = SocketIO(app)
@app.route("/")
def index():
return render_template("index.html")
# 出現(xiàn)消息后,率先執(zhí)行此處
@socketio.on("message",namespace="/Socket")
def socket(message):
print("接收到消息:",message['data'])
while True:
socketio.sleep(2)
data = time.strftime("%M:%S",time.localtime())
cpu = psutil.cpu_percent(interval=None,percpu=True)
socketio.emit("response",
{"datetime": data, "load1": cpu[0], "load5": cpu[1], "load15": cpu[2]},
namespace="/Socket")
# 當(dāng)websocket連接成功時(shí),自動觸發(fā)connect默認(rèn)方法
@socketio.on("connect",namespace="/Socket")
def connect():
print("鏈接建立成功..")
# 當(dāng)websocket連接失敗時(shí),自動觸發(fā)disconnect默認(rèn)方法
@socketio.on("disconnect",namespace="/Socket")
def disconnect():
print("鏈接建立失敗..")
if __name__ == '__main__':
socketio.run(app,debug=True)
運(yùn)行后，即可輸出當(dāng)前系統(tǒng)下CPU的負(fù)載情況，如下圖所示；
后端參數(shù)拼接
如上所示的代碼是在前端進(jìn)行的數(shù)據(jù)拼接，如果我們想要在后端進(jìn)行數(shù)據(jù)的拼接，則需要對代碼進(jìn)行一定的改進(jìn)。
前端編寫以下代碼，通過WebSocket建立通信隧道，而后臺則每隔2秒向前臺推送傳遞字典數(shù)據(jù)。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<script type="text/javascript" src="https://www.lyshark.com/javascript/jquery/3.5.1/jquery.min.js"></script>
<script type="text/javascript" src="https://www.lyshark.com/javascript/socket.io/socket.io.min.js"></script>
<script type="text/javascript" src="https://www.lyshark.com/javascript/echarts/5.3.0/echarts.min.js"></script>
</head>
<body>
<div id="Linechart" style="height:500px;width:1200px;border:1px solid #673ab7;padding:10px;"></div>
<!-- 執(zhí)行繪圖函數(shù)-->
<script type="text/javascript" charset="UTF-8">
var display = function(time,x,y,z)
{
var echo = echarts.init(document.getElementById("Linechart"));
var option = {
title: {
left: 'left',
text: 'CPU 利用表動態(tài)監(jiān)控',
},
// 調(diào)節(jié)大小
grid: {
left: '3%',
right: '4%',
bottom: '3%',
containLabel: true
},
// tooltip 鼠標(biāo)放上去之后會自動出現(xiàn)坐標(biāo)
tooltip: {
trigger: 'axis',
axisPointer: {
type: 'cross',
label: {
backgroundColor: '#6a7985'
}
}
},
legend: {
data: ['1分鐘負(fù)載', '5分鐘負(fù)載', '15分鐘負(fù)載']
},
xAxis: {
type: 'category',
// data: ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
data: time
},
yAxis: {
type: 'value'
},
series:
[
{
name: "1分鐘負(fù)載",
stack: "總量",
//data: [10, 25, 99, 87, 54, 66, 2],
data: x,
type: 'line',
areaStyle: {}
},
{
name: "5分鐘負(fù)載",
stack: "總量",
//data: [89, 57, 85, 44, 25, 4, 54],
data: y,
type: 'line',
areaStyle: {}
},
{
name: "15分鐘負(fù)載",
stack: "總量",
//data: [1, 43, 2, 12, 5, 4, 7],
data: z,
type: 'line',
areaStyle: {}
}
]
};
echo.setOption(option,true);
}
</script>
<!-- 負(fù)責(zé)接收目標(biāo)主機(jī)的CPU負(fù)載情況 -->
<script type="text/javascript" charset="UTF-8">
$(document).ready(function()
{
namespace = '/Socket';
var socket = io.connect("http://" + document.domain + ":" + location.port + namespace);
socket.emit("message",{"data":"hello lyshark"}); // 初始化完成后,發(fā)送一條消息.
socket.on('response', function(recv) {
console.log("時(shí)間: " + recv.datetime);
console.log("1分鐘: " + recv.load1);
console.log("5分鐘: " + recv.load5);
console.log("15分鐘: " + recv.load15);
// 調(diào)用繪圖函數(shù)
display(recv.datetime,recv.load1,recv.load5,recv.load15);
});
});
</script>
</body>
</html>
后臺代碼則是收集數(shù)據(jù)，并將數(shù)據(jù)通過socketio.emit函數(shù)，推送給前端。
from flask import Flask,render_template,request
from flask_socketio import SocketIO
import time,psutil
async_mode = None
app = Flask(__name__)
app.config['SECRET_KEY'] = "lyshark"
socketio = SocketIO(app)
# 填充數(shù)據(jù)表
local_time = ["", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", ""]
cpu_load1 = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
cpu_load5 = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
cpu_load15 = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
# 左移填充
def shift(Array, Size, Push):
if len(Array) <= Size and len(Array) >= 0:
Array.pop(0)
Array.append(Push)
return True
return False
# 右移填充
def unshift(Array, Size, Push):
if len(Array) <= Size and len(Array) >= 0:
Array.pop(Size-1)
Array.insert(0,Push)
@app.route("/")
def index():
return render_template("index.html")
# 出現(xiàn)消息后,率先執(zhí)行此處
@socketio.on("message",namespace="/Socket")
def socket(message):
print("接收到消息:",message['data'])
while True:
socketio.sleep(1)
data = time.strftime("%M:%S",time.localtime())
cpu = psutil.cpu_percent(interval=None,percpu=True)
# 實(shí)現(xiàn)數(shù)組最大35,每次左移覆蓋第一個(gè)
shift(local_time,35,data)
shift(cpu_load1,35,cpu[0])
shift(cpu_load5, 35, cpu[1])
shift(cpu_load15, 35, cpu[2])
socketio.emit("response",
{"datetime": local_time, "load1": cpu_load1, "load5": cpu_load5, "load15": cpu_load15},
namespace="/Socket")
# 當(dāng)websocket連接成功時(shí),自動觸發(fā)connect默認(rèn)方法
@socketio.on("connect",namespace="/Socket")
def connect():
print("鏈接建立成功..")
# 當(dāng)websocket連接失敗時(shí),自動觸發(fā)disconnect默認(rèn)方法
@socketio.on("disconnect",namespace="/Socket")
def disconnect():
print("鏈接建立失敗..")
if __name__ == '__main__':
socketio.run(app,debug=True)
運(yùn)行動態(tài)圖如下所示；
以上就是Flask SocketIO實(shí)現(xiàn)動態(tài)繪圖的示例詳解的詳細(xì)內(nèi)容，更多關(guān)于Flask SocketIO動態(tài)繪圖的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
Python內(nèi)置函數(shù)的用法實(shí)例教程
這篇文章主要介紹了Python內(nèi)置函數(shù)的用法,包括求絕對值的abs()函數(shù)及數(shù)值類型轉(zhuǎn)換函數(shù)等,需要的朋友可以參考下2014-09-09
關(guān)于爬蟲中scrapy.Request的更多參數(shù)用法
這篇文章主要介紹了關(guān)于爬蟲中scrapy.Request的更多參數(shù)用法說明，具有很好的參考價(jià)值，希望對大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-07-07
keras 獲取某層輸出 獲取復(fù)用層的多次輸出實(shí)例
這篇文章主要介紹了keras 獲取某層輸出 獲取復(fù)用層的多次輸出實(shí)例，具有很好的參考價(jià)值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-05-05
python 畫二維、三維點(diǎn)之間的線段實(shí)現(xiàn)方法
今天小編就為大家分享一篇python 畫二維、三維點(diǎn)之間的線段實(shí)現(xiàn)方法，具有很好的參考價(jià)值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-07-07
列表推導(dǎo)式形式較為簡潔，是利用其它列表創(chuàng)建新列表的一種方式，它的工作方式類似于for循環(huán)，也可以嵌套if條件判斷語句,需要的朋友可以參考下2023-04-04
python多線程實(shí)現(xiàn)動態(tài)圖繪制
這篇文章主要介紹了python多線程實(shí)現(xiàn)動態(tài)圖繪制，文章基于Python的相資料展開動態(tài)圖的繪制相關(guān)內(nèi)容，具有一定的參考價(jià)值，需要的小伙伴可以參考一下2022-04-04
Python3利用Dlib19.7實(shí)現(xiàn)攝像頭人臉識別的方法
這篇文章主要介紹了Python 3 利用 Dlib 19.7 實(shí)現(xiàn)攝像頭人臉識別 ，利用python開發(fā)，借助Dlib庫捕獲攝像頭中的人臉，提取人臉特征，通過計(jì)算歐氏距離來和預(yù)存的人臉特征進(jìn)行對比，達(dá)到人臉識別的目的，感興趣的小伙伴們可以參考一下2018-05-05
Python OpenCV學(xué)習(xí)之圖像形態(tài)學(xué)
形態(tài)學(xué)處理方法是基于對二進(jìn)制圖像進(jìn)行處理的，卷積核決定圖像處理后的效果。本文將為大家詳細(xì)介紹一下OpenCV中的圖像形態(tài)學(xué)，感興趣的可以了解一下2022-01-01
python bluetooth藍(lán)牙信息獲取藍(lán)牙設(shè)備類型的方法
這篇文章主要介紹了python bluetooth藍(lán)牙信息獲取藍(lán)牙設(shè)備類型的方法，具體轉(zhuǎn)化方法文中給大家介紹的非常詳細(xì)，非常不錯(cuò)，具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2019-11-11
python實(shí)現(xiàn)動態(tài)數(shù)組的示例代碼
這篇文章主要介紹了python實(shí)現(xiàn)動態(tài)數(shù)組的示例代碼，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-07-07