baiyushan_20240906.py 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196
  1. from multiprocessing import Pool
  2. from os import *
  3. import chardet
  4. import pandas as pd
  5. # 获取文件编码
  6. def detect_file_encoding(filename):
  7. # 读取文件的前1000个字节(足够用于大多数编码检测)
  8. with open(filename, 'rb') as f:
  9. rawdata = f.read(1000)
  10. result = chardet.detect(rawdata)
  11. encoding = result['encoding']
  12. if encoding is None:
  13. encoding = 'gb18030'
  14. if encoding and encoding.lower() == 'gb2312' or encoding.lower().startswith("windows"):
  15. encoding = 'gb18030'
  16. return encoding
  17. # 读取数据到df
  18. def read_file_to_df(file_path, read_cols=list(), header=0):
  19. df = pd.DataFrame()
  20. if str(file_path).lower().endswith("csv") or str(file_path).lower().endswith("gz"):
  21. encoding = detect_file_encoding(file_path)
  22. end_with_gz = str(file_path).lower().endswith("gz")
  23. if read_cols:
  24. if end_with_gz:
  25. df = pd.read_csv(file_path, encoding=encoding, usecols=read_cols, compression='gzip', header=header)
  26. else:
  27. df = pd.read_csv(file_path, encoding=encoding, usecols=read_cols, header=header, on_bad_lines='warn')
  28. else:
  29. if end_with_gz:
  30. df = pd.read_csv(file_path, encoding=encoding, compression='gzip', header=header)
  31. else:
  32. df = pd.read_csv(file_path, encoding=encoding, header=header, on_bad_lines='warn')
  33. else:
  34. xls = pd.ExcelFile(file_path)
  35. # 获取所有的sheet名称
  36. sheet_names = xls.sheet_names
  37. for sheet in sheet_names:
  38. if read_cols:
  39. df = pd.concat([df, pd.read_excel(xls, sheet_name=sheet, header=header, usecols=read_cols)])
  40. else:
  41. df = pd.concat([df, pd.read_excel(xls, sheet_name=sheet, header=header)])
  42. return df
  43. def __build_directory_dict(directory_dict, path, filter_types=None):
  44. # 遍历目录下的所有项
  45. for item in listdir(path):
  46. item_path = path.join(path, item)
  47. if path.isdir(item_path):
  48. __build_directory_dict(directory_dict, item_path, filter_types=filter_types)
  49. elif path.isfile(item_path):
  50. if path not in directory_dict:
  51. directory_dict[path] = []
  52. if filter_types is None or len(filter_types) == 0:
  53. directory_dict[path].append(item_path)
  54. elif str(item_path).split(".")[-1] in filter_types:
  55. if str(item_path).count("~$") == 0:
  56. directory_dict[path].append(item_path)
  57. # 读取所有文件
  58. # 读取路径下所有的excel文件
  59. def read_excel_files(read_path):
  60. directory_dict = {}
  61. __build_directory_dict(directory_dict, read_path, filter_types=['xls', 'xlsx', 'csv', 'gz'])
  62. return [path for paths in directory_dict.values() for path in paths if path]
  63. # 创建路径
  64. def create_file_path(path, is_file_path=False):
  65. if is_file_path:
  66. path = path.dirname(path)
  67. if not path.exists(path):
  68. makedirs(path, exist_ok=True)
  69. def read_status(status_path):
  70. all_files = read_excel_files(status_path)
  71. with Pool(20) as pool:
  72. dfs = pool.starmap(read_file_to_df, [(file, ['设备名称', '状态码', '开始时间'], 2) for file in all_files])
  73. df = pd.concat(dfs)
  74. df = df[df['状态码'].isin([3, 5])]
  75. df['开始时间'] = pd.to_datetime(df['开始时间'])
  76. df['处理后时间'] = (df['开始时间'] + pd.Timedelta(minutes=10)).apply(
  77. lambda x: f"{x.year}-{str(x.month).zfill(2)}-{str(x.day).zfill(2)} {str(x.hour).zfill(2)}:{x.minute // 10}0:00")
  78. df['处理后时间'] = pd.to_datetime(df['处理后时间'])
  79. df = df[(df['处理后时间'] >= '2023-09-01 00:00:00')]
  80. df[df['处理后时间'] >= '2024-09-01 00:00:00'] = '2024-09-01 00:00:00'
  81. df.sort_values(by=['设备名称', '处理后时间'], inplace=True)
  82. return df
  83. def read_fault_data(fault_path):
  84. all_files = read_excel_files(fault_path)
  85. with Pool(20) as pool:
  86. dfs = pool.starmap(read_file_to_df, [(file, ['设备名称', '故障开始时间'], 2) for file in all_files])
  87. df = pd.concat(dfs)
  88. df = df[df['设备名称'].str.startswith("#")]
  89. df['故障开始时间'] = pd.to_datetime(df['故障开始时间'])
  90. df['处理后故障开始时间'] = (df['故障开始时间'] + pd.Timedelta(minutes=10)).apply(
  91. lambda x: f"{x.year}-{str(x.month).zfill(2)}-{str(x.day).zfill(2)} {str(x.hour).zfill(2)}:{x.minute // 10}0:00")
  92. df['处理后故障开始时间'] = pd.to_datetime(df['处理后故障开始时间'])
  93. df = df[(df['处理后故障开始时间'] >= '2023-09-01 00:00:00') & (df['处理后故障开始时间'] < '2024-09-01 00:00:00')]
  94. df.sort_values(by=['设备名称', '处理后故障开始时间'], inplace=True)
  95. return df
  96. def read_10min_data(data_path):
  97. all_files = read_excel_files(data_path)
  98. with Pool(20) as pool:
  99. dfs = pool.starmap(read_file_to_df,
  100. [(file, ['设备名称', '时间', '平均风速(m/s)', '平均网侧有功功率(kW)'], 1) for file in all_files])
  101. df = pd.concat(dfs)
  102. df['时间'] = pd.to_datetime(df['时间'])
  103. df = df[(df['时间'] >= '2023-09-01 00:00:00') & (df['时间'] < '2024-09-01 00:00:00')]
  104. df.sort_values(by=['设备名称', '时间'], inplace=True)
  105. return df
  106. def select_data_and_save(name, fault_df, origin_df):
  107. df = pd.DataFrame()
  108. for i in range(fault_df.shape[0]):
  109. fault = fault_df.iloc[i]
  110. con1 = origin_df['时间'] >= fault['处理后故障开始时间']
  111. con2 = origin_df['时间'] <= fault['结束时间']
  112. df = pd.concat([df, origin_df[con1 & con2]])
  113. name = name.replace('#', 'F')
  114. df.drop_duplicates(inplace=True)
  115. df.to_csv(save_path + sep + name + '.csv', index=False, encoding='utf8')
  116. if __name__ == '__main__':
  117. base_path = r'/data/download/白玉山/需要整理的数据'
  118. save_path = base_path + sep + 'sele_data_202409261135'
  119. create_file_path(save_path)
  120. status_df = read_status(base_path + sep + '设备状态')
  121. fault_df = read_fault_data(base_path + sep + '故障')
  122. data_df = read_10min_data(base_path + sep + '十分钟')
  123. status_df.to_csv(base_path + sep + '设备状态' + '.csv', index=False, encoding='utf8')
  124. fault_df.to_csv(base_path + sep + '故障' + '.csv', index=False, encoding='utf8')
  125. data_df.to_csv(base_path + sep + '十分钟' + '.csv', index=False, encoding='utf8')
  126. print(status_df.shape)
  127. print(fault_df.shape)
  128. print(data_df.shape)
  129. fault_list = list()
  130. for i in range(fault_df.shape[0]):
  131. data = fault_df.iloc[i]
  132. con1 = status_df['设备名称'] == data['设备名称']
  133. con2 = status_df['处理后时间'] >= data['处理后故障开始时间']
  134. fault_list.append(status_df[con1 & con2]['处理后时间'].min())
  135. fault_df['结束时间'] = fault_list
  136. status_df.to_csv(base_path + sep + '设备状态' + '.csv', index=False, encoding='utf8')
  137. fault_df.to_csv(base_path + sep + '故障' + '.csv', index=False, encoding='utf8')
  138. data_df.to_csv(base_path + sep + '十分钟' + '.csv', index=False, encoding='utf8')
  139. names = set(fault_df['设备名称'])
  140. fault_map = dict()
  141. data_map = dict()
  142. for name in names:
  143. fault_map[name] = fault_df[fault_df['设备名称'] == name]
  144. data_map[name] = data_df[data_df['设备名称'] == name]
  145. with Pool(20) as pool:
  146. pool.starmap(select_data_and_save, [(name, fault_map[name], data_map[name]) for name in names])