DBManager.py 18 KB

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