1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
#!/usr/bin/env python
# coding=utf-8
"""A simple example demonstrating an application that asynchronously prints alerts and updates the prompt"""
import random
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!"]
def get_alerts() -> List[str]:
"""
Randomly reports alerts
:return: the list of alerts
"""
rand_num = random.randint(1, 20)
if rand_num > 3:
return []
alerts = []
for i in range(0, rand_num):
cur_alert = random.choice(ALERTS)
if cur_alert in alerts:
i -= 1
else:
alerts.append(cur_alert)
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
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._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"
# The thread that will asynchronously alert the user of events
self._stop_thread = False
self._alerter_thread = threading.Thread(name='alerter', target=self._alerter_thread_func)
# 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 function runs after cmdloop() locks _terminal_lock, which will be locked until the prompt appears.
# Therefore it is safe to start our thread since there is no risk of it alerting before the prompt is displayed.
self._stop_thread = False
self._alerter_thread.start()
def _postloop_hook(self) -> None:
""" Stops the alerter thread """
self._stop_thread = True
if self._alerter_thread.is_alive():
self._alerter_thread.join()
def _generate_new_prompt(self) -> str:
"""
Randomly generates a new 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._orig_prompt + Fore.RESET
def _alerter_thread_func(self) -> None:
""" Prints alerts and updates the prompt any time the prompt is showing """
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()
# Generate a new prompt
new_prompt = self._generate_new_prompt()
if alert_str:
self._async_alert(alert_str, new_prompt)
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()
|
