main.py 3.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104
  1. # -*- coding: utf-8 -*-
  2. """
  3. main
  4. ~~~~
  5. The module handles the main procedure of fan coil unit alarm test.
  6. """
  7. import logging.config
  8. from datetime import datetime
  9. import yaml
  10. from apscheduler.schedulers.blocking import BlockingScheduler
  11. import data_platform_api
  12. from alarm import FanCoilUnitAlarm
  13. with open('log_config.yaml', 'r') as f:
  14. config = yaml.safe_load(f.read())
  15. logging.config.dictConfig(config)
  16. logger = logging.getLogger(__name__)
  17. def get_alarm_params(project_id, equip_id):
  18. real_time_params = [
  19. 'ctm-info008',
  20. 'ReturnAirTemp',
  21. 'SupplyTemp',
  22. 'WaterOutTemp',
  23. 'WaterInTemp',
  24. ]
  25. real_time_data = data_platform_api.query_real_time_data(project_id, equip_id, real_time_params)
  26. history_params = [
  27. 'ctm-info008',
  28. 'ctm-info010',
  29. 'ctm-info011',
  30. 'ctm-info012'
  31. ]
  32. history_data = dict()
  33. for code in history_params:
  34. temp_data = data_platform_api.query_history_time_data(project_id, equip_id, code, 15)
  35. history_data.update({code: temp_data})
  36. tap_history_status = []
  37. for i in range(len(history_data.get('ctm-info010'))):
  38. if history_data.get('ctm-info010')[i] == 1.0:
  39. tap_history_status.append('low')
  40. if history_data.get('ctm-info011')[i] == 1.0:
  41. tap_history_status.append('mid')
  42. if history_data.get('ctm-info012')[i] == 1.0:
  43. tap_history_status.append('high')
  44. params_dic = {
  45. 'water_valve_switch': real_time_data.get('ctm-info008'),
  46. 'water_valve_history_status': history_data.get('ctm-info008'),
  47. 'tap_history_status': tap_history_status,
  48. 'return_air_temp': real_time_data.get('ReturnAirTemp'),
  49. 'supply_air_temp': real_time_data.get('SupplyTemp'),
  50. 'water_out_temp': real_time_data.get('WaterOutTemp'),
  51. 'water_in_temp': real_time_data.get('WaterInTemp'),
  52. }
  53. return params_dic
  54. def guard():
  55. project_id = 'Pj1101080002'
  56. devices = ['Eq1101080002001ACATFC032']
  57. for device in devices:
  58. alarm_params = get_alarm_params(project_id, device)
  59. # logger.info(json.dumps(alarm_params, indent=4))
  60. annunciator = FanCoilUnitAlarm(alarm_params)
  61. annunciator.procedure()
  62. if annunciator.get('alarm_type') == 1:
  63. type_code = '001'
  64. if not data_platform_api.is_alarmed(project_id, device, type_code):
  65. data_platform_api.send_alarm(project_id, device, type_code, '风量偏小')
  66. def recover():
  67. project_id = 'Pj1101080002'
  68. devices = ['Eq1101080002001ACATFC032']
  69. for device in devices:
  70. alarm_info = data_platform_api.query_redis_alarm(project_id, device)
  71. for k, v in alarm_info.items():
  72. if (datetime.now() - datetime.strptime(v, '%Y%m%d%H%M%S')).total_seconds() >= 30 * 60:
  73. data_platform_api.recover_alarm(project_id, k)
  74. if __name__ == '__main__':
  75. scheduler = BlockingScheduler()
  76. scheduler.add_job(guard, 'interval', seconds=5)
  77. scheduler.add_job(recover, 'interval', seconds=10)
  78. try:
  79. scheduler.start()
  80. except (KeyboardInterrupt, SystemExit):
  81. pass