Rename to Aergia

1 files changed, 0 insertions, 265 deletions

diff --git a/mini_scalene.py b/mini_scalene.py
deleted file mode 100644
index f7a8ca9..0000000
--- a/mini_scalene.py
+++ /dev/null
@@ -1,265 +0,0 @@
-import selectors
-
-import sys
-import argparse
-import threading
-import traceback
-import atexit
-import signal
-import asyncio
-import time
-from typing import (
-    Any,
-    Callable,
-    cast,
-)
-from types import FrameType
-from collections import defaultdict
-import replacement_epoll_selector
-
-the_globals = {
-    '__name__': '__main__',
-    '__doc__': None,
-    '__package__': None,
-    '__loader__': globals()['__loader__'],
-    '__spec__': None,
-    '__annotations__': {},
-    '__builtins__': globals()['__builtins__'],
-    '__file__': None,
-    '__cached__': None,
-}
-
-def parse_arguments():
-    '''Parse CLI args'''
-    parser = argparse.ArgumentParser()
-
-    parser.add_argument('-a', '--async_off',
-                        action='store_false',
-                        help='Turn off experimental async profiling.',
-                        default=True)
-
-    parser.add_argument('script', help='A python script to run')
-    parser.add_argument('s_args', nargs=argparse.REMAINDER,
-                        help='python script args')
-
-    return parser.parse_args()
-
-
-class MiniScalene(object):
-    '''A stripped-down version of SCALENE which tallies active lines during
-    execution.'''
-
-    # a key-value pair where keys represent frame metadata (see
-    # MiniScalene.frame_to_string) and values represent number of times
-    # sampled.
-    cpu_samples = defaultdict(lambda: 0)
-    cpu_samples_c = defaultdict(lambda: 0)
-    # number of times samples have been collected
-    total_cpu_samples = 0
-
-    # the time, in seconds, between samples
-    signal_interval = 0.01
-    # the timestamp recorded last signal
-    last_signal_time = 0.0
-
-    # if we should try to profile asynchronous code. Used to observe
-    # effectiveness of the implementation.
-    profile_async = True
-
-    def __init__(self):
-        signal.signal(signal.SIGPROF,
-                      self.cpu_signal_handler)
-        signal.setitimer(signal.ITIMER_PROF,
-                         self.signal_interval,
-                         self.signal_interval)
-        MiniScalene.last_signal_time = MiniScalene.gettime()
-
-    @staticmethod
-    def gettime():
-        '''get the wallclock time'''
-        return time.perf_counter()
-
-    @staticmethod
-    def start(profile_async):
-        MiniScalene.profile_async = profile_async
-
-    @staticmethod
-    def stop():
-        '''Turn off profiling signals'''
-        MiniScalene.disable_signals()
-        MiniScalene.exit_handler()
-
-    @staticmethod
-    def exit_handler():
-        '''Pretty-print profiling information.'''
-        # If we've collected any samples, dump them.
-        print("CPU usage (Python):")
-        if MiniScalene.total_cpu_samples > 0:
-            for key in MiniScalene.sort_samples(MiniScalene.cpu_samples):
-                print(f"{key} : "
-                      f"{MiniScalene.cpu_samples[key] * 100 / MiniScalene.total_cpu_samples:.3f} % "
-                      f"({MiniScalene.cpu_samples[key]:.1f} total samples)")
-            print("CPU usage (Native):")
-            for key in MiniScalene.sort_samples(MiniScalene.cpu_samples_c):
-                print(f"{key} : "
-                      f"{MiniScalene.cpu_samples_c[key] * 100 / MiniScalene.total_cpu_samples:.3f} % "
-                      f"({MiniScalene.cpu_samples_c[key]:.1f} total samples)")
-        else:
-            print("(did not run long enough to profile)")
-
-    @staticmethod
-    def disable_signals():
-        signal.signal(signal.ITIMER_PROF, signal.SIG_IGN)
-        signal.signal(signal.SIGVTALRM, signal.SIG_IGN)
-        signal.setitimer(signal.ITIMER_PROF, 0)
-
-    @staticmethod
-    def cpu_signal_handler(sig, frame):
-        elapsed_since_last_signal = MiniScalene.gettime() - \
-            MiniScalene.last_signal_time
-        c_time_norm = (elapsed_since_last_signal -
-                       MiniScalene.signal_interval) / \
-            MiniScalene.signal_interval
-
-        keys = MiniScalene.compute_frames_to_record(frame)
-        for key in keys:
-            MiniScalene.cpu_samples[MiniScalene.frame_to_string(key)] += 1
-            MiniScalene.cpu_samples_c[MiniScalene.frame_to_string(
-                key)] += c_time_norm
-            MiniScalene.total_cpu_samples += elapsed_since_last_signal / \
-                MiniScalene.signal_interval
-        MiniScalene.last_signal_time = MiniScalene.gettime()
-
-    @staticmethod
-    def compute_frames_to_record(this_frame):
-        '''Collects all stack frames that Scalene actually processes.'''
-        frames = [this_frame]
-        frames += [sys._current_frames().get(t.ident, None)
-                   for t in threading.enumerate()]
-        frames += MiniScalene.get_async_frames()
-
-        frames = MiniScalene.filter_duplicated_frames(frames)
-        # Process all the frames to remove ones we aren't going to track.
-        new_frames = []
-        for frame in frames:
-            if frame is None:
-                continue
-            fname = frame.f_code.co_filename
-            # Record samples only for files we care about.
-            if (len(fname)) == 0:
-                # 'eval/compile' gives no f_code.co_filename.  We have
-                # to look back into the outer frame in order to check
-                # the co_filename.
-                fname = frame.f_back.f_code.co_filename
-            while not MiniScalene.should_trace(fname):
-                # Walk the stack backwards until we hit a frame that
-                # IS one we should trace (if there is one).  i.e., if
-                # it's in the code being profiled, and it is just
-                # calling stuff deep in libraries.
-                if frame:
-                    frame = cast(FrameType, frame.f_back)
-                else:
-                    break
-                if frame:
-                    fname = frame.f_code.co_filename
-            if frame:
-                new_frames.append(frame)
-        return new_frames
-
-    @staticmethod
-    def frame_to_string(frame):
-        '''Pretty-prints a frame as a function/file name and a line number.
-        Additionally used a key for tallying lines.'''
-        co = frame.f_code
-        func_name = co.co_name
-        line_no = frame.f_lineno
-        filename = co.co_filename
-        return filename + '\t' + func_name + '\t' + str(line_no)
-
-    @staticmethod
-    def get_async_frames():
-        '''Obtains the stack frames of all currently executing tasks.'''
-        if MiniScalene.is_event_loop_running() and MiniScalene.profile_async:
-            return [task.get_coro().cr_frame for task in asyncio.all_tasks()]
-        return []
-
-    @staticmethod
-    def should_trace(filename):
-        '''Returns FALSE if filename is uninteresting to the user.'''
-        # FIXME Assume GuixSD. Makes filtering easy
-        if '/gnu/store' in filename:
-            return False
-        if filename[0] == '<':
-            return False
-        if 'mini_scalene.py' in filename:
-            return False
-        return True
-
-    @staticmethod
-    def is_event_loop_running() -> bool:
-        '''Returns TRUE if there is an exent loop running. This is what
-        `asyncio.get_event_loop()' did, before it was deprecated in 3.12'''
-        return asyncio.get_event_loop_policy()._local._loop is not None
-
-    @staticmethod
-    def sort_samples(sample_dict):
-        '''Returns SAMPLE_DICT in descending order by number of samples.'''
-        return {k: v for k, v in sorted(sample_dict.items(),
-                                        key=lambda item: item[1],
-                                        reverse=True)}
-
-    @staticmethod
-    def shim(func: Callable[[Any], Any]) -> Any:
-        """Provide a decorator that calls the wrapped function with the
-        Scalene variant.
-
-                Wrapped function must be of type (s: Scalene) -> Any.
-
-                This decorator allows for marking a function in a separate
-                file as a drop-in replacement for an existing library
-                function. The intention is for these functions to replace a
-                function that indefinitely blocks (which interferes with
-                Scalene) with a function that awakens periodically to allow
-                for signals to be delivered.
-
-        """
-        func(MiniScalene)
-        # Returns the function itself to the calling file for the sake
-        # of not displaying unusual errors if someone attempts to call
-        # it
-
-    @staticmethod
-    def filter_duplicated_frames(frames) -> bool:
-        s = set()
-        dup = []
-        for f in frames:
-            if f in s:
-                dup.append(f)
-            else:
-                s.add(f)
-        # TODO we probably have one because given get_async_frames returns the
-        # currently executing task. Would be an easy fix in that method.
-        # if there's more than one, I cannot explain it.
-        assert len(
-            dup) < 2, f"ERROR: More than 1 duplicate frame (shouldn't happen): {dup}"
-        if len(dup) != 0:
-            print(f"WARN: Duplicate frame found: {dup}", file=sys.stderr)
-        return list(s)
-
-
-def main():
-    args = parse_arguments()
-
-    sys.argv = [args.script] + args.s_args
-    try:
-        with open(args.script, 'rb') as fp:
-            code = compile(fp.read(), args.script, "exec")
-            MiniScalene().start(args.async_off)
-            exec(code, the_globals)
-            MiniScalene().stop()
-    except Exception:
-        traceback.print_exc()
-
-
-if __name__ == "__main__":
-    main()