schedule_service.py 5.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114
  1. # -*- coding: utf-8 -*-
  2. # @Time : 2024/6/11
  3. # @Author : 魏志亮
  4. from etl.base.TranseParam import TranseParam
  5. from etl.base.WindFarms import WindFarms
  6. from service.plt_service import get_exec_data, update_trans_status_error
  7. from service.trans_service import get_trans_conf
  8. from utils.conf.read_conf import read_conf
  9. from utils.log.trans_log import init_log, trans_print, logger
  10. def run_schedule(step=0, end=3):
  11. data = get_exec_data()
  12. if data is None:
  13. trans_print("当前有任务在执行")
  14. elif len(data.keys()) == 0:
  15. trans_print("当前无任务")
  16. else:
  17. batch_no = data['batch_code']
  18. transfer_type = data['transfer_type']
  19. transfer_file_addr = data['transfer_addr']
  20. field_code = data['field_code']
  21. field_name = data['field_name']
  22. __exec_trans(step, end, batch_no, transfer_type, transfer_file_addr, field_name, field_code, schedule_exec=True)
  23. def run_local(step=0, end=3, batch_no=None, transfer_type=None, transfer_file_addr=None, field_name=None,
  24. field_code="测试"):
  25. if batch_no is None or str(batch_no).strip() == '':
  26. return "批次编号不能为空"
  27. if transfer_type not in ['second', 'minute']:
  28. return "查询类型错误"
  29. if transfer_file_addr is None or str(transfer_file_addr).strip() == '':
  30. return "文件路径不能为空"
  31. __exec_trans(step, end, batch_no, transfer_type, transfer_file_addr, field_name, field_code, schedule_exec=False)
  32. def __exec_trans(step, end, batch_no, transfer_type, transfer_file_addr=None, field_name=None, field_code="测试",
  33. schedule_exec=False):
  34. init_log(batch_no, field_name, transfer_type)
  35. conf_map = get_trans_conf(field_name, transfer_type)
  36. if conf_map is None or type(conf_map) == tuple or len(conf_map.keys()) == 0:
  37. message = f"未找到{field_name}的{transfer_type}配置"
  38. trans_print(message)
  39. update_trans_status_error(batch_no, transfer_type, message, schedule_exec)
  40. else:
  41. resolve_col_prefix = read_conf(conf_map, 'resolve_col_prefix')
  42. wind_name_exec = read_conf(conf_map, 'wind_name_exec', None)
  43. wind_full_name = read_conf(conf_map, 'wind_full_name')
  44. is_vertical_table = read_conf(conf_map, 'is_vertical_table', False)
  45. merge_columns = read_conf(conf_map, 'merge_columns', False)
  46. vertical_cols = read_conf(conf_map, 'vertical_read_cols', '').split(',')
  47. index_cols = read_conf(conf_map, 'vertical_index_cols', '').split(',')
  48. vertical_key = read_conf(conf_map, 'vertical_col_key')
  49. vertical_value = read_conf(conf_map, 'vertical_col_value')
  50. need_valid_cols = not merge_columns
  51. cols_trans_all = dict()
  52. trans_cols = ['wind_turbine_number', 'time_stamp', 'active_power', 'rotor_speed', 'generator_speed',
  53. 'wind_velocity', 'pitch_angle_blade_1', 'pitch_angle_blade_2', 'pitch_angle_blade_3',
  54. 'cabin_position', 'true_wind_direction', 'yaw_error1', 'set_value_of_active_power',
  55. 'gearbox_oil_temperature', 'generatordrive_end_bearing_temperature',
  56. 'generatornon_drive_end_bearing_temperature', 'wind_turbine_status',
  57. 'wind_turbine_status2',
  58. 'cabin_temperature', 'twisted_cable_angle', 'front_back_vibration_of_the_cabin',
  59. 'side_to_side_vibration_of_the_cabin', 'actual_torque', 'given_torque',
  60. 'clockwise_yaw_count',
  61. 'counterclockwise_yaw_count', 'unusable', 'power_curve_available',
  62. 'required_gearbox_speed',
  63. 'inverter_speed_master_control', 'outside_cabin_temperature', 'main_bearing_temperature',
  64. 'gearbox_high_speed_shaft_bearing_temperature',
  65. 'gearboxmedium_speed_shaftbearing_temperature',
  66. 'gearbox_low_speed_shaft_bearing_temperature', 'generator_winding1_temperature',
  67. 'generator_winding2_temperature', 'generator_winding3_temperature',
  68. 'turbulence_intensity', 'param1',
  69. 'param2', 'param3', 'param4', 'param5', 'param6', 'param7', 'param8', 'param9', 'param10']
  70. for col in trans_cols:
  71. cols_trans_all[col] = read_conf(conf_map, col, '')
  72. trans_subject = WindFarms(field_name, batch_no=batch_no, field_code=field_code,
  73. wind_full_name=wind_full_name, schedule_exec=schedule_exec)
  74. params = TranseParam(read_type=transfer_type, read_path=transfer_file_addr,
  75. cols_tran=cols_trans_all,
  76. wind_name_exec=wind_name_exec, is_vertical_table=is_vertical_table,
  77. vertical_cols=vertical_cols, vertical_key=vertical_key,
  78. vertical_value=vertical_value, index_cols=index_cols, merge_columns=merge_columns,
  79. resolve_col_prefix=resolve_col_prefix, need_valid_cols=need_valid_cols)
  80. trans_subject.set_trans_param(params)
  81. try:
  82. trans_subject.run(step=step, end=end)
  83. except Exception as e:
  84. logger.exception(e)
  85. message = "系统返回错误:" + str(e)
  86. update_trans_status_error(batch_no, transfer_type, message, schedule_exec)
  87. if __name__ == '__main__':
  88. step = 0
  89. end = 3
  90. batch_no = 'hongtiguan-test'
  91. transfer_type = 'second'
  92. transfer_file_addr = r'/data/download/collection_data/1进行中/虹梯官风电场-山西-大唐/收资数据/秒级数据/20240527秒级数据'
  93. field_name = '虹梯官风电场'
  94. field_code = "测试"
  95. run_local(step, end, batch_no, transfer_type, transfer_file_addr, field_name, field_code)