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.
15 from __future__ import annotations
20 from o2common.domain.base import AgRoot, Serializer
22 from o2common.helper import o2logging
23 logger = o2logging.get_logger(__name__)
26 class FaultGenericModel(AgRoot):
27 def __init__(self, type: str,
28 api_response: dict = None, content_hash=None) -> None:
31 self.id = str(api_response.uuid)
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
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
48 self.hash = content_hash
50 if hasattr(api_response, 'filtered'):
51 self.filtered = api_response.filtered
52 self.hash = str(hash((self.id, str(self.filtered))))
54 self.hash = str(hash((self.id, self.updatetime)))
55 self.content = json.dumps(api_response.to_dict())
56 if EventTypeEnum.ALARM == type:
59 def is_outdated(self, newmodel) -> bool:
60 # return self.updatetime < newmodel.updatetime
61 # logger.warning("hash1: " + self.hash + " vs hash2: " + newmodel.hash)
62 return self.hash != newmodel.hash
64 def update_by(self, newmodel) -> None:
65 if self.id != newmodel.id:
67 # raise MismatchedModel("Mismatched model")
68 self.name = newmodel.name
69 self.createtime = newmodel.createtime
70 self.updatetime = newmodel.updatetime
71 self.content = newmodel.content
74 class EventTypeEnum(Enum):
79 class PerceivedSeverityEnum(str, Enum):
88 class AlarmEventRecord(AgRoot, Serializer):
89 def __init__(self, id: str, res_type_id: str, res_id: str,
90 alarm_def_id: str, probable_cause_id: str,
92 perc_severity: PerceivedSeverityEnum =
93 PerceivedSeverityEnum.WARNING
96 self.alarmEventRecordId = id
97 self.resourceTypeId = res_type_id
98 self.resourceId = res_id
99 self.alarmDefinitionId = alarm_def_id
100 self.probableCauseId = probable_cause_id
101 self.perceivedSeverity = perc_severity
102 self.alarmRaisedTime = raised_time
103 self.alarmChangedTime = ''
104 self.alarmAcknowledgeTime = ''
105 self.alarmAcknowledged = False
109 class ProbableCause(AgRoot, Serializer):
110 def __init__(self, id: str, name: str, desc: str = '') -> None:
112 self.probableCauseId = id
114 self.description = desc
117 class AlarmChangeTypeEnum(str, Enum):
120 MODIFYED = 'MODIFYED'
123 class ClearingTypeEnum(str, Enum):
124 AUTOMATIC = 'AUTOMATIC'
128 class AlarmEventRecordModifications(AgRoot):
129 def __init__(self, ack: bool = None,
130 clear: PerceivedSeverityEnum = None) -> None:
132 self.alarmAcknowledged = ack
133 self.perceivedSeverity = clear
136 class AlarmDefinition(AgRoot, Serializer):
137 def __init__(self, id: str, name: str, change_type: AlarmChangeTypeEnum,
138 desc: str, prop_action: str, clearing_type: ClearingTypeEnum,
139 pk_noti_field: str) -> None:
141 self.alarmDefinitionId = id
142 self.alarmName = name
143 self.alarmLastChange = '0.1'
144 self.alarmChangeType = change_type
145 self.alarmDescription = desc
146 self.proposedRepairActions = prop_action
147 self.clearingType = clearing_type
148 self.managementInterfaceId = "O2IMS"
149 self.pkNotificationField = pk_noti_field
150 self.alarmAdditionalFields = ""
153 class AlarmDictionary(AgRoot, Serializer):
154 def __init__(self, id: str) -> None:
157 self.alarmDictionaryVersion = ""
158 self.alarmDictionarySchemaVersion = ""
161 self.managementInterfaceId = "O2IMS"
162 self.pkNotificationField = ""
163 self.alarmDefinition = []
166 d = Serializer.serialize(self)
167 if 'alarmDefinition' in d and len(d['alarmDefinition']) > 0:
168 d['alarmDefinition'] = self.serialize_list(d['alarmDefinition'])
172 class AlarmNotificationEventEnum(str, Enum):
179 class AlarmEvent2SMO(Serializer):
180 def __init__(self, eventtype: AlarmNotificationEventEnum,
181 id: str, updatetime: str) -> None:
182 self.notificationEventType = eventtype
184 self.updatetime = updatetime
187 class AlarmSubscription(AgRoot, Serializer):
188 def __init__(self, id: str, callback: str, consumersubid: str = '',
189 filter: str = '') -> None:
191 self.alarmSubscriptionId = id
192 self.version_number = 0
193 self.callback = callback
194 self.consumerSubscriptionId = consumersubid
198 class AlarmEventNotification(AgRoot, Serializer):
199 def __init__(self, alarm: AlarmEventRecord, to_smo: AlarmEvent2SMO,
200 consumersubid: str) -> None:
202 self.globalCloudId = ''
203 self.consumerSubscriptionId = consumersubid
204 self._convert_params(alarm, to_smo)
206 def _convert_params(self, alarm: AlarmEventRecord, to_smo: AlarmEvent2SMO):
207 self.notificationEventType = to_smo.notificationEventType
208 self.objectRef = to_smo.objectRef
210 self.alarmEventRecordId = alarm.alarmEventRecordId
211 self.resourceTypeId = alarm.resourceTypeId
212 self.resourceId = alarm.resourceId
213 self.alarmDefinitionId = alarm.alarmDefinitionId
214 self.probableCauseId = alarm.probableCauseId
215 self.perceivedSeverity = alarm.perceivedSeverity
216 self.alarmRaisedTime = alarm.alarmRaisedTime
217 self.alarmChangedTime = alarm.alarmChangedTime
218 self.alarmAcknowledgeTime = alarm.alarmAcknowledgeTime
219 self.alarmAcknowledged = alarm.alarmAcknowledged
220 self.extensions = alarm.extensions