123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193 |
- import json
- import urllib.parse
- import boto3
- from csv import reader
- import pandas as pd
- import pymysql
- import numpy as np
- import awswrangler as wr
- rds_host_write = "database-ambt.cluster-cvnwxgmngsms.ap-northeast-2.rds.amazonaws.com"
- rds_host_read = "database-ambt.cluster-ro-cvnwxgmngsms.ap-northeast-2.rds.amazonaws.com"
- name = "admin"
- password = "hdci12#$"
- db_name = "ambt"
- PORT=3306
- def lambda_handler(event, context):
-
- conn = pymysql.connect(host=rds_host_read, user=name, passwd=password, port=PORT, database=db_name, charset='utf8')
-
-
- sql_statement = "SELECT * FROM `ambt.icos`.facility_type;"
- facility_type_id = pd.read_sql(sql=sql_statement, con=conn)
-
- sql_statement = "SELECT * FROM `ambt.icos`.facility_code;"
- material_code = pd.read_sql(sql=sql_statement, con=conn)
-
- sql_statement = "SELECT * FROM `ambt.icos`.control_value;"
- control_value = pd.read_sql(sql=sql_statement, con=conn)
-
- sql_statement = "SELECT * FROM `ambt.icos`.value_type;"
- value_type = pd.read_sql(sql=sql_statement, con=conn)
-
- sql_statement = "SELECT * FROM `ambt.icos`.ambt_anoicos_code;"
- mappingTable = pd.read_sql(sql=sql_statement, con=conn)
- print(mappingTable.head())
-
- for record in event['Records']:
-
- bucket = record['s3']['bucket']['name']
- key = urllib.parse.unquote_plus(record['s3']['object']['key'])
-
- print('bucket : ', bucket)
- print('key : ', key)
-
- s3 = boto3.client('s3')
- obj = s3.get_object(Bucket=bucket, Key=key)
- raw_data = pd.read_csv(obj['Body'], sep=',')
- raw_data = raw_data.drop(['cnt'], axis=1)
-
- raw_data = raw_data.dropna(axis=0)
- raw_data = raw_data.reset_index(drop=True)
- print('raw', raw_data.columns)
- print (raw_data[130:140])
- print('length of raw data', len(raw_data))
-
-
-
-
-
-
-
-
- processed_data = raw_data.copy()
-
- target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
- & ((mappingTable['control_value'] == 3) | (mappingTable['control_value'] == 4)
- | (mappingTable['control_value'] == 5) | (mappingTable['control_value'] == 6))]
-
- for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
- processed_data[list(set(target_row['raw_tag']))[c_idx]] \
- = [r_value if r_value > -15 and r_value <= 100 else np.nan \
- for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
-
-
- target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
- & (mappingTable['control_value'] == 1) ]
- for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
- processed_data[list(set(target_row['raw_tag']))[c_idx]] \
- = [r_value if r_value == 0 or r_value == 1 else np.nan \
- for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
-
-
- target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
- & ((mappingTable['control_value'] == 7) | (mappingTable['control_value'] == 8)) ]
- for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
- processed_data[list(set(target_row['raw_tag']))[c_idx]] \
- = [r_value if r_value >= 0 or r_value <= 100 else np.nan \
- for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
-
-
- target_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
- & (mappingTable['control_value'] == 30) ]
- for c_idx in range(raw_data[list(set(target_row['raw_tag']))].shape[1]):
- processed_data[list(set(target_row['raw_tag']))[c_idx]] \
- = [r_value if r_value >= 0 and r_value <= 10 else np.nan \
- for r_value in raw_data[list(set(target_row['raw_tag']))[c_idx]]]
- print('sum of null point:', processed_data.isnull().sum().sum())
-
- time_resolution = 5
-
- round_m_comp = 0
- missing_idx = []
- missing_date = []
- print('len(processed_data))',len(processed_data))
- for time_idx in range(len(processed_data)):
- str_date = processed_data['time'][time_idx]
- date = str_date.split()[0]
- round_m = int(int(str_date.split()[1].split(':')[1])/time_resolution)*time_resolution
- h, m, s = [str_date.split()[1].split(':')[0],
- str(round_m) if round_m >= 10 else '0'+str(round_m),
- '00']
-
- round_m_comp += time_resolution
-
- while round_m_comp <= int(h) * 60 + round_m:
- round_m_comp_tmp = round_m_comp - time_resolution
- round_h_tmp = int(round_m_comp_tmp/60)
- round_m_tmp = int(round_m_comp_tmp%60)
- h_tmp, m_tmp, s_tmp = [str(round_h_tmp) if round_h_tmp >= 10 else '0'+str(round_h_tmp),
- str(round_m_tmp) if round_m_tmp >= 10 else '0'+str(round_m_tmp),
- '00']
- missing_date.append(date + ' ' + h_tmp + ':' + m_tmp + ':' + s_tmp)
- missing_idx.append(time_idx + len(missing_idx))
- round_m_comp += time_resolution
-
- processed_data.at[time_idx,'time'] = date + ' ' + h + ':' + m + ':' + s
- print('len', len(processed_data))
- print('sum of missing row data:', len(missing_idx))
-
-
- idx = 0
- for miss_idx in missing_idx:
- tmp_data = np.zeros((1,len(raw_data.columns)))
- tmp_data[:] = np.nan
- reconstructed_data = pd.DataFrame(tmp_data, columns = raw_data.columns)
-
- idx_temp = miss_idx
- temp1 = processed_data[processed_data.index < idx_temp].copy()
- temp2 = processed_data[processed_data.index >= idx_temp].copy()
-
- temp1 = temp1.append(reconstructed_data, ignore_index=True)
- processed_data = temp1.append(temp2, ignore_index=True)
-
- processed_data.at[idx_temp, 'time'] = missing_date[idx]
- idx += 1
-
-
-
-
- TemHum_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
- & ((mappingTable['control_value'] == 3) | (mappingTable['control_value'] == 4)) ]
- temp1 = processed_data[list(processed_data[list(set(TemHum_row['raw_tag']))])].fillna(method='backfill').copy()
-
-
- StatusOpenRatio_row = mappingTable.loc[(mappingTable['site_id']==1) & (mappingTable['collect_status'] == 1)
- & ((mappingTable['control_value'] == 1) | (mappingTable['control_value'] == 7)
- | (mappingTable['control_value'] == 8) | (mappingTable['control_value'] == 19)
- | (mappingTable['control_value'] == 30)) ]
- temp2 = processed_data[list(processed_data[list(set(StatusOpenRatio_row['raw_tag']))])].fillna(method='backfill').copy()
- others_tag = [x for x in list(set(raw_data.columns)) if x not in list(set(TemHum_row['raw_tag']))]
- others_tag = [x for x in others_tag if x not in list(set(StatusOpenRatio_row['raw_tag']))]
-
- others = processed_data[others_tag].fillna(method='backfill').copy()
- processed_data = pd.concat([others, temp1, temp2], axis=1)
- processed_data = processed_data.round(1)
- processed_data = processed_data.fillna(method='pad')
-
- print('processed', processed_data)
-
-
-
- curated_file_key = key.replace(key.split('/')[-1], '')
- target_path = 's3://hdci-ambt-anoicos-prep/{}'.format(curated_file_key)
- print('target_path', target_path)
- wr.s3.to_csv(
- df=processed_data,
- path=target_path,
- mode='overwrite',
- dataset=True
- )
-
-
|