Merge pull request #539 from python-cmd2/alert_printer

Alert printer

1 files changed, 203 insertions, 0 deletions

diff --git a/examples/async_printing.py b/examples/async_printing.py
new file mode 100755
index 00000000..a4165ae8
--- /dev/null
+++ b/examples/async_printing.py
@@ -0,0 +1,203 @@
+#!/usr/bin/env python
+# coding=utf-8
+"""
+A simple example demonstrating an application that asynchronously prints alerts, updates the prompt
+and changes the window title
+"""
+
+import random
+import threading
+import time
+from typing import List
+
+from colorama import Fore
+
+import cmd2
+
+ALERTS = ["Watch as this application prints alerts and updates the prompt",
+          "This will only happen when the prompt is present",
+          "Notice how it doesn't interfere with your typing or cursor location",
+          "Go ahead and type some stuff and move the cursor throughout the line",
+          "Keep typing...",
+          "Move that cursor...",
+          "Pretty seamless, eh?",
+          "Feedback can also be given in the window title. Notice the arg count up there?",
+          "You can stop and start the alerts by typing stop_alerts and start_alerts",
+          "This demo will now continue to print alerts at random intervals"
+          ]
+
+
+class AlerterApp(cmd2.Cmd):
+    """ An app that shows off async_alert() and async_update_prompt() """
+
+    def __init__(self, *args, **kwargs) -> None:
+        """ Initializer """
+
+        super().__init__(*args, **kwargs)
+
+        self.prompt = "(APR)> "
+
+        # The thread that will asynchronously alert the user of events
+        self._stop_thread = False
+        self._alerter_thread = threading.Thread()
+        self._alert_count = 0
+        self._next_alert_time = 0
+
+        # Create some hooks to handle the starting and stopping of our thread
+        self.register_preloop_hook(self._preloop_hook)
+        self.register_postloop_hook(self._postloop_hook)
+
+    def _preloop_hook(self) -> None:
+        """ Start the alerter thread """
+        # This runs after cmdloop() acquires self.terminal_lock, which will be locked until the prompt appears.
+        # Therefore this is the best place to start the alerter thread since there is no risk of it alerting
+        # before the prompt is displayed. You can also start it via a command if its not something that should
+        # be running during the entire application. See do_start_alerts().
+        self._stop_thread = False
+
+        self._alerter_thread = threading.Thread(name='alerter', target=self._alerter_thread_func)
+        self._alerter_thread.start()
+
+    def _postloop_hook(self) -> None:
+        """ Stops the alerter thread """
+
+        # After this function returns, cmdloop() releases self.terminal_lock which could make the alerter
+        # thread think the prompt is on screen. Therefore this is the best place to stop the alerter thread.
+        # You can also stop it via a command. See do_stop_alerts().
+        self._stop_thread = True
+        if self._alerter_thread.is_alive():
+            self._alerter_thread.join()
+
+    def do_start_alerts(self, _):
+        """ Starts the alerter thread """
+        if self._alerter_thread.is_alive():
+            print("The alert thread is already started")
+        else:
+            self._stop_thread = False
+            self._alerter_thread = threading.Thread(name='alerter', target=self._alerter_thread_func)
+            self._alerter_thread.start()
+
+    def do_stop_alerts(self, _):
+        """ Stops the alerter thread """
+        self._stop_thread = True
+        if self._alerter_thread.is_alive():
+            self._alerter_thread.join()
+        else:
+            print("The alert thread is already stopped")
+
+    def _get_alerts(self) -> List[str]:
+        """
+        Reports alerts
+        :return: the list of alerts
+        """
+        global ALERTS
+
+        cur_time = time.monotonic()
+        if cur_time < self._next_alert_time:
+            return []
+
+        alerts = []
+
+        if self._alert_count < len(ALERTS):
+            alerts.append(ALERTS[self._alert_count])
+            self._alert_count += 1
+            self._next_alert_time = cur_time + 4
+
+        else:
+            rand_num = random.randint(1, 20)
+            if rand_num > 2:
+                return []
+
+            for i in range(0, rand_num):
+                self._alert_count += 1
+                alerts.append("Alert {}".format(self._alert_count))
+
+            self._next_alert_time = 0
+
+        return alerts
+
+    def _generate_alert_str(self) -> str:
+        """
+        Combines alerts into one string that can be printed to the terminal
+        :return: the alert string
+        """
+        global ALERTS
+
+        alert_str = ''
+        alerts = self._get_alerts()
+
+        longest_alert = max(ALERTS, key=len)
+        num_astericks = len(longest_alert) + 8
+
+        for i, cur_alert in enumerate(alerts):
+            # Use padding to center the alert
+            padding = ' ' * int((num_astericks - len(cur_alert)) / 2)
+
+            if i > 0:
+                alert_str += '\n'
+            alert_str += '*' * num_astericks + '\n'
+            alert_str += padding + cur_alert + padding + '\n'
+            alert_str += '*' * num_astericks + '\n'
+
+        return alert_str
+
+    def _generate_colored_prompt(self) -> str:
+        """
+        Randomly generates a colored the prompt
+        :return: the new prompt
+        """
+        rand_num = random.randint(1, 20)
+
+        status_color = Fore.RESET
+
+        if rand_num == 1:
+            status_color = Fore.LIGHTRED_EX
+        elif rand_num == 2:
+            status_color = Fore.LIGHTYELLOW_EX
+        elif rand_num == 3:
+            status_color = Fore.CYAN
+        elif rand_num == 4:
+            status_color = Fore.LIGHTGREEN_EX
+        elif rand_num == 5:
+            status_color = Fore.LIGHTBLUE_EX
+
+        return status_color + self.visible_prompt + Fore.RESET
+
+    def _alerter_thread_func(self) -> None:
+        """ Prints alerts and updates the prompt any time the prompt is showing """
+
+        self._alert_count = 0
+        self._next_alert_time = 0
+
+        while not self._stop_thread:
+            # Always acquire terminal_lock before printing alerts or updating the prompt
+            # To keep the app responsive, do not block on this call
+            if self.terminal_lock.acquire(blocking=False):
+
+                # Get any alerts that need to be printed
+                alert_str = self._generate_alert_str()
+
+                # Generate a new prompt
+                new_prompt = self._generate_colored_prompt()
+
+                # Check if we have alerts to print
+                if alert_str:
+                    # new_prompt is an optional parameter to async_alert()
+                    self.async_alert(alert_str, new_prompt)
+                    new_title = "Alerts Printed: {}".format(self._alert_count)
+                    self.set_window_title(new_title)
+
+                # No alerts needed to be printed, check if the prompt changed
+                elif new_prompt != self.prompt:
+                    self.async_update_prompt(new_prompt)
+
+                # Don't forget to release the lock
+                self.terminal_lock.release()
+
+            time.sleep(0.5)
+
+
+if __name__ == '__main__':
+    app = AlerterApp()
+    app.set_window_title("Asynchronous Printer Test")
+    app.cmdloop()