Как остановить поток, который вызывает длительную задачу из другого скрипта Python?Python

Программы на Python
Anonymous
Как остановить поток, который вызывает длительную задачу из другого скрипта Python?

Сообщение Anonymous »

Я хочу реализовать графический интерфейс, в котором я запускаю long_running_task с помощью кнопки «Пуск». Пока задача выполняется, я анимирую статус в графическом интерфейсе. Вот почему я использую многопоточность и создаю поток для анимации и long_running_task в start_task().
Теперь я хочу иметь возможность прерывать long_running_task во время его выполнения. Поэтому моя кнопка запуска переключается на кнопку отмены (update_button()), которая обновляет переменную состояния self.running, щелкая по ней.
Теперь я проверяю эту переменную состояния на каждой итерации в long_running_task. , который представляет собой просто os.walk() для вывода списка всех файлов в очень большом каталоге.
Если в моем классе определена функция long_running_task(), кнопка отмены работает отлично хорошо (просто раскомментируйте все прокомментированные строки кода, чтобы попробовать самостоятельно).
Моя проблема: на самом деле моя long_running_task содержит много строк кода. Вот почему я хочу сохранить его в отдельном скрипте Python. Поэтому я ввел и связал функцию back_end() с кнопкой старт. Он вызывает функцию long_running_task из другого скрипта, передавая ей переменную статуса. Это должно работать, потому что если вы передаете переменную, вы ссылаетесь только на ячейку памяти, поэтому переменная также всегда имеет текущее значение в функции. Тем не менее, если я вызываю функцию таким образом, моя кнопка отмены больше не останавливает long_running_task.
Я работаю в среде Windows.
Мой код:
Основной:
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()

Другой_скрипт_python:
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



Подробнее здесь: https://stackoverflow.com/questions/786 ... thon-scrip

Вернуться в «Python»