chenhaiyang 5 éve
szülő
commit
8b4819efb4
6 módosított fájl, 356 hozzáadás és 0 törlés
  1. 0 0
      __init__.py
  2. 48 0
      alarm.py
  3. 1 0
      alarm_info.json
  4. 165 0
      data_platform_api.py
  5. 38 0
      log_config.yaml
  6. 104 0
      main.py

+ 0 - 0
__init__.py


+ 48 - 0
alarm.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+
+"""
+alarm
+~~~~~
+
+This module contains the ATFC alarm logic.
+"""
+
+from abc import ABC, abstractmethod
+
+
+class Alarm(ABC):
+    """The base Alarm"""
+
+    def __init__(self, params):
+        self._params = params
+
+    def get(self, property_name):
+        return self._params.get(property_name)
+
+    def set_alarm(self, category):
+        self._params.update({'alarm_type': category})
+
+    @abstractmethod
+    def procedure(self):
+        raise NotImplementedError
+
+
+class FanCoilUnitAlarm(Alarm):
+
+    def __init__(self, params):
+        super().__init__(params)
+
+    def procedure(self):
+        if self.get('water_valve_switch') == 1:
+            if 0 not in self.get('water_valve_history_status'):
+                return
+
+        if len(set(self.get('tap_history_status'))) <= 1:
+            return
+
+        air_temp_diff = self.get('return_air_temp') - abs(self.get('supply_air_temp'))
+        water_temp_diff = self.get('water_out_temp') - abs(self.get('water_in_temp'))
+        if air_temp_diff > 12 and water_temp_diff < 5:
+            self.set_alarm(1)
+
+        return

+ 1 - 0
alarm_info.json

@@ -0,0 +1 @@
+{"Content": ["Test_id"]}

+ 165 - 0
data_platform_api.py

