data_analyze_platform/LIB/BACKEND/DBManager.py

'''
暂时采用http方式获取历史数据。

预留：后期若改用通过访问数据库的形式进行数据的获取，则本文件负责数据库的连接，sql指令的执行，数据获取等功能。
'''
__author__ = 'lmstack'

import time
import datetime
import time
import pandas as pd
import numpy as np
import json
import requests
import pymysql
import pdb
from collections import Counter



class DBManager():

    def __init__(self, host='', port='', auth='', db='', username='', password=''):
        self.host = host
        self.port = port
        self.auth = auth
        self.db = db
        self.username = username
        self.password = password
        pass

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def connect(self):
        conn_success_flag = 0
        while not conn_success_flag:
            try:
                self.conn = pymysql.connect(host=self.host, port=self.port, user=self.user, password=self.password, database=self.db)
            except Exception as e:
                conn_success_flag = 0
                print("数据库连接失败 :{}".format(e))
                time.sleep(5)
            else:
                conn_success_flag = 1
                self.cursor = self.conn.cursor()
    def close(self):
        try:
            self.conn.close()
        except Exception as e:
            print(e)
        else:
            print('数据库已断开连接')

    def add(table, keyvalue):
        fields_str = ''
        values_str = ''
        for k,v in keyvalue.items():

            fields_str += k+' '
        sql = 'insert into table {} ({}) values ({})'.format(table, fields_str, values_str)


    # 以下各个函数实现 通过http方式获取数据
    @staticmethod
    def _get_var_name(cellnum,Tempnum,Othernum):
        temp = []
        for i in range(cellnum):
            temp.append('单体电压'+str(i+1))
        for i in range(Tempnum):
            temp.append('单体温度'+str(i+1))
        for i in range(Othernum):
            temp.append('其他温度'+str(i+1))
        return temp

    @staticmethod
    def _download_json_data(url):
        '''
        返回json数据的生成器，一次一行
        '''
        i = 0
        while 1:
            try:
                r = requests.get(url, stream=True, timeout=(0.1, 1000),  headers={'Connection':'keep-alive', 'Accept':'*/*', 'Accept-Encoding':'gzip,deflate,br'})
                break
            except requests.exceptions.RequestException as e:
                # if (i == 0):
                #     print()
                # print('\r' + 'Server Error, retry {}......'.format(str(i)), end=" ")
                time.sleep(0.1)
                i+=1    
        # print(r.content)
        # pdb.set_trace()
        data = [] 
        for line in r.iter_lines():

            if line:
                data.append(json.loads(line))
                # yield json.loads(line)
        return data

    @staticmethod
    def _convert_to_dataframe_bms(data, mode=0, CellUNum=0, CellTNum=0, OtherTNum=0):
        CellU = []
        CellT = []
        OtherT = []
        CellU_Num = CellUNum
        CellT_Num = CellTNum
        OtherT_Num = OtherTNum
        # try:
        #     CellU_Num = len(data['ffBatteryStatus']['cellVoltageList'])
        #     CellU = np.array(data['ffBatteryStatus']['cellVoltageList']) * 1000
        #     CellU = CellU.tolist()
        # except:
        #     CellU_Num = 0
        # try:
        #     CellT_Num = len(data['ffBatteryStatus']['cellTempList'])
        #     CellU.extend(data['ffBatteryStatus']['cellTempList'])
        # except:
        #     CellT_Num = 0

        # try:
        #     OtherT_Num = len(data['ffBatteryStatus']['otherTempList'])
        #     CellU.extend(data['ffBatteryStatus']['otherTempList'])
        # except:
        #     OtherT_Num = 0
   
        CellU = np.array(data['ffBatteryStatus'].get('cellVoltageList',[]))
        CellT = data['ffBatteryStatus'].get('cellTempList',[])
        OtherT = data['ffBatteryStatus'].get('otherTempList',[])
        if (len(CellU) != CellU_Num or len(CellT) != CellT_Num or len(OtherT) != OtherT_Num):
            return pd.DataFrame()
        CellU = CellU * 1000


        # for i in range(CellU_Num):
        #     CellU.append(data['ffBatteryStatus']['cellVoltageList'][i]*1000)
        # for i in range(CellT_Num):
        #     CellU.append(data['ffBatteryStatus']['cellTempList'][i])
        # for i in range(OtherT_Num):
        #     CellU.append(data['ffBatteryStatus']['otherTempList'][i])
        if mode == 0:
            data_len = 16
            
            # data_block = np.array([data['info']['obdTime'],data['ffBatteryStatus']['rssi'],data['ffBatteryStatus']['errorLevel'],data['ffBatteryStatus']['errorCode']
            #         ,data['ffBatteryStatus']['current'],data['ffBatteryStatus']['voltageInner'],data['ffBatteryStatus']['voltageOutter'],
            #         data['ffBatteryStatus']['totalOutputState'],data['ffBatteryStatus']['lockedState'],
            #         data['ffBatteryStatus']['chargeState'],data['ffBatteryStatus']['heatState'],data['ffBatteryStatus']['cellVoltageDiff']
            #         ,data['ffBatteryStatus']['soc'],data['ffBatteryStatus']['soh'],data['ffBatteryStatus']['cellVolBalance']]).reshape(1,data_len)
            data_block = np.array([data['info']['obdTime'],data['ffBatteryStatus'].get('rssi',None),data['ffBatteryStatus'].get('errorLevel', None),data['ffBatteryStatus'].get('errorCode', None), 
            data['ffBatteryStatus'].get('current',None),data['ffBatteryStatus'].get('voltageInner', None),data['ffBatteryStatus'].get('voltageOutter', None),
                    data['ffBatteryStatus'].get('totalOutputState', None),data['ffBatteryStatus'].get('lockedState', None),
                    data['ffBatteryStatus'].get('chargeState', None),data['ffBatteryStatus'].get('heatState', None),data['ffBatteryStatus'].get('cellVoltageDiff', None)
                    ,data['ffBatteryStatus'].get('soc', None),data['ffBatteryStatus'].get('soh', None),data['ffBatteryStatus'].get('cellVolBalance', None),data['ffBatteryStatus'].get('insResis', None)]).reshape(1,data_len)
        elif mode == 1:
            data_len = 12
            
            data_block = np.array([data['info']['obdTime'],data['ffBatteryStatus'].get('rssi',None)
            ,data['ffBatteryStatus'].get('errorLevel', None),data['ffBatteryStatus'].get('errorCode', None),data['ffBatteryStatus'].get('switchState', None)
            ,data['ffBatteryStatus'].get('current',None),data['ffBatteryStatus'].get('voltageInner', None),data['ffBatteryStatus'].get('chargeState', None),
            data['ffBatteryStatus'].get('cellVoltageDiff', None),data['ffBatteryStatus'].get('soc', None),data['ffBatteryStatus'].get('soh', None),data['ffBatteryStatus'].get('insResis', None)]).reshape(1,data_len)
        data_block = np.append(data_block,CellU)
        data_block = np.append(data_block,CellT)
        data_block = np.append(data_block,OtherT)
        data_block = data_block.reshape(1,len(data_block))
        return data_block

    @staticmethod
    def _convert_to_dataframe_gps(data, mode=0):
        if mode == 0:
            # if data['info']['subType'] == 1:
                data_block = np.array([data['info'].get('obdTime',None),data['ffGps'].get('locationType',None), data['ffGps'].get('satellites',None),
                                       data['ffGps'].get('latitude',None),data['ffGps'].get('longitude',None),data['ffGps'].get('speed',None), 
                                       data['ffGps'].get('altitude',None), data['ffGps'].get('direction', None), data['ffGps'].get('mileage',None)]).reshape(1,9)
                df = pd.DataFrame(
                    columns=['时间戳','定位类型', '卫星数','纬度','经度','速度[km/h]','海拔','航向', '里程/m'],data=data_block)
            # elif data['info']['subType'] == 2:
            #     df = pd.DataFrame(
            #         columns=['时间戳','定位类型', '卫星数','纬度','经度','速度[km/h]','海拔','航向'])
        if mode == 1:
                data_block = np.array([data['info']['obdTime'],data['ffGps']['locationType'],data['ffGps']['latitude'],data['ffGps']['longitude']
                    ,data['ffGps']['speed'], data['ffGps']['isValid']]).reshape(1,6)
                df = pd.DataFrame(
                    columns=['时间戳','定位类型', '纬度','经度','速度[km/h]','有效位'],data=data_block)
        return df

    
    @staticmethod
    def _convert_to_dataframe_system(data, mode=0):
        if mode == 0:
            
            data_block = np.array([data['info'].get('obdTime', None),data['ffSystemInfo'].get('heatTargetTemp', None), data['ffSystemInfo'].get('heatTimeout',None),
                                    time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(data['ffSystemInfo'].get('rentalStartTime'))/1000)),
                                    data['ffSystemInfo'].get('rentalPeriodDays',None),data['ffSystemInfo'].get('bmsInterval',None), 
                                    data['ffSystemInfo'].get('gpsInterval', None)]).reshape(1,7)
            df = pd.DataFrame(
                columns=['时间戳','加热目标温度', '加热超时','租赁开始时间','租赁天数','bms上传周期','gps上传周期'],data=data_block)
        if mode == 1:
                df = pd.DataFrame()

        return df
    
    @staticmethod
    def _convert_to_dataframe_accum(data, mode=0):
        if mode == 0:
            
            data_block = np.array([data['info'].get('obdTime',None),data['ffBatteryAccum'].get('SOH_AlgUnexTime',None), data['ffBatteryAccum'].get('CHG_AHaccum',None),
                                    data['ffBatteryAccum'].get('CHG_PHaccum',None), data['ffBatteryAccum'].get('DSG_AHaccum',None),
                                    data['ffBatteryAccum'].get('DSG_PHaccum',None),data['ffBatteryAccum'].get('OverTemp_CHG_AHaccum',None), 
                                    data['ffBatteryAccum'].get('OverTemp_CHG_PHaccum',None)]).reshape(1,8)
            df = pd.DataFrame(
                columns=['时间戳','SOH未标定时间', '累计充电电量','累计充电能量','累计放电电量','累计放电能量',
                                               '累计高温充电电量', '累计高温充电能量'],data=data_block)

        if mode == 1:
                data_block = np.array([data['info'].get('obdTime',None), data['ffBatteryAccum'].get('CHG_AHaccum',None),
                                    data['ffBatteryAccum'].get('CHG_PHaccum',None), data['ffBatteryAccum'].get('DSG_AHaccum',None),
                                    data['ffBatteryAccum'].get('DSG_PHaccum',None),data['ffBatteryAccum'].get('totalMileage',None)]).reshape(1,6)
                df = pd.DataFrame(
                    columns=['时间戳','累计充电电量','累计充电能量','累计放电电量','累计放电能量', '累积里程'],data=data_block)
        return df
    
    @staticmethod
    def _convert_to_dataframe_storage_accum(data, mode=0):
        if mode == 0:
            
            data_block = np.array([data['info'].get('obdTime',None),data['ffEnergyStoreStat'].get('usedEnergy',None), data['ffEnergyStoreStat'].get('inputEnergy',None),
                                    data['ffEnergyStoreStat'].get('outputEnergy',None), data['ffEnergyStoreStat'].get('usedEnergy2',None), data['ffEnergyStoreStat'].get('inputEnergy2',None),
                                    data['ffEnergyStoreStat'].get('outputEnergy2',None)]).reshape(1,7)
            df = pd.DataFrame(
                columns=['时间戳','累计用电kWh数1', '累计正向输入kWh1','累计反向输出kWh数1',
                        '累计用电kWh数2', '累计正向输入kWh2','累计反向输出kWh数2'],data=data_block)

        if mode == 1:
                data_block = np.array([data['info'].get('obdTime',None),data['ffEnergyStoreStat'].get('usedEnergy',None), data['ffEnergyStoreStat'].get('inputEnergy',None),
                                    data['ffEnergyStoreStat'].get('outputEnergy',None), data['ffEnergyStoreStat'].get('usedEnergy2',None), data['ffEnergyStoreStat'].get('inputEnergy2',None),
                                    data['ffEnergyStoreStat'].get('outputEnergy2',None)]).reshape(1,7)
                df = pd.DataFrame(
                    columns=['时间戳','累计用电kWh数1', '累计正向输入kWh1','累计反向输出kWh数1',
                        '累计用电kWh数2', '累计正向输入kWh2','累计反向输出kWh数2'],data=data_block)
        return df
    
    
    @staticmethod
    def _get_data(urls,type_name,mode=0):
        data = DBManager._download_json_data(urls)
        if type_name == 'bms':
            if mode == 0:
                name_const = ['时间戳','GSM信号','故障等级','故障代码','总电流[A]','总电压[V]', '外电压', '总输出状态', '上锁状态', '充电状态','加热状态',
                              '单体压差', 'SOC[%]','SOH[%]','单体均衡状态', '绝缘电阻']
            elif mode == 1:
                name_const = ['时间戳','GSM信号','故障等级', '故障代码','开关状态', '总电流[A]','总电压[V]','充电状态', '单体压差', 'SOC[%]','SOH[%]', '绝缘电阻']
            i=0
            st = time.time()
            
            # 计算本次最大电芯数量
            CellUNum = 0
            CellTNum = 0
            OtherTNum = 0
            
            cellUNumList = []
            cellTNumList = []
            otherTNumList = []

            for line in data:
                temp = len(line['ffBatteryStatus'].get('cellVoltageList', []))
                cellUNumList.append(temp)
                # if (temp > CellUNum):
                #     CellUNum = temp
                temp = len(line['ffBatteryStatus'].get('cellTempList', []))
                cellTNumList.append(temp)
                # if (temp > CellTNum):
                #     CellTNum = temp
                temp = len(line['ffBatteryStatus'].get('otherTempList', []))
                otherTNumList.append(temp)
                # if (temp > OtherTNum):
                #     OtherTNum = temp
 
            if (len(data)>0):
                result = Counter(cellUNumList)
                CellUNum = max(result, key=result.get)
                
                result = Counter(cellTNumList)
                CellTNum = max(result, key=result.get)

                result = Counter(otherTNumList)
                OtherTNum = max(result, key=result.get)

                       
            data_blocks = pd.DataFrame()
            for line in data:
                et = time.time()
                try:
                    if i==0:
                        data_blocks = DBManager._convert_to_dataframe_bms(line, mode, CellUNum, CellTNum, OtherTNum)
                        if (len(data_blocks)>0):
                            i+=1
                        continue
                except Exception as e:
                    # print(e)
                    continue
                    
                try:
                    data_block = DBManager._convert_to_dataframe_bms(line, mode, CellUNum, CellTNum, OtherTNum)
                except Exception as e:
                    # print(e)
                    continue
           
                try:
                    if (len(data_block)>0):
                        data_blocks = np.concatenate((data_blocks,data_block),axis=0)
                except Exception as e:
                    print(e)
                    if 'all the input array dimensions for the concatenation axis must match exactly' in str(e) or \
                    'all the input array dimensions except for the concatenation axis must match exactly'  in str(e):
                        pass
                    else:
                        raise e
                i+=1
  
            name_var = DBManager._get_var_name(CellUNum,CellTNum,OtherTNum)
            name_const.extend(name_var)
            columns_name = name_const
            if i==0:
                data_blocks = []

            df_all = pd.DataFrame(columns=columns_name,data=data_blocks)

                
            if not df_all.empty:
                df_all.loc[:,'时间戳'] = df_all.loc[:,'时间戳'].apply(lambda x:time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(x)/1000)))
            return df_all
        elif type_name =='gps':
            if mode == 0:
                df_all = pd.DataFrame(columns=['时间戳','定位类型', '卫星数','纬度','经度','速度[km/h]','海拔','航向'])
            elif mode == 1:
                df_all = pd.DataFrame(columns=['时间戳','定位类型', '纬度','经度','速度[km/h]','有效位'])
            
            for line in data:
                df_add = DBManager._convert_to_dataframe_gps(line, mode)
                df_all = df_all.append(df_add,ignore_index=True)
            if not df_all.empty:
                df_all.loc[:,'时间戳'] = df_all.loc[:,'时间戳'].apply(lambda x:time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(x)/1000)))
            return df_all
        elif type_name =='system':
            if mode == 0:
                df_all = pd.DataFrame(columns=['时间戳','加热目标温度', '加热超时','租赁开始时间','租赁天数','bms上传周期','gps上传周期'])
            elif mode == 1:
                df_all = pd.DataFrame()
            for line in data:
                df_add = DBManager._convert_to_dataframe_system(line, mode)
                df_all = df_all.append(df_add,ignore_index=True)
            if not df_all.empty:
                df_all.loc[:,'时间戳'] = df_all.loc[:,'时间戳'].apply(lambda x:time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(x)/1000)))
            return df_all       
        elif type_name =='accum':
            if mode == 0:
                df_all = pd.DataFrame(columns=['时间戳','SOH未标定时间', '累计充电电量','累计充电能量','累计放电电量','累计放电能量',
                                               '累计高温充电电量', '累计高温充电能量'])
            elif mode == 1:
                df_all = pd.DataFrame(columns=['时间戳','累计充电电量','累计充电能量','累计放电电量','累计放电能量', '累积里程'])
            for line in data:
                df_add = DBManager._convert_to_dataframe_accum(line, mode)
                df_all = df_all.append(df_add,ignore_index=True)
            if not df_all.empty:
                df_all.loc[:,'时间戳'] = df_all.loc[:,'时间戳'].apply(lambda x:time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(x)/1000)))
            return df_all
        elif type_name =='storage_accum':
            if mode == 0:
                df_all = pd.DataFrame(columns=['时间戳','累计用电kWh数1', '累计正向输入kWh1','累计反向输出kWh数1',
                                               '累计用电kWh数2', '累计正向输入kWh2','累计反向输出kWh数2'])
            elif mode == 1:
                df_all = pd.DataFrame()
            for line in data:
                df_add = DBManager._convert_to_dataframe_storage_accum(line, mode)
                df_all = df_all.append(df_add,ignore_index=True)
            if not df_all.empty:
                df_all.loc[:,'时间戳'] = df_all.loc[:,'时间戳'].apply(lambda x:time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(x)/1000)))
            return df_all
    def get_data(self, url='http://172.16.126.13/store/load?dataType={}&limit=0&sn={}', sn='', start_time='', end_time='', 
                 data_groups=['bms', 'gps']):
        '''
        获取指定 sn 和起止日期的bms和gps数据.
        添加了重试机制。

        --------------输入参数------------
        url：数据获取url， 可采用默认值
        sn: str, 电池sn号
        start_time: str, 开始时间
        end_time: str, 结束时间
        data_groups:  选择需要获取的数据组，可填入多个字符串（默认只获取bms和gps数据）
                    bms: bms数据
                    gps：gps数据
                    system：system数据
                    accum：accum数据
        

        --------------输出参数------------
        df_data: {'bms':dataframe, 'gps':dataframe, 'system':dataframe, ;accum':dataframe}
        '''
        
        if len(set(data_groups) - (set(data_groups) and set(['bms', 'gps', 'system', 'accum', 'storage_accum']))) > 0:
            raise Exception("data_groups 参数错误")
        
        
        # mode: 0：正常取数； 1：7255 取数
        if sn[0:2] == 'UD' or sn[0:2] == 'MG':
            mode = 1
        else:
            mode = 0
        bms_all_data = pd.DataFrame()
        gps_all_data = pd.DataFrame()
        system_all_data = pd.DataFrame()
        accum_all_data = pd.DataFrame()
        storage_accum_all_data = pd.DataFrame()
        maxnum = (datetime.datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S") - datetime.datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")).days +1
        print("### start to get data {} from {} to {}".format(sn, start_time, end_time))
        # 为避免chunkEncodingError错误，数据每天获取一次，然后将每天的数据合并，得到最终的数据
        for j in  range(int(maxnum)):
            timefrom = datetime.datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")+ datetime.timedelta(days=j)
            timeto = datetime.datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")+ datetime.timedelta(days=j+1)
            #滴滴的数据sub=0
            if timefrom.strftime('%Y-%m-%d %H:%M:%S') >= end_time:
                break
            elif timeto.strftime('%Y-%m-%d %H:%M:%S') > end_time:
                timeto = datetime.datetime.strptime(end_time, '%Y-%m-%d %H:%M:%S')
            #print('{}_{}_----getting data----'.format(sn, timefrom))
            bms_data = pd.DataFrame()
            gps_data = pd.DataFrame()
            system_data = pd.DataFrame()
            accum_data = pd.DataFrame()
            storage_accum_data = pd.DataFrame()
            while True:
                try:
                    print('\r' + "# get data from {} to {}.........".format(str(timefrom), str(timeto)), end=" ")
                    for data_group in data_groups:
                        if data_group == 'bms':
                            file_url = url.format(12, sn) + "&from="+timefrom.strftime('%Y-%m-%d %H:%M:%S')+"&to="+timeto.strftime('%Y-%m-%d %H:%M:%S')
                            bms_data = DBManager._get_data(file_url,'bms',mode)
                        if data_group == 'gps':
                            file_url = url.format(16, sn) + "&from="+timefrom.strftime('%Y-%m-%d %H:%M:%S')+"&to="+timeto.strftime('%Y-%m-%d %H:%M:%S')
                            gps_data = DBManager._get_data(file_url,'gps',mode)
                        if data_group == 'system':
                            file_url = url.format(13, sn) + "&from="+timefrom.strftime('%Y-%m-%d %H:%M:%S')+"&to="+timeto.strftime('%Y-%m-%d %H:%M:%S')
                            system_data = DBManager._get_data(file_url,'system',mode)
                        if data_group == 'accum':
                            file_url = url.format(23, sn) + "&from="+timefrom.strftime('%Y-%m-%d %H:%M:%S')+"&to="+timeto.strftime('%Y-%m-%d %H:%M:%S')
                            accum_data = DBManager._get_data(file_url,'accum',mode)
                        if data_group == 'storage_accum':
                            file_url = url.format(29, sn) + "&from="+timefrom.strftime('%Y-%m-%d %H:%M:%S')+"&to="+timeto.strftime('%Y-%m-%d %H:%M:%S')
                            storage_accum_data = DBManager._get_data(file_url,'storage_accum',mode)                        
                except Exception as e:
                    if 'Connection broken' in str(e):
                        continue
                    elif 'Number of manager items must equal union of block items' in str(e):
                        break
                    else:
                        raise Exception
                else:
                    bms_all_data = pd.concat([bms_all_data, bms_data], ignore_index=True)
                    gps_all_data = pd.concat([gps_all_data, gps_data], ignore_index=True)
                    system_all_data = pd.concat([system_all_data, system_data], ignore_index=True)
                    accum_all_data = pd.concat([accum_all_data, accum_data], ignore_index=True)
                    storage_accum_all_data = pd.concat([storage_accum_all_data, storage_accum_data], ignore_index=True)
                    break

        bms_all_data = bms_all_data.reset_index(drop=True)
        gps_all_data = gps_all_data.reset_index(drop=True)
        system_all_data = system_all_data.reset_index(drop=True)
        accum_all_data = accum_all_data.reset_index(drop=True)
        storage_accum_all_data = storage_accum_all_data.reset_index(drop=True)
        print('\nall data-getting done, bms_count is {}, gps_count is {}, system_count is {}, accum_count is {} ,storage_accum_count is {}\n'.format(
            str(len(bms_all_data)), str(len(gps_all_data)), str(len(system_all_data)), str(len(accum_all_data)), str(len(storage_accum_all_data))))
        return {'bms':bms_all_data, 'gps':gps_all_data, 'system':system_all_data, 'accum':accum_all_data,'storage_accum':storage_accum_all_data}