Python/scheduling/cpuschedulingalgorithms.py

import copy
import threading
import time
import tkinter as tk
from collections.abc import Generator
from tkinter import messagebox, ttk

import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg


# ===================== Scheduler Engine ===================== #
class SchedulerEngine:
    def __init__(self, processes: list[dict], algorithm: str, quantum: int = 2) -> None:
        """
        Initializes the scheduler engine.

        >>> processes = [{"pid": "P1", "arrival": 0, "burst": 4}]
        >>> engine = SchedulerEngine(processes, "FCFS")
        >>> isinstance(engine, SchedulerEngine)
        True
        """
        self.original = copy.deepcopy(processes)
        self.processes = [p.copy() for p in processes]
        self.algorithm = algorithm
        self.quantum = quantum
        for process in self.processes:
            process["remaining"] = process["burst"]
        self.timeline: list[tuple[int, str]] = []  # [(time, pid)]
        self.stats: list[tuple] = []

    def simulate(self) -> Generator[tuple[int, str | None, list[str]]]:
        """
        Runs the selected CPU scheduling algorithm.

        >>> processes = [{"pid": "P1", "arrival": 0, "burst": 2}]
        >>> engine = SchedulerEngine(processes, "FCFS")
        >>> list(engine.simulate())[:2]
        [(0, 'P1', []), (1, 'P1', [])]
        """
        algo = self.algorithm.lower()
        if algo == "fcfs":
            yield from self._simulate_fcfs()
        elif algo == "sjf (non-preemptive)":
            yield from self._simulate_sjf_np()
        elif algo == "sjf (preemptive)":
            yield from self._simulate_sjf_p()
        elif algo == "priority (non-preemptive)":
            yield from self._simulate_priority_np()
        elif algo == "priority (preemptive)":
            yield from self._simulate_priority_p()
        elif algo == "round robin":
            yield from self._simulate_rr()
        self._calculate_stats()

    # first come first serve
    def _simulate_fcfs(self) -> Generator[tuple[int, str, list[str]]]:
        """
        Simulates First Come First Serve scheduling.

        >>> processes = [{"pid": "P1", "arrival": 0, "burst": 2}]
        >>> engine = SchedulerEngine(processes, "FCFS")
        >>> next(engine._simulate_fcfs())
        (0, 'P1', [])
        """
        t = 0
        processes = sorted(self.processes, key=lambda process: process["arrival"])
        for process in processes:
            t = max(t, process["arrival"])
            for _ in range(process["burst"]):
                self.timeline.append((t, process["pid"]))
                yield (t, process["pid"], [])
                t += 1
            process["completion"] = t

    # shortest job first non preemptive
    def _simulate_sjf_np(self) -> Generator[tuple[int, str | None, list[str]]]:
        """
        Simulates Shortest Job First (Non-Preemptive).

        >>> processes = [{"pid": "P1", "arrival": 0, "burst": 2}]
        >>> engine = SchedulerEngine(processes, "SJF (Non-Preemptive)")
        >>> isinstance(engine._simulate_sjf_np(), Generator)
        True
        """
        t = 0
        processes = sorted(self.processes, key=lambda process: process["arrival"])
        done = 0
        while done < len(processes):
            ready = [
                process
                for process in processes
                if process["arrival"] <= t and "completion" not in process
            ]
            if not ready:
                t += 1
                yield (t, None, [])
                continue
            process = min(ready, key=lambda x: x["burst"])
            for _ in range(process["burst"]):
                self.timeline.append((t, process["pid"]))
                yield (t, process["pid"], [])
                t += 1
            process["completion"] = t
            done += 1

    # shortest job first preemptive
    def _simulate_sjf_p(self) -> Generator[tuple[int, str | None, list[str]]]:
        """Simulates SJF Preemptive scheduling."""
        t = 0
        processes = sorted(self.processes, key=lambda process: process["arrival"])
        done = 0
        while done < len(processes):
            ready = [
                process
                for process in processes
                if process["arrival"] <= t and process["remaining"] > 0
            ]
            if not ready:
                t += 1
                yield (t, None, [])
                continue
            process = min(ready, key=lambda x: x["remaining"])
            self.timeline.append((t, process["pid"]))
            yield (t, process["pid"], [])
            process["remaining"] -= 1
            if process["remaining"] == 0:
                process["completion"] = t + 1
                done += 1
            t += 1

    # priority non preemptive
    def _simulate_priority_np(self) -> Generator[tuple[int, str | None, list[str]]]:
        """Simulates Priority (Non-Preemptive) scheduling."""
        t = 0
        done = 0
        while done < len(self.processes):
            ready = [
                process
                for process in self.processes
                if process["arrival"] <= t and "completion" not in process
            ]
            if not ready:
                t += 1
                yield (t, None, [])
                continue
            process = min(ready, key=lambda x: x["priority"])
            for _ in range(process["burst"]):
                self.timeline.append((t, process["pid"]))
                yield (t, process["pid"], [])
                t += 1
            process["completion"] = t
            done += 1

    # priority preemptive
    def _simulate_priority_p(self) -> Generator[tuple[int, str | None, list[str]]]:
        """Simulates Priority (Preemptive) scheduling."""
        t = 0
        done = 0
        while done < len(self.processes):
            ready = [
                process
                for process in self.processes
                if process["arrival"] <= t and process["remaining"] > 0
            ]
            if not ready:
                t += 1
                yield (t, None, [])
                continue
            process = min(ready, key=lambda x: x["priority"])
            self.timeline.append((t, process["pid"]))
            yield (t, process["pid"], [])
            process["remaining"] -= 1
            if process["remaining"] == 0:
                process["completion"] = t + 1
                done += 1
            t += 1

    # round robin
    def _simulate_rr(self) -> Generator[tuple[int, str | None, list[str]]]:
        """Simulates Round Robin scheduling."""
        t = 0
        q: list[dict] = []
        processes = sorted(self.processes, key=lambda process: process["arrival"])
        i = 0
        done = 0
        while done < len(processes):
            while i < len(processes) and processes[i]["arrival"] <= t:
                q.append(processes[i])
                i += 1
            if not q:
                t += 1
                yield (t, None, [])
                continue
            process = q.pop(0)
            burst = min(self.quantum, process["remaining"])
            for _ in range(burst):
                self.timeline.append((t, process["pid"]))
                yield (t, process["pid"], [x["pid"] for x in q])
                t += 1
                process["remaining"] -= 1
                while i < len(processes) and processes[i]["arrival"] <= t:
                    q.append(processes[i])
                    i += 1
            if process["remaining"] > 0:
                q.append(process)
            else:
                process["completion"] = t
                done += 1

    def _calculate_stats(self) -> None:
        """Calculates turnaround, waiting, and response times."""
        for process in self.processes:
            pid = process["pid"]
            arrival = process["arrival"]
            burst = process["burst"]
            completion = process["completion"]
            first_exec = next((t for t, pid2 in self.timeline if pid2 == pid), arrival)
            tat = completion - arrival
            wt = tat - burst
            rt = first_exec - arrival
            self.stats.append((pid, arrival, burst, completion, tat, wt, rt))


