o2ims/adapter/clients/alarm_dict_client.py

   1 # Copyright (C) 2022 Wind River Systems, Inc.
   2 #
   3 #  Licensed under the Apache License, Version 2.0 (the "License");
   4 #  you may not use this file except in compliance with the License.
   5 #  You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #  Unless required by applicable law or agreed to in writing, software
  10 #  distributed under the License is distributed on an "AS IS" BASIS,
  11 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #  See the License for the specific language governing permissions and
  13 #  limitations under the License.
  14
  15 import os
  16 import json
  17 import yaml
  18 import errno
  19 import collections
  20 import uuid as uuid_gen
  21
  22 from o2common.service import unit_of_work
  23 from o2common.config import config
  24 from o2ims.domain import alarm_obj as alarm
  25
  26 from o2common.helper import o2logging
  27 logger = o2logging.get_logger(__name__)
  28
  29
  30 def load_alarm_dictionary_from_conf_file(conf_path: str,
  31                                          uow: unit_of_work.AbstractUnitOfWork):
  32
  33     logger.info("Converting alarm.yaml to dict: ")
  34
  35     if not os.path.isfile(conf_path):
  36         logger.error("file %s doesn't exist. Ending execution" %
  37                      (conf_path))
  38         raise OSError(
  39             errno.ENOENT, os.strerror(errno.ENOENT), conf_path
  40         )
  41
  42     try:
  43         with open(conf_path, 'r') as stream:
  44             alarm_yaml = yaml.load(stream, Loader=yaml.FullLoader)
  45             dictionaries = alarm_yaml.get('dictionary')
  46     except Exception as exp:
  47         logger.error(exp)
  48         raise RuntimeError(exp)
  49
  50     for dictionary in list(dictionaries.keys()):
  51         with uow:
  52             # res_type = uow.resource_types.get_by_name(dictionary)
  53             # logger.info('res_type: ' + res_type.resourceTypeName)
  54             alarm_dict = alarm.AlarmDictionary(dictionary)
  55             alarm_dict.entityType = dictionary
  56             alarm_dict.alarmDictionaryVersion = \
  57                 dictionaries[dictionary]['version']
  58             alarm_dict.alarmDefinition = \
  59                 dictionaries[dictionary]['alarmDefinition']
  60             uow.alarm_dictionaries.add(alarm_dict)
  61
  62
  63 def prettyDict(dict):
  64     output = json.dumps(dict, sort_keys=True, indent=4)
  65     return output
  66
  67
  68 def load_alarm_definition(uow: unit_of_work.AbstractUnitOfWork):
  69     logger.info("Converting events.yaml to dict: ")
  70     EVENT_TYPES_FILE = config.get_events_yaml_filename()
  71
  72     if not os.path.isfile(EVENT_TYPES_FILE):
  73         logger.error("file %s doesn't exist. Ending execution" %
  74                      (EVENT_TYPES_FILE))
  75         raise OSError(
  76             errno.ENOENT, os.strerror(errno.ENOENT), EVENT_TYPES_FILE
  77         )
  78
  79     try:
  80         with open(EVENT_TYPES_FILE, 'r') as stream:
  81             event_types = yaml.load(stream, Loader=yaml.FullLoader)
  82     except Exception as exp:
  83         logger.error(exp)
  84         raise RuntimeError(exp)
  85
  86     for alarm_id in list(event_types.keys()):
  87         if isinstance(alarm_id, float):
  88             # force 3 digits after the decimal point,
  89             # to include trailing zero's (ex.: 200.010)
  90             formatted_alarm_id = "{:.3f}".format(alarm_id)
  91             event_types[formatted_alarm_id] = event_types.pop(alarm_id)
  92
  93     event_types = collections.OrderedDict(sorted(event_types.items()))
  94
  95     yaml_event_list = []
  96     uneditable_descriptions = {'100.114', '200.007',
  97                                '200.02', '200.021', '200.022', '800.002'}
  98
  99     # Parse events.yaml dict, and add any new alarm to definition table:
 100     logger.info(
 101         "Parsing events.yaml and adding any new alarm to definition table: ")
 102     for event_type in event_types:
 103
 104         if event_types.get(event_type).get('Type') == "Alarm":
 105             event_uuid = str(uuid_gen.uuid3(
 106                 uuid_gen.NAMESPACE_URL, str(event_type)))
 107
 108             string_event_type = str(event_type)
 109
 110             yaml_event_list.append(string_event_type)
 111
 112             if str(event_type) not in uneditable_descriptions:
 113                 event_description = (event_types.get(event_type)
 114                                      .get('Description'))
 115             else:
 116                 event_description = event_types.get(
 117                     event_type).get('Description')
 118
 119             event_description = str(event_description)
 120             event_description = (event_description[:250] + ' ...') \
 121                 if len(event_description) > 250 else event_description
 122             prop_action = event_types.get(
 123                 event_type).get("Proposed_Repair_Action")
 124
 125             with uow:
 126                 alarm_def = uow.alarm_definitions.get(event_uuid)
 127                 event_mgmt_affecting = str(event_types.get(event_type).get(
 128                     'Management_Affecting_Severity', 'warning'))
 129 #
 130                 event_degrade_affecting = str(event_types.get(event_type).get(
 131                     'Degrade_Affecting_Severity', 'none'))
 132
 133                 if alarm_def:
 134                     alarm_def.description = event_description
 135                     alarm_def.mgmt_affecting = event_mgmt_affecting
 136                     alarm_def.degrade_affecting = event_degrade_affecting
 137                 else:
 138                     alarm_def = alarm.AlarmDefinition(
 139                         id=event_uuid,
 140                         name=str(event_type),
 141                         last_change=alarm.AlarmLastChangeEnum.ADDED,
 142                         desc=event_description, prop_action=prop_action,
 143                         clearing_type=alarm.ClearingTypeEnum.MANUAL,
 144                         pk_noti_field=""
 145                     )
 146                     logger.info(str(event_type))
 147                     uow.alarm_definitions.add(alarm_def)
 148
 149                 uow.commit()
 150
 151             prob_cause = event_types.get(event_type).get("Probable_Cause")
 152             prob_cause_uuid = str(uuid_gen.uuid3(
 153                 uuid_gen.NAMESPACE_URL, prob_cause))
 154
 155             with uow:
 156                 probable_cause = uow.alarm_probable_causes.get(prob_cause_uuid)
 157                 if probable_cause is None:
 158                     pc = alarm.ProbableCause(
 159                         prob_cause_uuid, prob_cause, prob_cause)
 160                     uow.alarm_probable_causes.add(pc)
 161                     uow.commit()