import pymysql import pandas as pd import numpy as np from sqlalchemy import create_engine import matplotlib.pyplot as plt import pytz plt.rcParams['font.sans-serif'] = ['SimHei'] import utils.savedata from utils import Arg from norm import Normalize arg = Arg.Arg() norm = Normalize() def clear_data(): """ 删除所有csv :return: """ # 设置文件夹路径 import glob import os folder_path = arg.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("清除所有scv文件") def create_database(database): """ 创建数据库连接 :param database: 数据库地址 :return: """ engine = create_engine(database) return engine def exec_sql(sql,engine): """ 从数据库获取数据 :param sql: sql语句 :param engine: 数据库对象 :return: """ df = pd.read_sql_query(sql, engine) return df def get_process_NWP(database): """ 从数据库中获取NWP数据，并进行简单处理 :param database: :return: """ # NPW数据 engine = create_database(database) sql_NWP = "select C_PRE_TIME,C_T,C_RH,C_PRESSURE, C_SWR, 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" # 光的NWP字段 NWP = exec_sql(sql_NWP, engine) #删除后三位 NWP['C_PRE_TIME'] = NWP['C_PRE_TIME'].astype(str) NWP['C_PRE_TIME'] = NWP['C_PRE_TIME'].str[:-3] # 将 'timestamp' 列转换为日期时间格式 NWP['C_PRE_TIME'] = NWP['C_PRE_TIME'].astype(float) NWP['C_PRE_TIME'] = pd.to_datetime(NWP['C_PRE_TIME'], unit='s') # 将日期时间转换为本地时区 NWP['C_PRE_TIME'] = NWP['C_PRE_TIME'].dt.tz_localize(pytz.utc).dt.tz_convert('Asia/Shanghai') # 格式化日期时间为年月日时分秒 NWP['C_PRE_TIME'] = NWP['C_PRE_TIME'].dt.strftime('%Y-%m-%d %H:%M:%S') NWP = NWP.rename(columns={'C_PRE_TIME': 'C_TIME'}) utils.savedata.saveData("NWP.csv",NWP) norm.normalize(NWP) return NWP def get_process_weather(database): """ 获取环境检测仪数据 :param database: :return: """ engine = create_database(database) print("现有环境监测仪：{}".format(arg.weatherloc)) for i in arg.weatherloc: print("环境监测仪{}导出数据".format(i)) # 删除没用的列 drop_colmns = ["C_ID", "C_EQUIPMENT_NO", "C_DATA1","C_DATA2","C_DATA3","C_DATA4","C_DATA5","C_DATA6","C_DATA7","C_DATA8","C_DATA9","C_DATA10", "C_STATUS", "C_IS_GENERATED","C_ABNORMAL_CODE"] get_colmns = [] # 查询表的所有列名 result_set = exec_sql("SHOW COLUMNS FROM t_weather_station_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]) weather_sql = "select " + all_columns_str + " from t_weather_station_status_data where C_EQUIPMENT_NO="+str(i) weather = exec_sql(weather_sql, engine) utils.savedata.saveData("weather/weather-{}.csv".format(i), weather) norm.normalize(weather) def get_process_power(database): """ 获取整体功率数据 :param database: :return: """ engine = create_database(database) sql_cap = "select C_CAPACITY from t_electric_field" cap = exec_sql(sql_cap, engine)['C_CAPACITY'] sql_power = "select C_TIME,C_REAL_VALUE from t_power_station_status_data" powers = exec_sql(sql_power, engine) mask1 = powers['C_REAL_VALUE'] > float(cap) mask = powers['C_REAL_VALUE'] == -99 mask = mask | mask1 print("要剔除功率有{}条".format(mask.sum())) powers = powers[~mask] utils.savedata.saveData("power.csv", powers) power5, power_index = [], [0] # 功率表，索引表 ps = 0 # 获取5分钟一个间隔的功率数据 for i, power in powers.iterrows(): real_value = power['C_REAL_VALUE'] ps += real_value if str(power['C_TIME'].minute)[-1] in ('0', '5'): power_index.append(i) num = power_index[-1] - power_index[-2] num = num if num != 0 else 1 psa = round(ps / num, 2) power5.append([power['C_TIME'], psa]) ps = 0 power5 = pd.DataFrame(power5, columns=['C_TIME', 'C_REAL_VALUE']) utils.savedata.saveData("power5.csv", power5) norm.normalize(power5) def get_process_dq(database): """ 获取短期预测结果 :param database: :return: """ engine = create_database(database) sql_dq = "select C_FORECAST_TIME AS C_TIME, C_ABLE_VALUE from t_forecast_power_short_term_his" dq = exec_sql(sql_dq, engine) dq['C_TIME'] = pd.to_datetime(dq['C_TIME'], unit='ms') utils.savedata.saveData("dq.csv", dq) def get_process_cdq(database): """ 获取超短期预测结果 :param database: :return: """ engine = create_database(database) sql_cdq = "select C_FORECAST_TIME AS C_TIME, C_ABLE_VALUE from t_forecast_power_ultra_short_term_his" cdq = exec_sql(sql_cdq, engine) cdq['C_TIME'] = pd.to_datetime(cdq['C_TIME'], unit='ms') utils.savedata.saveData("cdq.csv", cdq) def indep_process(): """ 进一步数据处理：时间统一处理等 :return: """ # 环境监测仪数据处理 for i in arg.weatherloc: weather = utils.savedata.readData("/weather/weather-{}.csv".format(i)) # 判断每一列是否全是 -99 all_minus_99 = (weather == -99).all() # 获取全是 -99 的列的列名 cols_to_drop = all_minus_99[all_minus_99 == True].index.tolist() # 使用 drop() 方法删除列 weather = weather.drop(cols_to_drop, axis=1) # MBD: 将一部分是-99的列删除，把-99替换为nan weather_nan = weather.replace(-99, np.nan, inplace=False) # nan 超过80% 删除 weather = weather.dropna(axis=1, thresh=len(weather_nan) * 0.8) weather = weather.replace(np.nan, -99, inplace=False) # 删除取值全部相同的列 weather = weather.loc[:, (weather != weather.iloc[0]).any()] utils.savedata.saveData("/weather/weather-{}-process.csv".format(i), weather) # 时间统一 weather1 = utils.savedata.readData("/weather/weather-{}-process.csv".format(1)) # tower2 = utils.savedata.readData("/tower/tower-{}-process.csv".format(2)) # tower1 = tower1[tower1['C_TIME'].isin(tower2['C_TIME'])] # tower2 = tower2[tower2['C_TIME'].isin(tower1['C_TIME'])] utils.savedata.saveData("/weather/weather-{}-process.csv".format(1), weather1) # utils.savedata.saveData("/tower/tower-{}-process.csv".format(2), tower2) # 所有表时间统一 filenames = ["/NWP.csv","/power.csv", "power5.csv", "/dq.csv", "/cdq.csv", '/weather/weather-1-process.csv'] dataframes = [] for name in filenames: dataframes.append(utils.savedata.readData(name)) # 查找最大起始时间和最小结束时间 max_start_time = max(df['C_TIME'].min() for df in dataframes) min_end_time = min(df['C_TIME'].max() for df in dataframes) print(max_start_time) print(min_end_time) # 重新调整每个 DataFrame 的时间范围，只保留在 [max_start_time, min_end_time] 区间内的数据 for i, df in enumerate(dataframes): df['C_TIME'] = pd.to_datetime(df['C_TIME']) # 确保时间列是 datetime 类型 df_filtered = df[(df['C_TIME'] >= max_start_time) & (df['C_TIME'] <= min_end_time)] # 将结果保存到新文件，文件名为原文件名加上 "_filtered" 后缀 utils.savedata.saveData(filenames[i],df_filtered) def NWP_indep_process(): """ 将NWP数据按照缺失值数量划分为N个不同数据集 :return: """ # NWP数据进一步处理 NWP = utils.savedata.readData("NWP.csv") df = pd.to_datetime(NWP['C_TIME']) time_diff = df.diff() time_diff_threshold = pd.Timedelta(minutes=15) missing_values = df[time_diff > time_diff_threshold] print("NWP数据缺失的数量为:{}".format(len(missing_values))) print(missing_values) # 文件保存 utils.savedata.saveVar("NWP_miss.pickle", missing_values) split_indices = [] for i in range(len(missing_values)): if i == 0: split_indices.append((0, missing_values.index[i])) else: split_indices.append((missing_values.index[i - 1], missing_values.index[i])) split_indices.append((missing_values.index[-1], len(df))) # MBD: 分割少了一个点 split_datasets = [NWP.iloc[start:end,:] for start, end in split_indices] for i, split_df in enumerate(split_datasets): utils.savedata.saveData("Dataset_training/NWP/NWP_{}.csv".format(i),split_df) return split_datasets # def power_indep_process(): # NWP = utils.savedata.readData("power.csv") def Data_split(): """ 这个函数没用上,可以不看 :return: """ NWP = utils.savedata.readData("power_15min.csv") df = pd.to_datetime(NWP['C_TIME']) time_diff = df.diff() time_diff_threshold = pd.Timedelta(minutes=15) missing_values = df[time_diff > time_diff_threshold] print("NWP数据缺失的数量为:{}".format(len(missing_values))) print(missing_values) NWP_miss = utils.savedata.readVar("NWP_miss.pickle") for t in missing_values.index: a = t-1 b = t time1 = NWP['C_TIME'][a] time2 = NWP['C_TIME'][b] df = pd.to_datetime([time1, time2]) # 计算时间差 time_diff = (df[1] - df[0]) / pd.Timedelta(minutes=15) print(time_diff) time1 = "2022-10-27 14:00:00" time2 = "2023-04-16 12:00:00" df = pd.to_datetime([time1, time2]) # 计算时间差 time_diff = (df[1] - df[0]) / pd.Timedelta(minutes=15) print(time_diff) def time_all_in(): """ 这个函数暂时没用上，这个函数的目的是给机头数据进行填充，找到时间缺失的位置，填充为-99 :return: """ filenames = [] dataframes = [] for i in arg.turbineloc: filenames.append("/turbine-15/turbine-{}.csv".format(i)) for name in filenames: dataframes.append(utils.savedata.readData(name)) for df in dataframes: df['C_TIME'] = pd.to_datetime(df['C_TIME']) # 创建一个完整的时间序列索引，包括所有可能的时间点 start_time = df['C_TIME'].min() end_time = df['C_TIME'].max() full_time_range = pd.date_range(start_time, end_time, freq='15min') # 使用完整的时间序列索引创建一个空的 DataFrame full_df = pd.DataFrame(index=full_time_range) full_df.index.name = 'C_TIME' # 将原始数据与空的 DataFrame 合并 merged_df = full_df.merge(df, how='left', left_on='time', right_on='time') # 使用 -99 填充缺失值，除了时间列 merged_df.fillna(-99, inplace=True) merged_df.reset_index(inplace=True) def data_process(database): """ 数据导出+初步处理的总操控代码 :param database: :return: """ clear_data() get_process_power(database) get_process_dq(database) get_process_cdq(database) get_process_NWP(database) get_process_weather(database) indep_process() NWP_indep_process() norm.save_yml({'mean': norm.mean, 'std': norm.std}, arg.normloc) if __name__ == '__main__': import os import glob # 设置文件夹路径 folder_path = '../data' # 使用 glob 获取所有的 .csv 文件 csv_files = glob.glob(os.path.join(folder_path, '*.csv')) # 遍历所有 .csv 文件并删除 for file_path in csv_files: os.remove(file_path) # database = "mysql+pymysql://root:!QAZ2root@192.168.1.205:3306/ipfcst-sishui-a" # engine = create_database(database) # # # NPW数据 # sql_NWP = "select C_SC_DATE,C_SC_TIME,C_T,C_RH,C_PRESSURE,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" # NWP = exec_sql(sql_NWP,engine) # # # 分风机功率 # sql_wind = "select C_WS,C_ACTIVE_POWER from t_wind_turbine_status_data_15 WHERE C_EQUIPMENT_NO=2 and C_WS>0 and C_ACTIVE_POWER>0 " # df_wind = exec_sql(sql_wind,engine) # print(df_wind) # 总功率数据读取 # sql_power = "select * from t_power_station_status_data" # df_power = exec_sql(sql_power, engine) filenames = [] dataframes = [] for i in arg.turbineloc: filenames.append("../data/turbine-15/turbine-{}.csv".format(i)) for name in filenames: dataframes.append(pd.read_csv(name).iloc[:7000,:]) # for df in enumerate(dataframes): # df = mean_of_first_columns = pd.concat([df['C_WS'] for df in dataframes], axis=1).mean(axis=1) mean_of_second_columns = (pd.concat([df['C_ACTIVE_POWER'] for df in dataframes], axis=1).sum(axis=1)/1000).astype(int) print(len(mean_of_first_columns)) plt.scatter(mean_of_first_columns, mean_of_second_columns) plt.show()