Python coding

Using threading

last updated: 2024-01-30

Quick links

gui threading window

Intro

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.

Project Structure

├── 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

The GUI module 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()

The functions module 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)

The main program 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()

Downloads