@@ -0,0 +1,165 @@
+# -*- coding: utf-8 -*-
+
+"""
+data_platform_api
+~~~~~~~~~~~~~~~~~
+
+This module handle the data platform api request, query equipment data, creat alarm and so on.
+"""
+
+import datetime
+import logging
+
+import redis
+import requests
+from requests import exceptions
+
+logger = logging.getLogger(__name__)
+
+pool = redis.ConnectionPool(host='172.16.34.43', port=6379, db=12, password='deepsaga', decode_responses=True)
+r = redis.Redis(connection_pool=pool)
+
+DATA_PLATFORM_SERVER = 'http://api.sagacloud.cn/data-platform-3/'
+ALARM_SERVER = 'http://47.93.33.207:8890/alarm-system/'
+
+
+class PlatformRequest:
+    """The Base Request for Platform"""
+
+    def __init__(self, server, url, params, payload, timeout=5):
+        self._url = server + url
+        self._params = params
+        self._payload = payload
+        self._timeout = timeout
+
+    def send_request(self):
+        result = dict()
+        try:
+            response = requests.post(self._url, params=self._params, json=self._payload, timeout=self._timeout)
+        except exceptions as e:
+            raise Exception(f'Could not connect to platform server {e}')
+        else:
+            if response.status_code == 200:
+                result = response.json()
+            else:
+                logger.info(f'request error {response.status_code}, {response.reason}')
+        finally:
+            return result
+
+
+def query_real_time_data(project_id, equip_id, code_list):
+    logger.info(project_id)
+    url = 'parameter/batch_query_param'
+    params = {
+        'projectId': project_id,
+        'secret': 'saga123456'
+    }
+    request_param_list = [{'id': equip_id, 'code': item} for item in code_list]
+    payload = {
+        'criterias': request_param_list
+    }
+    porter = PlatformRequest(DATA_PLATFORM_SERVER, url, params, payload)
+    raw_result = porter.send_request()
+    result = {item.get('code'): item.get('data') for item in raw_result.get('Content')}
+
+    return result
+
+
+def query_history_time_data(project_id, equip_id, code, period):
+    url = 'hisdata/query_by_obj'
+    from_time = (datetime.datetime.now() - datetime.timedelta(minutes=period)).strftime('%Y%m%d%H%M%S')
+    params = {
+        'projectId': project_id,
+        'secret': 'saga123456'
+    }
+    payload = {
+        'criteria': {
+            'id': equip_id,
+            'code': code,
+            'receivetime': {
+                '$gte': from_time
+            }
+        }
+    }
+    porter = PlatformRequest(DATA_PLATFORM_SERVER, url, params, payload)
+    raw_result = porter.send_request()
+    result = [item.get('data') for item in raw_result.get('Content')]
+
+    return result
+
+
+def send_alarm(project_id, object_id, type_code, description):
+    url = 'alarm/create'
+    params = {
+        'projectId': project_id,
+        'secret': 'saga123456'
+    }
+    time_stamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
+    payload = {
+        'status': 'alarm',
+        'objects': [object_id],
+        'ctime': time_stamp,
+        'level': 's',
+        'type_code': type_code,
+        'description': description
+    }
+    porter = PlatformRequest(ALARM_SERVER, url, params, payload)
+    raw_result = porter.send_request()
+    result = raw_result.get('id')
+
+    send_alarm(project_id, object_id, result, time_stamp)
+
+    return result
+
+
+def is_alarmed(project_id, object_id, alarm_type):
+    url = 'alarm/query'
+    params = {
+        'projectId': project_id,
+        'secret': 'saga123456'
+    }
+    from_time = (datetime.datetime.now() - datetime.timedelta(minutes=30)).strftime('%Y%m%d%H%M%S')
+    payload = {
+        'criteria': {
+            'object': object_id,
+            'type_code': alarm_type,
+            'ctime': {
+                '$gte': from_time
+            }
+        },
+    }
+    porter = PlatformRequest(ALARM_SERVER, url, params, payload)
+    raw_result = porter.send_request()
+    flag = False
+    for item in raw_result.get('Content'):
+        if 'alarm' == item.get('status'):
+            flag = True
+
+    return flag
+
+
+def recover_alarm(project_id, alarm_id):
+    url = 'alarm/batchRecover'
+    params = {
+        'projectId': project_id,
+        'secret': 'saga123456'
+    }
+    payload = {
+        'alarmId': [alarm_id]
+    }
+    porter = PlatformRequest(ALARM_SERVER, url, params, payload)
+    raw_result = porter.send_request()
+
+    return raw_result.get('Result')
+
+
+def save_alarm_info(project_id, object_id, alarm_id, time_stamp):
+    name = f'alarm_{project_id}_{object_id}'
+    r.hset(name, alarm_id, time_stamp)
+
+
+def query_redis_alarm(project_id, object_id):
+    name = f'alarm_{project_id}_{object_id}'
+    result = r.hgetall(name)
+
+    return result

+ 38 - 0
log_config.yaml

@@ -0,0 +1,38 @@
+version: 1
+disable_existing_loggers: no
+
+formatters:
+    simple:
+        format: "%(asctime)s - %(name)s - %(message)s"
+    extended:
+        format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+
+handlers:
+    console:
+        class: logging.StreamHandler
+        level: DEBUG
+        formatter: extended
+        stream: ext://sys.stdout
+
+    file_handler:
+        class: logging.handlers.RotatingFileHandler
+        level: INFO
+        formatter: extended
+        filename: test.log
+        maxBytes: 5242880
+        backupCount: 1
+        encoding: utf8
+
+loggers:
+    dev:
+        level: DEBUG
+        handlers: [console, file_handler]
+        propagate: no
+    test:
+        level: INFO
+        handlers: [console]
+        propagate: no
+
+root:
+    level: DEBUG
+    handlers: [console]

+ 104 - 0
main.py

@@ -0,0 +1,104 @@
+# -*- 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