# ===================== Interface ===================== #
class CPUSchedulerGUI:
    def __init__(self, root: tk.Tk) -> None:
        """Initializes the GUI window."""
        self.root = root
        self.root.title("CPU Scheduling Visualizer")
        self.root.geometry("1000x700")
        self.processes: list[dict] = []
        self.setup_ui()

    def setup_ui(self) -> None:
        """Sets up GUI widgets."""
        top_frame = ttk.Frame(self.root)
        top_frame.pack(pady=10)

        self.tree = ttk.Treeview(
            top_frame,
            columns=("pid", "arrival", "burst", "priority"),
            show="headings",
        )
        for col in self.tree["columns"]:
            self.tree.heading(col, text=col.capitalize())
        self.tree.pack(side="left")

        form = ttk.Frame(top_frame)
        form.pack(side="left", padx=10)
        ttk.Label(form, text="PID").grid(row=0, column=0)
        ttk.Label(form, text="Arrival").grid(row=1, column=0)
        ttk.Label(form, text="Burst").grid(row=2, column=0)
        ttk.Label(form, text="Priority").grid(row=3, column=0)
        self.pid_e = ttk.Entry(form)
        self.arrival_e = ttk.Entry(form)
        self.burst_e = ttk.Entry(form)
        self.priority_e = ttk.Entry(form)
        self.pid_e.grid(row=0, column=1)
        self.arrival_e.grid(row=1, column=1)
        self.burst_e.grid(row=2, column=1)
        self.priority_e.grid(row=3, column=1)
        ttk.Button(
            form,
            text="Add",
            command=self.add_process,
        ).grid(row=4, column=0, pady=5)
        ttk.Button(
            form,
            text="Delete",
            command=self.delete_process,
        ).grid(row=4, column=1)
        algo_frame = ttk.Frame(self.root)
        algo_frame.pack(pady=10)
        ttk.Label(algo_frame, text="Algorithm:").pack(side="left")
        self.algo_cb = ttk.Combobox(
            algo_frame,
            values=[
                "FCFS",
                "SJF (Non-Preemptive)",
                "SJF (Preemptive)",
                "Priority (Non-Preemptive)",
                "Priority (Preemptive)",
                "Round Robin",
            ],
        )
        self.algo_cb.current(0)
        self.algo_cb.pack(side="left", padx=5)
        ttk.Label(algo_frame, text="Quantum:").pack(side="left")
        self.quantum_e = ttk.Entry(algo_frame, width=5)
        self.quantum_e.insert(0, "2")
        self.quantum_e.pack(side="left")
        ttk.Button(
            algo_frame,
            text="Run",
            command=self.run_scheduling,
        ).pack(side="left", padx=10)
        self.ready_label = ttk.Label(self.root, text="Ready Queue:")
        self.ready_list = tk.Listbox(self.root, height=3)
        self.ready_label.pack_forget()
        self.ready_list.pack_forget()

        self.figure, self.ax = plt.subplots(figsize=(8, 3))
        self.canvas = FigureCanvasTkAgg(self.figure, master=self.root)
        self.canvas.get_tk_widget().pack()

        self.result_box = ttk.Treeview(
            self.root,
            columns=("pid", "arrival", "burst", "completion", "tat", "wt", "rt"),
            show="headings",
            height=6,
        )
        for col in self.result_box["columns"]:
            self.result_box.heading(col, text=col.upper())
        self.result_box.pack(pady=10)

        self.avg_label = ttk.Label(self.root, text="", font=("Arial", 11, "bold"))
        self.avg_label.pack()

    def add_process(self) -> None:
        """Adds a new process entry to the table."""
        try:
            pid = self.pid_e.get()
            arrival = int(self.arrival_e.get())
            burst = int(self.burst_e.get())
            priority = int(self.priority_e.get() or 0)
            self.processes.append(
                {"pid": pid, "arrival": arrival, "burst": burst, "priority": priority}
            )
            self.tree.insert("", "end", values=(pid, arrival, burst, priority))
        except ValueError:
            messagebox.showerror("Error", "Invalid input")

    def delete_process(self) -> None:
        """Deletes a selected process."""
        if sel := self.tree.selection():
            pid = self.tree.item(sel[0])["values"][0]
            self.processes = [p for p in self.processes if p["pid"] != pid]
            self.tree.delete(sel[0])

    def run_scheduling(self) -> None:
        """Runs the selected scheduling algorithm."""
        algo = self.algo_cb.get()
        quantum = int(self.quantum_e.get() or 2)
        if algo.lower() == "round robin":
            self.ready_label.pack()
            self.ready_list.pack()
        else:
            self.ready_label.pack_forget()
            self.ready_list.pack_forget()

        self.engine = SchedulerEngine(self.processes, algo, quantum)
        threading.Thread(target=self.animate, daemon=True).start()

    def animate(self) -> None:
        """Animates the scheduling visualization."""
        self.ax.clear()
        x: int = 0
        colors: dict[str, str] = {}
        current_pid: str | None = None
        for step in self.engine.simulate():
            _, pid, rq = step
            if pid:
                if pid != current_pid:
                    self.ax.axvline(x, color="black", linewidth=0.8)
                    current_pid = pid
                colors.setdefault(pid, plt.cm.tab20(len(colors) % 20))
                self.ax.barh(0, 1, left=x, color=colors[pid])
                self.ax.text(
                    x + 0.5,
                    0,
                    pid,
                    ha="center",
                    va="center",
                    color="white",
                    fontsize=9,
                )
                x += 1
                self.ax.set_xticks(range(x + 1))
                self.ax.set_yticks([])
                self.ax.set_xlabel("Time")
                self.canvas.draw()
                if rq:
                    self.ready_list.delete(0, tk.END)
                    for pid_r in rq:
                        self.ready_list.insert(tk.END, pid_r)
            time.sleep(0.3)
        self.show_results()

    def show_results(self) -> None:
        """Displays scheduling results."""
        for item in self.result_box.get_children():
            self.result_box.delete(item)
        total_wt = total_tat = total_rt = 0
        for row in self.engine.stats:
            self.result_box.insert("", "end", values=row)
            total_wt += row[5]
            total_tat += row[4]
            total_rt += row[6]
        n = len(self.engine.stats) or 1
        self.avg_label.config(
            text=(
                f"AVG WT = {total_wt / n:.2f} | "
                f"AVG TAT = {total_tat / n:.2f} | "
                f"AVG RT = {total_rt / n:.2f}"
            )
        )


if __name__ == "__main__":
    root = tk.Tk()
    CPUSchedulerGUI(root)
    root.mainloop()