Cleaner CLI
Change-Id: I81916ee10450e6f2137a3ff72a693dcf521c8a85
diff --git a/cli/main.py b/cli/main.py
index e9e881a..7b4b241 100755
--- a/cli/main.py
+++ b/cli/main.py
@@ -14,10 +14,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
+
import readline
-import sys
from optparse import make_option
-from time import sleep
+from time import sleep, time
import grpc
import requests
@@ -27,21 +27,21 @@
from cli.device import DeviceCli
from cli.logical_device import LogicalDeviceCli
+from cli.table import TablePrinter, print_pb_table
from voltha.core.flow_decomposer import *
from voltha.protos import third_party
from voltha.protos import voltha_pb2
from voltha.protos.openflow_13_pb2 import FlowTableUpdate
_ = third_party
-from cli.utils import pb2dict
-
+from cli.utils import pb2dict, dict2line
banner = """\
- _ _ _ _ _
- __ _____| | |_| |_ __ _ __| (_)
- \ V / _ \ | _| ' \/ _` | / _| | |
- \_/\___/_|\__|_||_\__,_| \__|_|_|
-(to exit type q, exit or quit or hit Ctrl-D)
+ _ _ _ _ _
+__ _____| | |_| |_ __ _ __| (_)
+\ V / _ \ | _| ' \/ _` | / _| | |
+ \_/\___/_|\__|_||_\__,_| \__|_|_|
+(to exit type quit or hit Ctrl-D)
"""
class VolthaCli(Cmd):
@@ -67,11 +67,30 @@
'is specified',
))
+ # cleanup of superflous commands from cmd2
+ del Cmd.do_cmdenvironment
+ # del Cmd.do_eof
+ del Cmd.do_exit
+ del Cmd.do_q
+ del Cmd.do_hi
+ del Cmd.do_l
+ del Cmd.do_li
+ del Cmd.do_r
+ del Cmd.do__load
+ del Cmd.do__relative_load
+ Cmd.do_edit = Cmd.do_ed
+
+
def __init__(self, *args, **kw):
+
Cmd.__init__(self, *args, **kw)
self.prompt = '(' + self.colorize(
- self.colorize(self.prompt, 'red'), 'bold') + ') '
+ self.colorize(self.prompt, 'blue'), 'bold') + ') '
self.channel = None
+ self.device_ids_cache = None
+ self.device_ids_cache_ts = time()
+ self.logical_device_ids_cache = None
+ self.logical_device_ids_cache_ts = time()
def load_history(self):
"""Load saved command history from local history file"""
@@ -86,81 +105,144 @@
def save_history(self):
try:
- with file(self.history_file_name, 'w') as f:
+ with open(self.history_file_name, 'w') as f:
f.write('\n'.join(self.history[-self.max_history_lines:]))
- except IOError, e:
- print >> sys.stderr, 'Could not save history in {}: {}'.format(
- self.history_file_name, e.msg)
+ except IOError as e:
+ self.perror('Could not save history in {}: {}'.format(
+ self.history_file_name, e))
else:
- print >> sys.stderr, 'History saved as {}'.format(
- self.history_file_name)
+ self.perror('History saved as {}'.format(
+ self.history_file_name))
+
+ def perror(self, errmsg, statement=None):
+ # Touch it up to make sure error is prefixed and colored
+ Cmd.perror(self, self.colorize('***ERROR: ', 'red') + errmsg,
+ statement)
def get_channel(self):
if self.channel is None:
self.channel = grpc.insecure_channel(self.voltha_grpc)
return self.channel
- def preloop(self):
- self.poutput(banner)
+ # ~~~~~~~~~~~~~~~~~ ACTUAL COMMAND IMPLEMENTATIONS ~~~~~~~~~~~~~~~~~~~~~~~~
- def do_reset_history(self, arg):
+ def do_reset_history(self, line):
"""Reset CLI history"""
while self.history:
self.history.pop()
- def do_launch(self, arg):
+ def do_launch(self, line):
"""If Voltha is not running yet, launch it"""
- pass
+ raise NotImplementedError('not implemented yet')
- def do_restart(self, arg):
+ def do_restart(self, line):
"""Launch Voltha, but if it is already running, terminate it first"""
pass
- def do_devices(self, arg):
- """List devices registered in Voltha"""
+ def do_adapters(self, line):
+ """List loaded adapter"""
+ stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
+ res = stub.ListAdapters(Empty())
+ omit_fields = {}
+ print_pb_table('Adapters:', res.items, omit_fields, self.poutput)
+
+ def get_devices(self):
stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
res = stub.ListDevices(Empty())
- for device in res.items:
- print self.colorize('# ====== device {}'.format(device.id), 'blue')
- print dumps(pb2dict(device), indent=4, sort_keys=True)
+ return res.items
- def do_logical_devices(self, arg):
+ def get_logical_devices(self):
+ stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
+ res = stub.ListLogicalDevices(Empty())
+ return res.items
+
+ def do_devices(self, line):
+ """List devices registered in Voltha"""
+ devices = self.get_devices()
+ omit_fields = {
+ 'adapter',
+ 'vendor',
+ 'model',
+ 'hardware_version',
+ 'software_version',
+ 'firmware_version',
+ 'serial_number'
+ }
+ print_pb_table('Devices:', devices, omit_fields, self.poutput)
+
+ def do_logical_devices(self, line):
"""List logical devices in Voltha"""
stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
res = stub.ListLogicalDevices(Empty())
- for logical_device in res.items:
- print self.colorize('# ====== logical device {}'.format(
- logical_device.id), 'blue')
- print dumps(pb2dict(logical_device), indent=4, sort_keys=True)
+ omit_fields = {
+ 'desc.mfr_desc',
+ 'desc.hw_desc',
+ 'desc.sw_desc',
+ 'desc.dp_desc',
+ 'desc.serial_number',
+ 'switch_features.capabilities'
+ }
+ print_pb_table('Logical devices:', res.items, omit_fields,
+ self.poutput)
- def do_device(self, arg):
+ def do_device(self, line):
"""Enter device level command mode"""
- device_id = arg or self.default_device_id
+ device_id = line.strip() or self.default_device_id
if not device_id:
raise Exception('<device-id> parameter needed')
sub = DeviceCli(self.get_channel, device_id)
sub.cmdloop()
- def do_logical_device(self, arg):
+ def do_logical_device(self, line):
"""Enter logical device level command mode"""
- logical_device_id = arg or self.default_logical_device_id
+ logical_device_id = line.strip() or self.default_logical_device_id
if not logical_device_id:
raise Exception('<logical-device-id> parameter needed')
sub = LogicalDeviceCli(self.get_channel, logical_device_id)
sub.cmdloop()
- def do_debug(self, arg):
+ def device_ids(self, force_refresh=False):
+ if force_refresh or self.device_ids is None or \
+ (time() - self.device_ids_cache_ts) > 1:
+ self.device_ids_cache = [d.id for d in self.get_devices()]
+ self.device_ids_cache_ts = time()
+ return self.device_ids_cache
+
+ def logical_device_ids(self, force_refresh=False):
+ if force_refresh or self.logical_device_ids is None or \
+ (time() - self.logical_device_ids_cache_ts) > 1:
+ self.logical_device_ids_cache = [d.id for d
+ in self.get_logical_devices()]
+ self.logical_device_ids_cache_ts = time()
+ return self.logical_device_ids_cache
+
+ def complete_device(self, text, line, begidx, endidx):
+ if not text:
+ completions = self.device_ids()[:]
+ else:
+ completions = [d for d in self.device_ids() if d.startswith(text)]
+ return completions
+
+ def complete_logical_device(self, text, line, begidx, endidx):
+ if not text:
+ completions = self.logical_device_ids()[:]
+ else:
+ completions = [d for d in self.logical_device_ids()
+ if d.startswith(text)]
+ return completions
+
+ def do_pdb(self, line):
"""Launch PDB debug prompt in CLI (for CLI development)"""
from pdb import set_trace
set_trace()
- def do_health(self, arg):
+ def do_health(self, line):
"""Show connectivity status to Voltha status"""
stub = voltha_pb2.HealthServiceStub(self.get_channel())
res = stub.GetHealthStatus(Empty())
- print dumps(pb2dict(res), indent=4)
+ self.poutput(dumps(pb2dict(res), indent=4))
- def do_test(self, arg):
+ def do_test(self, line):
"""Enter test mode, which makes a bunch on new commands available"""
sub = TestCli(self.history, self.get_channel)
sub.cmdloop()
@@ -182,7 +264,7 @@
default='00:0c:e2:31:40:00'),
make_option('-i', '--ip-address', action='store', dest='ip_address'),
])
- def do_preprovision_olt(self, arg, opts):
+ def do_preprovision_olt(self, line, opts):
"""Preprovision a new OLT with given device type"""
stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
kw = dict(type=opts.device_type)
@@ -194,16 +276,16 @@
raise Exception('Either IP address or Mac Address is needed')
device = voltha_pb2.Device(**kw)
device = stub.CreateDevice(device)
- print 'success (device id = {})'.format(device.id)
+ self.poutput('success (device id = {})'.format(device.id))
self.default_device_id = device.id
- def do_activate_olt(self, arg):
+ def do_activate_olt(self, line):
"""
Activate an OLT. If the <id> is not provided, it will be on the last
pre-provisioned OLT.
"""
- device_id = arg or self.default_device_id
- print 'activating', device_id
+ device_id = line or self.default_device_id
+ self.poutput('activating {}'.format(device_id))
stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
stub.ActivateDevice(voltha_pb2.ID(id=device_id))
@@ -214,32 +296,36 @@
assert device.parent_id
self.default_logical_device_id = device.parent_id
break
- print 'waiting for device to be activated...'
- sleep(1)
- print 'success (logical device id = {})'.format(
- self.default_logical_device_id)
+ self.poutput('waiting for device to be activated...')
+ sleep(.5)
+ self.poutput('success (logical device id = {})'.format(
+ self.default_logical_device_id))
- def do_arrive_onus(self, arg):
+ complete_activate_olt = VolthaCli.complete_device
+
+ def do_arrive_onus(self, line):
"""
Simulate the arrival of ONUs
"""
- device_id = arg or self.default_device_id
+ device_id = line or self.default_device_id
requests.get('http://{}/devices/{}/detect_onus'.format(
self.voltha_sim_rest, device_id
))
- def do_install_eapol_flow(self, arg):
+ complete_arrive_onus = VolthaCli.complete_device
+
+ def do_install_eapol_flow(self, line):
"""
Install an EAPOL flow on the given logical device. If device is not
given, it will be applied to logical device of the last pre-provisioned
OLT device.
"""
- logical_device_id = arg or self.default_logical_device_id
+ logical_device_id = line or self.default_logical_device_id
update = FlowTableUpdate(
id=logical_device_id,
flow_mod = mk_simple_flow_mod(
priority=2000,
- match_fields=[in_port(101), eth_type(0x888e)],
+ match_fields=[in_port(241), eth_type(0x888e)],
actions=[
push_vlan(0x8100),
set_field(vlan_vid(4096 + 4000)),
@@ -249,18 +335,22 @@
)
stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
res = stub.UpdateLogicalDeviceFlowTable(update)
- print 'success', res
+ self.poutput('success ({})'.format(res))
- def do_send_simulated_upstream_eapol(self, arg):
+ complete_install_eapol_flow = VolthaCli.complete_logical_device
+
+ def do_send_simulated_upstream_eapol(self, line):
"""
Send an EAPOL upstream from a simulated OLT
"""
- device_id = arg or self.default_device_id
+ device_id = line or self.default_device_id
requests.get('http://{}/devices/{}/test_eapol_in'.format(
self.voltha_sim_rest, device_id
))
- def do_inject_eapol_start(self, arg):
+ complete_send_simulated_upstream_eapol = VolthaCli.complete_device
+
+ def do_inject_eapol_start(self, line):
"""
Send out an an EAPOL start message into the given Unix interface
"""
@@ -269,6 +359,7 @@
if __name__ == '__main__':
c = VolthaCli()
+ c.poutput(banner)
c.load_history()
c.cmdloop()
c.save_history()