Updated. (plasma-umass#831)

* Updated.

* Added imports.

scalene/find_browser.py

@@ -1,18 +1,25 @@
 import webbrowser
 from typing import Optional
 
+
 def find_browser(browserClass: Optional[str] = None) -> Optional[str]:
     """Find the default system browser, excluding text browsers.
-    
+
     If you want a specific browser, pass its class as an argument."""
     text_browsers = [
-        "browsh", "elinks", "links", "lynx", "w3m",
+        "browsh",
+        "elinks",
+        "links",
+        "lynx",
+        "w3m",
     ]
 
     try:
         # Get the default browser object
         browser = webbrowser.get(browserClass)
-        browser_name = browser.name if browser.name else browser.__class__.__name__
+        browser_name = (
+            browser.name if browser.name else browser.__class__.__name__
+        )
         return browser_name if browser_name not in text_browsers else None
     except AttributeError:
         # https://github.com/plasma-umass/scalene/issues/790
@@ -21,7 +28,7 @@ def find_browser(browserClass: Optional[str] = None) -> Optional[str]:
         # we need to refer to it as such to prevent this error:
         #   'MacOSXOSAScript' object has no attribute 'name'
         browser = webbrowser.get(browserClass)
-        return browser._name if browser._name not in text_browsers else None # type: ignore[attr-defined]
+        return browser._name if browser._name not in text_browsers else None  # type: ignore[attr-defined]
     except webbrowser.Error:
         # Return None if there is an error in getting the browser
         return None

scalene/get_module_details.py

@@ -3,18 +3,10 @@
 
 from importlib.abc import SourceLoader
 from importlib.machinery import ModuleSpec
-from types import CodeType, FrameType
+from types import CodeType
 from typing import (
-    Any,
-    Callable,
-    Dict,
-    List,
-    Optional,
-    Set,
     Tuple,
     Type,
-    Union,
-    cast,
 )
 
 

scalene/launchbrowser.py

@@ -4,7 +4,6 @@
 import shutil
 import socket
 import socketserver
-import subprocess
 import sys
 import tempfile
 import threading
@@ -15,48 +14,59 @@
 from jinja2 import Environment, FileSystemLoader
 from typing import Any, NewType
 
+
 def read_file_content(directory: str, subdirectory: str, filename: str) -> str:
     file_path = os.path.join(directory, subdirectory, filename)
     return pathlib.Path(file_path).read_text()
 
 
 def launch_browser_insecure(url: str) -> None:
-    if platform.system() == 'Windows':
-        chrome_path = 'C:\\Program Files (x86)\\Google\\Chrome\\Application\\chrome.exe'
-    elif platform.system() == 'Linux':
-        chrome_path = '/usr/bin/google-chrome'
-    elif platform.system() == 'Darwin':
-        chrome_path = '/Applications/Google\\ Chrome.app/Contents/MacOS/Google\\ Chrome'
+    if platform.system() == "Windows":
+        chrome_path = (
+            "C:\\Program Files (x86)\\Google\\Chrome\\Application\\chrome.exe"
+        )
+    elif platform.system() == "Linux":
+        chrome_path = "/usr/bin/google-chrome"
+    elif platform.system() == "Darwin":
+        chrome_path = (
+            "/Applications/Google\\ Chrome.app/Contents/MacOS/Google\\ Chrome"
+        )
 
     # Create a temporary directory
     with tempfile.TemporaryDirectory() as temp_dir:
         # Create a command with the required flags
         chrome_cmd = f'{chrome_path} %s --disable-web-security --user-data-dir="{temp_dir}"'
 
         # Register the new browser type
-        webbrowser.register('chrome_with_flags', None,
-                            webbrowser.Chrome(chrome_cmd), preferred=True)
+        webbrowser.register(
+            "chrome_with_flags",
+            None,
+            webbrowser.Chrome(chrome_cmd),
+            preferred=True,
+        )
 
         # Open a URL using the new browser type
         webbrowser.get(chrome_cmd).open(url)
 
 
-HOST = 'localhost'
+HOST = "localhost"
 shutdown_requested = False
 last_heartbeat = time.time()
 server_running = True
 
+
 class CustomHandler(http.server.SimpleHTTPRequestHandler):
     def do_GET(self) -> Any:
         global last_heartbeat
-        if self.path == '/heartbeat':
+        if self.path == "/heartbeat":
             last_heartbeat = time.time()
             self.send_response(200)
             self.end_headers()
             return
         else:
             return http.server.SimpleHTTPRequestHandler.do_GET(self)
 
+
 def monitor_heartbeat() -> None:
     global server_running
     while server_running:
@@ -66,15 +76,18 @@ def monitor_heartbeat() -> None:
             os._exit(0)
         time.sleep(1)
 
+
 def serve_forever(httpd: Any) -> None:
     while server_running:
         httpd.handle_request()
 
+
 def run_server(host: str, port: int) -> None:
     with socketserver.TCPServer((host, port), CustomHandler) as httpd:
         print(f"Serving at http://{host}:{port}")
         serve_forever(httpd)
 
+
 def is_port_available(port: int) -> bool:
     """
     Check if a given TCP port is available to start a server on the local machine.
@@ -84,14 +97,16 @@ def is_port_available(port: int) -> bool:
     """
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
         try:
-            s.bind(('localhost', port))
+            s.bind(("localhost", port))
             return True
         except socket.error:
             return False
 
+
 Filename = NewType("Filename", str)
 LineNumber = NewType("LineNumber", int)
-
+
+
 def generate_html(profile_fname: Filename, output_fname: Filename) -> None:
     """Apply a template to generate a single HTML payload containing the current profile."""
 
@@ -108,13 +123,23 @@ def generate_html(profile_fname: Filename, output_fname: Filename) -> None:
     scalene_dir = os.path.dirname(__file__)
 
     file_contents = {
-        'scalene_gui_js_text': read_file_content(scalene_dir, "scalene-gui", "scalene-gui.js"),
-        'prism_css_text': read_file_content(scalene_dir, "scalene-gui", "prism.css"),
-        'prism_js_text': read_file_content(scalene_dir, "scalene-gui", "prism.js"),
-        'tablesort_js_text': read_file_content(scalene_dir, "scalene-gui", "tablesort.js"),
-        'tablesort_number_js_text': read_file_content(scalene_dir, "scalene-gui", "tablesort.number.js")
+        "scalene_gui_js_text": read_file_content(
+            scalene_dir, "scalene-gui", "scalene-gui.js"
+        ),
+        "prism_css_text": read_file_content(
+            scalene_dir, "scalene-gui", "prism.css"
+        ),
+        "prism_js_text": read_file_content(
+            scalene_dir, "scalene-gui", "prism.js"
+        ),
+        "tablesort_js_text": read_file_content(
+            scalene_dir, "scalene-gui", "tablesort.js"
+        ),
+        "tablesort_number_js_text": read_file_content(
+            scalene_dir, "scalene-gui", "tablesort.number.js"
+        ),
     }
-    
+
     # Put the profile and everything else into the template.
     environment = Environment(
         loader=FileSystemLoader(os.path.join(scalene_dir, "scalene-gui"))
@@ -126,11 +151,11 @@ def generate_html(profile_fname: Filename, output_fname: Filename) -> None:
         import scalene.scalene_config as scalene_config
     rendered_content = template.render(
         profile=profile,
-        gui_js=file_contents['scalene_gui_js_text'],
-        prism_css=file_contents['prism_css_text'],
-        prism_js=file_contents['prism_js_text'],
-        tablesort_js=file_contents['tablesort_js_text'],
-        tablesort_number_js=file_contents['tablesort_number_js_text'],
+        gui_js=file_contents["scalene_gui_js_text"],
+        prism_css=file_contents["prism_css_text"],
+        prism_js=file_contents["prism_js_text"],
+        tablesort_js=file_contents["tablesort_js_text"],
+        tablesort_number_js=file_contents["tablesort_number_js_text"],
         scalene_version=scalene_config.scalene_version,
         scalene_date=scalene_config.scalene_date,
     )
@@ -146,32 +171,33 @@ def generate_html(profile_fname: Filename, output_fname: Filename) -> None:
 def start(filename: str, port: int) -> None:
     while not is_port_available(port):
         port += 1
-        
+
     cwd = os.getcwd()
     if filename == "demo":
         generate_html(Filename("demo"), Filename("demo.html"))
         filename = "demo.html"
-    shutil.copy(filename, os.path.join(tempfile.gettempdir(), 'index.html'))
+    shutil.copy(filename, os.path.join(tempfile.gettempdir(), "index.html"))
     os.chdir(tempfile.gettempdir())
     server_thread = threading.Thread(target=run_server, args=[HOST, port])
     server_thread.start()
     threading.Thread(target=monitor_heartbeat).start()
 
-    webbrowser.open_new(f'http://{HOST}:{port}/')
+    webbrowser.open_new(f"http://{HOST}:{port}/")
     server_thread.join()
 
     os.chdir(cwd)
-    
+
     # Optional: a delay to ensure all resources are released
     time.sleep(1)
     os._exit(0)  # Forcefully stops the program
 
-if __name__ == '__main__':
+
+if __name__ == "__main__":
     import sys
+
     if len(sys.argv) > 2:
         filename = sys.argv[1]
         port = int(sys.argv[2])
         start(filename, port)
     else:
         print("Need to supply filename and port arguments.")
-

scalene/redirect_python.py

@@ -1,11 +1,12 @@
 import os
 import pathlib
-import re
 import stat
 import sys
 
 
-def redirect_python(preface: str, cmdline: str, python_alias_dir: pathlib.Path) -> str:
+def redirect_python(
+    preface: str, cmdline: str, python_alias_dir: pathlib.Path
+) -> str:
     """
     Redirects Python calls to a different command with a preface and cmdline.
 
@@ -51,5 +52,5 @@ def redirect_python(preface: str, cmdline: str, python_alias_dir: pathlib.Path)
         sys_executable_path = sys_executable_path.with_suffix(".bat")
 
     sys.executable = str(sys_executable_path)
-    
+
     return orig_sys_executable

scalene/replacement_mp_lock.py

@@ -1,7 +1,4 @@
 import multiprocessing.synchronize
-import sys
-import threading
-from typing import Any
 
 from scalene.scalene_profiler import Scalene
 

scalene/replacement_sem_lock.py

@@ -1,13 +1,16 @@
-import multiprocessing
+import multiprocessing.context
+import multiprocessing.synchronize
 import random
 import sys
 import threading
-from multiprocessing.synchronize import Lock
 from scalene.scalene_profiler import Scalene
 from typing import Any, Callable, Optional, Tuple
 
+
 class ReplacementSemLock(multiprocessing.synchronize.Lock):
-    def __init__(self, ctx: Optional[multiprocessing.context.DefaultContext] = None) -> None:
+    def __init__(
+        self, ctx: Optional[multiprocessing.context.DefaultContext] = None
+    ) -> None:
         # Ensure to use the appropriate context while initializing
         if ctx is None:
             ctx = multiprocessing.get_context()

scalene/replacement_signal_fns.py

@@ -5,16 +5,17 @@
 from scalene.scalene_profiler import Scalene
 from typing import Any, Tuple
 
+
 @Scalene.shim
 def replacement_signal_fns(scalene: Scalene) -> None:
     scalene_signals = scalene.get_signals()
     expected_handlers_map = {
-            scalene_signals.malloc_signal: scalene.malloc_signal_handler,
-            scalene_signals.free_signal: scalene.free_signal_handler,
-            scalene_signals.memcpy_signal: scalene.memcpy_signal_handler,
-            signal.SIGTERM: scalene.term_signal_handler,
-            scalene_signals.cpu_signal: scalene.cpu_signal_handler,
-        }
+        scalene_signals.malloc_signal: scalene.malloc_signal_handler,
+        scalene_signals.free_signal: scalene.free_signal_handler,
+        scalene_signals.memcpy_signal: scalene.memcpy_signal_handler,
+        signal.SIGTERM: scalene.term_signal_handler,
+        scalene_signals.cpu_signal: scalene.cpu_signal_handler,
+    }
     old_signal = signal.signal
     if sys.version_info < (3, 8):
 
@@ -61,11 +62,14 @@ def replacement_signal(signum: int, handler: Any) -> Any:
         # a NOP-like, then we can ignore it. It can't have been set already, and the expected return value is the
         # previous handler, so this behavior is reasonable
         if signum in all_signals and (
-                handler is signal.SIG_IGN or handler is signal.SIG_DFL
+            handler is signal.SIG_IGN or handler is signal.SIG_DFL
         ):
             return handler
         # If trying to "reset" to a handler that we already set it to, ignore
-        if signal.Signals(signum) in expected_handlers_map and expected_handlers_map[signal.Signals(signum)] is handler:
+        if (
+            signal.Signals(signum) in expected_handlers_map
+            and expected_handlers_map[signal.Signals(signum)] is handler
+        ):
             return signal.SIG_IGN
         if signum in all_signals:
             print(
@@ -108,7 +112,9 @@ def replacement_siginterrupt(signum: int, flag: bool) -> None:
                 )
             return old_siginterrupt(signum, flag)
 
-        def replacement_setitimer(which: int, seconds: float, interval: float = 0.0) -> Tuple[float, float]:
+        def replacement_setitimer(
+            which: int, seconds: float, interval: float = 0.0
+        ) -> Tuple[float, float]:
             timer_signal, cpu_signal = scalene.get_timer_signals()
             if which == timer_signal:
                 old = scalene.client_timer.get_itimer()
@@ -122,7 +128,7 @@ def replacement_setitimer(which: int, seconds: float, interval: float = 0.0) ->
         signal.setitimer = replacement_setitimer
         signal.siginterrupt = replacement_siginterrupt
 
-    signal.signal = replacement_signal # type: ignore
+    signal.signal = replacement_signal  # type: ignore
     if sys.version_info >= (3, 8):
         signal.raise_signal = replacement_raise_signal
     os.kill = replacement_kill

scalene/scalene-gui/example-profile.js

@@ -1,9 +1,9 @@
 const example_profile = {
   elapsed_time_sec: 12.696948051452637,
-    memory: true,
+  memory: true,
   files: {
-      "./test/testme.py": {
-	  imports: [],
+    "./test/testme.py": {
+      imports: [],
       functions: [
         {
           line: "doit1",