Refactor REPL server to reduce complexity (#1499)

Co-authored-by: jan iversen <jancasacondor@gmail.com>

Author: alexrudd2

pymodbus/repl/server/cli.py

@@ -105,8 +105,8 @@ def print_help():
         )
 
 
-async def interactive_shell(server):  # pylint: disable=too-complex
-    """Run CLI interactive shell."""
+def print_title():
+    """Print title - large if there are sufficient columns, otherwise small."""
     col = get_terminal_width()
     max_len = max(  # pylint: disable=consider-using-generator
         [len(t) for t in TITLE.split("\n")]
@@ -117,16 +117,20 @@ async def interactive_shell(server):  # pylint: disable=too-complex
         print_formatted_text(
             HTML(f'<u><b><style color="green">{SMALL_TITLE}</style></b></u>')
         )
+
+
+async def interactive_shell(server):  # pylint: disable=too-complex
+    """Run CLI interactive shell."""
+    print_title()
     info("")
     completer = NestedCompleter.from_nested_dict(COMMANDS)
     session = PromptSession(
         "SERVER > ", completer=completer, bottom_toolbar=BOTTOM_TOOLBAR
     )
 
     # Run echo loop. Read text from stdin, and reply it back.
-    while True:  # pylint: disable=too-many-nested-blocks
+    while True:
         try:
-            invalid_command = False
             result = await session.prompt_async()
             if result == "exit":
                 await server.web_app.shutdown()
@@ -139,51 +143,13 @@ async def interactive_shell(server):  # pylint: disable=too-complex
                 continue
             if command := result.split():
                 if command[0] not in COMMANDS:
-                    invalid_command = True
-                if invalid_command:
                     warning(f"Invalid command or invalid usage of command - {command}")
                     continue
                 if len(command) == 1:
                     warning(f'Usage: "{USAGE}"')
                 else:
-                    args = command[1:]
-                    skip_next = False
-                    val_dict = {}
-                    for index, arg in enumerate(args):
-                        if skip_next:
-                            skip_next = False
-                            continue
-                        if "=" in arg:
-                            arg, value = arg.split("=")
-                        elif arg in COMMAND_ARGS:
-                            try:
-                                value = args[index + 1]
-                                skip_next = True
-                            except IndexError:
-                                error(f"Missing value for argument - {arg}")
-                                warning('Usage: "{USAGE}"')
-                                break
-                        valid = True
-                        if arg == "response_type":
-                            if value not in RESPONSE_TYPES:
-                                warning(f"Invalid response type request - {value}")
-                                warning(f"Choose from {RESPONSE_TYPES}")
-                                valid = False
-                        elif arg in {  # pylint: disable=confusing-consecutive-elif
-                            "error_code",
-                            "delay_by",
-                            "clear_after",
-                            "data_len",
-                        }:
-                            try:
-                                value = int(value)
-                            except ValueError:
-                                warning(f"Expected integer value for {arg}")
-                                valid = False
-
-                        if valid:
-                            val_dict[arg] = value
-                    if val_dict:
+                    val_dict = _process_args(command[1:])
+                    if val_dict:  # pylint: disable=consider-using-assignment-expr
                         server.update_manipulator_config(val_dict)
                         # server.manipulator_config = val_dict
                 # result = await run_command(tester, *command)
@@ -192,6 +158,44 @@ async def interactive_shell(server):  # pylint: disable=too-complex
             return
 
 
+def _process_args(args) -> dict:  # pylint: disable=too-complex
+    """Process arguments passed to CLI."""
+    skip_next = False
+    val_dict = {}
+    for index, arg in enumerate(args):
+        if skip_next:
+            skip_next = False
+            continue
+        if "=" in arg:
+            arg, value = arg.split("=")
+        elif arg in COMMAND_ARGS:
+            try:
+                value = args[index + 1]
+                skip_next = True
+            except IndexError:
+                error(f"Missing value for argument - {arg}")
+            warning('Usage: "{USAGE}"')
+            break
+        if arg == "response_type":
+            if value not in RESPONSE_TYPES:
+                warning(f"Invalid response type request - {value}")
+                warning(f"Choose from {RESPONSE_TYPES}")
+                continue
+        elif arg in {  # pylint: disable=confusing-consecutive-elif
+            "error_code",
+            "delay_by",
+            "clear_after",
+            "data_len",
+        }:
+            try:
+                value = int(value)
+            except ValueError:
+                warning(f"Expected integer value for {arg}")
+                continue
+        val_dict[arg] = value
+    return val_dict
+
+
 async def main(server):
     """Run main."""
     # with patch_stdout():