发帖
翼度科技»论坛 › 云主机 › LINUX › 查看内容

Flask SocketIO 实现动态绘图

追风猎人

13

主题

13

帖子

39

积分

新手上路

Rank: 1

积分
39
显示全部楼层
Flask-SocketIO 是基于 Flask 的一个扩展，用于简化在 Flask 应用中集成 WebSocket 功能。WebSocket 是一种在客户端和服务器之间实现实时双向通信的协议，常用于实现实时性要求较高的应用，如聊天应用、实时通知等，使得开发者可以更轻松地构建实时性要求较高的应用。通过定义事件处理函数，可以实现双向实时通信，为应用提供更加丰富和实时的用户体验。
前端参数拼接

Flask 提供了针对WebSocket的支持插件flask_socketio直接通过pip命令安装即可导入使用，同时前端也需要引入SocketIO.js库文件。
如下代码通过ECharts图表库和WebSocket技术实现了一个实时监控主机CPU负载的动态折线图。通过WebSocket连接到Flask应用中的Socket.IO命名空间，前端通过实时接收后端传来的CPU负载数据，动态更新折线图，展示1分钟、5分钟和15分钟的CPU负载趋势。同时，通过控制台打印实时数据，实现了方便的调试和监控功能。
  1. <!DOCTYPE html>
  2. <html lang="en">
  3. <head>
  4.     <meta charset="UTF-8">
  5.    
  6.    
  7.    
  8. </head>
  9. <body>
  13.    
  14.    
  16.    
  17.    
  19.    
  20.    
  21. </body>
  22. </html>
复制代码
后台代码使用Flask和Flask-SocketIO搭建了一个实时监控主机CPU负载的WebSocket应用，并将数据通过socketio.emit函数将数据推送给前端展示。
关键点概括如下：
Flask和SocketIO集成：

  • 使用Flask框架创建了一个Web应用，并通过Flask-SocketIO集成了WebSocket功能，实现了实时双向通信。
消息接收与实时推送：

  • 定义了socket事件处理函数，用于接收前端通过WebSocket发送的消息。在无限循环中，通过socketio.sleep方法设置每2秒推送一次实时的CPU负载数据给前端。
前端连接和断开事件：

  • 定义了connect和disconnect事件处理函数，分别在WebSocket连接建立和断开时触发。在控制台打印相应信息，用于监控连接状态。
实时数据推送：

  • 使用socketio.emit方法实时将CPU负载数据推送给前端，以更新折线图。推送的数据包括当前时间、1分钟负载、5分钟负载和15分钟负载。
前端页面渲染：

  • 通过Flask的render_template方法渲染了一个HTML页面，用于展示实时更新的CPU负载折线图。
调试信息输出：

  • 在每个事件处理函数中使用print语句输出调试信息，方便监测WebSocket连接和消息的传递过程。
总体来说，该应用实现了一个简单而实用的实时监控系统，通过WebSocket技术实时推送主机CPU负载数据至前端，为用户提供了实时可视化的监控体验。
  1. from flask import Flask,render_template,request
  2. from flask_socketio import SocketIO
  3. import time,psutil
  5. async_mode = None
  6. app = Flask(__name__)
  7. app.config['SECRET_KEY'] = "lyshark"
  8. socketio = SocketIO(app)
  10. @app.route("/")
  11. def index():
  12.     return render_template("index.html")
  14. # 出现消息后,率先执行此处
  15. @socketio.on("message",namespace="/Socket")
  16. def socket(message):
  17.     print("接收到消息:",message['data'])
  19.     while True:
  20.         socketio.sleep(2)
  21.         data = time.strftime("%M:%S",time.localtime())
  22.         cpu = psutil.cpu_percent(interval=None,percpu=True)
  24.         socketio.emit("response",
  25.                       {"datetime": data, "load1": cpu[0], "load5": cpu[1], "load15": cpu[2]},
  26.                       namespace="/Socket")
  28. # 当websocket连接成功时,自动触发connect默认方法
  29. @socketio.on("connect",namespace="/Socket")
  30. def connect():
  31.     print("链接建立成功..")
  33. # 当websocket连接失败时,自动触发disconnect默认方法
  34. @socketio.on("disconnect",namespace="/Socket")
  35. def disconnect():
  36.     print("链接建立失败..")
  38. if __name__ == '__main__':
  39.     socketio.run(app,debug=True)
