nemesis: Update nemesis with type annotations

Provided by righttyper

1 files changed, 32 insertions, 32 deletions

diff --git a/nemesis/nemesis.py b/nemesis/nemesis.py
index f4fa2d8..e7c5e7e 100755
--- a/nemesis/nemesis.py
+++ b/nemesis/nemesis.py
@@ -24,7 +24,7 @@ Copyright 2025 bdunahu
 Commentary:
 Code:
 '''
-
+from typing import Any, Self
 from causal_event_loop import CausalEventLoop
 from collections import defaultdict
 from html_gen import plot_results
@@ -46,57 +46,57 @@ CO_COROUTINE = inspect.CO_COROUTINE
 class Experiment:
 
     # event loops participating in this this experiment
-    _loops = []
+    _loops: list[Any] = []
 
-    def __init__(self, loops):
+    def __init__(self: Self, loops: list[Any]) -> None:
         self._loops = loops
 
-    def get_loops(self):
+    def get_loops(self: Self) -> list[Any]:
         return [l for l in self._loops if l.is_running()]
 
 class Nemesis(object):
 
     # the name of the target program
-    prog = None
+    prog: str
 
     # the (ideal) interval between samples
-    signal_interval = 0.0
+    signal_interval: float
     # the timestamp which the last sample was taken
-    last_sample = None
+    last_sample: float
 
     # the current experiment being run
-    experiment_data = None
+    experiment_data: Experiment
     # the coroutine the current experiment is speeding up
-    experiment_coro = None
+    experiment_coro: str
     # the speedup of the current experiment
-    experiment_spdp = None
+    experiment_spdp: float
     # the total time this experiment has been running
-    experiment_time = None
+    experiment_time: float
 
     # results from previous experiments. Keys represent names of coroutines.
     results = defaultdict(lambda: defaultdict(lambda: defaultdict(lambda: [])))
 
     # the file to write results to
-    filename = None
+    filename: str
 
     # The base duration of each performance experiment
-    e_duration = None
+    e_duration: int
 
     # A mapping of event loops to the previous running coroutine.
     prev_coro = defaultdict(lambda: None)
 
     # Path of files to profile.
-    path_include : []
+    path_include: list(Path)
     # Path of files to exclude
-    path_exclude = []
+    path_exclude: list(Path)
 
     @staticmethod
-    def __init__(e_duration,
-                 path_include,
-                 path_exclude,
-                 filename,
-                 prog,
-                 signal_interval=0.01):
+    def __init__(e_duration: int,
+                 path_include: None,
+                 path_exclude: None,
+                 filename: str,
+                 prog: str,
+                 signal_interval: float=0.01) -> None:
         Nemesis.signal_interval = signal_interval
         Nemesis.e_duration = e_duration
         Nemesis.path_include = path_include
@@ -104,7 +104,7 @@ class Nemesis(object):
         Nemesis.prog = prog
 
     @staticmethod
-    def start():
+    def start() -> None:
         Nemesis.last_sample = time.perf_counter()
         signal.signal(signal.SIGALRM,
                       Nemesis._signal_handler)
@@ -128,13 +128,13 @@ class Nemesis(object):
             print(f'')
 
     @staticmethod
-    def stop():
+    def stop() -> None:
         signal.setitimer(signal.ITIMER_REAL, 0)
         plot_results(Nemesis.results, Nemesis.filename, Nemesis.prog)
         print(f"Wrote {Nemesis.filename}")
 
     @staticmethod
-    def _start_experiment(coro, speedup):
+    def _start_experiment(coro: str, speedup: float) -> None:
         Nemesis.prev_coro = defaultdict(lambda: None)
         Nemesis.experiment_coro = coro
         Nemesis.experiment_spdp = speedup
@@ -149,7 +149,7 @@ class Nemesis(object):
         Nemesis.experiment_data = Experiment(loops)
 
     @staticmethod
-    def _stop_experiment():
+    def _stop_experiment() -> None:
         if Nemesis.experiment_data is not None:
             loops = Nemesis.experiment_data.get_loops()
 
@@ -163,7 +163,7 @@ class Nemesis(object):
             del Nemesis.experiment_data
 
     @staticmethod
-    def _signal_handler(sig, frame):
+    def _signal_handler(sig: int, frame: types.FrameType) -> None:
         curr_sample = time.perf_counter()
         passed_time = curr_sample - Nemesis.last_sample
         Nemesis.last_sample = curr_sample
@@ -209,7 +209,7 @@ class Nemesis(object):
             return [str(type(handle).__name__), cb.__name__]
 
     @staticmethod
-    def _get_current_frame(loop):
+    def _get_current_frame(loop: Any) -> types.FrameType:
         tid = loop._thread_id
         assert tid, f"{loop} is not running, yet we attempted to sample it!"
 
@@ -237,7 +237,7 @@ class Nemesis(object):
         return None
 
     @staticmethod
-    def _is_child_of_async(frame):
+    def _is_child_of_async(frame: types.FrameType) -> bool:
         '''Returns TRUE if an async method is a parent of FRAME.'''
         while frame:
             if bool(frame.f_code.co_flags & CO_COROUTINE):
@@ -245,7 +245,7 @@ class Nemesis(object):
             frame = frame.f_back
 
     @staticmethod
-    def _get_event_loops():
+    def _get_event_loops() -> list[Any]:
         '''Returns each thread's event loop, if it exists.'''
         loops = []
         for t in threading.enumerate():
@@ -257,7 +257,7 @@ class Nemesis(object):
         return loops
 
     @staticmethod
-    def _walk_back_until_loop(frame):
+    def _walk_back_until_loop(frame: types.FrameType) -> Any:
         '''Walks back the callstack until we are in a method named '_run_once'.
         If this is ever true, we assume we are in an Asyncio event loop method,
         and check to see if the 'self' variable is indeed and instance of
@@ -272,7 +272,7 @@ class Nemesis(object):
         return None
 
     @staticmethod
-    def _should_trace(filename):
+    def _should_trace(filename: str) -> bool:
         '''Returns FALSE if filename is uninteresting to the user.
         Don't depend on this. It kind of sucks.'''
         if not filename:
@@ -293,7 +293,7 @@ class Nemesis(object):
             return False
         return True
 
-    def _select_speedup():
+    def _select_speedup() -> float:
         '''
         Returns a random speedup between 0% to 100%, in multiples of 5%.
         Because a baseline is needed to calculate effect on program