Reformat all utility files

Author: nicholasjng

pybm/util/common.py

@@ -1,10 +1,10 @@
 import re
 from typing import List, Iterable, Tuple, Callable, TypeVar, Dict
 
-T = TypeVar('T')
-S = TypeVar('S')
-U = TypeVar('U')
-V = TypeVar('V')
+T = TypeVar("T")
+S = TypeVar("S")
+U = TypeVar("U")
+V = TypeVar("V")
 
 
 def lmap(fn: Callable[[S], T], iterable: Iterable[S]) -> List[T]:
@@ -23,17 +23,19 @@ def tfilter(fn: Callable[[S], bool], iterable: Iterable[S]) -> Tuple[S, ...]:
     return tuple(filter(fn, iterable))
 
 
-def lpartition(fn: Callable[[S], bool], iterable: Iterable[S]) -> \
-        Tuple[List[S], List[S]]:
+def lpartition(
+    fn: Callable[[S], bool], iterable: Iterable[S]
+) -> Tuple[List[S], List[S]]:
     """Partition list into two with a boolean function. Not particularly
     efficient because of the double pass, but fine for small lists."""
     true_list = lfilter(lambda x: fn(x), iterable)
     false_list = lfilter(lambda x: not fn(x), iterable)
     return true_list, false_list
 
 
-def tpartition(fn: Callable[[S], bool], iterable: Iterable[S]) -> \
-        Tuple[Tuple[S, ...], Tuple[S, ...]]:
+def tpartition(
+    fn: Callable[[S], bool], iterable: Iterable[S]
+) -> Tuple[Tuple[S, ...], Tuple[S, ...]]:
     """Partition tuple into two with a boolean function. Not particularly
     efficient because of the double pass, but fine for small tuples."""
     true_list = tfilter(lambda x: fn(x), iterable)
@@ -49,17 +51,15 @@ def dvmap(fn: Callable[[U], T], dictionary: Dict[S, U]) -> Dict[S, T]:
     return {k: fn(v) for k, v in dictionary.items()}
 
 
-def dmap(fn: Callable[[S, T], Tuple[U, V]], dictionary: Dict[S, T]) \
-        -> Dict[U, V]:
+def dmap(fn: Callable[[S, T], Tuple[U, V]], dictionary: Dict[S, T]) -> Dict[U, V]:
     return {k: v for k, v in map(fn, dictionary.keys(), dictionary.values())}
 
 
 def dkfilter(fn: Callable[[S], bool], dictionary: Dict[S, T]) -> Dict[S, T]:
     return {k: dictionary[k] for k in lfilter(fn, dictionary.keys())}
 
 
-def dfilter(fn: Callable[[Tuple[S, T]], bool], dictionary: Dict[S, T]) \
-        -> Dict[S, T]:
+def dfilter(fn: Callable[[Tuple[S, T]], bool], dictionary: Dict[S, T]) -> Dict[S, T]:
     return {k: v for k, v in tfilter(fn, dictionary.items())}
 
 
@@ -89,12 +89,12 @@ def flatten(t):
     return [item for sublist in t for item in sublist]
 
 
-def partition_n(n: int, fn: Callable[[T], int], listlike: Iterable[T]) \
-        -> List[List[T]]:
+def partition_n(n: int, fn: Callable[[T], int], listlike: Iterable[T]) -> List[List[T]]:
     return_obj: List[List[T]] = [[] for _ in range(n)]
     for elem in listlike:
         k = fn(elem)
-        assert 0 <= k < n, "partition function needs to return an integer " \
-                           "in the interval [0, {n})."
+        assert (
+            0 <= k < n
+        ), f"partition function needs to return an integer in the interval [0, {n})."
         return_obj[k].append(elem)
     return return_obj

pybm/util/extras.py

@@ -1,8 +1,6 @@
 def get_extras():
     """Extra pybm functionality, specified as a valid argument to
     setuptools.setup's 'extras_require' keyword argument."""
-    extra_features = {
-        "gbm": ["google-benchmark==0.2.0"]
-    }
+    extra_features = {"gbm": ["git+https://github.com/google/benchmark"]}
     extra_features["all"] = sum(extra_features.values(), start=[])
     return extra_features

pybm/util/git.py

@@ -12,7 +12,7 @@
 
 def get_git_version() -> Tuple[int, ...]:
     rc, output = git_subprocess(["git", "--version"])
