diff options
| -rwxr-xr-x | examples/async_printing.py | 195 |
1 files changed, 103 insertions, 92 deletions
diff --git a/examples/async_printing.py b/examples/async_printing.py index db2485fe..23147865 100755 --- a/examples/async_printing.py +++ b/examples/async_printing.py @@ -7,88 +7,20 @@ import threading import time from typing import List -import cmd2 from colorama import Fore -ALERTS = ["Mail server is down", - "Blockage detected in sector 4", - "Ants have overrun the break room", - "Your next appointment has arrived", - "Christmas bonuses are cancelled", - "Mandatory overtime this weekend", - "Jimmy quit. Jody got married.", - "Is there anyone on board who knows how to fly a plane?", - "Gentlemen, you can't fight in here. This is the War Room!", - "Your mom goes to college!", - "I'm an ugly stinking llama. Llama face!", - "That's some bad hat, Harry", - "I love the smell of napalm in the morning", - "You're gonna need a bigger boat", - "Snakes, why did it have to be snakes?", - "No, I am your father", - "I'm Batman", - "Goonies never say die", - "You're killing me Smalls", - "It's alive!! It's alive!!", - "Houston, we have a problem!", - "These go to 11", - "Soylent Green is people!!", - "You shall not pass!!", - "KHAAN!!", - "Get to tha choppa!", - "I've heard it both ways", - "Yo Adrian, I did it!!", - "Just what do you think you're doing, Dave?", - "Welcome... to the real world", - "See this? This... is my boomstick!", - "Shop smart. Shop S-MART!", - "It's not the years, it's the mileage", - "I am not an elephant! I am not an animal! I am a human being!", - "Great Scott!!", - "This is heavy", - "Do... or do not. There is no try.", - "Game over man, GAME OVER!", - "SHALL WE PLAY A GAME?"] - - -def get_alerts() -> List[str]: - """ - Randomly reports alerts - :return: the list of alerts - """ - rand_num = random.randint(1, 20) - if rand_num > 2: - return [] - - alerts = [] - for i in range(0, rand_num): - alerts.append(random.choice(ALERTS)) - - return alerts - - -def generate_alert_str() -> str: - """ - Combines alerts into one string that can be printed to the terminal - :return: the alert string - """ - alert_str = '' - alerts = 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 +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?", + "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): @@ -99,16 +31,13 @@ class AlerterApp(cmd2.Cmd): super().__init__(*args, **kwargs) - self._orig_prompt = "(APR)> " - self.prompt = self._orig_prompt - - self.intro = "\nWatch as this application prints alerts and updates the prompt.\n" - self.intro += "This will only happen when the prompt is present. Notice how it doesn't\n" - self.intro += "interfere with your typing.\n" + 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) @@ -118,7 +47,8 @@ class AlerterApp(cmd2.Cmd): """ 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. + # 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) @@ -129,13 +59,87 @@ class AlerterApp(cmd2.Cmd): # 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 _generate_new_prompt(self) -> str: + def _get_alerts(self) -> List[str]: + """ + Reports alerts + :return: the list of alerts """ - Randomly generates a new prompt + 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): + alerts.append("Alert {}".format(self._alert_count)) + self._alert_count += 1 + + 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) @@ -153,24 +157,31 @@ class AlerterApp(cmd2.Cmd): elif rand_num == 5: status_color = Fore.LIGHTBLUE_EX - return status_color + self._orig_prompt + Fore.RESET + 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 = generate_alert_str() + alert_str = self._generate_alert_str() # Generate a new prompt - new_prompt = self._generate_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) + + # No alerts needed to be printed, check if the prompt changed elif new_prompt != self.prompt: self._async_update_prompt(new_prompt) |