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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
|
#!/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!",
"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
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()
# 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.
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.
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()
|
