Das deutsche Python-Forum

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)

Code: Alles auswählen

import tkinter as tk


class CounterApp:
    def __init__(self, root):
        self.root = root
        self.counter_id = None
        self.counter_var = tk.IntVar(root, value=0)
        tk.Label(root, textvariable=self.counter_var).pack()
        self.start_button = tk.Button(root, text="Start", command=self.start_counter)
        self.start_button.pack()
        root.bind("<Control-c>", self.stop_counter)

    def start_counter(self):
        self.start_button.config(state=tk.DISABLED)
        self.update_counter()

    def stop_counter(self, event=None):
        self.start_button.config(state=tk.NORMAL)
        if self.counter_id is not None:
            self.root.after_cancel(self.counter_id)
            self.counter_id = None

    def update_counter(self):
        self.counter_id = self.root.after(100, self.update_counter)
        self.counter_var.set(self.counter_var.get() + 1)


def main():
    root = tk.Tk()
    app = CounterApp(root)
    root.mainloop()


if __name__ == "__main__":
    main()

Code: Alles auswählen

class ProgressWindow:
    """Manages a Tkinter window with a progress bar and status updates."""

    def __init__(self, root):
        self.root = root
        self.root.title("Progress")

        self.progress = ttk.Progressbar(
            root, orient="horizontal", length=300, mode="determinate"
        )
        self.progress.pack(pady=10)

        self.label_progress = tk.Label(root, text="Progress: 0.0%")
        self.label_progress.pack()

        self.label_status = tk.Label(root, text="—")
        self.label_status.pack()

        self.queue: Queue = Queue()
        self.stop_event = threading.Event()
        self.main_thread: threading.Thread | None = None

        self.start_button = tk.Button(root, text="Start", command=self.start_main)
        self.start_button.pack(pady=5)

        self.stop_button = tk.Button(root, text="Stop", command=self.stop_main, state=tk.DISABLED)
        self.stop_button.pack(pady=5)

        self.update_id = None

    def start_main(self):
        """Starts the main process in a separate thread."""
        self.start_button.config(state=tk.DISABLED)
        self.stop_button.config(state=tk.NORMAL)

        main_class = MainClass(self.queue, self.stop_event)
        self.main_thread = threading.Thread(target=main_class.main, daemon=True)
        self.main_thread.start()

        self.process_queue()

    def process_queue(self):
        """Processes messages from the queue and updates the UI accordingly."""
        while not self.queue.empty():
            message_type, *data = self.queue.get()

            if message_type == "progress":
                current, total, info = data
                percent = (current / total) * 100
                self.progress["value"] = percent
                self.label_progress.config(text=f"Progress: {percent:.1f}%")
                self.label_status.config(text=info if info else "Processing...")

            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!")
                self.stop_button.config(state=tk.DISABLED)

        if not self.stop_event.is_set():
            self.update_id = self.root.after(100, self.process_queue)

    def stop_main(self):
        """Stops the main process."""
        self.stop_event.set()
        self.start_button.config(state=tk.NORMAL)
        self.stop_button.config(state=tk.DISABLED)

        if self.main_thread and self.main_thread.is_alive():
            self.main_thread.join(timeout=2)

        if self.update_id is not None:
            self.root.after_cancel(self.update_id)

    def run(self):
        """Starts the Tkinter event loop."""
        self.root.mainloop()


def main():
    root = tk.Tk()
    app = ProgressApp(root)
    app.run()


if __name__ == "__main__":
    main()