编程宝库 - 技术改变世界

编程宝库

Search

Python编程技术

Python Flask 转换器的使用详解

将Python脚本打包成MACOSAPP程序过程

Django动态随机生成温度前端实时动态展示源码示例

动态随机生成温度前端动态实时展示

一、django APScheduler定时任务

简介

APScheduler的全称是Advanced Python Scheduler。
它是一个轻量级的 Python 定时任务调度框架。
APScheduler 支持三种调度任务：
固定时间间隔，
固定时间点（日期），
Linux 下的 Crontab命令。
同时，它还支持异步执行、后台执行调度任务。

安装

pip install APScheduler

使用步骤

新建一个 schedulers (调度器) 。
添加一个调度任务(job stores)。
运行调度任务。

基础组件

APScheduler 有四种组件，分别是：调度器(scheduler)，作业存储(job store)，触发器(trigger)，执行器(executor)。

schedulers（调度器）
它是任务调度器，属于控制器角色。它配置作业存储器和执行器可以在调度器中完成，例如添加、修改和移除作业。

triggers（触发器）
描述调度任务被触发的条件。不过触发器完全是无状态的。

job stores（作业存储器）
任务持久化仓库，默认保存任务在内存中，也可将任务保存都各种数据库中，任务中的数据序列化后保存到持久化数据库，从数据库加载后又反序列化。

executors（执行器）
负责处理作业的运行，它们通常通过在作业中提交指定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。

7个调度器

BlockingScheduler
调度器在当前进程的主线程中运行，也就是会阻塞当前线程。

BackgroundScheduler
调度器在后台线程中运行，不会阻塞当前线程。

AsyncIOScheduler
结合 asyncio 模块（一个异步框架）一起使用。

GeventScheduler
程序中使用 gevent（高性能的Python并发框架）作为IO模型，和 GeventExecutor 配合使用。

TornadoScheduler
程序中使用 Tornado（一个web框架）的IO模型，用 ioloop.add_timeout 完成定时唤醒。

TwistedScheduler
配合 TwistedExecutor，用 reactor.callLater 完成定时唤醒。

QtScheduler
你的应用是一个 Qt 应用，需使用QTimer完成定时唤醒。

二、dwebsocket

简介

WebSocket它实现了浏览器与服务器的全双工通信，扩展了浏览器与服务端的通信功能，使服务端也能主动向客户端发送数据。

安装

pip install dwebsocket

使用方法

如果你想为一个单独的视图处理一个websocklet连接可以使用accept_websocket装饰器，它会将标准的HTTP请求路由到视图中。使用require_websocke装饰器只允许使用WebSocket连接，会拒绝正常的HTTP请求。

属性和方法

1.request.is_websocket()
如果是个websocket请求返回True，如果是个普通的http请求返回False,可以用这个方法区分它们。

2.request.websocket
在一个websocket请求建立之后，这个请求将会有一个websocket属性，用来给客户端提供一个简单的api通讯，如果request.is_websocket()是False，这个属性将是None。

3.WebSocket.wait()
返回一个客户端发送的信息，在客户端关闭连接之前他不会返回任何值，这种情况下，方法将返回None

4.WebSocket.read()
如果没有从客户端接收到新的消息，read方法会返回一个新的消息，如果没有，就不返回。这是一个替代wait的非阻塞方法

5.WebSocket.count_messages()
返回消息队列数量

6.WebSocket.has_messages()
如果有新消息返回True，否则返回False

7.WebSocket.send(message)
向客户端发送消息

8.WebSocket.__iter__()
websocket迭代器

ok,现在我们进入到项目中

为了模拟随机的温度显示，在models中的表中的数据

from django.db import models
# Create your models here.
class Temperature(models.Model):
  wendu = models.IntegerField()
  add_time = models.DateTimeField(auto_now_add=True)
  class Meta:
      db_table='temperature'

views

from django.shortcuts import render,HttpResponse
from dwebsocket.decorators import accept_websocket
from apscheduler.schedulers.background import BackgroundScheduler
from . models import *
import random
import datetime
import json
def temperature():
  # 随机生成1-50之间的温度，加入到temperature表中
  data = random.randint(1,50)
  Temperature.objects.create(wendu=data)   
# 定时任务三部曲 一秒钟执行一次
scheduler = BackgroundScheduler()
scheduler .add_job(temperature, 'interval', seconds=1)
scheduler.start()
clients = []
# 使用websocket中的 @accept_websocket装饰器
@accept_websocket
def websocket_conn(request):
  # 获取连接
  if request.is_websocket:
      # 新增 用户  连接信息
      clients.append(request.websocket)
      # 监听接收客户端发送的消息 或者 客户端断开连接
      for message in request.websocket:
          break