-    version_str = re.search(r'([\d.]+)', output)
+    version_str = re.search(r"([\d.]+)", output)
     if version_str is not None:
         version = version_str.group()
         return version_tuple(version)
@@ -30,19 +30,22 @@ def get_git_version() -> Tuple[int, ...]:
 
 # ---------------------------------------
 
+
 def _feature_guard(min_git: Tuple[int, int, int]):
     if GIT_VERSION < min_git:
         min_git_str = version_string(min_git)
-        msg = f"Command `git restore` requires at minimum git version " \
-              f"{min_git_str}, "
+        msg = f"Command `git restore` requires at minimum git version {min_git_str}, "
 
         if GIT_VERSION == (0, 0, 0):
-            msg += "but no git installation was found on your system. " \
-                   "Please assure that git is installed and added to PATH."
+            msg += (
+                "but no git installation was found on your system. "
+                "Please assure that git is installed and added to PATH."
+            )
         else:
             curr_git_str = version_string(GIT_VERSION)
-            msg += f"but your installed git was found to be only version " \
-                   f"{curr_git_str}."
+            msg += (
+                f"but your installed git was found to be only version {curr_git_str}."
+            )
         raise GitError(msg)
 
 
@@ -80,8 +83,10 @@ def resolve_ref(commit_ish: str, resolve_commits: bool):
     elif is_valid_sha1_part(commit_ish):
         ref_type = "commit"
     else:
-        msg = f"Input {commit_ish!r} did not resolve to any known local " \
-              f"branch, tag or valid commit SHA1."
+        msg = (
+            f"Input {commit_ish!r} did not resolve to any known local "
+            f"branch, tag or valid commit SHA1."
+        )
         raise GitError(msg)
     # force commit resolution, leads to detached HEAD
     if resolve_commits:
@@ -105,10 +110,10 @@ def process_line(line: str) -> Tuple[str, str]:
 
     tag_marker = "^{}"
     # show-ref exits with 1 if no tags exist in the repo
-    rc, tags = git_subprocess(["git", "show-ref", "--tags", "-d"],
-                              allowed_statuscodes=[1])
-    commits_and_tags = lfilter(lambda x: x.endswith(tag_marker),
-                               tags.splitlines())
+    rc, tags = git_subprocess(
+        ["git", "show-ref", "--tags", "-d"], allowed_statuscodes=[1]
+    )
+    commits_and_tags = lfilter(lambda x: x.endswith(tag_marker), tags.splitlines())
     # closure above is required here to keep mypy happy
     split_list = lmap(process_line, commits_and_tags)
     return dict(split_list)
@@ -151,8 +156,7 @@ def checkout(ref: str, cwd: Union[str, Path]):
     git_subprocess(command=command, cwd=cwd)
 
 
-def get_from_history(ref: str, resource: Union[str, Path],
-                     directory: Union[str, Path]):
+def get_from_history(ref: str, resource: Union[str, Path], directory: Union[str, Path]):
     """Check out a file or directory from another git reference."""
     # Source:
     # https://stackoverflow.com/questions/307579/how-do-i-copy-a-version-of-a-single-file-from-one-git-branch-to-another

pybm/util/imports.py

@@ -35,7 +35,8 @@ def import_module_from_source(source_path: Union[str, Path]) -> ModuleType:
         # return loaded module
         return sys.modules[py_name]
     spec: Optional[ModuleSpec] = importlib.util.spec_from_file_location(
-        py_name, source_path)
+        py_name, source_path
+    )
     if spec is not None:
         module = importlib.util.module_from_spec(spec)
         assert isinstance(spec.loader, Loader)
@@ -53,8 +54,9 @@ def import_func_from_source(source_path: str, fn_name: str) -> Callable:
         module = import_module_from_source(source_path)
         return getattr(module, fn_name)
     except (AttributeError, PybmError) as e:
-        raise PybmError(f"Failed to import function {fn_name} from source "
-                        f"file {source_path}.") from e
+        raise PybmError(
+            f"Failed to import function {fn_name} from source " f"file {source_path}."
+        ) from e
 
 
 def import_from_module(identifier: str) -> Callable:
@@ -63,8 +65,10 @@ def import_from_module(identifier: str) -> Callable:
     *module_parts, name = identifier.split(".")
     module_name = ".".join(module_parts)
     if not module_exists(module_name):
