OiO.lk Community platform!

Oio.lk is an excellent forum for developers, providing a wide range of resources, discussions, and support for those in the developer community. Join oio.lk today to connect with like-minded professionals, share insights, and stay updated on the latest trends and technologies in the development field.
  You need to log in or register to access the solved answers to this problem.
  • You have reached the maximum number of guest views allowed
  • Please register below to remove this limitation

How do I stop a thread which calls a long running task from another python script?

  • Thread starter Thread starter user25662262
  • Start date Start date
U

user25662262

Guest
I want to implement a GUI in which I start a long_running_task with a start button. While the task is running, I animate the status in the GUI. That's why I use multi-threading and create a thread for animation and the long_running_task in start_task(). Now I want to be able to abort the long_running_task while it's running. Therefore my start button switches to an abort button (update_button()), which updates a status variable self.running by clicking it.

Now I check that status variable in each iteration in the long_running_task, which is just an os.walk() to list all files in a very big directory.

If I have long_running_task() defined as a function in my class, the abort button works perfectly fine (just uncomment all commented lines of code to try it yourself).

My Problem is: In reality, my long_running_task has many lines of code. That's why I want to safe it in a separate python script. That's why I introduced and connected the function back_end() with the start button. It calls the long_running_task function from an other script, providing the status variable to it. This should work because if you pass a variable, you only reference the memory location, therefore the variable is always at the current value in the function as well. Nevertheless, if I call the function like that, my abort button doesn't stop the long_running_task anymore.

I'm working on a Windows Environment.

Thank you a lot for your help!

My code:

Main:​


Code:
import os
import tkinter as tk
from tkinter import ttk
import threading
import itertools
import time

from other_python_script import long_running_task

class APP():
    def __init__(self, root):
        self.root = root
        self.root.title("Task Manager")

        self.running = False
        self.task_thread = None
        self.anim_thread = None
        self.status_var = tk.StringVar(value="Ready")

        self.directory = "Any directory with great depth"
        
        self.status_label = ttk.Label(self.root, textvariable=self.status_var)
        self.status_label.pack(padx=10, pady=10)

        self.start_abort_button = ttk.Button(self.root, text="Start", command=self.start_task)
        self.start_abort_button.pack(padx=10, pady=5)

    # def long_running_task(self):
    #     self.status_var.set("Running...")
    #     for root, dirs, files in os.walk(self.directory):
    #         if not self.running:
    #             break
    #         else:
    #             self.file_list.append(files)
    #             print(self.file_list)
    #     if self.running:
    #         self.status_var.set("Finished")
    #     else:
    #         self.status_var.set("Aborted")
    #     self.update_button()

    def back_end(self):
        self.status_var.set("Running...")
        file_list = long_running_task(self.running, self.directory)

        if self.running:
            self.status_var.set("Finished")
        else:
            self.status_var.set("Aborted")

        print("FINISHED with total List: ", file_list)
        self.update_button()

            
    def start_task(self):
        print(self.task_thread)
        if self.task_thread is None or not self.task_thread.is_alive():
            self.running = True
            self.status_var.set("Running...")
            self.task_thread = threading.Thread(target=self.back_end, daemon=True)
            #self.task_thread = threading.Thread(target=self.long_running_task, daemon=True)
            self.task_thread.start()
            self.anim_thread = threading.Thread(target=self.animate_status, daemon=True)
            self.anim_thread.start()  
            self.update_button()

            
    def abort_task(self):
        self.running = False
        self.file_list = []
        self.update_button()
        
    def animate_status(self):
        while self.running:
            for status in itertools.cycle(["Running.", "Running..", "Running..."]):
                if not self.running:
                    self.status_var.set("Aborted")
                    return
                self.status_var.set(status)
                time.sleep(0.5)
                
    def update_button(self):
        if self.running:
            self.start_abort_button.configure(text="Abort", command=self.abort_task)
        else:
            self.start_abort_button.configure(text="Start", command=self.start_task)    


if __name__ == '__main__':
    root = tk.Tk()
    APP(root)
    root.mainloop()

