post_process.py 3.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  1. import pandas as pd
  2. from flask import Flask, request, jsonify
  3. import time
  4. import logging
  5. import traceback
  6. from common.database_dml import get_data_from_mongo, insert_data_into_mongo
  7. app = Flask('post_processing——service')
  8. def get_data(args):
  9. df = get_data_from_mongo(args)
  10. col_time = args['col_time']
  11. if not df.empty:
  12. print("预测数据加载成功!")
  13. df[col_time] = pd.to_datetime(df[col_time])
  14. df.set_index(col_time, inplace=True)
  15. df.sort_index(inplace=True)
  16. else:
  17. raise ValueError("未获取到预测数据。")
  18. return df
  19. def predict_result_adjustment(df, args):
  20. """
  21. 光伏/风电 数据后处理 主要操作
  22. 1. 光伏 (夜间 置零 + 平滑)
  23. 2. 风电 (平滑)
  24. 3. cap 封顶
  25. """
  26. mongodb_database, plant_type, cap, col_time = args['mongodb_database'], args['plant_type'], float(args['cap']), \
  27. args['col_time']
  28. if 'smooth_window' in args.keys():
  29. smooth_window = int(args['smooth_window'])
  30. else:
  31. smooth_window = 3
  32. # 平滑
  33. df_cp = df.copy()
  34. df_cp['power_forecast'] = df_cp['power_forecast'].rolling(window=smooth_window, min_periods=1,
  35. center=True).mean().clip(0, 0.985 * cap)
  36. print("smooth processed")
  37. # 光伏晚上置零
  38. if plant_type == 'solar' and 'mongodb_nwp_table' in args.keys():
  39. nwp_param = {
  40. 'mongodb_database': mongodb_database,
  41. 'mongodb_read_table': args['mongodb_nwp_table'],
  42. 'col_time': col_time
  43. }
  44. nwp = get_data(nwp_param)
  45. df_cp = df_cp.join(nwp['radiation'])
  46. df_cp.loc[nwp['radiation'] == 0, 'power_forecast'] = 0
  47. df_cp['power_forecast'] = round(df_cp['power_forecast'], 2)
  48. df_cp.drop(columns=['radiation'], inplace=True)
  49. print("solar processed")
  50. df_cp.reset_index(inplace=True)
  51. df_cp[col_time] = df_cp[col_time].dt.strftime('%Y-%m-%d %H:%M:%S')
  52. return df_cp
  53. @app.route('/post_process', methods=['POST'])
  54. def data_join():
  55. # 获取程序开始时间
  56. start_time = time.time()
  57. result = {}
  58. success = 0
  59. print("Program starts execution!")
  60. try:
  61. args = request.values.to_dict()
  62. print('args', args)
  63. logger.info(args)
  64. df_pre = get_data(args)
  65. res_df = predict_result_adjustment(df_pre, args)
  66. insert_data_into_mongo(res_df, args)
  67. success = 1
  68. except Exception as e:
  69. my_exception = traceback.format_exc()
  70. my_exception.replace("\n", "\t")
  71. result['msg'] = my_exception
  72. end_time = time.time()
  73. result['success'] = success
  74. result['args'] = args
  75. result['start_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(start_time))
  76. result['end_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(end_time))
  77. print("Program execution ends!")
  78. return result
  79. if __name__ == "__main__":
  80. print("Program starts execution!")
  81. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  82. logger = logging.getLogger("post_processing")
  83. from waitress import serve
  84. serve(app, host="0.0.0.0", port=10130)
  85. print("server start!")