From 50ce333750396a4addd6ed56d5edd1b1b6a454db Mon Sep 17 00:00:00 2001 From: shashwat-126 Date: Sat, 8 Nov 2025 00:36:42 +0530 Subject: [PATCH 1/8] CPU Scheduling Algorithm with Interfaces --- scheduling/cpuschedulingalgos.py | 409 +++++++++++++++++++++++++++++++ 1 file changed, 409 insertions(+) create mode 100644 scheduling/cpuschedulingalgos.py diff --git a/scheduling/cpuschedulingalgos.py b/scheduling/cpuschedulingalgos.py new file mode 100644 index 000000000000..3d449655103d --- /dev/null +++ b/scheduling/cpuschedulingalgos.py @@ -0,0 +1,409 @@ +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, colors, current_pid = 0, {}, 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() From 88801bf00169264f6581d0d3c104a8e319521ba7 Mon Sep 17 00:00:00 2001 From: shashwat-126 Date: Sat, 8 Nov 2025 01:00:16 +0530 Subject: [PATCH 2/8] CPU Scheduling Algorithm with Interfaces --- scheduling/cpuschedulingalgos.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/scheduling/cpuschedulingalgos.py b/scheduling/cpuschedulingalgos.py index 3d449655103d..429432aabe3a 100644 --- a/scheduling/cpuschedulingalgos.py +++ b/scheduling/cpuschedulingalgos.py @@ -355,7 +355,9 @@ def run_scheduling(self) -> None: def animate(self) -> None: """Animates the scheduling visualization.""" self.ax.clear() - x, colors, current_pid = 0, {}, None + x: int = 0 + colors: dict[str, any] = {} + current_pid: str | None = None for step in self.engine.simulate(): _, pid, rq = step if pid: From 50c1f9febde5c21c208d338fd334c64ca4750974 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Fri, 7 Nov 2025 19:33:21 +0000 Subject: [PATCH 3/8] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- scheduling/cpuschedulingalgos.py | 82 ++++++++++++++++---------------- 1 file changed, 42 insertions(+), 40 deletions(-) diff --git a/scheduling/cpuschedulingalgos.py b/scheduling/cpuschedulingalgos.py index 429432aabe3a..2a9a2d1cf996 100644 --- a/scheduling/cpuschedulingalgos.py +++ b/scheduling/cpuschedulingalgos.py @@ -88,10 +88,10 @@ def _simulate_sjf_np(self) -> Generator[tuple[int, str | None, list[str]]]: done = 0 while done < len(processes): ready = [ - process - for process in processes - if process["arrival"] <= t and "completion" not in process - ] + process + for process in processes + if process["arrival"] <= t and "completion" not in process + ] if not ready: t += 1 yield (t, None, []) @@ -112,10 +112,10 @@ def _simulate_sjf_p(self) -> Generator[tuple[int, str | None, list[str]]]: done = 0 while done < len(processes): ready = [ - process - for process in processes - if process["arrival"] <= t and process["remaining"] > 0 - ] + process + for process in processes + if process["arrival"] <= t and process["remaining"] > 0 + ] if not ready: t += 1 yield (t, None, []) @@ -136,10 +136,10 @@ def _simulate_priority_np(self) -> Generator[tuple[int, str | None, list[str]]]: done = 0 while done < len(self.processes): ready = [ - process - for process in self.processes - if process["arrival"] <= t and "completion" not in process - ] + process + for process in self.processes + if process["arrival"] <= t and "completion" not in process + ] if not ready: t += 1 yield (t, None, []) @@ -159,10 +159,10 @@ def _simulate_priority_p(self) -> Generator[tuple[int, str | None, list[str]]]: done = 0 while done < len(self.processes): ready = [ - process - for process in self.processes - if process["arrival"] <= t and process["remaining"] > 0 - ] + process + for process in self.processes + if process["arrival"] <= t and process["remaining"] > 0 + ] if not ready: t += 1 yield (t, None, []) @@ -261,15 +261,15 @@ def setup_ui(self) -> None: 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) + 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) + 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") @@ -291,10 +291,10 @@ def setup_ui(self) -> None: 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) + 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() @@ -367,14 +367,14 @@ def animate(self) -> None: 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 + 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([]) @@ -400,11 +400,13 @@ def show_results(self) -> None: 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}" - ) + 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) From e84d3c0cef806a219e6ebe42ef7eec858a594719 Mon Sep 17 00:00:00 2001 From: Shashwat Kumar Date: Sat, 8 Nov 2025 10:52:28 +0530 Subject: [PATCH 4/8] Add typing import for type hinting --- scheduling/cpuschedulingalgos.py | 1 + 1 file changed, 1 insertion(+) diff --git a/scheduling/cpuschedulingalgos.py b/scheduling/cpuschedulingalgos.py index 2a9a2d1cf996..e3ccc22676a3 100644 --- a/scheduling/cpuschedulingalgos.py +++ b/scheduling/cpuschedulingalgos.py @@ -2,6 +2,7 @@ import threading import time import tkinter as tk +from typing import Any from collections.abc import Generator from tkinter import messagebox, ttk From 37b31acaffb95ad3dd6c98cb573a81e4b2dd6dd9 Mon Sep 17 00:00:00 2001 From: Shashwat Kumar Date: Sat, 8 Nov 2025 10:53:29 +0530 Subject: [PATCH 5/8] Remove unused import statement from cpuschedulingalgos.py --- scheduling/cpuschedulingalgos.py | 1 - 1 file changed, 1 deletion(-) diff --git a/scheduling/cpuschedulingalgos.py b/scheduling/cpuschedulingalgos.py index e3ccc22676a3..2a9a2d1cf996 100644 --- a/scheduling/cpuschedulingalgos.py +++ b/scheduling/cpuschedulingalgos.py @@ -2,7 +2,6 @@ import threading import time import tkinter as tk -from typing import Any from collections.abc import Generator from tkinter import messagebox, ttk From 1e860f5bee77f2cd9fe2cf7b4dfbbbda48d41799 Mon Sep 17 00:00:00 2001 From: Shashwat Kumar Date: Sat, 8 Nov 2025 12:30:49 +0530 Subject: [PATCH 6/8] Update cpuschedulingalgos.py --- scheduling/cpuschedulingalgos.py | 102 ++++++++++++++++++++++++++++--- 1 file changed, 94 insertions(+), 8 deletions(-) diff --git a/scheduling/cpuschedulingalgos.py b/scheduling/cpuschedulingalgos.py index 2a9a2d1cf996..8a32c7b23db7 100644 --- a/scheduling/cpuschedulingalgos.py +++ b/scheduling/cpuschedulingalgos.py @@ -106,7 +106,15 @@ def _simulate_sjf_np(self) -> Generator[tuple[int, str | None, list[str]]]: # shortest job first preemptive def _simulate_sjf_p(self) -> Generator[tuple[int, str | None, list[str]]]: - """Simulates SJF Preemptive scheduling.""" + """ + Simulates SJF Preemptive scheduling. + + >>> processes = [{"pid": "P1", "arrival": 0, "burst": 3}] + >>> engine = SchedulerEngine(processes, "SJF (Preemptive)") + >>> gen = engine._simulate_sjf_p() + >>> isinstance(gen, Generator) + True + """ t = 0 processes = sorted(self.processes, key=lambda process: process["arrival"]) done = 0 @@ -131,7 +139,15 @@ def _simulate_sjf_p(self) -> Generator[tuple[int, str | None, list[str]]]: # priority non preemptive def _simulate_priority_np(self) -> Generator[tuple[int, str | None, list[str]]]: - """Simulates Priority (Non-Preemptive) scheduling.""" + """ + Simulates Priority (Non-Preemptive) scheduling. + + >>> processes = [{"pid": "P1", "arrival": 0, "burst": 2, "priority": 1}] + >>> engine = SchedulerEngine(processes, "Priority (Non-Preemptive)") + >>> gen = engine._simulate_priority_np() + >>> next(gen) + (0, 'P1', []) + """ t = 0 done = 0 while done < len(self.processes): @@ -154,7 +170,14 @@ def _simulate_priority_np(self) -> Generator[tuple[int, str | None, list[str]]]: # priority preemptive def _simulate_priority_p(self) -> Generator[tuple[int, str | None, list[str]]]: - """Simulates Priority (Preemptive) scheduling.""" + """ + Simulates Priority (Preemptive) scheduling. + + >>> processes = [{"pid": "P1", "arrival": 0, "burst": 2, "priority": 1}] + >>> engine = SchedulerEngine(processes, "Priority (Preemptive)") + >>> isinstance(engine._simulate_priority_p(), Generator) + True + """ t = 0 done = 0 while done < len(self.processes): @@ -318,7 +341,21 @@ def setup_ui(self) -> None: self.avg_label.pack() def add_process(self) -> None: - """Adds a new process entry to the table.""" + """ + Adds a new process entry to the table. + + >>> import tkinter as tk + >>> root = tk.Tk() + >>> gui = CPUSchedulerGUI(root) + >>> gui.pid_e.insert(0, 'P1') + >>> gui.arrival_e.insert(0, '0') + >>> gui.burst_e.insert(0, '3') + >>> gui.priority_e.insert(0, '1') + >>> gui.add_process() + >>> len(gui.processes) > 0 + True + >>> root.destroy() + """ try: pid = self.pid_e.get() arrival = int(self.arrival_e.get()) @@ -332,14 +369,40 @@ def add_process(self) -> None: messagebox.showerror("Error", "Invalid input") def delete_process(self) -> None: - """Deletes a selected process.""" + """ + Deletes a selected process. + + >>> import tkinter as tk + >>> root = tk.Tk() + >>> gui = CPUSchedulerGUI(root) + >>> gui.processes = [{"pid": "P1", "arrival": 0, "burst": 2, "priority": 1}] + >>> gui.tree.insert("", "end", values=("P1", 0, 2, 1)) + >>> sel = gui.tree.get_children()[0] + >>> gui.tree.selection_set(sel) + >>> gui.delete_process() + >>> gui.processes + [] + >>> root.destroy() + """ 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.""" + """ + Runs the selected scheduling algorithm. + + >>> import tkinter as tk + >>> root = tk.Tk() + >>> gui = CPUSchedulerGUI(root) + >>> gui.processes = [{"pid": "P1", "arrival": 0, "burst": 2, "priority": 1}] + >>> gui.algo_cb.set("FCFS") + >>> gui.run_scheduling() + >>> hasattr(gui, "engine") + True + >>> root.destroy() + """ algo = self.algo_cb.get() quantum = int(self.quantum_e.get() or 2) if algo.lower() == "round robin": @@ -353,7 +416,18 @@ def run_scheduling(self) -> None: threading.Thread(target=self.animate, daemon=True).start() def animate(self) -> None: - """Animates the scheduling visualization.""" + """ + Animates the scheduling visualization. + + >>> import tkinter as tk + >>> root = tk.Tk() + >>> gui = CPUSchedulerGUI(root) + >>> gui.processes = [{"pid": "P1", "arrival": 0, "burst": 1, "priority": 1}] + >>> gui.engine = SchedulerEngine(gui.processes, "FCFS") + >>> hasattr(gui, "animate") + True + >>> root.destroy() + """ self.ax.clear() x: int = 0 colors: dict[str, any] = {} @@ -388,7 +462,19 @@ def animate(self) -> None: self.show_results() def show_results(self) -> None: - """Displays scheduling results.""" + """ + Displays scheduling results. + + >>> import tkinter as tk + >>> root = tk.Tk() + >>> gui = CPUSchedulerGUI(root) + >>> gui.engine = SchedulerEngine([{"pid": "P1", "arrival": 0, "burst": 1, "priority": 1}], "FCFS") + >>> gui.engine.stats = [("P1", 0, 1, 1, 1, 0, 0)] + >>> gui.show_results() + >>> "AVG" in gui.avg_label.cget("text") + True + >>> root.destroy() + """ for item in self.result_box.get_children(): self.result_box.delete(item) total_wt = total_tat = total_rt = 0 From 4178fe6e5e20e5385235e70a3a964f07902a18bc Mon Sep 17 00:00:00 2001 From: Shashwat Kumar Date: Sat, 8 Nov 2025 12:43:40 +0530 Subject: [PATCH 7/8] Update cpuschedulingalgos.py --- scheduling/cpuschedulingalgos.py | 42 ++++++++++++++++---------------- 1 file changed, 21 insertions(+), 21 deletions(-) diff --git a/scheduling/cpuschedulingalgos.py b/scheduling/cpuschedulingalgos.py index 8a32c7b23db7..a8291b8c1155 100644 --- a/scheduling/cpuschedulingalgos.py +++ b/scheduling/cpuschedulingalgos.py @@ -96,7 +96,7 @@ def _simulate_sjf_np(self) -> Generator[tuple[int, str | None, list[str]]]: t += 1 yield (t, None, []) continue - process = min(ready, key=lambda x: x["burst"]) + process = min(ready, key=lambda proc: x["burst"]) for _ in range(process["burst"]): self.timeline.append((t, process["pid"])) yield (t, process["pid"], []) @@ -106,7 +106,7 @@ def _simulate_sjf_np(self) -> Generator[tuple[int, str | None, list[str]]]: # shortest job first preemptive def _simulate_sjf_p(self) -> Generator[tuple[int, str | None, list[str]]]: - """ + """ Simulates SJF Preemptive scheduling. >>> processes = [{"pid": "P1", "arrival": 0, "burst": 3}] @@ -128,7 +128,7 @@ def _simulate_sjf_p(self) -> Generator[tuple[int, str | None, list[str]]]: t += 1 yield (t, None, []) continue - process = min(ready, key=lambda x: x["remaining"]) + process = min(ready, key=lambda proc: x["remaining"]) self.timeline.append((t, process["pid"])) yield (t, process["pid"], []) process["remaining"] -= 1 @@ -139,7 +139,7 @@ def _simulate_sjf_p(self) -> Generator[tuple[int, str | None, list[str]]]: # priority non preemptive def _simulate_priority_np(self) -> Generator[tuple[int, str | None, list[str]]]: - """ + """ Simulates Priority (Non-Preemptive) scheduling. >>> processes = [{"pid": "P1", "arrival": 0, "burst": 2, "priority": 1}] @@ -160,7 +160,7 @@ def _simulate_priority_np(self) -> Generator[tuple[int, str | None, list[str]]]: t += 1 yield (t, None, []) continue - process = min(ready, key=lambda x: x["priority"]) + process = min(ready, key=lambda proc: x["priority"]) for _ in range(process["burst"]): self.timeline.append((t, process["pid"])) yield (t, process["pid"], []) @@ -170,7 +170,7 @@ def _simulate_priority_np(self) -> Generator[tuple[int, str | None, list[str]]]: # priority preemptive def _simulate_priority_p(self) -> Generator[tuple[int, str | None, list[str]]]: - """ + """ Simulates Priority (Preemptive) scheduling. >>> processes = [{"pid": "P1", "arrival": 0, "burst": 2, "priority": 1}] @@ -190,7 +190,7 @@ def _simulate_priority_p(self) -> Generator[tuple[int, str | None, list[str]]]: t += 1 yield (t, None, []) continue - process = min(ready, key=lambda x: x["priority"]) + process = min(ready, key=lambda proc: x["priority"]) self.timeline.append((t, process["pid"])) yield (t, process["pid"], []) process["remaining"] -= 1 @@ -299,12 +299,12 @@ def setup_ui(self) -> None: self.algo_cb = ttk.Combobox( algo_frame, values=[ - "FCFS", - "SJF (Non-Preemptive)", - "SJF (Preemptive)", - "Priority (Non-Preemptive)", - "Priority (Preemptive)", - "Round Robin", + "FCFS", + "SJF (Non-Preemptive)", + "SJF (Preemptive)", + "Priority (Non-Preemptive)", + "Priority (Preemptive)", + "Round Robin", ], ) self.algo_cb.current(0) @@ -355,7 +355,7 @@ def add_process(self) -> None: >>> len(gui.processes) > 0 True >>> root.destroy() - """ + """ try: pid = self.pid_e.get() arrival = int(self.arrival_e.get()) @@ -369,7 +369,7 @@ def add_process(self) -> None: messagebox.showerror("Error", "Invalid input") def delete_process(self) -> None: - """ + """ Deletes a selected process. >>> import tkinter as tk @@ -383,14 +383,14 @@ def delete_process(self) -> None: >>> gui.processes [] >>> root.destroy() - """ + """ 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. >>> import tkinter as tk @@ -402,7 +402,7 @@ def run_scheduling(self) -> None: >>> hasattr(gui, "engine") True >>> root.destroy() - """ + """ algo = self.algo_cb.get() quantum = int(self.quantum_e.get() or 2) if algo.lower() == "round robin": @@ -427,7 +427,7 @@ def animate(self) -> None: >>> hasattr(gui, "animate") True >>> root.destroy() - """ + """ self.ax.clear() x: int = 0 colors: dict[str, any] = {} @@ -462,7 +462,7 @@ def animate(self) -> None: self.show_results() def show_results(self) -> None: - """ + """ Displays scheduling results. >>> import tkinter as tk @@ -474,7 +474,7 @@ def show_results(self) -> None: >>> "AVG" in gui.avg_label.cget("text") True >>> root.destroy() - """ + """ for item in self.result_box.get_children(): self.result_box.delete(item) total_wt = total_tat = total_rt = 0 From d3144a546d85291769f516ee658d201bc6ae36f2 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Sat, 8 Nov 2025 07:13:58 +0000 Subject: [PATCH 8/8] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- scheduling/cpuschedulingalgos.py | 138 +++++++++++++++---------------- 1 file changed, 69 insertions(+), 69 deletions(-) diff --git a/scheduling/cpuschedulingalgos.py b/scheduling/cpuschedulingalgos.py index a8291b8c1155..b9969d29e669 100644 --- a/scheduling/cpuschedulingalgos.py +++ b/scheduling/cpuschedulingalgos.py @@ -299,12 +299,12 @@ def setup_ui(self) -> None: self.algo_cb = ttk.Combobox( algo_frame, values=[ - "FCFS", - "SJF (Non-Preemptive)", - "SJF (Preemptive)", - "Priority (Non-Preemptive)", - "Priority (Preemptive)", - "Round Robin", + "FCFS", + "SJF (Non-Preemptive)", + "SJF (Preemptive)", + "Priority (Non-Preemptive)", + "Priority (Preemptive)", + "Round Robin", ], ) self.algo_cb.current(0) @@ -342,20 +342,20 @@ def setup_ui(self) -> None: def add_process(self) -> None: """ - Adds a new process entry to the table. - - >>> import tkinter as tk - >>> root = tk.Tk() - >>> gui = CPUSchedulerGUI(root) - >>> gui.pid_e.insert(0, 'P1') - >>> gui.arrival_e.insert(0, '0') - >>> gui.burst_e.insert(0, '3') - >>> gui.priority_e.insert(0, '1') - >>> gui.add_process() - >>> len(gui.processes) > 0 - True - >>> root.destroy() - """ + Adds a new process entry to the table. + + >>> import tkinter as tk + >>> root = tk.Tk() + >>> gui = CPUSchedulerGUI(root) + >>> gui.pid_e.insert(0, 'P1') + >>> gui.arrival_e.insert(0, '0') + >>> gui.burst_e.insert(0, '3') + >>> gui.priority_e.insert(0, '1') + >>> gui.add_process() + >>> len(gui.processes) > 0 + True + >>> root.destroy() + """ try: pid = self.pid_e.get() arrival = int(self.arrival_e.get()) @@ -370,20 +370,20 @@ def add_process(self) -> None: def delete_process(self) -> None: """ - Deletes a selected process. - - >>> import tkinter as tk - >>> root = tk.Tk() - >>> gui = CPUSchedulerGUI(root) - >>> gui.processes = [{"pid": "P1", "arrival": 0, "burst": 2, "priority": 1}] - >>> gui.tree.insert("", "end", values=("P1", 0, 2, 1)) - >>> sel = gui.tree.get_children()[0] - >>> gui.tree.selection_set(sel) - >>> gui.delete_process() - >>> gui.processes - [] - >>> root.destroy() - """ + Deletes a selected process. + + >>> import tkinter as tk + >>> root = tk.Tk() + >>> gui = CPUSchedulerGUI(root) + >>> gui.processes = [{"pid": "P1", "arrival": 0, "burst": 2, "priority": 1}] + >>> gui.tree.insert("", "end", values=("P1", 0, 2, 1)) + >>> sel = gui.tree.get_children()[0] + >>> gui.tree.selection_set(sel) + >>> gui.delete_process() + >>> gui.processes + [] + >>> root.destroy() + """ 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] @@ -391,18 +391,18 @@ def delete_process(self) -> None: def run_scheduling(self) -> None: """ - Runs the selected scheduling algorithm. - - >>> import tkinter as tk - >>> root = tk.Tk() - >>> gui = CPUSchedulerGUI(root) - >>> gui.processes = [{"pid": "P1", "arrival": 0, "burst": 2, "priority": 1}] - >>> gui.algo_cb.set("FCFS") - >>> gui.run_scheduling() - >>> hasattr(gui, "engine") - True - >>> root.destroy() - """ + Runs the selected scheduling algorithm. + + >>> import tkinter as tk + >>> root = tk.Tk() + >>> gui = CPUSchedulerGUI(root) + >>> gui.processes = [{"pid": "P1", "arrival": 0, "burst": 2, "priority": 1}] + >>> gui.algo_cb.set("FCFS") + >>> gui.run_scheduling() + >>> hasattr(gui, "engine") + True + >>> root.destroy() + """ algo = self.algo_cb.get() quantum = int(self.quantum_e.get() or 2) if algo.lower() == "round robin": @@ -417,17 +417,17 @@ def run_scheduling(self) -> None: def animate(self) -> None: """ - Animates the scheduling visualization. - - >>> import tkinter as tk - >>> root = tk.Tk() - >>> gui = CPUSchedulerGUI(root) - >>> gui.processes = [{"pid": "P1", "arrival": 0, "burst": 1, "priority": 1}] - >>> gui.engine = SchedulerEngine(gui.processes, "FCFS") - >>> hasattr(gui, "animate") - True - >>> root.destroy() - """ + Animates the scheduling visualization. + + >>> import tkinter as tk + >>> root = tk.Tk() + >>> gui = CPUSchedulerGUI(root) + >>> gui.processes = [{"pid": "P1", "arrival": 0, "burst": 1, "priority": 1}] + >>> gui.engine = SchedulerEngine(gui.processes, "FCFS") + >>> hasattr(gui, "animate") + True + >>> root.destroy() + """ self.ax.clear() x: int = 0 colors: dict[str, any] = {} @@ -463,18 +463,18 @@ def animate(self) -> None: def show_results(self) -> None: """ - Displays scheduling results. - - >>> import tkinter as tk - >>> root = tk.Tk() - >>> gui = CPUSchedulerGUI(root) - >>> gui.engine = SchedulerEngine([{"pid": "P1", "arrival": 0, "burst": 1, "priority": 1}], "FCFS") - >>> gui.engine.stats = [("P1", 0, 1, 1, 1, 0, 0)] - >>> gui.show_results() - >>> "AVG" in gui.avg_label.cget("text") - True - >>> root.destroy() - """ + Displays scheduling results. + + >>> import tkinter as tk + >>> root = tk.Tk() + >>> gui = CPUSchedulerGUI(root) + >>> gui.engine = SchedulerEngine([{"pid": "P1", "arrival": 0, "burst": 1, "priority": 1}], "FCFS") + >>> gui.engine.stats = [("P1", 0, 1, 1, 1, 0, 0)] + >>> gui.show_results() + >>> "AVG" in gui.avg_label.cget("text") + True + >>> root.destroy() + """ for item in self.result_box.get_children(): self.result_box.delete(item) total_wt = total_tat = total_rt = 0