last updated: 2024-01-30
If my python program is using a GUI and other modules, it would be nice to run the modules in different threads using threading.
This project is a simple application that demonstrates the use of threading and a graphical user interface (GUI) in Python (Copilot helped :)). The application consists of three main components: a main program logic that reacts to a button press and passes text from entry fields, a GUI that allows the user to interact with the application and functions that are executed. I want to use this as a template for other projects.
├── src
│ ├── main.py # Entry point of the application
│ ├── functions.py # module with our functions
│ └── gui.py # GUI module with a button
├── requirements.txt # Dependencies for the project
└── README.md # Project documentation
gui.py
I wanted three buttons, two entry fields and a text window. The GUI methods are in a class.
All the widget texts are in one dictionary to facilitate the changing of these texts. The module is communicating with flags and text queues with the main program. The start_gui(flags_2_main, queue_2_main, queue_2_gui)
function creates the GUI object and runs the GUI loop. It is invoked in the main program.
import tkinter as tk
import queue
class GUI:
def __init__(self, flags_2_main, queues_2_main, queues_2_gui):
self.flags_2_main = flags_2_main
self.queues_2_main = queues_2_main
self.queues_2_gui = queues_2_gui
self.root = tk.Tk()
self.root.title("Simple GUI using threading")
self.widget_texts_dict = { # Dictionary for button texts (button:message)
"Press Me 1 ": "Button 1 was pressed!",
"Press Me 2" : "Button 2 was pressed!",
"Press Me 3" : "Button 3 was pressed!",
"Submit 1" : "Hello",
"Submit 2" : "Do it now!"}
self.widget_texts_list = list(self.widget_texts_dict.keys())
self.button1 = tk.Button(self.root, # Button 1
text=self.widget_texts_list[0],
command=lambda: self.set_flag(self.flags_2_main[0],
self.widget_texts_dict[self.widget_texts_list[0]]))
self.button1.pack(pady=10)
self.button2 = tk.Button(self.root, # Button 2
text=self.widget_texts_list[1],
command=lambda: self.set_flag(self.flags_2_main[1],
self.widget_texts_dict[self.widget_texts_list[1]]))
self.button2.pack(pady=10)
self.button3 = tk.Button(self.root, # Button 3
text=self.widget_texts_list[2],
command=lambda: self.set_flag(self.flags_2_main[2],
self.widget_texts_dict[self.widget_texts_list[2]]))
self.button3.pack(pady=10)
self.text_window = tk.Text(self.root, height=10, width=60) # Text window
self.text_window.pack(pady=20)
self.entry1 = tk.Entry(self.root) # Entry window 1
self.entry1.insert(0, self.widget_texts_dict[self.widget_texts_list[3]])
self.entry1.pack(pady=10)
self.submit_button1 = tk.Button(self.root,
text=self.widget_texts_list[3],
command=lambda: self.submit_text(0))
self.submit_button1.pack(pady=10)
self.entry2 = tk.Entry(self.root) # Entry window 2
self.entry2.insert(0, self.widget_texts_dict[self.widget_texts_list[4]])
self.entry2.pack(pady=10)
self.submit_button2 = tk.Button(self.root,
text=self.widget_texts_list[4],
command=lambda: self.submit_text(1))
self.submit_button2.pack(pady=10)
self.root.protocol("WM_DELETE_WINDOW", self.on_closing) # Handle window close event
self.root.after(100, self.check_queue_from_main) # Periodically check the GUI queue
def set_flag(self, flag, message):
flag.set()
self.text_window.insert(tk.END, message + "\n")
def submit_text(self, queue_index):
if queue_index == 0:
text = self.entry1.get()
else:
text = self.entry2.get()
self.text_window.insert(tk.END, f"Entry {queue_index + 1} submitted: " + text + "\n")
self.queues_2_main.put(text) # Add the text to the appropriate queue
def check_queue_from_main(self):
try:
while True:
message = self.queues_2_gui.get_nowait()
self.text_window.insert(tk.END, message + "\n")
except queue.Empty:
pass
self.root.after(100, self.check_queue_from_main) # Check the queue again after 100ms
def on_closing(self):
self.flags_2_main[3].set() # Set the exit flag to stop the main loop
self.root.destroy()
def run(self):
self.root.mainloop()
def start_gui(flags_2_main, queues_2_main, queues_2_gui):
gui = GUI(flags_2_main, queues_2_main, queues_2_gui)
gui.run()
functions.py
The functions invoked by pressing a button are in an own file and class.
class MyFunctions:
def __init__(self, queue_2_gui):
self.queue_2_gui = queue_2_gui
def function_1(self):
text = "Function 1 executed, text transported in queues_2_gui"
self.queue_2_gui.put(text)
def function_2(self):
text = "Function 2 executed, text transported in queues_2_gui"
self.queue_2_gui.put(text)
def function_3(self):
text = "Function 3 executed, text transported in queues_2_gui"
self.queue_2_gui.put(text)
main.py
The flags are threading events objects to signal between threads. Flags and Queues are thread-safe and provide a simple way to manage data exchange between threads. They handle synchronization internally. We use 3 button flags and an exit flag to close main when the GUI window is closed. Two queues handle messages from the GUI to main and from Main to GUI.
After creating the threads they need to be started. The join
command is used to ensure that the main program waits for the threads to complete before exiting. It blocks the calling thread (in this case, the main thread) until the thread whose join
method is called terminates. This is important to ensure that the program does not exit prematurely while the threads are still running.
import threading
import time
import queue
from gui import start_gui
from functions import MyFunctions
def main_loop(myfunc, flags_2_main, queue_2_main, queue_2_gui):
""" Main loop that checks if any flag is set """
while not flags_2_main[3].is_set(): # Check the exit flag
if flags_2_main[0].is_set():
myfunc.function_1()
flags_2_main[0].clear()
if flags_2_main[1].is_set():
myfunc.function_2()
flags_2_main[1].clear()
if flags_2_main[2].is_set():
myfunc.function_3()
flags_2_main[2].clear()
try:
text = queue_2_main.get_nowait()
print(f"Received text from gui: {text}")
queue_2_gui.put(f"Text '{text}' well received from Main")
# Here you can add code to handle the text
except queue.Empty:
pass
time.sleep(0.1)
def main():
flag_b1 = threading.Event() # Create Event objects to signal between threads
flag_b2 = threading.Event()
flag_b3 = threading.Event()
flag_exit = threading.Event() # Exit flag
flags_2_main = [flag_b1, flag_b2, flag_b3, flag_exit]
queue_2_main = queue.Queue() # Create queue for text communication between GUI and main_loop
queue_2_gui = queue.Queue() # Create a queue for communication from main_loop to GUI
myfunc = MyFunctions(queue_2_gui) # Create an instance of MyFunctions
gui_thread = threading.Thread(target=start_gui, # Create GUI thread
args=(flags_2_main, queue_2_main, queue_2_gui))
main_thread = threading.Thread(target=main_loop, # Create main_loop thread
args=(myfunc, flags_2_main, queue_2_main, queue_2_gui))
gui_thread.start() # Start both threads
main_thread.start()
gui_thread.join() # Wait for both threads to complete
main_thread.join()
if __name__ == "__main__":
main()