diff --git a/python/state-cli.md b/python/state-cli.md new file mode 100644 index 0000000..1157e01 --- /dev/null +++ b/python/state-cli.md @@ -0,0 +1,191 @@ +Implementation of a command line interface that provides different commands +based on the state of a state machine. + +``` +import prompt_toolkit +from prompt_toolkit.styles import Style +from prompt_toolkit.patch_stdout import StdoutProxy, patch_stdout +import time +import threading +import logging + + +# +# StateCli +# + + +class StateCommandBase: + """Base class for state-specific commands.""" + + def __init__(self, cli, state_machine): + self.cli = cli + self.state_machine = state_machine + + +class StateCli: + """ + Implementation of a CLI working in conjuction with a state machine. + Depending on the state, the CLI provides different commands. + """ + + def __init__(self, state_machine, command_map): + self.state_machine = state_machine + + self.state_commands = { + key: value(self, state_machine) for key, value in command_map.items() + } + + self.update_prompt() + + self._prompt_style = Style.from_dict( + { + "prompt": "ansiblue bold", + "input": "", + } + ) + + self.session = prompt_toolkit.PromptSession() + + def update_prompt(self): + self.prompt = f"{self.state_machine.current_state}> " + + def _process_command(self, command, args): + current_state = self.state_machine.current_state + if current_state not in self.state_commands: + print(f"No commands defined for state {current_state}") + return + + state_cmd_obj = self.state_commands[current_state] + if not hasattr(state_cmd_obj, command): + print(f"Unknown command: '{command}'") + self._do_help() + return + + getattr(state_cmd_obj, command)(" ".join(args)) + + def cmdloop(self): + with patch_stdout(): + while True: + try: + user_input = self.session.prompt( + self.prompt, style=self._prompt_style + ) + + parts = user_input.split() + command, args = parts[0], parts[1:] + + if command == "help": + self._do_help() + continue + elif command == "exit": + break + + self._process_command(command, args) + except KeyboardInterrupt: # ctrl+c + break + except EOFError: # ctrl+d + break + + def _do_help(self): + current_state = self.state_machine.current_state + print("=" * 50) + print("Available commands in any state") + print(f" {'help':15} - Show this help") + print(f" {'exit':15} - Exit the application") + + print("=" * 50) + print(f"Available commands in state {current_state}") + + if current_state not in self.state_commands: + print(f" No commands defined for state {current_state}") + print("=" * 50) + return + + state_cmd_obj = self.state_commands[current_state] + commands = [ + name + for name in dir(state_cmd_obj) + if not name.startswith("_") and callable(getattr(state_cmd_obj, name)) + ] + + if not commands: + print(" No commands available") + print("=" * 50) + return + + for cmd in sorted(commands): + method = getattr(state_cmd_obj, cmd) + if method.__doc__: + doc = method.__doc__.strip() + first_line = doc.split("\n")[0] + print(f" {cmd:15} - {first_line}") + else: + print(f" {cmd:15} - No documentation available") + + print("=" * 50) + + +# +# StateMachine +# + + +class StateMachine: + def __init__(self): + self.current_state = "idle" + + def start(self): + self.current_state = "running" + + def stop(self): + self.current_state = "idle" + + +# +# Usage +# + + +class IdleCommands(StateCommandBase): + def start(self, args): + """Start the machine""" + print("Starting the machine") + self.state_machine.start() + self.cli.update_prompt() + + +class RunningCommands(StateCommandBase): + def _stop(self, args): + """Stop the machine""" + print("Stopping the the machine") + self.state_machine.stop() + self.cli.update_prompt() + + +def background_task(): + while True: + time.sleep(2) + logging.info("This is a log message from some background task") + + +def main(): + sm = StateMachine() + cli = StateCli(sm, {"idle": IdleCommands, "running": RunningCommands}) + + logging.basicConfig( + format="%(asctime)s %(levelname)s %(message)s", + datefmt="%H:%M:%S", + level=logging.INFO, + stream=StdoutProxy(), + ) + + thread = threading.Thread(target=background_task, daemon=True) + thread.start() + + cli.cmdloop() + + +if __name__ == "__main__": + main() +```