# 发送消息
def websocketMsg(client, msg):
  b1 = json.dumps(msg,ensure_ascii=False).encode('utf-8')
  client.send(b1)
# 服务端发送消息
def send():
  if clients:
      for i in clients:
          wendu = Temperature.objects.order_by('-id').first()
          websocketMsg(i, {'wendu': wendu.wendu})
  return HttpResponse("ok")
# 定时任务 1秒/次
scheduler = BackgroundScheduler()
scheduler.add_job(send,'interval',seconds = 1)
scheduler.start()

路由

from django.urls import path
from . import views
urlpatterns = [
  path('websocket_conn/',views.websocket_conn)
]

VUE中的代码

<template>
  <div>
      当前的温度是:{{wendu}}
  </div>
</template>
<script>
import axios from 'axios'
export default {
  data() {
      return {
          wendu:0
      }
  },
  methods: {
      getWenDu(){
          var socket = new WebSocket("ws:127.0.0.1:8000/user/websocket_conn/");
             socket.onopen = function () {
              console.log('连接成功啦');//成功连接上Websocket
          };
          socket.onmessage = (e=>{
              let data = JSON.parse(e.data);//打印服务端返回的数据
              this.wendu = data['wendu']
              console.log(data['wendu'])
          })                            
          socket.onclose=function(e){
              console.log(e);
              socket.close(); //关闭TCP连接
          };      
      }
  },
  created() {
      this.getWenDu()
  }
}
</script>
<style scoped>
</style>

效果如下：

在这里插入图片描述

然后我们写一个接口展示出10秒钟的温度变化

class TestAPIView(APIView):
  def get(self,request):
      # 当前时间
      now = datetime.datetime.now()
      # 显示10秒前的温度
      datelist = []
      wendulist = []
      for i in range(10,0,-1):
          start_time = datetime.datetime.strftime(now-datetime.timedelta(seconds=i),"%Y-%m-%d %H:%M:%S")    # 当前时间
          end_time = datetime.datetime.strftime(now-datetime.timedelta(seconds=i-1),"%Y-%m-%d %H:%M:%S")   # 结束时间
          # 获取在开始时间之后 结束时间之前的的温度
          wendu = Temperature.objects.filter(add_time__gte=start_time,add_time__lte=end_time).all()
          dict = {}
          for i in wendu:
              dict['wendu'] = i.wendu
          for k,v in dict.items():
              wendulist.append(v)
          # print(wendulist)
          datelist.append(start_time)
      return Response({'datelist':datelist,'wendulist':wendulist})

这样呢我们就拿到数据如下图

在这里插入图片描述

最后，我们利用echarts在前端渲染最终效果

<template>
  <div>
      当前的温度是:{{wendu}}
      <div id="main" style="width: 800px;height:400px;">            
      </div>
  </div>
</template>
<script>
import axios from 'axios'
import * as echarts from 'echarts';
export default {
  data() {
      return {
          wendu:0,
          datelist:[],
          wendulist:[]
          }
  },
  methods: {
  // echarts
      getTest1(){
          var chartDom = document.getElementById('main');
          var myChart = echarts.init(chartDom);
            let  option = {
              xAxis: {
                  type: 'category',
                  data: this.datelist
              },
              yAxis: {
                  type: 'value',                    
              },
              series: [{
                  data: this.wendulist,
                  type: 'line',
                  smooth: true
              }]
          };
          myChart.setOption(option);
      },
      // 获取接口 
      getTest(){
          this.axios.get('user/test/').then(res=>{
              console.log(res.data)
              this.datelist = res.data.datelist
              this.wendulist = res.data.wendulist
              this.getTest1()
              this.getWenDu()
          })
      },
      getWenDu(){
          var socket = new WebSocket("ws:127.0.0.1:8000/user/websocket_conn/");
              socket.onopen = function () {
              console.log('连接成功啦');//成功连接上Websocket
          };
          socket.onmessage = (e=>{
              let data = JSON.parse(e.data);//打印服务端返回的数据
              this.wendu = data['wendu']
              console.log(data['wendu'])
          })                           
          socket.onclose=function(e){
              console.log(e);
              socket.close(); //关闭TCP连接
          };      
      }
  },
  mounted() {
      this.getWenDu()
      this.getTest()
      // this.getTest1()       
  }
}
</script>
<style scoped>
</style>

在这里插入图片描述