VOL-1556 Add alarm simulation to voltha-go core
Change-Id: I23dcd720909a3e23cb203fd1ae32eada5fc4e34e
diff --git a/python/cli/device.py b/python/cli/device.py
old mode 100644
new mode 100755
index 484185f..25ad72b
--- a/python/cli/device.py
+++ b/python/cli/device.py
@@ -508,6 +508,116 @@
self.poutput('response: {}'.format(name))
self.poutput('{}'.format(response))
+ def help_simulate_alarm(self):
+ self.poutput(
+'''
+simulate_alarm <alarm_name> [-b <bit rate>] [-c] [-d <drift>] [-e <eqd>]
+ [-i <interface id>] [-o <onu device id>] [-p <port type name>]
+
+<name> is the name of the alarm to raise. Other rguments are alarm specific
+and only have meaning in the context of a particular alarm. Below is a list
+of the alarms that may be raised:
+
+simulate_alarm los -i <interface_id> -p <port_type_name>
+simulate_alarm dying_gasp -i <interface_id> -o <onu_device_id>
+simulate_alarm onu_los -i <interface_id> -o <onu_device_id>
+simulate_alarm onu_lopc_miss -i <interface_id> -o <onu_device_id>
+simulate_alarm onu_lopc_mic -i <interface_id> -o <onu_device_id>
+simulate_alarm onu_lob -i <interface_id> -o <onu_device_id>
+simulate_alarm onu_signal_degrade -i <interface_id> -o <onu_device_id>
+ -b <bit_rate>
+simulate_alarm onu_drift_of_window -i <interface_id>
+ -o <onu_device_id> -d <drift> -e <eqd>
+simulate_alarm onu_signal_fail -i <interface_id> -o <onu_device_id>
+ -b <bit_rate>
+simulate_alarm onu_activation -i <interface_id> -o <onu_device_id>
+simulate_alarm onu_startup -i <interface_id> -o <onu_device_id>
+simulate_alarm onu_discovery -i <interface_id> -s <onu_serial_number>
+
+If the -c option is specified then the alarm will be cleared. By default,
+it will be raised. Note that only some alarms can be cleared.
+'''
+ )
+
+ @options([
+ make_option('-c', '--clear', action='store_true', default=False,
+ help="Clear alarm instead of raising"),
+ make_option('-b', '--inverse_bit_error_rate', action='store', dest='inverse_bit_error_rate',
+ help="Inverse bit error rate", default=0, type="int"),
+ make_option('-d', '--drift', action='store', dest='drift',
+ help="Drift", default=0, type="int"),
+ make_option('-e', '--new_eqd', action='store', dest='new_eqd',
+ help="New EQD", default=0, type="int"),
+ make_option('-i', '--intf_id', action='store', dest='intf_id',
+ help="Interface ID", default=""),
+ make_option('-o', '--onu_device_id', action='store', dest='onu_device_id',
+ help="ONU device ID", default=""),
+ make_option('-p', '--port_type_name', action='store', dest='port_type_name',
+ help="Port type name", default=""),
+ make_option('-s', '--onu_serial_number', action='store', dest='onu_serial_number',
+ help="ONU Serial Number", default=""),
+ ])
+ def do_simulate_alarm(self, line, opts):
+ indicator = line
+ device = self.get_device(depth=-1)
+ device_id = device.id
+
+ alarm_args = {"los": ["intf_id", "port_type_name"],
+ "dying_gasp": ["intf_id", "onu_device_id"],
+ "onu_los": ["intf_id", "onu_device_id"],
+ "onu_lopc_miss": ["intf_id", "onu_device_id"],
+ "onu_lopc_mic": ["intf_id", "onu_device_id"],
+ "onu_lob": ["intf_id", "onu_device_id"],
+ "onu_signal_degrade": ["intf_id", "onu_device_id", "inverse_bit_error_rate"],
+ "onu_drift_of_window": ["intf_id", "onu_device_id", "drift", "new_eqd"],
+ "onu_signal_fail": ["intf_id", "onu_device_id", "inverse_bit_error_rate"],
+ "onu_activation": ["intf_id", "onu_device_id"],
+ "onu_startup": ["intf_id", "onu_device_id"],
+ "onu_discovery": ["intf_id", "onu_serial_number"]
+ }
+ try:
+ if indicator not in alarm_args:
+ self.poutput("Unknown alarm indicator %s. Valid choices are %s." % (indicator,
+ ", ".join(alarm_args.keys())))
+ raise Exception("Unknown alarm indicator %s" % indicator)
+
+ for arg_name in alarm_args[indicator]:
+ if not getattr(opts, arg_name):
+ self.poutput("Option %s is required for alarm %s. See help." % (arg_name, indicator))
+ raise Exception("Option %s is required for alarm %s" % (arg_name, indicator))
+
+ # TODO: check for required arguments
+ kw = dict(id=device_id)
+
+ kw["indicator"] = indicator
+ kw["intf_id"] = opts.intf_id
+ kw["onu_device_id"] = opts.onu_device_id
+ kw["port_type_name"] = opts.port_type_name
+ kw["inverse_bit_error_rate"] = opts.inverse_bit_error_rate
+ kw["drift"] = opts.drift
+ kw["new_eqd"] = opts.new_eqd
+ kw["onu_serial_number"] = opts.onu_serial_number
+
+ if opts.clear:
+ kw["operation"] = voltha_pb2.SimulateAlarmRequest.CLEAR
+ else:
+ kw["operation"] = voltha_pb2.SimulateAlarmRequest.RAISE
+ except Exception as e:
+ self.poutput('Error simulate alarm {}. Error:{}'.format(device_id, e))
+ return
+ response = None
+ try:
+ simulate_alarm = voltha_pb2.SimulateAlarmRequest(**kw)
+ stub = self.get_stub()
+ response = stub.SimulateAlarm(simulate_alarm)
+ except Exception as e:
+ self.poutput('Error simulate alarm {}. Error:{}'.format(kw['id'], e))
+ return
+ name = enum2name(common_pb2.OperationResp,
+ 'OperationReturnCode', response.code)
+ self.poutput('response: {}'.format(name))
+ self.poutput('{}'.format(response))
+
@options([
make_option('-n', '--name', action='store', dest='name',
help="Image name"),