Type refactoring

scalene/find_browser.py

@@ -1,17 +1,32 @@
 import webbrowser
 from typing import Optional
 
+
 def find_browser() -> Optional[webbrowser.BaseBrowser]:
     """Find the default browser if possible and if compatible."""
     # Names of known graphical browsers as per Python's webbrowser documentation
-    graphical_browsers = ["windowsdefault", "macosx", "safari", "google-chrome", 
-                          "chrome", "chromium", "firefox", "opera", "edge", "mozilla", "netscape"]
+    graphical_browsers = [
+        "windowsdefault",
+        "macosx",
+        "safari",
+        "google-chrome",
+        "chrome",
+        "chromium",
+        "firefox",
+        "opera",
+        "edge",
+        "mozilla",
+        "netscape",
+    ]
     try:
         # Get the default browser object
         browser = webbrowser.get()
         # Check if the browser's class name matches any of the known graphical browsers
         browser_class_name = str(type(browser)).lower()
-        if any(graphical_browser in browser_class_name for graphical_browser in graphical_browsers):
+        if any(
+            graphical_browser in browser_class_name
+            for graphical_browser in graphical_browsers
+        ):
             return browser
         else:
             return None

scalene/get_module_details.py

@@ -92,4 +92,3 @@ def _get_module_details(
     if code is None:
         raise error("No code object available for %s" % mod_name)
     return mod_name, spec, code
-

scalene/redirect_python.py

@@ -3,19 +3,26 @@
 import stat
 import sys
 
-def redirect_python(preface: str, cmdline: str, python_alias_dir: pathlib.Path) -> None:
+
+def redirect_python(
+    preface: str, cmdline: str, python_alias_dir: pathlib.Path
+) -> None:
     # Likely names for the Python interpreter.
     base_python_extension = ".exe" if sys.platform == "win32" else ""
     all_python_names = [
         "python" + base_python_extension,
         "python" + str(sys.version_info.major) + base_python_extension,
-        "python" + str(sys.version_info.major) + "." + str(sys.version_info.minor) + base_python_extension
+        "python"
+        + str(sys.version_info.major)
+        + "."
+        + str(sys.version_info.minor)
+        + base_python_extension,
     ]
     # if sys.platform == "win32":
     #     base_python_name = re.sub(r'\.exe$', '', os.path.basename(sys.executable))
     # else:
     #     base_python_name = sys.executable
-    
+
     # Don't show commands on Windows; regular shebang for
     # shell scripts on Linux/OS X
     shebang = "@echo off" if sys.platform == "win32" else "#!/bin/bash"
@@ -31,23 +38,20 @@ def redirect_python(preface: str, cmdline: str, python_alias_dir: pathlib.Path)
     for name in all_python_names:
         fname = os.path.join(python_alias_dir, name)
         if sys.platform == "win32":
-            fname = re.sub(r'\.exe$', '.bat', fname)
+            fname = re.sub(r"\.exe$", ".bat", fname)
         with open(fname, "w") as file:
             file.write(payload)
         os.chmod(fname, stat.S_IXUSR | stat.S_IRUSR | stat.S_IWUSR)
 
     # Finally, insert this directory into the path.
     sys.path.insert(0, str(python_alias_dir))
     os.environ["PATH"] = (
-        str(python_alias_dir)
-        + os.pathsep
-        + os.environ["PATH"]
+        str(python_alias_dir) + os.pathsep + os.environ["PATH"]
     )
     # Force the executable (if anyone invokes it later) to point to one of our aliases.
     sys.executable = os.path.join(
         python_alias_dir,
         all_python_names[0],
     )
     if sys.platform == "win32" and sys.executable.endswith(".exe"):
-        sys.executable = re.sub(r'\.exe$', '.bat', sys.executable)
-
+        sys.executable = re.sub(r"\.exe$", ".bat", sys.executable)

scalene/replacement_mp_lock.py

@@ -15,10 +15,9 @@
 # timeout_obj is parsed as a double
 
 from scalene.replacement_sem_lock import ReplacementSemLock
-
+
+
 @Scalene.shim
 def replacement_mp_semlock(scalene: Scalene) -> None:
-    ReplacementSemLock.__qualname__ = (
-        "replacement_semlock.ReplacementSemLock"
-    )
+    ReplacementSemLock.__qualname__ = "replacement_semlock.ReplacementSemLock"
     multiprocessing.synchronize.Lock = ReplacementSemLock  # type: ignore

scalene/replacement_pjoin.py

@@ -43,9 +43,9 @@ def replacement_process_join(self, timeout: float = -1) -> None:  # type: ignore
             if timeout != -1:
                 end_time = time.perf_counter()
                 if end_time - start_time >= timeout:
-                    from multiprocessing.process import (
+                    from multiprocessing.process import (  # type: ignore
                         _children,
-                    )  # type: ignore
+                    )
 
                     _children.discard(self)
                     return

scalene/replacement_sem_lock.py

@@ -6,15 +6,18 @@
 from scalene.scalene_profiler import Scalene
 from typing import Any
 
+
 def _recreate_replacement_sem_lock():
     return ReplacementSemLock()
 
+
 class ReplacementSemLock(multiprocessing.synchronize.Lock):
     def __init__(self, ctx=None):
         # Ensure to use the appropriate context while initializing
         if ctx is None:
             ctx = multiprocessing.get_context()
         super().__init__(ctx=ctx)
+
     def __enter__(self) -> bool:
         max_timeout = sys.getswitchinterval()
         tident = threading.get_ident()
@@ -26,8 +29,10 @@ def __enter__(self) -> bool:
             if acquired:
                 return True
             else:
-                max_timeout *= 2 # Exponential backoff
+                max_timeout *= 2  # Exponential backoff
+
     def __exit__(self, *args: Any) -> None:
         super().__exit__(*args)
+
     def __reduce__(self):
         return (_recreate_replacement_sem_lock, ())

scalene/scalene_analysis.py

@@ -110,15 +110,18 @@ def find_regions(src: str) -> Dict[int, Tuple[int, int]]:
         classes = {}
         for node in ast.walk(tree):
             if isinstance(node, ast.ClassDef):
+                assert node.end_lineno
                 for line in range(node.lineno, node.end_lineno + 1):
                     classes[line] = (node.lineno, node.end_lineno)
             if isinstance(node, (ast.For, ast.While)):
+                assert node.end_lineno
                 for line in range(node.lineno, node.end_lineno + 1):
                     loops[line] = (node.lineno, node.end_lineno)
             if isinstance(node, ast.FunctionDef):
+                assert node.end_lineno
                 for line in range(node.lineno, node.end_lineno + 1):
                     functions[line] = (node.lineno, node.end_lineno)
-        for lineno, line in enumerate(srclines, 1):
+        for lineno, _ in enumerate(srclines, 1):
             if lineno in loops:
                 regions[lineno] = loops[lineno]
             elif lineno in functions:
@@ -156,7 +159,11 @@ def walk(node, current_outermost_region, outer_class):
             for child_node in ast.iter_child_nodes(node):
                 walk(child_node, current_outermost_region, outer_class)
             if isinstance(node, ast.stmt):
-                outermost_is_loop = outer_class in [ast.For, ast.AsyncFor, ast.While]
+                outermost_is_loop = outer_class in [
+                    ast.For,
+                    ast.AsyncFor,
+                    ast.While,
+                ]
                 curr_is_block_not_loop = node.__class__ in [
                     ast.With,
                     ast.If,
@@ -174,9 +181,7 @@ def walk(node, current_outermost_region, outer_class):
                             # NOTE: this additionally accounts for the case in which `node`
                             # is a loop. Any loop within another loop should take on the entirety of the
                             # outermost loop too
-                            regions[
-                                line
-                            ] = current_outermost_region
+                            regions[line] = current_outermost_region
                         elif (
                             curr_is_block_not_loop
                             and len(srclines[line - 1].strip()) > 0

scalene/scalene_gpu.py

@@ -117,7 +117,9 @@ def gpu_memory_usage(self, pid: int) -> float:
         for i in range(self.__ngpus):
             handle = self.__handle[i]
             with contextlib.suppress(Exception):
-                for proc in pynvml.nvmlDeviceGetComputeRunningProcesses(handle):
+                for proc in pynvml.nvmlDeviceGetComputeRunningProcesses(
+                    handle
+                ):
                     # Only accumulate memory stats for the current pid.
                     if proc.usedGpuMemory and proc.pid == pid:
                         # First check is to protect against return of None

scalene/scalene_json.py

@@ -235,8 +235,8 @@ def output_profiles(
         pid: int,
         profile_this_code: Callable[[Filename, LineNumber], bool],
         python_alias_dir: Path,
-        program_path: Path,
-        entrypoint_dir: Path,
+        program_path: Filename,
+        entrypoint_dir: Filename,
         program_args: Optional[List[str]],
         profile_memory: bool = True,
         reduced_profile: bool = False,
@@ -289,7 +289,7 @@ def output_profiles(
         # Convert stacks into a representation suitable for JSON dumping.
         stks = []
         for stk in stats.stacks.keys():
-            this_stk : List[str] = []
+            this_stk: List[str] = []
             this_stk.extend(stk)
             stks.append((this_stk, stats.stacks[stk]))
 
@@ -449,7 +449,9 @@ def output_profiles(
                     profile_line["end_region_line"] = enclosing_regions[
                         lineno
                     ][1]
-                    profile_line["start_outermost_loop"] = outer_loop[lineno][0]
+                    profile_line["start_outermost_loop"] = outer_loop[lineno][
+                        0
+                    ]
                     profile_line["end_outermost_loop"] = outer_loop[lineno][1]
                     # When reduced-profile set, only output if the payload for the line is non-zero.
                     if reduced_profile:

scalene/scalene_jupyter.py

@@ -3,6 +3,7 @@
 from threading import Thread
 from typing import Any, Optional
 
+
 class ScaleneJupyter:
     @staticmethod
     def find_available_port(start_port: int, end_port: int) -> Optional[int]:

scalene/scalene_output.py

@@ -306,7 +306,7 @@ def output_profiles(
         pid: int,
         profile_this_code: Callable[[Filename, LineNumber], bool],
         python_alias_dir: Path,
-        program_path: Path,
+        program_path: Filename,
         program_args: Optional[List[str]],
         profile_memory: bool = True,
         reduced_profile: bool = False,

scalene/scalene_parseargs.py

@@ -208,16 +208,22 @@ def parse_args() -> Tuple[argparse.Namespace, List[str]]:
             + " [/blue])",
         )
         if sys.platform == "win32":
-            memory_profile_message = "profile memory (not supported on this platform)" 
+            memory_profile_message = (
+                "profile memory (not supported on this platform)"
+            )
         else:
-            memory_profile_message = "profile memory (default: [blue]" + (str(defaults.memory)) + " [/blue])"
+            memory_profile_message = (
+                "profile memory (default: [blue]"
+                + (str(defaults.memory))
+                + " [/blue])"
+            )
         parser.add_argument(
             "--memory",
             dest="memory",
             action="store_const",
             const=True,
             default=None,
-            help= memory_profile_message
+            help=memory_profile_message,
         )
         parser.add_argument(
             "--profile-all",