lambda_function.py 9.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193
  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. # control_value = 3(온도), 4(습도), 5(온도 설정값), 6(습도 설정값)
  52. target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  53. & ((mappingTable['control_value'] == 3) | (mappingTable['control_value'] == 4)
  54. | (mappingTable['control_value'] == 5) | (mappingTable['control_value'] == 6))]
  55. for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
  56. processed_data[list(set(target_row['raw_tag']))[c_idx]] \
  57. = [r_value if r_value > -15 and r_value <= 100 else np.nan \
  58. for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
  59. # control_value = 1(운전상태)
  60. target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  61. & (mappingTable['control_value'] == 1) ]
  62. for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
  63. processed_data[list(set(target_row['raw_tag']))[c_idx]] \
  64. = [r_value if r_value == 0 or r_value == 1 else np.nan \
  65. for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
  66. # control_value = 7(개도율), 8(개도율 설정값)
  67. target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  68. & ((mappingTable['control_value'] == 7) | (mappingTable['control_value'] == 8)) ]
  69. for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
  70. processed_data[list(set(target_row['raw_tag']))[c_idx]] \
  71. = [r_value if r_value >= 0 or r_value <= 100 else np.nan \
  72. for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
  73. # control_value = 30(운전모드)
  74. target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  75. & (mappingTable['control_value'] == 30) ]
  76. for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
  77. processed_data[list(set(target_row['raw_tag']))[c_idx]] \
  78. = [r_value if r_value >= 0 and r_value <= 10 else np.nan \
  79. for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
  80. print('sum of null point:', processed_data.isnull().sum().sum())
  81. # --------------- Fill the missing data and round off time index
  82. time_resolution = 5
  83. #data = raw_data
  84. round_m_comp = 0
  85. missing_idx = []
  86. missing_date = []
  87. print('len(processed_data))',len(processed_data))
  88. for time_idx in range(len(processed_data)):
  89. str_date = processed_data['time'][time_idx]
  90. date = str_date.split()[0]
  91. round_m = int(int(str_date.split()[1].split(':')[1])/time_resolution)*time_resolution
  92. h, m, s = [str_date.split()[1].split(':')[0],
  93. str(round_m) if round_m >= 10 else '0'+str(round_m),
  94. '00']
  95. round_m_comp += time_resolution # for compare time index
  96. # ----- find the missing row ----- #
  97. while round_m_comp <= int(h) * 60 + round_m:
  98. round_m_comp_tmp = round_m_comp - time_resolution
  99. round_h_tmp = int(round_m_comp_tmp/60)
  100. round_m_tmp = int(round_m_comp_tmp%60)
  101. h_tmp, m_tmp, s_tmp = [str(round_h_tmp) if round_h_tmp >= 10 else '0'+str(round_h_tmp),
  102. str(round_m_tmp) if round_m_tmp >= 10 else '0'+str(round_m_tmp),
  103. '00']
  104. missing_date.append(date + ' ' + h_tmp + ':' + m_tmp + ':' + s_tmp)
  105. missing_idx.append(time_idx + len(missing_idx)) # save missing index considering append index
  106. round_m_comp += time_resolution
  107. # ----- find the missing row ----- #
  108. processed_data.at[time_idx,'time'] = date + ' ' + h + ':' + m + ':' + s # round off time index for rows of normal data
  109. print('len', len(processed_data))
  110. print('sum of missing row data:', len(missing_idx))
  111. # ----- fill nan on the missing row ----- #
  112. idx = 0
  113. for miss_idx in missing_idx:
  114. tmp_data = np.zeros((1,len(raw_data.columns)))
  115. tmp_data[:] = np.nan
  116. reconstructed_data = pd.DataFrame(tmp_data, columns = raw_data.columns)
  117. idx_temp = miss_idx
  118. temp1 = processed_data[processed_data.index < idx_temp].copy()
  119. temp2 = processed_data[processed_data.index >= idx_temp].copy()
  120. temp1 = temp1.append(reconstructed_data, ignore_index=True)
  121. processed_data = temp1.append(temp2, ignore_index=True)
  122. processed_data.at[idx_temp, 'time'] = missing_date[idx]
  123. idx += 1
  124. # ----- fill nan on the missing row ----- #
  125. # ----- interpolation ----- #
  126. # ControlValue = 3(온도), 4(습도)
  127. TemHum_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  128. & ((mappingTable['control_value'] == 3) | (mappingTable['control_value'] == 4)) ]
  129. temp1 = processed_data[list(processed_data[list(set(TemHum_row['raw_tag']))])].fillna(method='backfill').copy()
  130. # ControlValue = 1(운전상태(On/off)), 7(개도율), 8(개도율 설정값), 19(차압), 30(운전모드)
  131. StatusOpenRatio_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
  132. & ((mappingTable['control_value'] == 1) | (mappingTable['control_value'] == 7)
  133. | (mappingTable['control_value'] == 8) | (mappingTable['control_value'] == 19)
  134. | (mappingTable['control_value'] == 30)) ]
  135. temp2 = processed_data[list(processed_data[list(set(StatusOpenRatio_row['raw_tag']))])].fillna(method='backfill').copy()
  136. others_tag = [x for x in list(set(raw_data.columns)) if x not in list(set(TemHum_row['raw_tag']))]
  137. others_tag = [x for x in others_tag if x not in list(set(StatusOpenRatio_row['raw_tag']))]
  138. #print('iik',list(set(raw_data.columns)).remove(list(set(TemHum_row['raw_tag'])) + list(set(StatusOpenRatio_row['raw_tag']))))
  139. others = processed_data[others_tag].fillna(method='backfill').copy()
  140. processed_data = pd.concat([others, temp1, temp2], axis=1)
  141. processed_data = processed_data.round(1)
  142. processed_data = processed_data.fillna(method='pad')
  143. # ----- interpolation ----- #
  144. print('processed', processed_data)
  145. # ----- load to S3 prep bucket ----- #
  146. curated_file_key = key.replace(key.split('/')[-1], '')
  147. target_path = 's3://hdci-ambt-anoicos-prep/{}'.format(curated_file_key)
  148. print('target_path', target_path)
  149. wr.s3.to_csv(
  150. df=processed_data,
  151. path=target_path,
  152. mode='overwrite',
  153. dataset=True
  154. )
  155. # ----- load to S3 prep bucket ----- #