lambda_function.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201
  1. import json
  2. import urllib.parse
  3. import boto3
  4. from csv import reader
  5. import pandas as pd
  6. import pymysql
  7. import numpy as np
  8. import awswrangler as wr
  9. rds_host_write = "database-ambt.cluster-cvnwxgmngsms.ap-northeast-2.rds.amazonaws.com"
  10. rds_host_read = "database-ambt.cluster-ro-cvnwxgmngsms.ap-northeast-2.rds.amazonaws.com"
  11. name = "admin"
  12. password = "hdci12#$"
  13. db_name = "ambt"
  14. PORT=3306
  15. def lambda_handler(event, context):
  16. conn = pymysql.connect(host=rds_host_read, user=name, passwd=password, port=PORT, database=db_name, charset='utf8')
  17. #conn = pymysql.connect(host=rds_host, user=name, passwd=password, db=db_name, use_unicode=True, charset='utf8', connect_timeout=5)
  18. sql_statement = "SELECT * FROM `ambt.icos`.facility_type;"
  19. facility_type_id = pd.read_sql(sql=sql_statement, con=conn)
  20. sql_statement = "SELECT * FROM `ambt.icos`.facility_code;"
  21. material_code = pd.read_sql(sql=sql_statement, con=conn)
  22. sql_statement = "SELECT * FROM `ambt.icos`.control_value;"
  23. control_value = pd.read_sql(sql=sql_statement, con=conn)
  24. sql_statement = "SELECT * FROM `ambt.icos`.value_type;"
  25. value_type = pd.read_sql(sql=sql_statement, con=conn)
  26. sql_statement = "SELECT * FROM `ambt.icos`.ambt_anoicos_code;"
  27. mappingTable = pd.read_sql(sql=sql_statement, con=conn)
  28. print(mappingTable.head())
  29. for record in event['Records']:
  30. #print("Received event: " + json.dumps(event, indent=2))
  31. bucket = record['s3']['bucket']['name']
  32. key = urllib.parse.unquote_plus(record['s3']['object']['key'])
  33. print('bucket : ', bucket)
  34. print('key : ', key)
  35. s3 = boto3.client('s3')
  36. obj = s3.get_object(Bucket=bucket, Key=key)
  37. raw_data = pd.read_csv(obj['Body'], sep=',')
  38. raw_data = raw_data.drop(['cnt'], axis=1)
  39. # raw_data = raw_data.drop(['Unnamed: 0'], axis=1)
  40. raw_data = raw_data.dropna(axis=0) # drop the row including null
  41. raw_data = raw_data.reset_index(drop=True)
  42. print('raw', raw_data.columns)
  43. print (raw_data[130:140])
  44. print('length of raw data', len(raw_data))
  45. # --------------- Data pre-processing (considering data minimum properties)
  46. # temperature, humidity (set point, ) - -15 ~ 100
  47. # operating status - 0/1
  48. # open ratio (SP) - 0 ~ 100
  49. # operating mode - 0 ~ 10
  50. processed_data = raw_data.copy()
  51. null_sum = pd.DataFrame(np.zeros((1,len(raw_data.columns))), columns = raw_data.columns)
  52. # control_value = 3(온도), 4(습도), 5(온도 설정값), 6(습도 설정값)
  53. target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  54. & ((mappingTable['control_value'] == 3) | (mappingTable['control_value'] == 4)
  55. | (mappingTable['control_value'] == 5) | (mappingTable['control_value'] == 6))]
  56. for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
  57. processed_data[list(set(target_row['raw_tag']))[c_idx]] \
  58. = [r_value if r_value > -15 and r_value <= 100 else np.nan \
  59. for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
  60. null_sum[list(set(target_row['raw_tag']))[c_idx]][0] \
  61. = processed_data[list(set(target_row['raw_tag']))[c_idx]].isnull().sum()
  62. # control_value = 1(운전상태)
  63. target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  64. & (mappingTable['control_value'] == 1) ]
  65. for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
  66. processed_data[list(set(target_row['raw_tag']))[c_idx]] \
  67. = [r_value if r_value == 0 or r_value == 1 else np.nan \
  68. for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
  69. null_sum[list(set(target_row['raw_tag']))[c_idx]][0] \
  70. = processed_data[list(set(target_row['raw_tag']))[c_idx]].isnull().sum()
  71. # control_value = 7(개도율), 8(개도율 설정값)
  72. target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  73. & ((mappingTable['control_value'] == 7) | (mappingTable['control_value'] == 8)) ]
  74. for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
  75. processed_data[list(set(target_row['raw_tag']))[c_idx]] \
  76. = [r_value if r_value >= 0 or r_value <= 100 else np.nan \
  77. for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
  78. null_sum[list(set(target_row['raw_tag']))[c_idx]][0] \
  79. = processed_data[list(set(target_row['raw_tag']))[c_idx]].isnull().sum()
  80. # control_value = 30(운전모드)
  81. target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  82. & (mappingTable['control_value'] == 30) ]
  83. for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
  84. processed_data[list(set(target_row['raw_tag']))[c_idx]] \
  85. = [r_value if r_value >= 0 and r_value <= 10 else np.nan \
  86. for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
  87. null_sum[list(set(target_row['raw_tag']))[c_idx]][0] \
  88. = processed_data[list(set(target_row['raw_tag']))[c_idx]].isnull().sum()
  89. print('sum of null point:', null_sum.sum())
  90. # --------------- Fill the missing data and round off time index
  91. time_resolution = 5
  92. #data = raw_data
  93. round_m_comp = 0
  94. missing_idx = []
  95. missing_date = []
  96. print('len(processed_data))',len(processed_data))
  97. for time_idx in range(len(processed_data)):
  98. str_date = processed_data['time'][time_idx]
  99. date = str_date.split()[0]
  100. round_m = int(int(str_date.split()[1].split(':')[1])/time_resolution)*time_resolution
  101. h, m, s = [str_date.split()[1].split(':')[0],
  102. str(round_m) if round_m >= 10 else '0'+str(round_m),
  103. '00']
  104. round_m_comp += time_resolution # for compare time index
  105. # ----- find the missing row ----- #
  106. while round_m_comp <= int(h) * 60 + round_m:
  107. round_m_comp_tmp = round_m_comp - time_resolution
  108. round_h_tmp = int(round_m_comp_tmp/60)
  109. round_m_tmp = int(round_m_comp_tmp%60)
  110. h_tmp, m_tmp, s_tmp = [str(round_h_tmp) if round_h_tmp >= 10 else '0'+str(round_h_tmp),
  111. str(round_m_tmp) if round_m_tmp >= 10 else '0'+str(round_m_tmp),
  112. '00']
  113. missing_date.append(date + ' ' + h_tmp + ':' + m_tmp + ':' + s_tmp)
  114. missing_idx.append(time_idx + len(missing_idx)) # save missing index considering append index
  115. round_m_comp += time_resolution
  116. # ----- find the missing row ----- #
  117. processed_data.at[time_idx,'time'] = date + ' ' + h + ':' + m + ':' + s # round off time index for rows of normal data
  118. print('len', len(processed_data))
  119. print('sum of missing row data:', len(missing_idx))
  120. # ----- fill nan on the missing row ----- #
  121. idx = 0
  122. for miss_idx in missing_idx:
  123. tmp_data = np.zeros((1,len(raw_data.columns)))
  124. tmp_data[:] = np.nan
  125. reconstructed_data = pd.DataFrame(tmp_data, columns = raw_data.columns)
  126. idx_temp = miss_idx
  127. temp1 = processed_data[processed_data.index < idx_temp].copy()
  128. temp2 = processed_data[processed_data.index >= idx_temp].copy()
  129. temp1 = temp1.append(reconstructed_data, ignore_index=True)
  130. processed_data = temp1.append(temp2, ignore_index=True)
  131. processed_data.at[idx_temp, 'time'] = missing_date[idx]
  132. idx += 1
  133. # ----- fill nan on the missing row ----- #
  134. # ----- interpolation ----- #
  135. # ControlValue = 3(온도), 4(습도)
  136. TemHum_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  137. & ((mappingTable['control_value'] == 3) | (mappingTable['control_value'] == 4)) ]
  138. temp1 = processed_data[list(processed_data[list(set(TemHum_row['raw_tag']))])].fillna(method='pad').copy()
  139. # ControlValue = 1(운전상태(On/off)), 7(개도율), 8(개도율 설정값), 19(차압), 30(운전모드)
  140. StatusOpenRatio_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  141. & ((mappingTable['control_value'] == 1) | (mappingTable['control_value'] == 7)
  142. | (mappingTable['control_value'] == 8) | (mappingTable['control_value'] == 19)
  143. | (mappingTable['control_value'] == 30)) ]
  144. temp2 = processed_data[list(processed_data[list(set(StatusOpenRatio_row['raw_tag']))])].fillna(method='pad').copy()
  145. others_tag = [x for x in list(set(raw_data.columns)) if x not in list(set(TemHum_row['raw_tag']))]
  146. others_tag = [x for x in others_tag if x not in list(set(StatusOpenRatio_row['raw_tag']))]
  147. #print('iik',list(set(raw_data.columns)).remove(list(set(TemHum_row['raw_tag'])) + list(set(StatusOpenRatio_row['raw_tag']))))
  148. others = processed_data[others_tag].fillna(method='pad').copy()
  149. processed_data = pd.concat([others, temp1, temp2], axis=1)
  150. processed_data = processed_data.round(1)
  151. # ----- interpolation ----- #
  152. print('processed', processed_data)
  153. # ----- load to S3 prep bucket ----- #
  154. curated_file_key = key.replace(key.split('/')[-1], '')
  155. target_path = 's3://hdci-ambt-anoicos-prep/{}'.format(curated_file_key)
  156. print('target_path', target_path)
  157. wr.s3.to_csv(
  158. df=processed_data,
  159. path=target_path,
  160. mode='overwrite',
  161. dataset=True
  162. )
  163. # ----- load to S3 prep bucket ----- #