forked from PySimpleGUI/PySimpleGUI
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathDemo_Multithreaded_Logging.py
More file actions
98 lines (75 loc) · 2.57 KB
/
Demo_Multithreaded_Logging.py
File metadata and controls
98 lines (75 loc) · 2.57 KB
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
import sys
if sys.version_info[0] >= 3:
import PySimpleGUI as sg
else:
import PySimpleGUI27 as sg
import queue
import logging
import threading
import time
"""
This code originated in this project:
https://github.com/john144/MultiThreading
Thanks to John for writing this in the early days of PySimpleGUI
Demo program showing one way that a threaded application can function with PySimpleGUI
Events are sent from the ThreadedApp thread to the main thread, the GUI, by using a queue
"""
logger = logging.getLogger('mymain')
def externalFunction():
logger.info('Hello from external app')
logger.info('External app sleeping 5 seconds')
time.sleep(5)
logger.info('External app waking up and exiting')
class ThreadedApp(threading.Thread):
def __init__(self):
super().__init__()
self._stop_event = threading.Event()
def run(self):
externalFunction()
def stop(self):
self._stop_event.set()
class QueueHandler(logging.Handler):
def __init__(self, log_queue):
super().__init__()
self.log_queue = log_queue
def emit(self, record):
self.log_queue.put(record)
def main():
window = sg.FlexForm('Log window', default_element_size=(30, 2), font=('Helvetica', ' 10'), default_button_element_size=(8, 2), return_keyboard_events=True)
layout = \
[
[sg.Multiline(size=(50, 15), key='Log')],
[sg.Button('Start', bind_return_key=True, key='_START_'), sg.Button('Exit')]
]
window.LayoutAndRead(layout, non_blocking=True)
appStarted = False
# Setup logging and start app
logging.basicConfig(level=logging.DEBUG)
log_queue = queue.Queue()
queue_handler = QueueHandler(log_queue)
logger.addHandler(queue_handler)
threadedApp = ThreadedApp()
# Loop taking in user input and querying queue
while True:
# Wake every 100ms and look for work
event, values = window.Read(timeout=100)
if event == '_START_':
if appStarted is False:
threadedApp.start()
logger.debug('App started')
window.FindElement('_START_').Update(disabled=True)
appStarted = True
elif event in (None, 'Exit'):
break
# Poll queue
try:
record = log_queue.get(block=False)
except queue.Empty:
pass
else:
msg = queue_handler.format(record)
window.FindElement('Log').Update(msg+'\n', append=True)
window.Close()
exit()
if __name__ == '__main__':
main()