1 # Copyright (C) 2022 Wind River Systems, Inc.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
20 import uuid as uuid_gen
22 from o2common.service import unit_of_work
23 from o2common.config import config
24 from o2ims.domain import alarm_obj as alarm
26 from o2common.helper import o2logging
27 logger = o2logging.get_logger(__name__)
30 def load_alarm_dictionary_from_conf_file(conf_path: str,
31 uow: unit_of_work.AbstractUnitOfWork):
33 logger.info("Converting alarm.yaml to dict: ")
35 if not os.path.isfile(conf_path):
36 logger.error("file %s doesn't exist. Ending execution" %
39 errno.ENOENT, os.strerror(errno.ENOENT), conf_path
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:
48 raise RuntimeError(exp)
50 for dictionary in list(dictionaries.keys()):
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)
64 output = json.dumps(dict, sort_keys=True, indent=4)
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()
72 if not os.path.isfile(EVENT_TYPES_FILE):
73 logger.error("file %s doesn't exist. Ending execution" %
76 errno.ENOENT, os.strerror(errno.ENOENT), EVENT_TYPES_FILE
80 with open(EVENT_TYPES_FILE, 'r') as stream:
81 event_types = yaml.load(stream, Loader=yaml.FullLoader)
82 except Exception as exp:
84 raise RuntimeError(exp)
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)
93 event_types = collections.OrderedDict(sorted(event_types.items()))
96 uneditable_descriptions = {'100.114', '200.007',
97 '200.02', '200.021', '200.022', '800.002'}
99 # Parse events.yaml dict, and add any new alarm to definition table:
101 "Parsing events.yaml and adding any new alarm to definition table: ")
102 for event_type in event_types:
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)))
108 string_event_type = str(event_type)
110 yaml_event_list.append(string_event_type)
112 if str(event_type) not in uneditable_descriptions:
113 event_description = (event_types.get(event_type)
116 event_description = event_types.get(
117 event_type).get('Description')
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")
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'))
130 event_degrade_affecting = str(event_types.get(event_type).get(
131 'Degrade_Affecting_Severity', 'none'))
134 alarm_def.description = event_description
135 alarm_def.mgmt_affecting = event_mgmt_affecting
136 alarm_def.degrade_affecting = event_degrade_affecting
138 alarm_def = alarm.AlarmDefinition(
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,
146 logger.info(str(event_type))
147 uow.alarm_definitions.add(alarm_def)
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))
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)