From feb152b7d5eb531d439a4950d27cbc44031daf92 Mon Sep 17 00:00:00 2001 From: imarom Date: Tue, 8 Dec 2015 07:43:24 -0500 Subject: text based TUI (does not use ncurses) --- .../client/trex_stateless_client.py | 8 +- .../trex_control_plane/console/old_console.py | 958 --------------------- .../trex_control_plane/console/trex_console.py | 6 +- .../trex_control_plane/console/trex_tui.py | 309 +++++++ 4 files changed, 319 insertions(+), 962 deletions(-) delete mode 100644 scripts/automation/trex_control_plane/console/old_console.py create mode 100644 scripts/automation/trex_control_plane/console/trex_tui.py (limited to 'scripts/automation/trex_control_plane') diff --git a/scripts/automation/trex_control_plane/client/trex_stateless_client.py b/scripts/automation/trex_control_plane/client/trex_stateless_client.py index 0c6d9855..0a583f46 100755 --- a/scripts/automation/trex_control_plane/client/trex_stateless_client.py +++ b/scripts/automation/trex_control_plane/client/trex_stateless_client.py @@ -321,6 +321,8 @@ class Port(object): self.streams[stream_id] = None + self.state = self.STATE_STREAMS if len(self.streams > 0) else self.STATE_IDLE + return self.ok() # remove all the streams @@ -335,6 +337,8 @@ class Port(object): self.streams = {} + self.state = self.STATE_IDLE + return self.ok() # get a specific stream @@ -1364,12 +1368,13 @@ class CTRexStatelessClient(object): # set to show all stats if no filter was given mask = trex_stats.ALL_STATS_OPTS - # get stats objects, as dictionary stats = self.cmd_stats(opts.ports, mask) + # print stats to screen for stat_type, stat_data in stats.iteritems(): text_tables.print_table_with_header(stat_data.text_table, stat_type) + return RC_OK() @@ -1481,6 +1486,7 @@ class CTRexStatelessClient(object): def _filter_namespace_args(namespace, ok_values): return {k: v for k, v in namespace.__dict__.items() if k in ok_values} + ################################# # ------ private classes ------ # class CCommLink(object): diff --git a/scripts/automation/trex_control_plane/console/old_console.py b/scripts/automation/trex_control_plane/console/old_console.py deleted file mode 100644 index 9d61a3a6..00000000 --- a/scripts/automation/trex_control_plane/console/old_console.py +++ /dev/null @@ -1,958 +0,0 @@ - -# main console object -class TRexConsole1(cmd.Cmd): - """Trex Console""" - - def __init__(self, stateless_client, verbose): - cmd.Cmd.__init__(self) - - self.stateless_client = stateless_client - - self.do_connect("") - - self.intro = "\n-=TRex Console v{ver}=-\n".format(ver=__version__) - self.intro += "\nType 'help' or '?' for supported actions\n" - - self.verbose = False - self._silent = True - - self.postcmd(False, "") - - self.user_streams = {} - self.streams_db = CStreamsDB() - - - # a cool hack - i stole this function and added space - def completenames(self, text, *ignored): - dotext = 'do_'+text - return [a[3:]+' ' for a in self.get_names() if a.startswith(dotext)] - - - # set verbose on / off - def do_verbose(self, line): - '''Shows or set verbose mode\n''' - if line == "": - print "\nverbose is " + ("on\n" if self.verbose else "off\n") - - elif line == "on": - self.verbose = True - self.stateless_client.set_verbose(True) - print green("\nverbose set to on\n") - - elif line == "off": - self.verbose = False - self.stateless_client.set_verbose(False) - print green("\nverbose set to off\n") - - else: - print magenta("\nplease specify 'on' or 'off'\n") - - # query the server for registered commands - def do_query_server(self, line): - '''query the RPC server for supported remote commands\n''' - - res_ok, msg = self.stateless_client.get_supported_cmds() - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - return - print "\nRPC server supports the following commands:\n" - for func in msg: - if func: - print func - print '' - print format_text("[SUCCESS]\n", 'green', 'bold') - return - - def do_ping(self, line): - '''Pings the RPC server\n''' - - print "\n-> Pinging RPC server" - - res_ok, msg = self.stateless_client.ping() - if res_ok: - print format_text("[SUCCESS]\n", 'green', 'bold') - else: - print "\n*** " + msg + "\n" - return - - def do_force_acquire(self, line): - '''Acquires ports by force\n''' - - self.do_acquire(line, True) - - def complete_force_acquire(self, text, line, begidx, endidx): - return self.port_auto_complete(text, line, begidx, endidx, acquired=False) - - def extract_port_ids_from_line(self, line): - return {int(x) for x in line.split()} - - def extract_port_ids_from_list(self, port_list): - return {int(x) for x in port_list} - - def parse_ports_from_line (self, line): - port_list = set() - if line: - for port_id in line.split(' '): - if (not port_id.isdigit()) or (int(port_id) < 0) or (int(port_id) >= self.stateless_client.get_port_count()): - print "Please provide a list of ports separated by spaces between 0 and {0}".format(self.stateless_client.get_port_count() - 1) - return None - - port_list.add(int(port_id)) - - port_list = list(port_list) - - else: - port_list = [i for i in xrange(0, self.stateless_client.get_port_count())] - - return port_list - - - def do_acquire(self, line, force=False): - '''Acquire ports\n''' - - # make sure that the user wants to acquire all - args = line.split() - if len(args) < 1: - print magenta("Please provide a list of ports separated by spaces, or specify 'all' to acquire all available ports") - return - - if args[0] == "all": - ask = ConfirmMenu('Are you sure you want to acquire all ports ? ') - rc = ask.show() - if rc == False: - print yellow("[ABORTED]\n") - return - else: - port_list = self.stateless_client.get_port_ids() - else: - port_list = self.extract_port_ids_from_line(line) - - # rc, resp_list = self.stateless_client.take_ownership(port_list, force) - try: - res_ok, log = self.stateless_client.acquire(port_list, force) - self.prompt_response(log) - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - return - print format_text("[SUCCESS]\n", 'green', 'bold') - except ValueError as e: - print magenta(str(e)) - print format_text("[FAILED]\n", 'red', 'bold') - - - def port_auto_complete(self, text, line, begidx, endidx, acquired=True, active=False): - if acquired: - if not active: - ret_list = [x - for x in map(str, self.stateless_client.get_acquired_ports()) - if x.startswith(text)] - else: - ret_list = [x - for x in map(str, self.stateless_client.get_active_ports()) - if x.startswith(text)] - else: - ret_list = [x - for x in map(str, self.stateless_client.get_port_ids()) - if x.startswith(text)] - ret_list.append("all") - return ret_list - - - def complete_acquire(self, text, line, begidx, endidx): - return self.port_auto_complete(text, line, begidx, endidx, acquired=False) - - def do_release (self, line): - '''Release ports\n''' - - # if line: - # port_list = self.parse_ports_from_line(line) - # else: - # port_list = self.stateless_client.get_owned_ports() - args = line.split() - if len(args) < 1: - print "Please provide a list of ports separated by spaces, or specify 'all' to acquire all available ports" - if args[0] == "all": - ask = ConfirmMenu('Are you sure you want to release all acquired ports? ') - rc = ask.show() - if rc == False: - print yellow("[ABORTED]\n") - return - else: - port_list = self.stateless_client.get_acquired_ports() - else: - port_list = self.extract_port_ids_from_line(line) - - try: - res_ok, log = self.stateless_client.release(port_list) - self.prompt_response(log) - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - return - print format_text("[SUCCESS]\n", 'green', 'bold') - except ValueError as e: - print magenta(str(e)) - print format_text("[FAILED]\n", 'red', 'bold') - return - - def complete_release(self, text, line, begidx, endidx): - return self.port_auto_complete(text, line, begidx, endidx) - - def do_connect (self, line): - '''Connects to the server\n''' - - if line == "": - res_ok, msg = self.stateless_client.connect() - else: - sp = line.split() - if (len(sp) != 2): - print "\n[usage] connect [server] [port] or without parameters\n" - return - - res_ok, msg = self.stateless_client.connect(sp[0], sp[1]) - - if res_ok: - print format_text("[SUCCESS]\n", 'green', 'bold') - else: - print "\n*** " + msg + "\n" - print format_text("[FAILED]\n", 'red', 'bold') - return - - self.supported_rpc = self.stateless_client.get_supported_cmds().data - - # def do_rpc (self, line): - # '''Launches a RPC on the server\n''' - # - # if line == "": - # print "\nUsage: [method name] [param dict as string]\n" - # print "Example: rpc test_add {'x': 12, 'y': 17}\n" - # return - # - # sp = line.split(' ', 1) - # method = sp[0] - # - # params = None - # bad_parse = False - # if len(sp) > 1: - # - # try: - # params = ast.literal_eval(sp[1]) - # if not isinstance(params, dict): - # bad_parse = True - # - # except ValueError as e1: - # bad_parse = True - # except SyntaxError as e2: - # bad_parse = True - # - # if bad_parse: - # print "\nValue should be a valid dict: '{0}'".format(sp[1]) - # print "\nUsage: [method name] [param dict as string]\n" - # print "Example: rpc test_add {'x': 12, 'y': 17}\n" - # return - # - # res_ok, msg = self.stateless_client.transmit(method, params) - # if res_ok: - # print "\nServer Response:\n\n" + pretty_json(json.dumps(msg)) + "\n" - # else: - # print "\n*** " + msg + "\n" - # #print "Please try 'reconnect' to reconnect to server" - # - # - # def complete_rpc (self, text, line, begidx, endidx): - # return [x - # for x in self.supported_rpc - # if x.startswith(text)] - - def do_status (self, line): - '''Shows a graphical console\n''' - - if not self.stateless_client.is_connected(): - print "Not connected to server\n" - return - - self.do_verbose('off') - trex_status.show_trex_status(self.stateless_client) - - def do_quit(self, line): - '''Exit the client\n''' - return True - - def do_disconnect (self, line): - '''Disconnect from the server\n''' - if not self.stateless_client.is_connected(): - print "Not connected to server\n" - return - - res_ok, msg = self.stateless_client.disconnect() - if res_ok: - print format_text("[SUCCESS]\n", 'green', 'bold') - else: - print msg + "\n" - - def do_whoami (self, line): - '''Prints console user name\n''' - print "\n" + self.stateless_client.user + "\n" - - def postcmd(self, stop, line): - if self.stateless_client.is_connected(): - self.prompt = "TRex > " - else: - self.supported_rpc = None - self.prompt = "TRex (offline) > " - - return stop - - def default(self, line): - print "'{0}' is an unrecognized command. type 'help' or '?' for a list\n".format(line) - - # def do_help (self, line): - # '''Shows This Help Screen\n''' - # if line: - # try: - # func = getattr(self, 'help_' + line) - # except AttributeError: - # try: - # doc = getattr(self, 'do_' + line).__doc__ - # if doc: - # self.stdout.write("%s\n"%str(doc)) - # return - # except AttributeError: - # pass - # self.stdout.write("%s\n"%str(self.nohelp % (line,))) - # return - # func() - # return - # - # print "\nSupported Console Commands:" - # print "----------------------------\n" - # - # cmds = [x[3:] for x in self.get_names() if x.startswith("do_")] - # for cmd in cmds: - # if cmd == "EOF": - # continue - # - # try: - # doc = getattr(self, 'do_' + cmd).__doc__ - # if doc: - # help = str(doc) - # else: - # help = "*** Undocumented Function ***\n" - # except AttributeError: - # help = "*** Undocumented Function ***\n" - # - # print "{:<30} {:<30}".format(cmd + " - ", help) - - def do_stream_db_add(self, line): - '''Loads a YAML stream list serialization into user console \n''' - args = line.split() - if len(args) >= 2: - name = args[0] - yaml_path = args[1] - try: - multiplier = args[2] - except IndexError: - multiplier = 1 - stream_list = CStreamList() - loaded_obj = stream_list.load_yaml(yaml_path, multiplier) - # print self.stateless_client.pretty_json(json.dumps(loaded_obj)) - try: - compiled_streams = stream_list.compile_streams() - res_ok = self.streams_db.load_streams(name, LoadedStreamList(loaded_obj, - [StreamPack(v.stream_id, v.stream.dump()) - for k, v in compiled_streams.items()])) - if res_ok: - print green("Stream pack '{0}' loaded and added successfully\n".format(name)) - else: - print magenta("Picked name already exist. Please pick another name.\n") - except Exception as e: - print "adding new stream failed due to the following error:\n", str(e) - print format_text("[FAILED]\n", 'red', 'bold') - - return - else: - print magenta("please provide load name and YAML path, separated by space.\n" - "Optionally, you may provide a third argument to specify multiplier.\n") - - @staticmethod - def tree_autocomplete(text): - dir = os.path.dirname(text) - if dir: - path = dir - else: - path = "." - start_string = os.path.basename(text) - return [x - for x in os.listdir(path) - if x.startswith(start_string)] - - - def complete_stream_db_add(self, text, line, begidx, endidx): - arg_num = len(line.split()) - 1 - if arg_num == 2: - return TRexConsole.tree_autocomplete(line.split()[-1]) - else: - return [text] - - def do_stream_db_show(self, line): - '''Shows the loaded stream list named [name] \n''' - args = line.split() - if args: - list_name = args[0] - try: - stream = self.streams_db.get_stream_pack(list_name)#user_streams[list_name] - if len(args) >= 2 and args[1] == "full": - print pretty_json(json.dumps(stream.compiled)) - else: - print pretty_json(json.dumps(stream.loaded)) - except KeyError as e: - print "Unknown stream list name provided" - else: - print "Available stream packs:\n{0}".format(', '.join(sorted(self.streams_db.get_loaded_streams_names()))) - - def complete_stream_db_show(self, text, line, begidx, endidx): - return [x - for x in self.streams_db.get_loaded_streams_names() - if x.startswith(text)] - - def do_stream_db_remove(self, line): - '''Removes a single loaded stream packs from loaded stream pack repository\n''' - args = line.split() - if args: - removed_streams = self.streams_db.remove_stream_packs(*args) - if removed_streams: - print green("The following stream packs were removed:") - print bold(", ".join(sorted(removed_streams))) - print format_text("[SUCCESS]\n", 'green', 'bold') - else: - print red("No streams were removed. Make sure to provide valid stream pack names.") - else: - print magenta("Please provide stream pack name(s), separated with spaces.") - - def do_stream_db_clear(self, line): - '''Clears all loaded stream packs from loaded stream pack repository\n''' - self.streams_db.clear() - print format_text("[SUCCESS]\n", 'green', 'bold') - - - def complete_stream_db_remove(self, text, line, begidx, endidx): - return [x - for x in self.streams_db.get_loaded_streams_names() - if x.startswith(text)] - - - def do_attach(self, line): - '''Assign loaded stream pack into specified ports on TRex\n''' - args = line.split() - if len(args) >= 2: - stream_pack_name = args[0] - stream_list = self.streams_db.get_stream_pack(stream_pack_name) #user_streams[args[0]] - if not stream_list: - print "Provided stream list name '{0}' doesn't exists.".format(stream_pack_name) - print format_text("[FAILED]\n", 'red', 'bold') - return - if args[1] == "all": - ask = ConfirmMenu('Are you sure you want to release all acquired ports? ') - rc = ask.show() - if rc == False: - print yellow("[ABORTED]\n") - return - else: - port_list = self.stateless_client.get_acquired_ports() - else: - port_list = self.extract_port_ids_from_line(' '.join(args[1:])) - owned = set(self.stateless_client.get_acquired_ports()) - try: - if set(port_list).issubset(owned): - res_ok, log = self.stateless_client.add_stream_pack(stream_list.compiled, port_id=port_list) - # res_ok, msg = self.stateless_client.add_stream(port_list, stream_list.compiled) - self.prompt_response(log) - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - return - print format_text("[SUCCESS]\n", 'green', 'bold') - return - else: - print "Not all desired ports are acquired.\n" \ - "Acquired ports are: {acq}\n" \ - "Requested ports: {req}\n" \ - "Missing ports: {miss}".format(acq=list(owned), - req=port_list, - miss=list(set(port_list).difference(owned))) - print format_text("[FAILED]\n", 'red', 'bold') - except ValueError as e: - print magenta(str(e)) - print format_text("[FAILED]\n", 'red', 'bold') - else: - print magenta("Please provide list name and ports to attach to, " - "or specify 'all' to attach all owned ports.\n") - - def complete_attach(self, text, line, begidx, endidx): - arg_num = len(line.split()) - 1 - if arg_num == 1: - # return optional streams packs - if line.endswith(" "): - return self.port_auto_complete(text, line, begidx, endidx) - return [x - for x in self.streams_db.get_loaded_streams_names() - if x.startswith(text)] - elif arg_num >= 2: - # return optional ports to attach to - return self.port_auto_complete(text, line, begidx, endidx) - else: - return [text] - - def prompt_response(self, response_obj): - resp_list = response_obj if isinstance(response_obj, list) else [response_obj] - def format_return_status(return_status): - if return_status: - return green("OK") - else: - return red("FAIL") - - for response in resp_list: - response_str = "{id:^3} - {msg} ({stat})".format(id=response.id, - msg=response.msg, - stat=format_return_status(response.success)) - print response_str - return - - def do_remove_all_streams(self, line): - '''Acquire ports\n''' - - # make sure that the user wants to acquire all - args = line.split() - if len(args) < 1: - print magenta("Please provide a list of ports separated by spaces, " - "or specify 'all' to remove from all acquired ports") - return - if args[0] == "all": - ask = ConfirmMenu('Are you sure you want to remove all stream packs from all acquired ports? ') - rc = ask.show() - if rc == False: - print yellow("[ABORTED]\n") - return - else: - port_list = self.stateless_client.get_acquired_ports() - else: - port_list = self.extract_port_ids_from_line(line) - - # rc, resp_list = self.stateless_client.take_ownership(port_list, force) - try: - res_ok, log = self.stateless_client.remove_all_streams(port_list) - self.prompt_response(log) - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - return - print format_text("[SUCCESS]\n", 'green', 'bold') - except ValueError as e: - print magenta(str(e)) - print format_text("[FAILED]\n", 'red', 'bold') - - def complete_remove_all_streams(self, text, line, begidx, endidx): - return self.port_auto_complete(text, line, begidx, endidx) - - def do_start(self, line): - '''Start selected traffic in specified ports on TRex\n''' - # make sure that the user wants to acquire all - parser = parsing_opts.gen_parser("start", self.do_start.__doc__, - parsing_opts.PORT_LIST_WITH_ALL, - parsing_opts.FORCE, - parsing_opts.STREAM_FROM_PATH_OR_FILE, - parsing_opts.DURATION, - parsing_opts.MULTIPLIER) - opts = parser.parse_args(line.split()) - if opts is None: - # avoid further processing in this command - return - # print opts - port_list = self.extract_port_list(opts) - # print port_list - if opts.force: - # stop all active ports, if any - res_ok = self.stop_traffic(set(self.stateless_client.get_active_ports()).intersection(port_list)) - if not res_ok: - print yellow("[ABORTED]\n") - return - # remove all traffic from ports - res_ok = self.remove_all_streams(port_list) - if not res_ok: - print yellow("[ABORTED]\n") - return - # decide which traffic to use - stream_pack_name = None - if opts.db: - # use pre-loaded traffic - print format_text('{:<30}'.format("Load stream pack (from DB):"), 'bold'), - if opts.db not in self.streams_db.get_loaded_streams_names(): - print format_text("[FAILED]\n", 'red', 'bold') - print yellow("[ABORTED]\n") - return - else: - stream_pack_name = opts.db - else: - # try loading a YAML file - print format_text('{:<30}'.format("Load stream pack (from file):"), 'bold'), - stream_list = CStreamList() - loaded_obj = stream_list.load_yaml(opts.file[0]) - # print self.stateless_client.pretty_json(json.dumps(loaded_obj)) - try: - compiled_streams = stream_list.compile_streams() - res_ok = self.streams_db.load_streams(opts.file[1], - LoadedStreamList(loaded_obj, - [StreamPack(v.stream_id, v.stream.dump()) - for k, v in compiled_streams.items()])) - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - print yellow("[ABORTED]\n") - return - print format_text("[SUCCESS]\n", 'green', 'bold') - stream_pack_name = opts.file[1] - except Exception as e: - print format_text("[FAILED]\n", 'red', 'bold') - print yellow("[ABORTED]\n") - res_ok = self.attach_to_port(stream_pack_name, port_list) - if not res_ok: - print yellow("[ABORTED]\n") - return - # finally, start the traffic - res_ok = self.start_traffic(opts.mult, port_list) - if not res_ok: - print yellow("[ABORTED]\n") - return - return - - def help_start(self): - self.do_start("-h") - - def do_stop(self, line): - '''Stop active traffic in specified ports on TRex\n''' - parser = parsing_opts.gen_parser("stop", self.do_stop.__doc__, - parsing_opts.PORT_LIST_WITH_ALL) - opts = parser.parse_args(line.split()) - if opts is None: - # avoid further processing in this command - return - port_list = self.extract_port_list(opts) - res_ok = self.stop_traffic(port_list) - return - - def do_pause(self, line): - '''Pause active traffic in specified ports on TRex\n''' - parser = parsing_opts.gen_parser("stop", self.do_stop.__doc__, - parsing_opts.PORT_LIST_WITH_ALL) - opts = parser.parse_args(line.split()) - if opts is None: - # avoid further processing in this command - return - port_list = self.extract_port_list(opts) - res_ok = self.stop_traffic(port_list) - return - - - def help_stop(self): - self.do_stop("-h") - - - def do_debug(self, line): - '''Enter DEBUG mode of the console to invoke smaller building blocks with server''' - i = DebugTRexConsole(self) - i.prompt = self.prompt[:-3] + ':' + blue('debug') + ' > ' - i.cmdloop() - - # aliasing - do_exit = do_EOF = do_q = do_quit - - # ----- utility methods ----- # - - def start_traffic(self, multiplier, port_list):#, silent=True): - print format_text('{:<30}'.format("Start traffic:"), 'bold'), - try: - res_ok, log = self.stateless_client.start_traffic(multiplier, port_id=port_list) - if not self._silent: - print '' - self.prompt_response(log) - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - return False - print format_text("[SUCCESS]\n", 'green', 'bold') - return True - except ValueError as e: - print '' - print magenta(str(e)) - print format_text("[FAILED]\n", 'red', 'bold') - return False - - def attach_to_port(self, stream_pack_name, port_list): - print format_text('{:<30}'.format("Attaching traffic to ports:"), 'bold'), - stream_list = self.streams_db.get_stream_pack(stream_pack_name) #user_streams[args[0]] - if not stream_list: - print "Provided stream list name '{0}' doesn't exists.".format(stream_pack_name) - print format_text("[FAILED]\n", 'red', 'bold') - return - try: - res_ok, log = self.stateless_client.add_stream_pack(stream_list.compiled, port_id=port_list) - if not self._silent: - print '' - self.prompt_response(log) - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - return False - print format_text("[SUCCESS]\n", 'green', 'bold') - return True - except ValueError as e: - print '' - print magenta(str(e)) - print format_text("[FAILED]\n", 'red', 'bold') - return False - - def stop_traffic(self, port_list): - print format_text('{:<30}'.format("Stop traffic:"), 'bold'), - try: - res_ok, log = self.stateless_client.stop_traffic(port_id=port_list) - if not self._silent: - print '' - self.prompt_response(log) - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - return - print format_text("[SUCCESS]\n", 'green', 'bold') - return True - except ValueError as e: - print '' - print magenta(str(e)) - print format_text("[FAILED]\n", 'red', 'bold') - - def remove_all_streams(self, port_list): - '''Remove all streams from given port_list''' - print format_text('{:<30}'.format("Remove all streams:"), 'bold'), - try: - res_ok, log = self.stateless_client.remove_all_streams(port_id=port_list) - if not self._silent: - print '' - self.prompt_response(log) - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - return - print format_text("[SUCCESS]\n", 'green', 'bold') - return True - except ValueError as e: - print '' - print magenta(str(e)) - print format_text("[FAILED]\n", 'red', 'bold') - - - - - - def extract_port_list(self, opts): - if opts.all_ports or "all" in opts.ports: - # handling all ports - port_list = self.stateless_client.get_acquired_ports() - else: - port_list = self.extract_port_ids_from_list(opts.ports) - return port_list - - def decode_multiplier(self, opts_mult): - pass - - -class DebugTRexConsole(cmd.Cmd): - - def __init__(self, trex_main_console): - cmd.Cmd.__init__(self) - self.trex_console = trex_main_console - self.stateless_client = self.trex_console.stateless_client - self.streams_db = self.trex_console.streams_db - self.register_main_console_methods() - self.do_silent("on") - pass - - # ----- super methods overriding ----- # - def completenames(self, text, *ignored): - dotext = 'do_'+text - return [a[3:]+' ' for a in self.get_names() if a.startswith(dotext)] - - def get_names(self): - result = cmd.Cmd.get_names(self) - result += self.trex_console.get_names() - return list(set(result)) - - def register_main_console_methods(self): - main_names = set(self.trex_console.get_names()).difference(set(dir(self.__class__))) - for name in main_names: - for prefix in 'do_', 'help_', 'complete_': - if name.startswith(prefix): - self.__dict__[name] = getattr(self.trex_console, name) - - # if (name[:3] == 'do_') or (name[:5] == 'help_') or (name[:9] == 'complete_'): - # chosen.append(name) - # self.__dict__[name] = getattr(self.trex_console, name) - # # setattr(self, name, classmethod(getattr(self.trex_console, name))) - - # print chosen - # self.get_names() - - # return result - - - # ----- DEBUGGING methods ----- # - # set silent on / off - def do_silent(self, line): - '''Shows or set silent mode\n''' - if line == "": - print "\nsilent mode is " + ("on\n" if self.trex_console._silent else "off\n") - - elif line == "on": - self.verbose = True - self.stateless_client.set_verbose(True) - print green("\nsilent set to on\n") - - elif line == "off": - self.verbose = False - self.stateless_client.set_verbose(False) - print green("\nsilent set to off\n") - - else: - print magenta("\nplease specify 'on' or 'off'\n") - - def do_quit(self, line): - '''Exit the debug client back to main console\n''' - self.do_silent("off") - return True - - def do_start_traffic(self, line): - '''Start pre-submitted traffic in specified ports on TRex\n''' - # make sure that the user wants to acquire all - parser = parsing_opts.gen_parser("start_traffic", self.do_start_traffic.__doc__, - parsing_opts.PORT_LIST_WITH_ALL, parsing_opts.MULTIPLIER) - opts = parser.parse_args(line.split()) - # print opts - # return - if opts is None: - # avoid further processing in this command - return - try: - port_list = self.trex_console.extract_port_list(opts) - return self.trex_console.start_traffic(opts.mult, port_list) - except Exception as e: - print e - return - - def do_stop_traffic(self, line): - '''Stop active traffic in specified ports on TRex\n''' - parser = parsing_opts.gen_parser("stop_traffic", self.do_stop_traffic.__doc__, - parsing_opts.PORT_LIST_WITH_ALL) - opts = parser.parse_args(line.split()) - # print opts - # return - if opts is None: - # avoid further processing in this command - return - try: - port_list = self.trex_console.extract_port_list(opts) - return self.trex_console.stop_traffic(port_list) - except Exception as e: - print e - return - - - def complete_stop_traffic(self, text, line, begidx, endidx): - return self.port_auto_complete(text, line, begidx, endidx, active=True) - - # return - # # return - # # if not opts.port_list: - # # print magenta("Please provide a list of ports separated by spaces, " - # # "or specify 'all' to start traffic on all acquired ports") - # # return - # - - - return - args = line.split() - if len(args) < 1: - print magenta("Please provide a list of ports separated by spaces, " - "or specify 'all' to start traffic on all acquired ports") - return - if args[0] == "all": - ask = ConfirmMenu('Are you sure you want to start traffic at all acquired ports? ') - rc = ask.show() - if rc == False: - print yellow("[ABORTED]\n") - return - else: - port_list = self.stateless_client.get_acquired_ports() - else: - port_list = self.extract_port_ids_from_line(line) - - try: - res_ok, log = self.stateless_client.start_traffic(1.0, port_id=port_list) - self.prompt_response(log) - if not res_ok: - print format_text("[FAILED]\n", 'red', 'bold') - return - print format_text("[SUCCESS]\n", 'green', 'bold') - except ValueError as e: - print magenta(str(e)) - print format_text("[FAILED]\n", 'red', 'bold') - - def complete_start_traffic(self, text, line, begidx, endidx): - # return self.port_auto_complete(text, line, begidx, endidx) - return [text] - - def help_start_traffic(self): - self.do_start_traffic("-h") - - def help_stop_traffic(self): - self.do_stop_traffic("-h") - - # def do_help(self): - - def do_rpc (self, line): - '''Launches a RPC on the server\n''' - - if line == "": - print "\nUsage: [method name] [param dict as string]\n" - print "Example: rpc test_add {'x': 12, 'y': 17}\n" - return - - sp = line.split(' ', 1) - method = sp[0] - - params = None - bad_parse = False - if len(sp) > 1: - - try: - params = ast.literal_eval(sp[1]) - if not isinstance(params, dict): - bad_parse = True - - except ValueError as e1: - bad_parse = True - except SyntaxError as e2: - bad_parse = True - - if bad_parse: - print "\nValue should be a valid dict: '{0}'".format(sp[1]) - print "\nUsage: [method name] [param dict as string]\n" - print "Example: rpc test_add {'x': 12, 'y': 17}\n" - return - - res_ok, msg = self.stateless_client.transmit(method, params) - if res_ok: - print "\nServer Response:\n\n" + pretty_json(json.dumps(msg)) + "\n" - else: - print "\n*** " + msg + "\n" - #print "Please try 'reconnect' to reconnect to server" - - - def complete_rpc (self, text, line, begidx, endidx): - return [x - for x in self.trex_console.supported_rpc - if x.startswith(text)] - - # aliasing - do_exit = do_EOF = do_q = do_quit - -# diff --git a/scripts/automation/trex_control_plane/console/trex_console.py b/scripts/automation/trex_control_plane/console/trex_console.py index e187c8c2..8d52b2a4 100755 --- a/scripts/automation/trex_control_plane/console/trex_console.py +++ b/scripts/automation/trex_control_plane/console/trex_console.py @@ -34,8 +34,7 @@ from client.trex_stateless_client import CTRexStatelessClient from common.text_opts import * from client_utils.general_utils import user_input, get_current_user from client_utils import parsing_opts -import trex_status -import parsing_opts +import trex_tui from functools import wraps @@ -119,6 +118,7 @@ class TRexConsole(TRexGeneralCmd): self.stateless_client = stateless_client TRexGeneralCmd.__init__(self) + self.tui = trex_tui.TrexTUI(stateless_client) self.verbose = verbose self.acquire_all_ports = acquire_all_ports @@ -425,7 +425,7 @@ class TRexConsole(TRexGeneralCmd): '''Shows a graphical console\n''' self.do_verbose('off') - trex_status.show_trex_status(self.stateless_client) + self.tui.show() # quit function def do_quit(self, line): diff --git a/scripts/automation/trex_control_plane/console/trex_tui.py b/scripts/automation/trex_control_plane/console/trex_tui.py new file mode 100644 index 00000000..e835ed24 --- /dev/null +++ b/scripts/automation/trex_control_plane/console/trex_tui.py @@ -0,0 +1,309 @@ +import termios +import sys +import os +import time +from common.text_opts import * +from common import trex_stats +from client_utils import text_tables +from collections import OrderedDict +import datetime + +# base type of a panel +class TrexTUIPanel(object): + def __init__ (self, mng, name): + + self.mng = mng + self.name = name + self.stateless_client = mng.stateless_client + + + def show (self): + raise Exception("must implement this") + + def get_key_actions (self): + raise Exception("must implement this") + + def get_name (self): + return self.name + + +# dashboard panel +class TrexTUIDashBoard(TrexTUIPanel): + def __init__ (self, mng): + super(TrexTUIDashBoard, self).__init__(mng, "dashboard") + + self.key_actions = OrderedDict() + self.key_actions['p'] = {'action': self.action_pause, 'legend': 'pause', 'show': True} + self.key_actions['r'] = {'action': self.action_resume, 'legend': 'resume', 'show': True} + self.key_actions['+'] = {'action': self.action_raise, 'legend': 'up 5%', 'show': True} + self.key_actions['-'] = {'action': self.action_lower, 'legend': 'low 5%', 'show': True} + + def show (self): + ports = self.stateless_client.get_acquired_ports() + stats = self.stateless_client.cmd_stats(ports, trex_stats.ALL_STATS_OPTS) + # print stats to screen + for stat_type, stat_data in stats.iteritems(): + text_tables.print_table_with_header(stat_data.text_table, stat_type) + + def get_key_actions (self): + return self.key_actions + + # actions + def action_pause (self): + rc = self.stateless_client.pause_traffic(self.mng.acquired_ports) + if rc.good(): + return "paused traffic on all ports" + else: + return "" + + def action_resume (self): + rc = self.stateless_client.resume_traffic(self.mng.acquired_ports) + if rc.good(): + return "resumed traffic on all ports" + else: + return "" + + def action_raise (self): + mul = {'type': 'percentage', 'value': 5, 'op': 'add'} + rc = self.stateless_client.update_traffic(mul, self.mng.acquired_ports) + + if rc.good(): + return "raised B/W by 5% on all ports" + else: + return "" + + def action_lower (self): + mul = {'type': 'percentage', 'value': 5, 'op': 'sub'} + rc = self.stateless_client.update_traffic(mul, self.mng.acquired_ports) + + if rc.good(): + return "lowered B/W by 5% on all ports" + else: + return "" + + +# port panel +class TrexTUIPort(TrexTUIPanel): + def __init__ (self, mng, port_id): + super(TrexTUIPort, self).__init__(mng, "port {0}".format(port_id)) + self.port_id = port_id + + self.key_actions = OrderedDict() + self.key_actions['p'] = {'action': self.action_pause, 'legend': 'pause', 'show': True} + self.key_actions['r'] = {'action': self.action_resume, 'legend': 'resume', 'show': True} + self.key_actions['+'] = {'action': self.action_raise, 'legend': 'up 5%', 'show': True} + self.key_actions['-'] = {'action': self.action_lower, 'legend': 'low 5%', 'show': True} + + def show (self): + stats = self.stateless_client.cmd_stats([self.port_id], trex_stats.ALL_STATS_OPTS) + # print stats to screen + for stat_type, stat_data in stats.iteritems(): + text_tables.print_table_with_header(stat_data.text_table, stat_type) + + def get_key_actions (self): + return self.key_actions + + # actions + def action_pause (self): + rc = self.stateless_client.pause_traffic([self.port_id]) + if rc.good(): + return "paused traffic on port {0}".format(self.port_id) + else: + return "" + + def action_resume (self): + rc = self.stateless_client.resume_traffic([self.port_id]) + if rc.good(): + return "resumed traffic on port {0}".format(self.port_id) + else: + return "" + + def action_raise (self): + mul = {'type': 'percentage', 'value': 5, 'op': 'add'} + rc = self.stateless_client.update_traffic(mul, [self.port_id]) + + if rc.good(): + return "raised B/W by 5% on port {0}".format(self.port_id) + else: + return "" + + def action_lower (self): + mul = {'type': 'percentage', 'value': 5, 'op': 'sub'} + rc = self.stateless_client.update_traffic(mul, [self.port_id]) + + if rc.good(): + return "lowered B/W by 5% on port {0}".format(self.port_id) + else: + return "" + +# log +class TrexTUILog(): + def __init__ (self): + self.log = [] + + def add_event (self, msg): + self.log.append("[{0}] {1}".format(str(datetime.datetime.now().time()), msg)) + + def show (self, max_lines = 4): + cut = len(self.log) - max_lines + if cut < 0: + cut = 0 + + print format_text("\nLog:", 'bold', 'underline') + + for msg in self.log[cut:]: + print msg + + +# Panels manager (contains server panels) +class TrexTUIPanelManager(): + def __init__ (self, tui): + self.tui = tui + self.stateless_client = tui.stateless_client + self.acquired_ports = self.stateless_client.get_acquired_ports() + + self.panels = {} + self.panels['dashboard'] = TrexTUIDashBoard(self) + + self.key_actions = OrderedDict() + self.key_actions['q'] = {'action': self.action_quit, 'legend': 'quit', 'show': True} + self.key_actions['g'] = {'action': self.action_show_dash, 'legend': 'dashboard', 'show': True} + + for port_id in self.acquired_ports: + self.key_actions[str(port_id)] = {'action': self.action_show_port(port_id), 'legend': 'port {0}'.format(port_id), 'show': False} + self.panels['port {0}'.format(port_id)] = TrexTUIPort(self, port_id) + + # start with dashboard + self.main_panel = self.panels['dashboard'] + + self.generate_legend() + + self.log = TrexTUILog() + + + def generate_legend (self): + self.legend = "{:<12}".format("browse:") + + for k, v in self.key_actions.iteritems(): + if v['show']: + x = "'{0}' - {1}, ".format(k, v['legend']) + self.legend += "{:}".format(x) + + self.legend += "'0-{0}' - port display".format(len(self.acquired_ports) - 1) + + panel_actions = self.main_panel.get_key_actions() + if not panel_actions: + return + + self.legend += "\n{:<12}".format(self.main_panel.get_name() + ":") + for k, v in self.main_panel.get_key_actions().iteritems(): + if v['show']: + x = "'{0}' - {1}, ".format(k, v['legend']) + self.legend += "{:}".format(x) + + + def print_legend (self): + print format_text(self.legend, 'bold') + + + def show (self): + self.main_panel.show() + self.print_legend() + self.log.show() + + + def handle_key (self, ch): + # check for the manager registered actions + if ch in self.key_actions: + msg = self.key_actions[ch]['action']() + + # check for main panel actions + elif ch in self.main_panel.get_key_actions(): + msg = self.main_panel.get_key_actions()[ch]['action']() + + else: + msg = "" + + self.generate_legend() + + if msg == None: + return False + else: + if msg: + self.log.add_event(msg) + return True + + + # actions + + def action_quit (self): + return None + + def action_show_dash (self): + self.main_panel = self.panels['dashboard'] + return "" + + def action_show_port (self, port_id): + def action_show_port_x (): + self.main_panel = self.panels['port {0}'.format(port_id)] + return "" + + return action_show_port_x + + + +# shows a textual top style window +class TrexTUI(): + def __init__ (self, stateless_client): + self.stateless_client = stateless_client + + self.pm = TrexTUIPanelManager(self) + + + + def handle_key_input (self): + # try to read a single key + ch = os.read(sys.stdin.fileno(), 1) + if ch != None and len(ch) > 0: + return self.pm.handle_key(ch) + + else: + return True + + + def clear_screen (self): + os.system('clear') + + + + def show (self): + # init termios + old_settings = termios.tcgetattr(sys.stdin) + new_settings = termios.tcgetattr(sys.stdin) + new_settings[3] = new_settings[3] & ~(termios.ECHO | termios.ICANON) # lflags + new_settings[6][termios.VMIN] = 0 # cc + new_settings[6][termios.VTIME] = 0 # cc + termios.tcsetattr(sys.stdin, termios.TCSADRAIN, new_settings) + + try: + while True: + self.clear_screen() + + cont = self.handle_key_input() + self.pm.show() + + if not cont: + break + + time.sleep(0.1) + + finally: + # restore + termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings) + + print "" + + # key actions + def action_quit (self): + return False + -- cgit 1.2.3-korg