-        raise PybmError(f"Module {module_name} does not exist in the current "
-                        f"Python environment.")
+        raise PybmError(
+            f"Module {module_name} does not exist in the current "
+            f"Python environment."
+        )
     module = importlib.import_module(module_name)
     if not hasattr(module, name):
         raise PybmError(f"Module {module_name} has no member {name}.")
@@ -79,8 +83,7 @@ def import_function(name: str):
         pass
     try:
         source_path, fn_name = name.split(":")
-        func = import_func_from_source(source_path=source_path,
-                                       fn_name=fn_name)
+        func = import_func_from_source(source_path=source_path, fn_name=fn_name)
     except (PybmError, AttributeError):
         pass
     return func

pybm/util/path.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Union, List
+from typing import Union, List, Optional
 
 from pybm.util.common import lmap
 
@@ -14,11 +14,23 @@ def get_subdirs(path: Union[str, Path]):
     return [f.stem for f in filter(Path.is_dir, p.iterdir())]
 
 
-def list_contents(path: Union[str, Path], file_suffix: str = "",
-                  names_only: bool = True) -> List[str]:
-    p = Path(path).resolve()
+def list_contents(
+    path: Union[str, Path],
+    file_suffix: str = "",
+    rel_path: Optional[str] = None,
+    names_only: bool = False,
+) -> List[str]:
+
+    resolved_path = Path(path).resolve()
+
     glob_pattern = "*" if file_suffix == "" else "*" + file_suffix
+    matching_files = resolved_path.glob(glob_pattern)
+
     if names_only:
-        return lmap(lambda x: str(x.name), p.glob(glob_pattern))
+        return lmap(lambda x: x.name, matching_files)
+    elif rel_path is not None:
+        return lmap(
+            lambda p: str(p.relative_to(rel_path)), matching_files  # type: ignore
+        )
     else:
-        return lmap(str, p.glob(glob_pattern))
+        return lmap(str, matching_files)

pybm/util/print.py

@@ -18,15 +18,12 @@ def calculate_column_widths(data: Iterable[Iterable[str]]) -> List[int]:
     return lmap(max, data_lengths)
 
 
-def make_line(values: Iterable[str], column_widths: Iterable[int],
-              padding: int) -> str:
+def make_line(values: Iterable[str], column_widths: Iterable[int], padding: int) -> str:
     pad_char = " " * padding
     sep = "|".join([pad_char] * 2)
     offset = pad_char
 
-    line = offset + sep.join(
-        f"{n:<{w}}" for n, w in zip(values, column_widths)
-    )
+    line = offset + sep.join(f"{n:<{w}}" for n, w in zip(values, column_widths))
     return line
 
 

pybm/util/subprocess.py

@@ -7,27 +7,36 @@
 Handling = Literal["raise", "ignore"]
 
 
-def run_subprocess(command: List[str],
-                   allowed_statuscodes: List[int] = None,
-                   ex_type: type = PybmError,
-                   errors: Handling = "raise",
-                   cwd: Union[str, Path] = None) -> Tuple[int, str]:
+def run_subprocess(
+    command: List[str],
+    allowed_statuscodes: List[int] = None,
+    ex_type: type = PybmError,
+    errors: Handling = "raise",
+    cwd: Union[str, Path] = None,
+) -> Tuple[int, str]:
     full_command = " ".join(command)
+
     allowed_statuscodes = allowed_statuscodes or []
     allowed_statuscodes.append(0)
     # logger.debug(f"Running command {full_command}...")
-    p = subprocess.run(command,
-                       stdout=subprocess.PIPE,
-                       stderr=subprocess.PIPE,
-                       encoding="utf-8",
-                       cwd=cwd)
+    p = subprocess.run(
+        command,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        encoding="utf-8",
+        cwd=cwd,
+    )
+
     rc = p.returncode
     if rc not in allowed_statuscodes:
         if errors == "raise":
-            msg = f"The command `{full_command}` returned the non-zero " \
-                  f"exit code {rc}.\nFurther information (stderr " \
-                  f"output of the subprocess):\n{p.stderr}"
+            msg = (
+                f"The command `{full_command}` returned the non-zero "
+                f"exit code {rc}.\nFurther information (stderr "
+                f"output of the subprocess):\n{p.stderr}"
+            )
             raise ex_type(msg)
         else:
             return rc, p.stderr
+
     return rc, p.stdout