X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=o2ims%2Fadapter%2Fclients%2Ffault_client.py;h=74abe57c7c3c5b8e9e4e1464ded1ba0826eb6682;hb=54ca64640ce149364fcfa91666c01edfe5c02e72;hp=989edfc693237e05627fb3b4715dcdc8faba6c98;hpb=0d104e73ad08103d1c983c25841ada15acf7dadb;p=pti%2Fo2.git

diff --git a/o2ims/adapter/clients/fault_client.py b/o2ims/adapter/clients/fault_client.py
index 989edfc..74abe57 100644
--- a/o2ims/adapter/clients/fault_client.py
+++ b/o2ims/adapter/clients/fault_client.py
@@ -1,4 +1,4 @@
-# Copyright (C) 2022 Wind River Systems, Inc.
+# Copyright (C) 2022-2024 Wind River Systems, Inc.
 #
 #  Licensed under the Apache License, Version 2.0 (the "License");
 #  you may not use this file except in compliance with the License.
@@ -97,11 +97,15 @@ class StxAlarmClient(BaseClient):
     def _set_stx_client(self):
         self.driver.setFaultClient(self._pool_id)
 
+    def delete(self, id) -> alarmModel.FaultGenericModel:
+        return self.driver.deleteAlarm(id)
+
 
 class StxEventClient(BaseClient):
-    def __init__(self, driver=None):
+    def __init__(self, uow: unit_of_work.AbstractUnitOfWork, driver=None):
         super().__init__()
         self.driver = driver if driver else StxFaultClientImp()
+        self.uow = uow
 
     def _get(self, id) -> alarmModel.FaultGenericModel:
         return self.driver.getEventInfo(id)
@@ -112,6 +116,12 @@ class StxEventClient(BaseClient):
     def _set_stx_client(self):
         self.driver.setFaultClient(self._pool_id)
 
+    def suppression_list(self, alarm_id) -> List[alarmModel.FaultGenericModel]:
+        return self.driver.getSuppressionList(alarm_id)
+
+    def suppress(self, id) -> alarmModel.FaultGenericModel:
+        return self.driver.suppressEvent(id)
+
 
 # internal driver which implement client call to Stx Fault Management instance
 class StxFaultClientImp(object):
@@ -212,6 +222,12 @@ class StxFaultClientImp(object):
         return alarmModel.FaultGenericModel(
             alarmModel.EventTypeEnum.ALARM, self._alarmconverter(alarm))
 
+    def deleteAlarm(self, id) -> alarmModel.FaultGenericModel:
+        alarm = self.fmclient.alarm.delete(id)
+        logger.debug('delete alarm id ' + id + ':' + str(alarm.to_dict()))
+        return alarmModel.FaultGenericModel(
+            alarmModel.EventTypeEnum.ALARM, self._alarmconverter(alarm))
+
     def getEventList(self, **filters) -> List[alarmModel.FaultGenericModel]:
         events = self.fmclient.event_log.list(alarms=True, expand=True)
         [logger.debug(
@@ -226,6 +242,27 @@ class StxFaultClientImp(object):
         return alarmModel.FaultGenericModel(
             alarmModel.EventTypeEnum.EVENT, self._eventconverter(event))
 
+    def suppressEvent(self, id) -> alarmModel.FaultGenericModel:
+        patch = [dict(path='/' + 'suppression_status', value='suppressed',
+                      op='replace')]
+        event = self.fmclient.event_suppression.update(id, patch)
+        logger.debug('suppressed event id ' + id + ':' + str(event.to_dict()))
+        return alarmModel.FaultGenericModel(
+            alarmModel.EventTypeEnum.EVENT, self._suppression_converter(event))
+
+    def getSuppressionList(self, alarm_id) -> alarmModel.FaultGenericModel:
+        suppression_list = []
+        query_as_array = []
+        events = self.fmclient.event_suppression.list(q=query_as_array)
+        for event in events:
+            if event.alarm_id == alarm_id:
+                # logger.debug('suppression event:' + str(event.to_dict()))
+                suppression_list.append(
+                    alarmModel.FaultGenericModel(
+                        alarmModel.EventTypeEnum.EVENT,
+                        self._suppression_converter(event)))
+        return suppression_list
+
     @ staticmethod
     def _alarmconverter(alarm):
         selected_keys = [
@@ -271,6 +308,27 @@ class StxFaultClientImp(object):
                 uuid.NAMESPACE_URL, event.probable_cause)))
         return event
 
+    @ staticmethod
+    def _suppression_converter(event, clear=False):
+        selected_keys = [
+            'alarm_id', 'description', 'suppression_status',
+            'links'
+        ]
+        content = event.to_dict()
+        filtered = dict(
+            filter(lambda item: item[0] in selected_keys, content.items()))
+        setattr(event, 'filtered', filtered)
+        setattr(event, 'uuid', event.uuid)
+        setattr(event, 'alarm_id', event.alarm_id)
+        setattr(event, 'description', event.description)
+        setattr(event, 'suppression_status', event.suppression_status)
+        setattr(event, 'alarm_type', None)
+        setattr(event, 'alarm_def_id', None)
+        setattr(event, 'probable_cause_id', None)
+        setattr(event, 'state', None)
+        setattr(event, 'timestamp', None)
+        return event
+
     @ staticmethod
     def _alarmeventhasher(event, state=''):
         # The event model and the alarm model have different parameter name