import pandas as pd import datetime, time import pytz from savedata import saveData, readData from Arg import Arg from sqlalchemy import create_engine import pytz from data_cleaning import cleaning, rm_duplicated # def readData(name): # """ # 读取数据 # :param name: 名字 # :return: # """ # path = dataloc + r"/" + name # return pd.read_csv(path) # # # def saveData(name, data): # """ # 存放数据 # :param name: 名字 # :param data: 数据 # :return: # """ # path = dataloc + r"/" + name # os.makedirs(os.path.dirname(path), exist_ok=True) # data.to_csv(path, index=False) def timestamp_to_datetime(ts): local_timezone = pytz.timezone('Asia/Shanghai') if type(ts) is not int: raise ValueError("timestamp-时间格式必须是整型") if len(str(ts)) == 13: dt = datetime.datetime.fromtimestamp(ts/1000, tz=pytz.utc).astimezone(local_timezone) return dt elif len(str(ts)) == 10: dt = datetime.datetime.fromtimestamp(ts, tz=pytz.utc).astimezone(local_timezone) return dt else: raise ValueError("timestamp-时间格式错误") def timestr_to_timestamp(time_str): """ 将时间戳或时间字符串转换为datetime.datetime类型 :param time_data: int or str :return:datetime.datetime """ if isinstance(time_str, str): if len(time_str) == 10: dt = datetime.datetime.strptime(time_str, '%Y-%m-%d') return int(round(time.mktime(dt.timetuple())) * 1000) elif len(time_str) in {17, 18, 19}: dt = datetime.datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S') # strptime字符串解析必须严格按照字符串中的格式 return int(round(time.mktime(dt.timetuple())) * 1000) # 转换成毫秒级的时间戳 else: raise ValueError("时间字符串长度不满足要求！") else: return time_str class DataBase(object): def __init__(self, arg): self.begin = datetime.datetime.strptime(arg.begin, '%Y-%m-%d') self.end = datetime.datetime.strptime(arg.end, '%Y-%m-%d') - pd.Timedelta(minutes=15) self.begin_stamp = timestr_to_timestamp(str(arg.begin)) self.end_stamp = timestr_to_timestamp(str(self.end)) self.database = arg.database self.towerloc = arg.towerloc self.turbineloc = arg.turbineloc self.dataloc = arg.dataloc def clear_data(self): """ 删除所有csv :return: """ # 设置文件夹路径 import glob import os folder_path = self.dataloc # 使用 glob 获取所有的 .csv 文件路径 csv_files = glob.glob(os.path.join(folder_path, '**/*.csv'), recursive=True) # 遍历所有 .csv 文件并删除 for file_path in csv_files: os.remove(file_path) print("清除所有csv文件") def create_database(self): """ 创建数据库连接 :param database: 数据库地址 :return: """ engine = create_engine(self.database) return engine def exec_sql(self, sql, engine): """ 从数据库获取数据 :param sql: sql语句 :param engine: 数据库对象 :return: """ df = pd.read_sql_query(sql, engine) return df def split_time(self, data): data.set_index('C_TIME', inplace=True) data = data.sort_index().loc[self.begin: self.end] data.reset_index(drop=False, inplace=True) return data def get_process_NWP(self): """ 从数据库中获取NWP数据，并进行简单处理 :param database: :return: """ # NPW数据 engine = self.create_database() sql_NWP = "select C_PRE_TIME,C_T,C_RH,C_PRESSURE, C_SWR," \ "C_DIFFUSE_RADIATION, C_DIRECT_RADIATION, " \ "C_WD10,C_WD30,C_WD50,C_WD70,C_WD80,C_WD90,C_WD100,C_WD170," \ "C_WS10,C_WS30,C_WS50,C_WS70,C_WS80,C_WS90,C_WS100,C_WS170 from t_nwp" \ " where C_PRE_TIME between {} and {}".format(self.begin_stamp, self.end_stamp) # 风的NWP字段 NWP = self.exec_sql(sql_NWP, engine) NWP['C_PRE_TIME'] = NWP['C_PRE_TIME'].apply(timestamp_to_datetime) NWP = NWP.rename(columns={'C_PRE_TIME': 'C_TIME'}) NWP = cleaning(NWP, 'NWP') # NWP = self.split_time(NWP) NWP['C_TIME'] = NWP['C_TIME'].dt.strftime('%Y-%m-%d %H:%M:%S') saveData("NWP.csv", NWP) return NWP def get_process_tower(self): """ 获取环境检测仪数据 :param database: :return: """ engine = self.create_database() print("提取测风塔：{}".format(self.towerloc)) for i in self.towerloc: # 删除没用的列 drop_colmns = ["C_DATA1","C_DATA2","C_DATA3","C_DATA4","C_DATA5","C_DATA6","C_DATA7","C_DATA8","C_DATA9","C_DATA10","C_IS_GENERATED","C_ABNORMAL_CODE"] get_colmns = [] # 查询表的所有列名 result_set = self.exec_sql("SHOW COLUMNS FROM t_wind_tower_status_data", engine) for name in result_set.iloc[:, 0]: if name not in drop_colmns: get_colmns.append(name) all_columns_str = ", ".join([f'{col}' for col in get_colmns]) tower_sql = "select " + all_columns_str + " from t_wind_tower_status_data where C_EQUIPMENT_NO="+str(i) + " and C_TIME between '{}' and '{}'".format(self.begin, self.end) tower = self.exec_sql(tower_sql, engine) tower['C_TIME'] = pd.to_datetime(tower['C_TIME']) saveData("/tower-{}.csv".format(i), tower) print("测风塔{}导出数据".format(i)) def get_process_power(self): """ 获取整体功率数据 :param database: :return: """ engine = self.create_database() sql_cap = "select C_CAPACITY from t_electric_field" cap = self.exec_sql(sql_cap, engine)['C_CAPACITY'] sql_power = "select C_TIME,C_REAL_VALUE, C_ABLE_VALUE, C_IS_RATIONING_BY_MANUAL_CONTROL, C_IS_RATIONING_BY_AUTO_CONTROL" \ " from t_power_station_status_data where C_TIME between '{}' and '{}'".format(self.begin, self.end) # if self.opt.usable_power["clean_power_by_signal"]: # sql_power += " and C_IS_RATIONING_BY_MANUAL_CONTROL=0 and C_IS_RATIONING_BY_AUTO_CONTROL=0" powers = self.exec_sql(sql_power, engine) powers['C_TIME'] = pd.to_datetime(powers['C_TIME']) mask1 = powers['C_REAL_VALUE'].astype(float) > float(cap) mask = powers['C_REAL_VALUE'] == -99 mask = mask | mask1 print("实际功率共{}条，要剔除功率有{}条".format(len(powers), mask.sum())) powers = powers[~mask] print("剔除完后还剩{}条".format(len(powers))) binary_map = {b'\x00': 0, b'\x01': 1} powers['C_IS_RATIONING_BY_AUTO_CONTROL'] = powers['C_IS_RATIONING_BY_AUTO_CONTROL'].map(binary_map) powers = rm_duplicated(powers) saveData("power.csv", powers) def get_process_dq(self): """ 获取短期预测结果 :param database: :return: """ engine = self.create_database() sql_dq = "select C_FORECAST_TIME AS C_TIME, C_FP_VALUE from t_forecast_power_short_term " \ "where C_FORECAST_TIME between {} and {}".format(self.begin_stamp, self.end_stamp) dq = self.exec_sql(sql_dq, engine) # dq['C_TIME'] = pd.to_datetime(dq['C_TIME'], unit='ms') dq['C_TIME'] = dq['C_TIME'].apply(timestamp_to_datetime) # dq = dq[dq['C_FORECAST_HOW_LONG_AGO'] == 1] # dq.drop('C_FORECAST_HOW_LONG_AGO', axis=1, inplace=True) dq = cleaning(dq, 'dq', cols=['C_FP_VALUE']) dq['C_TIME'] = dq['C_TIME'].dt.strftime('%Y-%m-%d %H:%M:%S') saveData("dq.csv", dq) def indep_process(self): """ 进一步数据处理：时间统一处理等 :return: """ # 测风塔数据处理 for i in self.towerloc: tower = readData("/tower-{}.csv".format(i)) tower['C_TIME'] = pd.to_datetime(tower['C_TIME']) # 判断每一列是否全是 -99 all_minus_99 = (tower == -99).all() # 获取全是 -99 的列的列名 cols_to_drop = all_minus_99[all_minus_99 == True].index.tolist() # 使用 drop() 方法删除列 tower = tower.drop(cols_to_drop, axis=1) tower = cleaning(tower, 'tower', ['C_WS_INST_HUB_HEIGHT']) saveData("/tower-{}-process.csv".format(i), tower) def get_process_cdq(self): """ 获取超短期预测结果 :param database: :return: """ engine = self.create_database() sql_cdq = "select C_FORECAST_TIME AS C_TIME, C_ABLE_VALUE, C_FORECAST_HOW_LONG_AGO from t_forecast_power_ultra_short_term_his" \ " where C_FORECAST_TIME between {} and {}".format(self.begin_stamp, self.end_stamp) cdq = self.exec_sql(sql_cdq, engine) cdq['C_TIME'] = cdq['C_TIME'].apply(timestamp_to_datetime) cdq = cleaning(cdq, 'cdq', cols=['C_ABLE_VALUE'], dup=False) # cdq = cdq[cdq['C_FORECAST_HOW_LONG_AGO'] == int(str(self.opt.predict_point)[1:])] cdq['C_TIME'] = cdq['C_TIME'].dt.strftime('%Y-%m-%d %H:%M:%S') saveData("cdq.csv", cdq) def get_process_turbine(self): """ 从数据库中获取风头数据，并进行简单处理 :param database: :return: """ for number in self.turbineloc: # number = self.opt.usable_power['turbine_id'] # 机头数据 engine = self.create_database() print("导出风机{}的数据".format(number)) sql_turbine = "select C_TIME, C_WS, C_WD, C_ACTIVE_POWER from t_wind_turbine_status_data " \ "WHERE C_EQUIPMENT_NO=" + str(number) + " and C_TIME between '{}' and '{}'".format(self.begin, self.end) # + " and C_WS>0 and C_ACTIVE_POWER>0" turbine = self.exec_sql(sql_turbine, engine) turbine = cleaning(turbine, 'cdq', cols=['C_WS', 'C_ACTIVE_POWER'], dup=False) turbine['C_TIME'] = pd.to_datetime(turbine['C_TIME']) turbine = turbine[turbine['C_TIME'].dt.strftime('%M').isin(['00', '15', '30', '45'])] # 直接导出所有数据 saveData("turbine-15/turbine-{}.csv".format(number), turbine) def data_process(self): """ 数据导出+初步处理的总操控代码 :param database: :return: """ self.clear_data() try: self.get_process_power() self.get_process_dq() self.get_process_cdq() self.get_process_NWP() self.get_process_tower() # self.get_process_turbine() self.indep_process() except Exception as e: print("导出数据出错：{}".format(e.args))