Fix INF-364 alarmEventRecord does not comply with spec
[pti/o2.git] / o2ims / domain / alarm_obj.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 from __future__ import annotations
16 from enum import Enum
17 import json
18 import datetime
19
20 from o2common.domain.base import AgRoot, Serializer
21
22 from o2common.helper import o2logging
23 logger = o2logging.get_logger(__name__)
24
25
26 class FaultGenericModel(AgRoot):
27     def __init__(self, type: str,
28                  api_response: dict = None, content_hash=None) -> None:
29         super().__init__()
30         if api_response:
31             self.id = str(api_response.uuid)
32             self.name = self.id
33             self.alarm_type = api_response.alarm_type
34             self.alarm_def_name = api_response.alarm_id
35             self.alarm_def_id = api_response.alarm_def_id
36             self.probable_cause_id = api_response.probable_cause_id
37             self.status = api_response.state
38             # TODO: time less than second
39             self.timestamp = datetime.datetime.strptime(
40                 api_response.timestamp.split('.')[0], "%Y-%m-%dT%H:%M:%S") \
41                 if api_response.timestamp else None
42
43             # if hasattr(api_response, 'alarm_id'):
44             #     self.alarm_id = api_response.alarm_id
45             # elif hasattr(api_response, 'event_log_id'):
46             #     self.alarm_id = api_response.event_log_id
47
48             self.hash = content_hash if content_hash \
49                 else str(hash((self.id, self.timestamp, self.status)))
50             self.content = json.dumps(api_response.to_dict())
51             if EventTypeEnum.ALARM == type:
52                 pass
53
54     def is_outdated(self, newmodel) -> bool:
55         # return self.updatetime < newmodel.updatetime
56         # logger.warning("hash1: " + self.hash + " vs hash2: " + newmodel.hash)
57         return self.hash != newmodel.hash
58
59     def update_by(self, newmodel) -> None:
60         if self.id != newmodel.id:
61             pass
62             # raise MismatchedModel("Mismatched model")
63         self.name = newmodel.name
64         self.createtime = newmodel.createtime
65         self.updatetime = newmodel.updatetime
66         self.content = newmodel.content
67
68
69 class EventTypeEnum(Enum):
70     ALARM = 'alarm'
71     EVENT = 'event'
72
73
74 class PerceivedSeverityEnum(str, Enum):
75     CRITICAL = 0
76     MAJOR = 1
77     MINOR = 2
78     WARNING = 3
79     INDETERMINATE = 4
80     CLEARED = 5
81
82
83 class AlarmEventRecord(AgRoot, Serializer):
84     def __init__(self, id: str, res_type_id: str, res_id: str,
85                  alarm_def_id: str, probable_cause_id: str,
86                  raised_time: str,
87                  perc_severity: PerceivedSeverityEnum =
88                  PerceivedSeverityEnum.WARNING
89                  ) -> None:
90         super().__init__()
91         self.alarmEventRecordId = id
92         self.resourceTypeId = res_type_id
93         self.resourceId = res_id
94         self.alarmDefinitionId = alarm_def_id
95         self.probableCauseId = probable_cause_id
96         self.perceivedSeverity = perc_severity
97         self.alarmRaisedTime = raised_time
98         self.alarmChangedTime = ''
99         self.alarmAcknowledgeTime = ''
100         self.alarmAcknowledged = False
101         self.extensions = []
102
103
104 class ProbableCause(AgRoot, Serializer):
105     def __init__(self, id: str, name: str, desc: str = '') -> None:
106         super().__init__()
107         self.probableCauseId = id
108         self.name = name
109         self.description = desc
110
111
112 class AlarmChangeTypeEnum(str, Enum):
113     ADDED = 'ADDED'
114     DELETED = 'DELETED'
115     MODIFYED = 'MODIFYED'
116
117
118 class ClearingTypeEnum(str, Enum):
119     AUTOMATIC = 'AUTOMATIC'
120     MANUAL = 'MANUAL'
121
122
123 class AlarmDefinition(AgRoot, Serializer):
124     def __init__(self, id: str, name: str, change_type: AlarmChangeTypeEnum,
125                  desc: str, prop_action: str, clearing_type: ClearingTypeEnum,
126                  pk_noti_field: str) -> None:
127         super().__init__()
128         self.alarmDefinitionId = id
129         self.alarmName = name
130         self.alarmLastChange = '0.1'
131         self.alarmChangeType = change_type
132         self.alarmDescription = desc
133         self.proposedRepairActions = prop_action
134         self.clearingType = clearing_type
135         self.managementInterfaceId = "O2IMS"
136         self.pkNotificationField = pk_noti_field
137         self.alarmAdditionalFields = ""
138
139
140 class AlarmDictionary(AgRoot, Serializer):
141     def __init__(self, id: str) -> None:
142         super().__init__()
143         self.id = id
144         self.alarmDictionaryVersion = ""
145         self.alarmDictionarySchemaVersion = ""
146         self.entityType = ""
147         self.vendor = ""
148         self.managementInterfaceId = "O2IMS"
149         self.pkNotificationField = ""
150         self.alarmDefinition = []
151
152     def serialize(self):
153         d = Serializer.serialize(self)
154         if 'alarmDefinition' in d and len(d['alarmDefinition']) > 0:
155             d['alarmDefinition'] = self.serialize_list(d['alarmDefinition'])
156         return d
157
158
159 class AlarmNotificationEventEnum(str, Enum):
160     NEW = 0
161     CHANGE = 1
162     CLEAR = 2
163     ACKNOWLEDGE = 3
164
165
166 class AlarmEvent2SMO(Serializer):
167     def __init__(self, eventtype: AlarmNotificationEventEnum,
168                  id: str, ref: str, updatetime: str) -> None:
169         self.notificationEventType = eventtype
170         self.objectRef = ref
171         self.id = id
172         self.updatetime = updatetime
173
174
175 class AlarmSubscription(AgRoot, Serializer):
176     def __init__(self, id: str, callback: str, consumersubid: str = '',
177                  filter: str = '') -> None:
178         super().__init__()
179         self.alarmSubscriptionId = id
180         self.version_number = 0
181         self.callback = callback
182         self.consumerSubscriptionId = consumersubid
183         self.filter = filter
184
185
186 class AlarmEventNotification(AgRoot, Serializer):
187     def __init__(self, alarm: AlarmEventRecord, to_smo: AlarmEvent2SMO,
188                  consumersubid: str) -> None:
189         super().__init__()
190         self.globalCloudId = ''
191         self.consumerSubscriptionId = consumersubid
192         self._convert_params(alarm, to_smo)
193
194     def _convert_params(self, alarm: AlarmEventRecord, to_smo: AlarmEvent2SMO):
195         self.notificationEventType = to_smo.notificationEventType
196         self.objectRef = to_smo.objectRef
197
198         self.alarmEventRecordId = alarm.alarmEventRecordId
199         self.resourceTypeId = alarm.resourceTypeId
200         self.resourceId = alarm.resourceId
201         self.alarmDefinitionId = alarm.alarmDefinitionId
202         self.probableCauseId = alarm.probableCauseId
203         self.perceivedSeverity = alarm.perceivedSeverity
204         self.alarmRaisedTime = alarm.alarmRaisedTime
205         self.alarmChangedTime = alarm.alarmChangedTime
206         self.alarmAcknowledgeTime = alarm.alarmAcknowledgeTime
207         self.alarmAcknowledged = alarm.alarmAcknowledged
208         self.extensions = []