# -*- coding: utf-8 -*- """ main ~~~~ The module handles the main procedure of fan coil unit alarm test. """ import logging.config from datetime import datetime import yaml from apscheduler.schedulers.blocking import BlockingScheduler import data_platform_api from alarm import FanCoilUnitAlarm with open('log_config.yaml', 'r') as f: config = yaml.safe_load(f.read()) logging.config.dictConfig(config) logger = logging.getLogger(__name__) def get_alarm_params(project_id, equip_id): real_time_params = [ 'ctm-info008', 'ReturnAirTemp', 'SupplyTemp', 'WaterOutTemp', 'WaterInTemp', ] real_time_data = data_platform_api.query_real_time_data(project_id, equip_id, real_time_params) history_params = [ 'ctm-info008', 'ctm-info010', 'ctm-info011', 'ctm-info012' ] history_data = dict() for code in history_params: temp_data = data_platform_api.query_history_time_data(project_id, equip_id, code, 15) history_data.update({code: temp_data}) tap_history_status = [] for i in range(len(history_data.get('ctm-info010'))): if history_data.get('ctm-info010')[i] == 1.0: tap_history_status.append('low') if history_data.get('ctm-info011')[i] == 1.0: tap_history_status.append('mid') if history_data.get('ctm-info012')[i] == 1.0: tap_history_status.append('high') params_dic = { 'water_valve_switch': real_time_data.get('ctm-info008'), 'water_valve_history_status': history_data.get('ctm-info008'), 'tap_history_status': tap_history_status, 'return_air_temp': real_time_data.get('ReturnAirTemp'), 'supply_air_temp': real_time_data.get('SupplyTemp'), 'water_out_temp': real_time_data.get('WaterOutTemp'), 'water_in_temp': real_time_data.get('WaterInTemp'), } return params_dic def guard(): project_id = 'Pj1101080002' devices = ['Eq1101080002001ACATFC032'] for device in devices: alarm_params = get_alarm_params(project_id, device) # logger.info(json.dumps(alarm_params, indent=4)) annunciator = FanCoilUnitAlarm(alarm_params) annunciator.procedure() if annunciator.get('alarm_type') == 1: type_code = '001' if not data_platform_api.is_alarmed(project_id, device, type_code): data_platform_api.send_alarm(project_id, device, type_code, '风量偏小') def recover(): project_id = 'Pj1101080002' devices = ['Eq1101080002001ACATFC032'] for device in devices: alarm_info = data_platform_api.query_redis_alarm(project_id, device) for k, v in alarm_info.items(): if (datetime.now() - datetime.strptime(v, '%Y%m%d%H%M%S')).total_seconds() >= 30 * 60: data_platform_api.recover_alarm(project_id, k) if __name__ == '__main__': scheduler = BlockingScheduler() scheduler.add_job(guard, 'interval', seconds=5) scheduler.add_job(recover, 'interval', seconds=10) try: scheduler.start() except (KeyboardInterrupt, SystemExit): pass