DBManager.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383
  1. '''
  2. 暂时采用http方式获取历史数据。
  3. 预留:后期若改用通过访问数据库的形式进行数据的获取,则本文件负责数据库的连接,sql指令的执行,数据获取等功能。
  4. '''
  5. __author__ = 'lmstack'
  6. import time
  7. import datetime
  8. import time
  9. import pandas as pd
  10. import numpy as np
  11. import json
  12. import requests
  13. import pymysql
  14. import pdb
  15. class DBManager():
  16. def __init__(self, host='', port='', auth='', db='', username='', password=''):
  17. self.host = host
  18. self.port = port
  19. self.auth = auth
  20. self.db = db
  21. self.username = username
  22. self.password = password
  23. pass
  24. def __enter__(self):
  25. self.connect()
  26. return self
  27. def __exit__(self, exc_type, exc_val, exc_tb):
  28. self.close()
  29. def connect(self):
  30. conn_success_flag = 0
  31. while not conn_success_flag:
  32. try:
  33. self.conn = pymysql.connect(host=self.host, port=self.port, user=self.user, password=self.password, database=self.db)
  34. except Exception as e:
  35. conn_success_flag = 0
  36. print("数据库连接失败 :{}".format(e))
  37. time.sleep(5)
  38. else:
  39. conn_success_flag = 1
  40. self.cursor = self.conn.cursor()
  41. def close(self):
  42. try:
  43. self.conn.close()
  44. except Exception as e:
  45. print(e)
  46. else:
  47. print('数据库已断开连接')
  48. def add(table, keyvalue):
  49. fields_str = ''
  50. values_str = ''
  51. for k,v in keyvalue.items():
  52. fields_str += k+' '
  53. sql = 'insert into table {} ({}) values ({})'.format(table, fields_str, values_str)
  54. # 以下各个函数实现 通过http方式获取数据
  55. @staticmethod
  56. def _get_var_name(cellnum,Tempnum,Othernum):
  57. temp = []
  58. for i in range(cellnum):
  59. temp.append('单体电压'+str(i+1))
  60. for i in range(Tempnum):
  61. temp.append('单体温度'+str(i+1))
  62. for i in range(Othernum):
  63. temp.append('其他温度'+str(i+1))
  64. return temp
  65. @staticmethod
  66. def _download_json_data(url):
  67. '''
  68. 返回json数据的生成器,一次一行
  69. '''
  70. i = 0
  71. while 1:
  72. try:
  73. r = requests.get(url, stream=True, timeout=(0.1, 1000), headers={'Connection':'keep-alive', 'Accept':'*/*', 'Accept-Encoding':'gzip,deflate,br'})
  74. break
  75. except requests.exceptions.RequestException as e:
  76. if (i == 0):
  77. print()
  78. print('\r' + 'Server Error, retry {}......'.format(str(i)), end=" ")
  79. time.sleep(0.1)
  80. i+=1
  81. # print(r.content)
  82. # pdb.set_trace()
  83. data = []
  84. for line in r.iter_lines():
  85. if line:
  86. data.append(json.loads(line))
  87. # yield json.loads(line)
  88. return data
  89. @staticmethod
  90. def _convert_to_dataframe_bms(data, mode=0):
  91. CellU = []
  92. CellT = []
  93. OtherT = []
  94. CellU_Num = 0
  95. CellT_Num = 0
  96. OtherT_Num = 0
  97. CellU_Num = len(data['ffBatteryStatus']['cellVoltageList'])
  98. CellT_Num = len(data['ffBatteryStatus']['cellTempList'])
  99. try:
  100. OtherT_Num = len(data['ffBatteryStatus']['otherTempList'])
  101. except:
  102. OtherT_Num = 0
  103. for i in range(CellU_Num):
  104. CellU.append(data['ffBatteryStatus']['cellVoltageList'][i]*1000)
  105. for i in range(CellT_Num):
  106. CellU.append(data['ffBatteryStatus']['cellTempList'][i])
  107. for i in range(OtherT_Num):
  108. CellU.append(data['ffBatteryStatus']['otherTempList'][i])
  109. if mode == 0:
  110. data_len = 16
  111. # data_block = np.array([data['info']['obdTime'],data['ffBatteryStatus']['rssi'],data['ffBatteryStatus']['errorLevel'],data['ffBatteryStatus']['errorCode']
  112. # ,data['ffBatteryStatus']['current'],data['ffBatteryStatus']['voltageInner'],data['ffBatteryStatus']['voltageOutter'],
  113. # data['ffBatteryStatus']['totalOutputState'],data['ffBatteryStatus']['lockedState'],
  114. # data['ffBatteryStatus']['chargeState'],data['ffBatteryStatus']['heatState'],data['ffBatteryStatus']['cellVoltageDiff']
  115. # ,data['ffBatteryStatus']['soc'],data['ffBatteryStatus']['soh'],data['ffBatteryStatus']['cellVolBalance']]).reshape(1,data_len)
  116. data_block = np.array([data['info']['obdTime'],data['ffBatteryStatus'].get('rssi',None),data['ffBatteryStatus'].get('errorLevel', None),data['ffBatteryStatus'].get('errorCode', None),
  117. data['ffBatteryStatus'].get('current',None),data['ffBatteryStatus'].get('voltageInner', None),data['ffBatteryStatus'].get('voltageOutter', None),
  118. data['ffBatteryStatus'].get('totalOutputState', None),data['ffBatteryStatus'].get('lockedState', None),
  119. data['ffBatteryStatus'].get('chargeState', None),data['ffBatteryStatus'].get('heatState', None),data['ffBatteryStatus'].get('cellVoltageDiff', None)
  120. ,data['ffBatteryStatus'].get('soc', None),data['ffBatteryStatus'].get('soh', None),data['ffBatteryStatus'].get('cellVolBalance', None),data['ffBatteryStatus'].get('insResis', None)]).reshape(1,data_len)
  121. elif mode == 1:
  122. data_len = 12
  123. data_block = np.array([data['info']['obdTime'],data['ffBatteryStatus'].get('rssi',None)
  124. ,data['ffBatteryStatus'].get('errorLevel', None),data['ffBatteryStatus'].get('errorCode', None),data['ffBatteryStatus'].get('switchState', None)
  125. ,data['ffBatteryStatus'].get('current',None),data['ffBatteryStatus'].get('voltageInner', None),data['ffBatteryStatus'].get('chargeState', None),
  126. data['ffBatteryStatus'].get('cellVoltageDiff', None),data['ffBatteryStatus'].get('soc', None),data['ffBatteryStatus'].get('soh', None),data['ffBatteryStatus'].get('insResis', None)]).reshape(1,data_len)
  127. data_block = np.append(data_block,CellU)
  128. data_block = np.append(data_block,CellT)
  129. data_block = np.append(data_block,OtherT)
  130. data_block = data_block.reshape(1,len(data_block))
  131. return data_block,CellU_Num,CellT_Num,OtherT_Num
  132. @staticmethod
  133. def _convert_to_dataframe_gps(data, mode=0):
  134. if mode == 0:
  135. if data['info']['subType'] == 1:
  136. data_block = np.array([data['info']['obdTime'],data['ffGps']['locationType'], data['ffGps']['satellites'],
  137. data['ffGps']['latitude'],data['ffGps']['longitude'],data['ffGps']['speed'],
  138. data['ffGps']['altitude'], data['ffGps']['direction']]).reshape(1,8)
  139. df = pd.DataFrame(
  140. columns=['时间戳','定位类型', '卫星数','纬度','经度','速度[km/h]','海拔','航向'],data=data_block)
  141. elif data['info']['subType'] == 2:
  142. df = pd.DataFrame(
  143. columns=['时间戳','定位类型', '卫星数','纬度','经度','速度[km/h]','海拔','航向'])
  144. if mode == 1:
  145. data_block = np.array([data['info']['obdTime'],data['ffGps']['locationType'],data['ffGps']['latitude'],data['ffGps']['longitude']
  146. ,data['ffGps']['speed'], data['ffGps']['isValid']]).reshape(1,6)
  147. df = pd.DataFrame(
  148. columns=['时间戳','定位类型', '纬度','经度','速度[km/h]','有效位'],data=data_block)
  149. return df
  150. @staticmethod
  151. def _convert_to_dataframe_system(data, mode=0):
  152. if mode == 0:
  153. data_block = np.array([data['info'].get('obdTime', None),data['ffSystemInfo'].get('heatTargetTemp', None), data['ffSystemInfo'].get('heatTimeout',None),
  154. time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(data['ffSystemInfo'].get('rentalStartTime'))/1000)),
  155. data['ffSystemInfo'].get('rentalPeriodDays',None),data['ffSystemInfo'].get('bmsInterval',None),
  156. data['ffSystemInfo'].get('gpsInterval', None)]).reshape(1,7)
  157. df = pd.DataFrame(
  158. columns=['时间戳','加热目标温度', '加热超时','租赁开始时间','租赁天数','bms上传周期','gps上传周期'],data=data_block)
  159. if mode == 1:
  160. df = pd.DataFrame()
  161. return df
  162. @staticmethod
  163. def _convert_to_dataframe_accum(data, mode=0):
  164. if mode == 0:
  165. data_block = np.array([data['info'].get('obdTime',None),data['ffBatteryAccum'].get('SOH_AlgUnexTime',None), data['ffBatteryAccum'].get('CHG_AHaccum',None),
  166. data['ffBatteryAccum'].get('CHG_PHaccum',None), data['ffBatteryAccum'].get('DSG_AHaccum',None),
  167. data['ffBatteryAccum'].get('DSG_PHaccum',None),data['ffBatteryAccum'].get('OverTemp_CHG_AHaccum',None),
  168. data['ffBatteryAccum'].get('OverTemp_CHG_PHaccum',None)]).reshape(1,8)
  169. df = pd.DataFrame(
  170. columns=['时间戳','SOH未标定时间', '累计充电电量','累计充电能量','累计放电电量','累计放电能量',
  171. '累计高温充电电量', '累计高温充电能量'],data=data_block)
  172. if mode == 1:
  173. data_block = np.array([data['info'].get('obdTime',None), data['ffBatteryAccum'].get('CHG_AHaccum',None),
  174. data['ffBatteryAccum'].get('CHG_PHaccum',None), data['ffBatteryAccum'].get('DSG_AHaccum',None),
  175. data['ffBatteryAccum'].get('DSG_PHaccum',None),data['ffBatteryAccum'].get('totalMileage',None)]).reshape(1,6)
  176. df = pd.DataFrame(
  177. columns=['时间戳','累计充电电量','累计充电能量','累计放电电量','累计放电能量', '累积里程'],data=data_block)
  178. return df
  179. @staticmethod
  180. def _get_data(urls,type_name,mode=0):
  181. data = DBManager._download_json_data(urls)
  182. if type_name == 'bms':
  183. if mode == 0:
  184. name_const = ['时间戳','GSM信号','故障等级','故障代码','总电流[A]','总电压[V]', '外电压', '总输出状态', '上锁状态', '充电状态','加热状态',
  185. '单体压差', 'SOC[%]','SOH[%]','单体均衡状态', '绝缘电阻']
  186. elif mode == 1:
  187. name_const = ['时间戳','GSM信号','故障等级', '故障代码','开关状态', '总电流[A]','总电压[V]','充电状态', '单体压差', 'SOC[%]','SOH[%]', '绝缘电阻']
  188. i=0
  189. CellUNum = 0
  190. CellTNum = 0
  191. OtherTNumm = 0
  192. st = time.time()
  193. for line in data:
  194. et = time.time()
  195. try:
  196. if i==0:
  197. data_blocks,CellUNum,CellTNum,OtherTNumm = DBManager._convert_to_dataframe_bms(line, mode)
  198. i+=1
  199. continue
  200. except Exception as e:
  201. print(str(e))
  202. i = 0
  203. try:
  204. data_block,CellUNum,CellTNum,OtherTNumm = DBManager._convert_to_dataframe_bms(line, mode)
  205. except Exception as e:
  206. print(str(e))
  207. continue
  208. try:
  209. data_blocks = np.concatenate((data_blocks,data_block),axis=0)
  210. except Exception as e:
  211. if 'all the input array dimensions for the concatenation axis must match exactly' in str(e) or \
  212. 'all the input array dimensions except for the concatenation axis must match exactly' in str(e):
  213. pass
  214. else:
  215. raise e
  216. # print('\r'+str(i),end=" ")
  217. # print(data_block)
  218. # print(urls)
  219. # print(time.time()-et)
  220. i+=1
  221. name_var = DBManager._get_var_name(CellUNum,CellTNum,OtherTNumm)
  222. name_const.extend(name_var)
  223. columns_name = name_const
  224. if i==0:
  225. data_blocks = []
  226. df_all = pd.DataFrame(columns=columns_name,data=data_blocks)
  227. if not df_all.empty:
  228. df_all.loc[:,'时间戳'] = df_all.loc[:,'时间戳'].apply(lambda x:time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(x)/1000)))
  229. return df_all
  230. elif type_name =='gps':
  231. if mode == 0:
  232. df_all = pd.DataFrame(columns=['时间戳','定位类型', '卫星数','纬度','经度','速度[km/h]','海拔','航向'])
  233. elif mode == 1:
  234. df_all = pd.DataFrame(columns=['时间戳','定位类型', '纬度','经度','速度[km/h]','有效位'])
  235. for line in data:
  236. df_add = DBManager._convert_to_dataframe_gps(line, mode)
  237. df_all = df_all.append(df_add,ignore_index=True)
  238. if not df_all.empty:
  239. df_all.loc[:,'时间戳'] = df_all.loc[:,'时间戳'].apply(lambda x:time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(x)/1000)))
  240. return df_all
  241. elif type_name =='system':
  242. if mode == 0:
  243. df_all = pd.DataFrame(columns=['时间戳','加热目标温度', '加热超时','租赁开始时间','租赁天数','bms上传周期','gps上传周期'])
  244. elif mode == 1:
  245. df_all = pd.DataFrame()
  246. for line in data:
  247. df_add = DBManager._convert_to_dataframe_system(line, mode)
  248. df_all = df_all.append(df_add,ignore_index=True)
  249. if not df_all.empty:
  250. df_all.loc[:,'时间戳'] = df_all.loc[:,'时间戳'].apply(lambda x:time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(x)/1000)))
  251. return df_all
  252. elif type_name =='accum':
  253. if mode == 0:
  254. df_all = pd.DataFrame(columns=['时间戳','SOH未标定时间', '累计充电电量','累计充电能量','累计放电电量','累计放电能量',
  255. '累计高温充电电量', '累计高温充电能量'])
  256. elif mode == 1:
  257. df_all = pd.DataFrame(columns=['时间戳','累计充电电量','累计充电能量','累计放电电量','累计放电能量', '累积里程'])
  258. for line in data:
  259. df_add = DBManager._convert_to_dataframe_accum(line, mode)
  260. df_all = df_all.append(df_add,ignore_index=True)
  261. if not df_all.empty:
  262. df_all.loc[:,'时间戳'] = df_all.loc[:,'时间戳'].apply(lambda x:time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(int(x)/1000)))
  263. return df_all
  264. def get_data(self, url='http://172.16.121.236/store/load?dataType={}&limit=0&sn={}', sn='', start_time='', end_time='',
  265. data_groups=['bms', 'gps']):
  266. '''
  267. 获取指定 sn 和起止日期的bms和gps数据.
  268. 添加了重试机制。
  269. --------------输入参数------------
  270. url:数据获取url, 可采用默认值
  271. sn: str, 电池sn号
  272. start_time: str, 开始时间
  273. end_time: str, 结束时间
  274. data_groups: 选择需要获取的数据组,可填入多个字符串(默认只获取bms和gps数据)
  275. bms: bms数据
  276. gps:gps数据
  277. system:system数据
  278. accum:accum数据
  279. --------------输出参数------------
  280. df_data: {'bms':dataframe, 'gps':dataframe, 'system':dataframe, ;accum':dataframe}
  281. '''
  282. if len(set(data_groups) - (set(data_groups) and set(['bms', 'gps', 'system', 'accum']))) > 0:
  283. raise Exception("data_groups 参数错误")
  284. # mode: 0:正常取数; 1:7255 取数
  285. if sn[0:2] == 'UD' or sn[0:2] == 'MG':
  286. mode = 1
  287. else:
  288. mode = 0
  289. bms_all_data = pd.DataFrame()
  290. gps_all_data = pd.DataFrame()
  291. system_all_data = pd.DataFrame()
  292. accum_all_data = pd.DataFrame()
  293. 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
  294. print("### start to get data {} from {} to {}".format(sn, start_time, end_time))
  295. # 为避免chunkEncodingError错误,数据每天获取一次,然后将每天的数据合并,得到最终的数据
  296. for j in range(int(maxnum)):
  297. timefrom = datetime.datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")+ datetime.timedelta(days=j)
  298. timeto = datetime.datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")+ datetime.timedelta(days=j+1)
  299. #滴滴的数据sub=0
  300. if timefrom.strftime('%Y-%m-%d %H:%M:%S') >= end_time:
  301. break
  302. elif timeto.strftime('%Y-%m-%d %H:%M:%S') > end_time:
  303. timeto = datetime.datetime.strptime(end_time, '%Y-%m-%d %H:%M:%S')
  304. #print('{}_{}_----getting data----'.format(sn, timefrom))
  305. bms_data = pd.DataFrame()
  306. gps_data = pd.DataFrame()
  307. system_data = pd.DataFrame()
  308. accum_data = pd.DataFrame()
  309. while True:
  310. try:
  311. print('\r' + "# get data from {} to {}.........".format(str(timefrom), str(timeto)), end=" ")
  312. for data_group in data_groups:
  313. if data_group == 'bms':
  314. file_url = url.format(12, sn) + "&from="+timefrom.strftime('%Y-%m-%d %H:%M:%S')+"&to="+timeto.strftime('%Y-%m-%d %H:%M:%S')
  315. bms_data = DBManager._get_data(file_url,'bms',mode)
  316. if data_group == 'gps':
  317. file_url = url.format(16, sn) + "&from="+timefrom.strftime('%Y-%m-%d %H:%M:%S')+"&to="+timeto.strftime('%Y-%m-%d %H:%M:%S')
  318. gps_data = DBManager._get_data(file_url,'gps',mode)
  319. if data_group == 'system':
  320. file_url = url.format(13, sn) + "&from="+timefrom.strftime('%Y-%m-%d %H:%M:%S')+"&to="+timeto.strftime('%Y-%m-%d %H:%M:%S')
  321. system_data = DBManager._get_data(file_url,'system',mode)
  322. if data_group == 'accum':
  323. file_url = url.format(23, sn) + "&from="+timefrom.strftime('%Y-%m-%d %H:%M:%S')+"&to="+timeto.strftime('%Y-%m-%d %H:%M:%S')
  324. accum_data = DBManager._get_data(file_url,'accum',mode)
  325. except Exception as e:
  326. if 'Connection broken' in str(e):
  327. continue
  328. else:
  329. raise Exception
  330. else:
  331. bms_all_data = pd.concat([bms_all_data, bms_data], ignore_index=True)
  332. gps_all_data = pd.concat([gps_all_data, gps_data], ignore_index=True)
  333. system_all_data = pd.concat([system_all_data, system_data], ignore_index=True)
  334. accum_all_data = pd.concat([accum_all_data, accum_data], ignore_index=True)
  335. break
  336. bms_all_data = bms_all_data.reset_index(drop=True)
  337. gps_all_data = gps_all_data.reset_index(drop=True)
  338. system_all_data = system_all_data.reset_index(drop=True)
  339. accum_all_data = accum_all_data.reset_index(drop=True)
  340. print('\nall data-getting done, bms_count is {}, gps_count is {}, system_count is {}, accum_count is {} \n'.format(
  341. str(len(bms_all_data)), str(len(gps_all_data)), str(len(system_all_data)), str(len(accum_all_data))))
  342. return {'bms':bms_all_data, 'gps':gps_all_data, 'system':system_all_data, 'accum':accum_all_data}