[Misc][Lex] Linting

Author: LaplaceXD

main.py

@@ -1,6 +1,6 @@
 import os
 
-from scheduling import main as scheduling_main
+from process_scheduling import main as scheduling_main
 from memory_managing import main as memory_managing_main
 from utils.io import input_bounded_num
 from views import View
@@ -32,4 +32,4 @@ def main():
             is_running = False
 
 if __name__ == "__main__":
-    main()
+    main()

memory_managing.py

@@ -23,7 +23,7 @@ def main():
     pages = [int(page) if page.isdigit() else page for page in pages]
 
     # Selection of memory algorithms to use for simulation
-    memory_choices: List[Memory] = [FIFO, LRU, LFU, Optimal]
+    memory_choices = [FIFO, LRU, LFU, Optimal]
     print(View.numbered_list(m.name for m in memory_choices), end="\n\n")
 
     is_selecting = True
@@ -74,7 +74,7 @@ def main():
 
         print("## Memory State Visualization")
         print("Legend: F -> Page Fault | H -> Page Hit")
-        table = TableView(header=["Time"] + [s.snapped_on for s in paging_timeline], footer=["Status"] + [s.status for s in paging_timeline])
+        table = TableView(header=["Time"] + [str(s.snapped_on) for s in paging_timeline], footer=["Status"] + [s.status for s in paging_timeline])
         for frame in range(memory.capacity):
             row_header = "Frame {}".format(memory.frame_label_of(memory[frame]))
             table.add_item(row_header, *(s.snapshot[frame] if s.snapshot[frame] is not None else "--" for s in paging_timeline))
@@ -103,7 +103,7 @@ def main():
     print()
     print("===== Summary Statistics =====")
     print("Legend: Ranked by page hits")
-    print("Page References:", page_ref)
+    print("Page References:", " ".join(str(page) for page in pages))
 
     memory_metrics.sort(key=lambda metric : metric.hits, reverse=True)
     summary_table = TableView(min_cell_width=8, header=["Rank#", "Page Replacement Algorithm", "Hits", "Faults", "Hit%", "Fault%"])
@@ -112,4 +112,4 @@ def main():
     summary_table.render()
 
 if __name__ == "__main__":
-    main()
+    main()

modules/memories/memory.py

@@ -12,7 +12,7 @@ class Memory(ABC, Generic[T]):
     def __init__(self, frame_size: int):
         self._memory: List[Optional[T]] = [None] * frame_size
         self._capacity: int = frame_size
-        self._iter_ptr: int = 0
+        self.__iter_ptr: int = 0
 
     def __eq__(self, other: 'Memory'):
         return self.name == other.name and self.size == other.size and self._capacity == other.capacity and all(a == b for a, b in zip(self._memory, other))
@@ -21,13 +21,13 @@ def __getitem__(self, idx: int):
         return self._memory[idx]
 
     def __iter__(self):
-        self._iter_ptr = 0
+        self.__iter_ptr = 0
         return self
 
     def __next__(self):
-        if self._iter_ptr < self._capacity:
-            value = self._memory[self._iter_ptr]
-            self._iter_ptr += 1
+        if self.__iter_ptr < self._capacity:
+            value = self._memory[self.__iter_ptr]
+            self.__iter_ptr += 1
             return value
 
         raise StopIteration

modules/processor.py

@@ -27,7 +27,7 @@ def __last_log_end_time(self):
 
     def __record_cleared_process(self, cleared_process: Process):
         """ Records the cleared process to the logs. """
-        log = ProcessLog(cleared_process.pid, self.__last_log_end_time, self.__clock.time, tag=cleared_process.queue_level)
+        log = ProcessLog(str(cleared_process.pid), self.__last_log_end_time, self.__clock.time, tag=cleared_process.queue_level)
         self.__process_logs.append(log)
 
     def __record_idle_time(self, loaded_process: Process):
@@ -48,7 +48,7 @@ def is_occupied(self):
 
     @property
     def is_finished(self):
-        return self.is_occupied and self.__current_process.is_depleted
+        return self.__current_process is not None and self.__current_process.is_depleted
 
     @property
     def idle_time(self):
@@ -65,11 +65,10 @@ def current_process(self):
 
     def run(self, time_quantum = 1):
         """ Runs the processor by a given time quantum. """
