diff --git a/pymodbus/repl/client/main.py b/pymodbus/repl/client/main.py index 3f2fa1fc8..e231b1dc9 100644 --- a/pymodbus/repl/client/main.py +++ b/pymodbus/repl/client/main.py @@ -151,90 +151,103 @@ def _parse_val(arg_name, val): return kwargs, execute -def cli(client): # pylint: disable=too-complex - """Run client definition.""" - use_keys = KeyBindings() - history_file = pathlib.Path.home().joinpath(".pymodhis") - - @use_keys.add("c-space") - def _(event): - """Initialize autocompletion, or select the next completion.""" - buff = event.app.current_buffer - if buff.complete_state: - buff.complete_next() - else: - buff.start_completion(select_first=False) - - @use_keys.add("enter", filter=has_selected_completion) - def _(event): - """Make the enter key work as the tab key only when showing the menu.""" - event.current_buffer.complete_state = None - buffer = event.cli.current_buffer - buffer.complete_state = None - - session = PromptSession( - lexer=PygmentsLexer(PythonLexer), - completer=CmdCompleter(client), - style=style, - complete_while_typing=True, - bottom_toolbar=bottom_toolbar, - key_bindings=use_keys, - history=FileHistory(history_file), - auto_suggest=AutoSuggestFromHistory(), - ) - click.secho(TITLE, fg="green") - result = None - while True: # pylint: disable=too-many-nested-blocks - try: - - text = session.prompt("> ", complete_while_typing=True) - if text.strip().lower() == "help": - print_formatted_text(HTML("Available commands:")) - for cmd, obj in sorted(session.completer.commands.items()): - if cmd != "help": - print_formatted_text( - HTML( - "{:45s}" # pylint: disable=consider-using-f-string - "{:100s}" - "".format(cmd, obj.help_text) - ) - ) - - continue - if text.strip().lower() == "exit": - raise EOFError() - if text.strip().lower().startswith("client."): - try: - text = text.strip().split() - cmd = text[0].split(".")[1] - args = text[1:] - kwargs, execute = _process_args(args, string=False) - if execute: - if text[0] in CLIENT_ATTRIBUTES: - result = Result(getattr(client, cmd)) - else: - result = Result(getattr(client, cmd)(**kwargs)) - result.print_result() - except Exception as exc: # pylint: disable=broad-except - click.secho(repr(exc), fg="red") - elif text.strip().lower().startswith("result."): - if result: - words = text.lower().split() - if words[0] == "result.raw": - result.raw() - if words[0] == "result.decode": - args = words[1:] - kwargs, execute = _process_args(args) - if execute: - result.decode(**kwargs) - except KeyboardInterrupt: - continue # Control-C pressed. Try again. - except EOFError: - break # Control-D pressed. - except Exception as exc: # pylint: disable=broad-except - click.secho(str(exc), fg="red") - - click.secho("GoodBye!", fg="blue") +class CLI: # pylint: disable=too-few-public-methods + """Client definition.""" + + def __init__(self, client): + """Set up client and keybindings.""" + + use_keys = KeyBindings() + history_file = pathlib.Path.home().joinpath(".pymodhis") + self.client = client + + @use_keys.add("c-space") + def _(event): + """Initialize autocompletion, or select the next completion.""" + buff = event.app.current_buffer + if buff.complete_state: + buff.complete_next() + else: + buff.start_completion(select_first=False) + + @use_keys.add("enter", filter=has_selected_completion) + def _(event): + """Make the enter key work as the tab key only when showing the menu.""" + event.current_buffer.complete_state = None + buffer = event.cli.current_buffer + buffer.complete_state = None + + self.session = PromptSession( + lexer=PygmentsLexer(PythonLexer), + completer=CmdCompleter(client), + style=style, + complete_while_typing=True, + bottom_toolbar=bottom_toolbar, + key_bindings=use_keys, + history=FileHistory(history_file), + auto_suggest=AutoSuggestFromHistory(), + ) + click.secho(TITLE, fg="green") + + def _print_command_help(self, commands): + """Print a list of commands with help text.""" + for cmd, obj in sorted(commands.items()): + if cmd != "help": + print_formatted_text( + HTML( + "{:45s}" # pylint: disable=consider-using-f-string + "{:100s}" + "".format(cmd, obj.help_text) + ) + ) + + def _process_client(self, text, client): + """Process client commands.""" + text = text.strip().split() + cmd = text[0].split(".")[1] + args = text[1:] + kwargs, execute = _process_args(args, string=False) + if execute: + if text[0] in CLIENT_ATTRIBUTES: + result = Result(getattr(client, cmd)) + else: + result = Result(getattr(client, cmd)(**kwargs)) + result.print_result() + + def _process_result(self, text, result): + """Process result commands.""" + words = text.split() + if words[0] == "result.raw": + result.raw() + if words[0] == "result.decode": + args = words[1:] + kwargs, execute = _process_args(args) + if execute: + result.decode(**kwargs) + + def run(self): + """Run the REPL.""" + result = None + while True: + try: + text = self.session.prompt("> ", complete_while_typing=True) + if text.strip().lower() == "help": + print_formatted_text(HTML("Available commands:")) + self._print_command_help(self.session.completer.commands) + elif text.strip().lower() == "exit": + raise EOFError() + elif text.strip().lower().startswith("client."): + self._process_client(text, self.client) + elif text.strip().lower().startswith("result.") and result: + self._process_result(text, result) + except KeyboardInterrupt: + continue # Control-C pressed. Try again. + except EOFError: + break # Control-D pressed. + except Exception as exc: # pylint: disable=broad-except + click.secho(str(exc), fg="red") + + click.secho("GoodBye!", fg="blue") @click.group("pymodbus-repl") @@ -307,7 +320,8 @@ def tcp(ctx, host, port, framer): if framer == "rtu": kwargs["framer"] = ModbusRtuFramer client = ModbusTcpClient(**kwargs) - cli(client) + cli = CLI(client) + cli.run() @main.command("serial") @@ -427,7 +441,8 @@ def serial( # pylint: disable=too-many-arguments write_timeout=write_timeout, **ctx.obj, ) - cli(client) + cli = CLI(client) + cli.run() if __name__ == "__main__":