高质量编码--使用Pandas和Tornado构建高性能数据查询服务

#-*- coding:utf-8 -*-
import os
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
from tornado.options import define,options
import json
import datetime
from tools import *

define('port',default=8096,help='run on the given port',type=int)

class cross_originAllowed_Handler(tornado.web.RequestHandler):
    
    def initialize(self):
       
        self.set_header('Access-Control-Allow-Origin','*')
        self.set_header('Access-Control-Allow-Methods','POST,GET')
        self.set_header('Access-Control-Max-Age',1000)
        self.set_header('Access-Control-Allow-Headers','*')

class indexHandler(tornado.web.RequestHandler):
    def get(self):      
        self.render('index.html')   

class devsHandler(cross_originAllowed_Handler):
    def get(self):
        self.write({'result':devs.to_json()})  
              
class dataHandler(cross_originAllowed_Handler):
    def get(self):
        dev=self.get_argument('dev',None)
        startDate=self.get_argument('startDate',None)
        endDate=self.get_argument('endDate',None)
        if not (dev and startDate and endDate):
            self.write({'result':'{}'})
            return     
        data=getByDev2DateRange(int(dev),startDate,endDate)
        self.write({'result':data})

class Application(tornado.web.Application):
    def __init__(self, handlers, **settings):
        global df
        global devs
        df,devs=initDf()
        tornado.web.Application.__init__(self, handlers, **settings)    
        
if __name__=='__main__':
    tornado.options.parse_command_line()
    app=Application(
        handlers=[
                 
                (r'/',indexHandler),
                (r'/index',indexHandler),
                
                 (r'/devs',devsHandler),
                (r'/data',dataHandler),
                 (r"/assets/(.*)", tornado.web.StaticFileHandler, {"path": "static/assets"}),

                (r"/dist/(.*)", tornado.web.StaticFileHandler, {"path": "static/dist"}),
                (r"/css/(.*)", tornado.web.StaticFileHandler, {"path": "static/css"}),
                (r"/js/(.*)", tornado.web.StaticFileHandler, {"path": "static/js"}),
                 
                  ],

        template_path=os.path.join(os.path.curdir,'templates'),static_path=os.path.join(os.path.curdir,'static'),cookie_secret='miaojiyue',debug=True )
         
    http_server=tornado.httpserver.HTTPServer(app)
    http_server.listen(options.port)
    tornado.ioloop.IOLoop.instance().start()
    

#-*-coding:utf-8 -*-
import os
import numpy as np
import pandas as pd
def initData():
    dataDir='D:\wc\data'
    csvs=os.listdir(dataDir)
    dfs=[]
    names=['ID', 'DevID', 'DevData', 'DevUnit', 'voltage',
       'DevAlarm', 'RecDateTime', 'StrInfo', 'ErrorCode',
       'dflag', 'AddTime']

    #遍历文件夹中所有csv文件,将数据拼合到一个dataframe中
    for csv in csvs:
        #由于csv中首行没有存储列名，指定数据对应的列名称
        df0=pd.read_csv(os.path.join(dataDir,csv),names=names)
        print(csv)
        print(df0)
        dfs.append(df0)
    df=pd.concat(dfs)
    df['devId']=df['DevID']
    df['devUnit']=df['DevUnit']
    #根据小数点分隔字符串，将时间格式化到整数秒，并将字符串转为时间格式。
    df.RecDateTime=df.RecDateTime.apply(lambda x:x[:x.find('.')])
    df.RecDateTime=pd.to_datetime(df.RecDateTime)
    df['recDateTime']=df['RecDateTime']
    #为dataframe指定设备ID和接收时间作为索引并根据索引排序
    df=df.set_index(['DevID','RecDateTime']).drop(columns=['ID']).sort_index()
    return df

def initDf():
    print('begin to read data from pkl!')
    global df
    df=pd.read_pickle('data.pkl')
    print('read complete!')
    global df2
    print('begin to read devs!')
    #devs=df.index.map(lambda x:x[0]).unique().to_list()
    #devs=np.load('data.npz')['devs'].tolist()
    #df2=pd.read_excel('data.xls',names=["devId","devName","devUnit"])
    df2=pd.read_pickle('data2.pkl')
    
    print('devs complete!')
    return df,df2

def getByDev2DateRange(dev,startDate,endDate):
    print(dev)
    print(startDate)
    print(endDate)
    if dev not in df2['devId']:
        return '{}'
    dates=pd.date_range(start=startDate, end=endDate)
    dates=dates.map(lambda x:x.strftime('%Y-%m-%d'))
    #使用传感器ID索引过滤数据
    d=df.loc[dev]
    #取得该传感器所有数据的时间列表
    dates2=d.index.map(lambda x:x.strftime('%Y-%m-%d'))
    #用是否包含布尔索引来过滤查询时间范围的数据，也可以使用dates.contains(date2)
    data=d[dates2.isin(dates)]
    #只返回'recDateTime','DevData'两列
    data=data.loc[:,['recDateTime','DevData']]
    data.recDateTime=data.recDateTime.apply(lambda x:x.strftime('%Y-%m-%d %H:%M:%S'))
    data=data.reset_index()
    data=data.drop(columns=['RecDateTime'])
    return data.to_json()