Other_python_skript:​


Code:
import os

def long_running_task(running, directory):
    file_list = []
    for root, dirs, files in os.walk(directory):
        if not running:
            return[]
        else:
            file_list.append(files)
            print(file_list)
    return file_list
<p>I want to implement a GUI in which I start a long_running_task with a start button. While the task is running, I animate the status in the GUI. That's why I use multi-threading and create a thread for animation and the long_running_task in start_task().
Now I want to be able to abort the long_running_task while it's running. Therefore my start button switches to an abort button (update_button()), which updates a status variable self.running by clicking it.</p>
<p>Now I check that status variable in each iteration in the long_running_task, which is just an os.walk() to list all files in a very big directory.</p>
<p>If I have long_running_task() defined as a function in my class, the abort button works perfectly fine (just uncomment all commented lines of code to try it yourself).</p>
<p>My Problem is: In reality, my long_running_task has many lines of code. That's why I want to safe it in a separate python script. That's why I introduced and connected the function back_end() with the start button. It calls the long_running_task function from an other script, providing the status variable to it. This should work because if you pass a variable, you only reference the memory location, therefore the variable is always at the current value in the function as well. Nevertheless, if I call the function like that, my abort button doesn't stop the long_running_task anymore.</p>
<p>I'm working on a Windows Environment.</p>
<p>Thank you a lot for your help!</p>
<p>My code:</p>
<h1>Main:</h1>
<pre><code>import os
import tkinter as tk
from tkinter import ttk
import threading
import itertools
import time

from other_python_script import long_running_task

class APP():
def __init__(self, root):
self.root = root
self.root.title("Task Manager")

self.running = False
self.task_thread = None
self.anim_thread = None
self.status_var = tk.StringVar(value="Ready")

self.directory = "Any directory with great depth"

self.status_label = ttk.Label(self.root, textvariable=self.status_var)
self.status_label.pack(padx=10, pady=10)

self.start_abort_button = ttk.Button(self.root, text="Start", command=self.start_task)
self.start_abort_button.pack(padx=10, pady=5)

# def long_running_task(self):
# self.status_var.set("Running...")
# for root, dirs, files in os.walk(self.directory):
# if not self.running:
# break
# else:
# self.file_list.append(files)
# print(self.file_list)
# if self.running:
# self.status_var.set("Finished")
# else:
# self.status_var.set("Aborted")
# self.update_button()

def back_end(self):
self.status_var.set("Running...")
file_list = long_running_task(self.running, self.directory)

if self.running:
self.status_var.set("Finished")
else:
self.status_var.set("Aborted")

print("FINISHED with total List: ", file_list)
self.update_button()


def start_task(self):
print(self.task_thread)
if self.task_thread is None or not self.task_thread.is_alive():
self.running = True
self.status_var.set("Running...")
self.task_thread = threading.Thread(target=self.back_end, daemon=True)
#self.task_thread = threading.Thread(target=self.long_running_task, daemon=True)
self.task_thread.start()
self.anim_thread = threading.Thread(target=self.animate_status, daemon=True)
self.anim_thread.start()
self.update_button()


def abort_task(self):
self.running = False
self.file_list = []
self.update_button()

def animate_status(self):
while self.running:
for status in itertools.cycle(["Running.", "Running..", "Running..."]):
if not self.running:
self.status_var.set("Aborted")
return
self.status_var.set(status)
time.sleep(0.5)

def update_button(self):
if self.running:
self.start_abort_button.configure(text="Abort", command=self.abort_task)
else:
self.start_abort_button.configure(text="Start", command=self.start_task)


if __name__ == '__main__':
root = tk.Tk()
APP(root)
root.mainloop()
</code></pre>
<h1>Other_python_skript:</h1>
<pre><code>import os

def long_running_task(running, directory):
file_list = []
for root, dirs, files in os.walk(directory):
if not running:
return[]
else:
file_list.append(files)
print(file_list)
return file_list

</code></pre>
 

Latest posts

Top