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