-        if self.is_occupied:
-            for _ in range(time_quantum):
-                if not self.__current_process.is_depleted:
-                    self.__current_process.tick()
-                    self.__tick_signal.emit(self.__current_process)
+        for _ in range(time_quantum):
+            if self.__current_process is not None and not self.__current_process.is_depleted:
+                self.__current_process.tick()
+                self.__tick_signal.emit(self.__current_process)
 
     def on_tick(self, fn: Callable[[Process], None]):
         """ Adds a function to listen whenever the processor does a tick. """
@@ -96,7 +95,7 @@ def clear(self):
         """ Removes a loaded process from the processor, and returns that removed process to the caller. """
         cleared_process = None
 
-        if self.is_occupied:
+        if self.__current_process is not None:
             cleared_process = self.__current_process
             self.__current_process = None
             self.__clear_signal.emit(cleared_process)
@@ -109,4 +108,4 @@ def on_clear(self, fn: Callable[[Process], None]):
 
     def off_clear(self, fn: Callable[[Process], None]):
         """ Removes a function listening to the processor clear. """
-        self.__clear_signal.ignore(fn)
+        self.__clear_signal.ignore(fn)

modules/schedulers/mlfq.py

@@ -29,7 +29,7 @@ def __init__(self, processes: List[Process], processor: Processor, time_quantums
         self.__layers.append(last_layer(processes, processor)) 
 
     @staticmethod
-    def last_layer_choices() -> List[Scheduler]:
+    def last_layer_choices():
         return [FCFS, SJF, PriorityNP]
 
     @classmethod
@@ -40,7 +40,7 @@ def factory(cls, time_quantums: List[int], last_layer: Callable[[List[Process],
 
     @property
     def round_robin_layers(self) -> List[RoundRobin]:
-        return self.__layers[:-1]
+        return self.__layers[:-1] # pyright: ignore[reportGeneralTypeIssues]
 
     @property
     def last_layer(self):
@@ -75,4 +75,4 @@ def run(self, timestamp: int, is_allowed_to_preempt: bool = True):
             if len(self._ready_queue) == 0:
                 self._ready_queue = self.last_layer._ready_queue
 
-        return self._ready_queue
+        return self._ready_queue

modules/schedulers/mlq.py

@@ -25,7 +25,7 @@ def __init__(self, processes: List[Process], processor: Processor, layers: List[
             self.__layers.append(layer_instance)
 
     @staticmethod
-    def layer_choices() -> List[Scheduler]:
+    def layer_choices():
         return [FCFS, SJF, PriorityNP, Priority, RoundRobin, SRTF]
 
     @classmethod
@@ -68,4 +68,4 @@ def run(self, timestamp: int, is_allowed_to_preempt: bool = True) -> List[Proces
                         self._processor.on_tick(layer.decrement_time_window)
                     break 
 
-        return self._ready_queue
+        return self._ready_queue

scheduling.py renamed to process_scheduling.py

@@ -1,15 +1,15 @@
 import os
-from typing import List
+from typing import List, Any
 
 from models import Process, ProcessLog
 from modules import Processor, Clock
-from modules.schedulers import Scheduler, FCFS, SJF, PriorityNP, Priority, RoundRobin, SRTF, MLQ, MLFQ
+from modules.schedulers import FCFS, SJF, PriorityNP, Priority, RoundRobin, SRTF, MLQ, MLFQ
 from views import View, TableView, GanttView
 from utils.io import input_bounded_num
 
 def create_process_execution_gantt(process_dump: List[ProcessLog], layers: List[str] = []):
     layer_gantts = [GanttView(name="[{}]".format(num), show_timestamps=num == len(layers)) for num, _ in enumerate(layers, start=1)]
-    merged_gantt = GanttView(name="[A]" if len(layers) > 0 else None)
+    merged_gantt = GanttView(name="[A]" if len(layers) > 0 else "")
     for log in process_dump:
         name = "P" + str(log.name) if type(log.name) == int else log.name
         time = log.end
@@ -25,8 +25,8 @@ def create_os_metrics(processes: List[Process], total_run_time: int, total_idle_
     metrics = ""
 
     metrics += "CPU Utilization: {:.2f}%\n".format((float(total_run_time - total_idle_time) / float(total_run_time)) * 100)
-    metrics += "Average TAT: {:.2f}\n".format(sum(p.turnaround for p in processes) / len(processes))
-    metrics += "Average WT: {:.2f}".format(sum(p.waiting for p in processes) / len(processes))
+    metrics += "Average TAT: {:.2f}\n".format(sum(p.turnaround for p in processes if p.turnaround is not None) / len(processes))
+    metrics += "Average WT: {:.2f}".format(sum(p.waiting for p in processes if p.waiting is not None) / len(processes))
 
     return metrics
 
@@ -74,7 +74,7 @@ def configure_mlq(num_layers: int):
         if layer_scheduler == RoundRobin:
             time_quantum = input_bounded_num("Time quantum: ") 
             layer_names.append(layer_scheduler.name + " | q=" + str(time_quantum))
-            layers.append(layer_scheduler.factory(time_quantum, False))
+            layers.append(RoundRobin.factory(time_quantum, False))
         else:
             layer_names.append(layer_scheduler.name)
             layers.append(layer_scheduler.factory())
@@ -108,23 +108,23 @@ def main():
     print("===== CPU Scheduling Simulator =====")
 
     # Select a scheduler
-    scheduler_choices: List[Scheduler] = [FCFS, SJF, PriorityNP, Priority, RoundRobin, SRTF, MLQ, MLFQ] 
+    scheduler_choices = [FCFS, SJF, PriorityNP, Priority, RoundRobin, SRTF, MLQ, MLFQ] 
     print(View.numbered_list(s.name for s in scheduler_choices), end="\n\n")
     scheduler_choice = input_bounded_num("Select a scheduler: ", max=len(scheduler_choices))
 
-    chosen_scheduler: Scheduler = scheduler_choices[scheduler_choice - 1]
+    chosen_scheduler = scheduler_choices[scheduler_choice - 1]
     has_priority_field: bool = chosen_scheduler.has_priority_field
     has_queue_level_field: bool = chosen_scheduler.has_queue_level_field
 
     # Configure the chosen scheduler
     time_quantum: int = 0
     layer_names: List[str] = []
-    scheduler_factory = None
+    scheduler_factory: Any = None
 
     print("\n=====", chosen_scheduler.name, "Configuration =====")
     if chosen_scheduler == RoundRobin:
         time_quantum = input_bounded_num("Time quantum: ") 
-        scheduler_factory = chosen_scheduler.factory(time_quantum, True)
+        scheduler_factory = RoundRobin.factory(time_quantum, True)
     elif chosen_scheduler.is_multilevel:
         num_layers = input_bounded_num("Number of Layers: ")
 
@@ -205,4 +205,4 @@ def main():
     print(metrics)
 
 if __name__ == "__main__":
-    main()
+    main()

utils/io.py

@@ -1,6 +1,6 @@
 from typing import List, Optional
 
-def input_bounded_num(prompt: object = "", min: int = 1, max: Optional[int] = None):
+def input_bounded_num(prompt: str = "", min: int = 1, max: Optional[int] = None):
     """ Get a number input from the user that satisfies a certain minimum and maximum. """
     result: Optional[int] = None
 
@@ -21,7 +21,7 @@ def input_bounded_num(prompt: object = "", min: int = 1, max: Optional[int] = No
 
     return result
 
-def input_choice(prompt: object = "", choices: List[str] = [], default: str = ""):
+def input_choice(prompt: str = "", choices: List[str] = [], default: str = ""):
     """ Make a user select between a set of values. """
     result: Optional[str] = None
     prompt_suffix = "(" + "/".join(choices) + (", default: " + default if default else "") + ")"
@@ -42,4 +42,4 @@ def input_choice(prompt: object = "", choices: List[str] = [], default: str = ""
         except:
             print("Please input {} only.".format("(" + "/".join(choices) + ")"))
 
-    return result
+    return result

views/gantt.py

@@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import List
 from views import View
 
 class GanttView(View):
@@ -35,4 +35,4 @@ def add_item(self, name: str, time: int):
         return self
 
     def render(self):
-        print(self)
+        print(self)

views/table.py

@@ -35,8 +35,8 @@ def __str__(self):
 
     def add_item(self, *data: Any):
         self._adjust_cell_sizes_to_fit(*data)
-        self.__data.append(data)
+        self.__data.append(list(data))
         return self
 
     def render(self):
-        print(self)
+        print(self)

views/view.py

@@ -36,7 +36,8 @@ def _format_row(self, items: List[Any], sep: str = "|"):
 
     def _create_separator_line(self, joint: str = "+", line: str = "-"):
         return joint + joint.join(line * w for w in self._cell_widths) + joint
-
+    
+    @staticmethod
     def numbered_list(items_iter: Iterable[Any], start_at: int = 1, is_reversed: bool = False):
         """ 
             Turns a set of items into a string of numbered list. The numbering is sequential,
@@ -72,4 +73,4 @@ def add_item(self):
     @abstractmethod
     def render(self):
         """ Render the given view to the console. """
-        pass 
+        pass