复制代码
运行后，即可输出当前系统下CPU的负载情况，如下图所示；

后端参数拼接

如上所示的代码是在前端进行的数据拼接，如果我们想要在后端进行数据的拼接，则需要对代码进行一定的改进。
前端编写以下代码，通过WebSocket建立通信隧道，而后台则每隔2秒向前台推送传递字典数据。
  1. <!DOCTYPE html>
  2. <html lang="en">
  3. <head>
  4.     <meta charset="UTF-8">
  5.    
  6.    
  7.    
  8. </head>
  9. <body>
  13.    
  14.    
  16.    
  17.    
  18. </body>
  19. </html>
复制代码
后台代码则是收集数据，并将数据通过socketio.emit函数，推送给前端。
  1. from flask import Flask,render_template,request
  2. from flask_socketio import SocketIO
  3. import time,psutil
  5. async_mode = None
  6. app = Flask(__name__)
  7. app.config['SECRET_KEY'] = "lyshark"
  8. socketio = SocketIO(app)
  10. # 填充数据表
  11. local_time = ["", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", ""]
  12. 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]
  13. 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]
  14. 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]
  16. # 左移填充
  17. def shift(Array, Size, Push):
  18.     if len(Array) <= Size and len(Array) >= 0:
  19.         Array.pop(0)
  20.         Array.append(Push)
  21.         return True
  22.     return False
  24. # 右移填充
  25. def unshift(Array, Size, Push):
  26.     if len(Array) <= Size and len(Array) >= 0:
  27.         Array.pop(Size-1)
  28.         Array.insert(0,Push)
  30. @app.route("/")
  31. def index():
  32.     return render_template("index.html")
  34. # 出现消息后,率先执行此处
  35. @socketio.on("message",namespace="/Socket")
  36. def socket(message):
  37.     print("接收到消息:",message['data'])
  38.     while True:
  39.         socketio.sleep(1)
  40.         data = time.strftime("%M:%S",time.localtime())
  41.         cpu = psutil.cpu_percent(interval=None,percpu=True)
  43.         # 实现数组最大35,每次左移覆盖第一个
  44.         shift(local_time,35,data)
  45.         shift(cpu_load1,35,cpu[0])
  46.         shift(cpu_load5, 35, cpu[1])
  47.         shift(cpu_load15, 35, cpu[2])
  49.         socketio.emit("response",
  50.                       {"datetime": local_time, "load1": cpu_load1, "load5": cpu_load5, "load15": cpu_load15},
  51.                       namespace="/Socket")
  53. # 当websocket连接成功时,自动触发connect默认方法
  54. @socketio.on("connect",namespace="/Socket")
  55. def connect():
  56.     print("链接建立成功..")
  58. # 当websocket连接失败时,自动触发disconnect默认方法
  59. @socketio.on("disconnect",namespace="/Socket")
  60. def disconnect():
  61.     print("链接建立失败..")
  63. if __name__ == '__main__':
  64.     socketio.run(app,debug=True)
复制代码
运行动态图如下所示；


来源:https://www.cnblogs.com/LyShark/p/17860158.html
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x

上一篇: Flask SocketIO 实现动态绘图

下一篇: GMAC网卡Fixed-Link模式

发表于 2023-11-27 22:22:56

举报 回复 使用道具

返回列表 发新帖

Archiver|手机版|小黑屋|翼度科技

Copyright © 2001-2020, Tencent Cloud.　Powered by Discuz! X3.4