Refactor REPL client to reduce complexity (#1489)

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("<u>Available commands:</u>"))
-                for cmd, obj in sorted(session.completer.commands.items()):
-                    if cmd != "help":
-                        print_formatted_text(
-                            HTML(
-                                "<skyblue>{:45s}</skyblue>"  # pylint: disable=consider-using-f-string
-                                "<seagreen>{:100s}"
-                                "</seagreen>".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(
+                        "<skyblue>{:45s}</skyblue>"  # pylint: disable=consider-using-f-string
+                        "<seagreen>{:100s}"
+                        "</seagreen>".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("<u>Available commands:</u>"))
+                    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__":