X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=o2ims%2Fadapter%2Fclients%2Ffault_client.py;h=53602b44ddb066487dbbfed2f27964a4b9f3ed2c;hb=6841e2e68f72068374d333723f5b5275bfb98933;hp=c9e1e95cf531f941009942200fda39c388adb7c4;hpb=e00040ca7c6149995c5883bcd9dd81e02b25d0da;p=pti%2Fo2.git diff --git a/o2ims/adapter/clients/fault_client.py b/o2ims/adapter/clients/fault_client.py index c9e1e95..53602b4 100644 --- a/o2ims/adapter/clients/fault_client.py +++ b/o2ims/adapter/clients/fault_client.py @@ -1,4 +1,4 @@ -# Copyright (C) 2022-2024 Wind River Systems, Inc. +# Copyright (C) 2022 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. @@ -94,10 +94,9 @@ class StxAlarmClient(BaseClient): class StxEventClient(BaseClient): - def __init__(self, uow: unit_of_work.AbstractUnitOfWork, driver=None): + def __init__(self, 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) @@ -108,12 +107,6 @@ 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): @@ -226,27 +219,6 @@ 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 = [] - queryAsArray = [] - events = self.fmclient.event_suppression.list(q=queryAsArray) - 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 = [ @@ -292,27 +264,6 @@ 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