ich habe ein zeitintensives Programm und nutze deshalb Tkinter. Nach viel Mühe und mit Hilfe von ChatGPT funktioniert es nun einigermaßen.
Ziel ist, dass das Programm stoppt, wenn ich einen KeyboardInterrupt auslöse. Das Tkinter-Fenster soll dann weder einfrieren noch schließen. Ich möchte es stets manuell schließen. Aktuell schließt es aber, wenn ich einen KeyboardInterrupt mache.
Unten ist ein funktionierendes Minimalbeispiel, für das ich mir ebenfalls viel Mühe gegeben habe. Die Klasse ProgressWindow ist dabei die originale Klasse aus dem Skript.
Bitte lasst euch NICHT von der Länge des Codes abschrecken. Nur ProgressWindow ist kompliziert, und zwar so sehr, dass ich selbst nicht mehr durchblicke. Ich vermute aber stark, dass man dort einiges verschlanken könnte ohne Funktionalität einzubüßen. Helft ihr mir bitte bei diesen beiden Punkten? Danke im Voraus.
Code: Alles auswählen
import threading
from queue import Queue
import signal
import time
import random
import string
import tkinter as tk
from tkinter import ttk
class DummyClass:
"""Dummy function for simulating processes."""
def dummy_method1(self) -> None:
"""First simulated method for processing."""
time.sleep(random.randint(1, 2))
def dummy_method2(self) -> None:
"""Second simulated method for processing."""
time.sleep(random.uniform(0.5, 5))
class MainClass:
"""Simulates the main operations for testing purposes."""
def __init__(self, progress_queue: Queue, stop_event: threading.Event):
self.progress_queue = progress_queue
self.stop_event = stop_event
self.dummy_class = DummyClass()
def main(self) -> None:
"""Main simulation function."""
task_dict = {
i: ''.join(random.choices(string.ascii_letters, k=7)) for i in range(10)
}
try:
# Calculate the total number of subtasks
subtask_counts = {task: random.randint(1, 5) for task in task_dict.values()}
total_tasks = sum(subtask_counts.values())
completed_tasks = 0
self.progress_queue.put(("progress", completed_tasks, total_tasks, None))
for task in task_dict.values():
if self.stop_event.is_set():
self.progress_queue.put(("info", "Program stopped by user."))
return
subtask_list = [f"{task}_{i}" for i in range(subtask_counts[task])]
for subtask in subtask_list:
print(subtask) # Simulated subtask logging
self.dummy_class.dummy_method1()
if self.stop_event.is_set():
self.progress_queue.put(("info", "Program stopped by user."))
return
# Update progress
completed_tasks += 1
self.progress_queue.put(("progress", completed_tasks, total_tasks, subtask))
self.dummy_class.dummy_method2()
self.progress_queue.put(("done", None))
except Exception as e:
self.progress_queue.put(("error", str(e)))
class ProgressWindow:
"""Manages a Tkinter window with a progress bar and status updates."""
def __init__(self):
self.root = tk.Tk()
self.root.title("Progress")
self.progress = ttk.Progressbar(
self.root, orient="horizontal", length=300, mode="determinate"
)
self.progress.pack(pady=10)
self.label_progress = tk.Label(self.root, text="Progress: 0.0%")
self.label_progress.pack()
self.label_status = tk.Label(self.root, text="—")
self.label_status.pack()
self.queue: Queue = Queue()
self.stop_event = threading.Event()
self.main_thread: threading.Thread | None
# Ensure proper cleanup on window close
self.root.protocol("WM_DELETE_WINDOW", self.close)
self.root.after(100, self.process_queue)
def start_main(self) -> None:
"""Starts the main process in a separate thread."""
main_class = MainClass(self.queue, self.stop_event)
self.main_thread = threading.Thread(target=main_class.main, daemon=True)
self.main_thread.start()
def process_queue(self) -> None:
"""Processes messages from the queue and updates the UI accordingly."""
if self.root is None or not self.root.winfo_exists():
return
while not self.queue.empty():
message_type, *data = self.queue.get()
if message_type == "progress":
current, total, info = data
percent = (current / total) * 100
if self.root.winfo_exists():
self.progress["value"] = percent
self.label_progress.config(text=f"Progress: {percent:.1f}%")
self.label_status.config(text=info)
elif message_type == "info":
info = data[0]
self.label_status.config(text=info)
elif message_type == "error":
error_msg = data[0]
self.label_status.config(text=f"Error: {error_msg}")
elif message_type == "done":
self.label_status.config(text="Program Complete!")
# Schedule the next queue processing if the window exists
if self.root and self.root.winfo_exists():
self.root.after(100, self.process_queue)
def close(self) -> None:
"""Stops the main process and closes the application."""
self.stop_event.set()
if self.main_thread and self.main_thread.is_alive():
self.main_thread.join(timeout=2)
# Cancel planned callbacks and destroy the Tkinter window
if self.root:
self.root.after_cancel(self.process_queue)
self.root.destroy()
self.root = None
def run(self) -> None:
"""Starts the Tkinter event loop."""
self.start_main()
self.root.mainloop()
def handle_keyboard_interrupt(progress_window: ProgressWindow) -> None:
"""Handles KeyboardInterrupt and ensures a clean shutdown."""
print("\nKeyboardInterrupt detected. Closing application...")
progress_window.close()
if __name__ == "__main__":
progress_window = ProgressWindow()
# Register a signal handler for KeyboardInterrupt
signal.signal(signal.SIGINT, lambda *args: handle_keyboard_interrupt(progress_window))
try:
progress_window.run()
except KeyboardInterrupt:
handle_keyboard_interrupt(progress_window)
