| William Kurkian | 6f436d0 | 2019-02-06 16:25:01 -0500 | [diff] [blame^] | 1 | # |
| 2 | # Copyright 2017 the original author or authors. |
| 3 | # |
| 4 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | # you may not use this file except in compliance with the License. |
| 6 | # You may obtain a copy of the License at |
| 7 | # |
| 8 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | # |
| 10 | # Unless required by applicable law or agreed to in writing, software |
| 11 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | # See the License for the specific language governing permissions and |
| 14 | # limitations under the License. |
| 15 | |
| 16 | from voltha.extensions.alarms.olt.olt_los_alarm import OltLosAlarm |
| 17 | from voltha.extensions.alarms.onu.onu_dying_gasp_alarm import OnuDyingGaspAlarm |
| 18 | from voltha.extensions.alarms.onu.onu_los_alarm import OnuLosAlarm |
| 19 | from voltha.extensions.alarms.onu.onu_lopc_miss_alarm import OnuLopcMissAlarm |
| 20 | from voltha.extensions.alarms.onu.onu_lopc_mic_error_alarm import OnuLopcMicErrorAlarm |
| 21 | from voltha.extensions.alarms.onu.onu_lob_alarm import OnuLobAlarm |
| 22 | |
| 23 | from voltha.extensions.alarms.onu.onu_startup_alarm import OnuStartupAlarm |
| 24 | from voltha.extensions.alarms.onu.onu_signal_degrade_alarm import OnuSignalDegradeAlarm |
| 25 | from voltha.extensions.alarms.onu.onu_signal_fail_alarm import OnuSignalFailAlarm |
| 26 | from voltha.extensions.alarms.onu.onu_window_drift_alarm import OnuWindowDriftAlarm |
| 27 | from voltha.extensions.alarms.onu.onu_activation_fail_alarm import OnuActivationFailAlarm |
| 28 | |
| 29 | from voltha.extensions.alarms.onu.onu_discovery_alarm import OnuDiscoveryAlarm |
| 30 | |
| 31 | class AdapterAlarmSimulator(object): |
| 32 | def __init__(self, adapter_alarms): |
| 33 | self.adapter_alarms = adapter_alarms |
| 34 | |
| 35 | def simulate_alarm(self, alarm): |
| 36 | if alarm.indicator == "los": |
| 37 | alarm_obj = OltLosAlarm(self.adapter_alarms, intf_id=alarm.intf_id, port_type_name=alarm.port_type_name) |
| 38 | elif alarm.indicator == "dying_gasp": |
| 39 | alarm_obj = OnuDyingGaspAlarm(self.adapter_alarms, onu_id=alarm.onu_device_id, intf_id=alarm.intf_id) |
| 40 | elif alarm.indicator == "onu_los": |
| 41 | alarm_obj = OnuLosAlarm(self.adapter_alarms, onu_id=alarm.onu_device_id, intf_id=alarm.intf_id) |
| 42 | elif alarm.indicator == "onu_lopc_miss": |
| 43 | alarm_obj = OnuLopcMissAlarm(self.adapter_alarms, onu_id=alarm.onu_device_id, intf_id=alarm.intf_id) |
| 44 | elif alarm.indicator == "onu_lopc_mic": |
| 45 | alarm_obj = OnuLopcMicErrorAlarm(self.adapter_alarms, onu_id=alarm.onu_device_id, intf_id=alarm.intf_id) |
| 46 | elif alarm.indicator == "onu_lob": |
| 47 | alarm_obj = OnuLobAlarm(self.adapter_alarms, onu_id=alarm.onu_device_id, intf_id=alarm.intf_id) |
| 48 | elif alarm.indicator == "onu_startup": |
| 49 | alarm_obj = OnuStartupAlarm(self.adapter_alarms, intf_id=alarm.intf_id, onu_id=alarm.onu_device_id) |
| 50 | elif alarm.indicator == "onu_signal_degrade": |
| 51 | alarm_obj = OnuSignalDegradeAlarm(self.adapter_alarms, intf_id=alarm.intf_id, onu_id=alarm.onu_device_id, |
| 52 | inverse_bit_error_rate=alarm.inverse_bit_error_rate) |
| 53 | elif alarm.indicator == "onu_drift_of_window": |
| 54 | alarm_obj = OnuWindowDriftAlarm(self.adapter_alarms, intf_id=alarm.intf_id, |
| 55 | onu_id=alarm.onu_device_id, |
| 56 | drift=alarm.drift, |
| 57 | new_eqd=alarm.new_eqd) |
| 58 | elif alarm.indicator == "onu_signal_fail": |
| 59 | alarm_obj = OnuSignalFailAlarm(self.adapter_alarms, intf_id=alarm.intf_id, |
| 60 | onu_id=alarm.onu_device_id, |
| 61 | inverse_bit_error_rate=alarm.inverse_bit_error_rate) |
| 62 | elif alarm.indicator == "onu_activation": |
| 63 | alarm_obj = OnuActivationFailAlarm(self.adapter_alarms, intf_id=alarm.intf_id, |
| 64 | onu_id=alarm.onu_device_id) |
| 65 | elif alarm.indicator == "onu_discovery": |
| 66 | alarm_obj = OnuDiscoveryAlarm(self.adapter_alarms, pon_id=alarm.intf_id, |
| 67 | serial_number=alarm.onu_serial_number) |
| 68 | else: |
| 69 | raise Exception("Unknown alarm indicator %s" % alarm.indicator) |
| 70 | |
| 71 | if alarm.operation == alarm.RAISE: |
| 72 | alarm_obj.raise_alarm() |
| 73 | elif alarm.operation == alarm.CLEAR: |
| 74 | alarm_obj.clear_alarm() |
| 75 | else: |
| 76 | # This shouldn't happen |
| 77 | raise Exception("Unknown alarm operation") |