-
Notifications
You must be signed in to change notification settings - Fork 11
/
loading.py
70 lines (63 loc) · 2.3 KB
/
loading.py
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
import sys
import time
import threading
class Loading:
def __init__(self):
self.spinner = '|/-\\'
self.spinner_index = 0
self.running = False
self.thread = None
def hide_cursor(self):
try:
sys.stdout.write('\033[?25l')
sys.stdout.flush()
except Exception as e:
print(f"[ WARNING ]: Failed to hide cursor: {str(e)}")
def show_cursor(self):
try:
sys.stdout.write('\033[?25h')
sys.stdout.flush()
except Exception as e:
print(f"[ WARNING ]: Failed to show cursor: {str(e)}")
def clear_line(self):
try:
sys.stdout.write('\r\033[K')
sys.stdout.flush()
except Exception as e:
print(f"[ WARNING ]: Failed to clear line: {str(e)}")
def update(self):
while self.running:
try:
sys.stdout.write('\rWaiting for assistant response... ' + self.spinner[self.spinner_index])
sys.stdout.flush()
self.spinner_index = (self.spinner_index + 1) % len(self.spinner)
time.sleep(0.1)
except Exception as e:
print(f"[ WARNING ]: Error updating spinner: {str(e)}")
break
self.clear_line()
self.show_cursor()
def start(self):
if not self.running:
try:
self.running = True
self.hide_cursor()
self.thread = threading.Thread(target=self.update)
self.thread.start()
except Exception as e:
print(f"[ ERROR ]: Failed to start loading animation: {str(e)}")
self.running = False
self.show_cursor()
def stop(self):
if self.running:
try:
self.running = False
if self.thread:
self.thread.join(timeout=1) # Wait for up to 1 second for the thread to finish
if self.thread.is_alive():
print("[ WARNING ]: Loading animation thread did not terminate as expected.")
except Exception as e:
print(f"[ ERROR ]: Failed to stop loading animation: {str(e)}")
finally:
self.show_cursor()
self.clear_line()