51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
"""Common utilities."""
|
|
|
|
import itertools
|
|
import logging
|
|
import sys
|
|
from collections.abc import Iterable
|
|
from subprocess import PIPE, Popen
|
|
from types import FunctionType
|
|
|
|
|
|
def configure_logger(level: str = "INFO") -> None:
|
|
"""Configure the logger.
|
|
|
|
Args:
|
|
level (str, optional): The logging level. Defaults to "INFO".
|
|
"""
|
|
logging.basicConfig(
|
|
level=level,
|
|
datefmt="%Y-%m-%dT%H:%M:%S%z",
|
|
format="%(asctime)s %(levelname)s %(filename)s:%(lineno)d - %(message)s",
|
|
handlers=[logging.StreamHandler(sys.stdout)],
|
|
)
|
|
|
|
|
|
def partition(predicate: FunctionType, iterable: Iterable) -> tuple[Iterable, Iterable]:
|
|
"""Partition entries into false entries and true entries.
|
|
|
|
If *predicate* is slow, consider wrapping it with functools.lru_cache().
|
|
"""
|
|
# partition(is_odd, range(10)) → 0 2 4 6 8 and 1 3 5 7 9
|
|
t1, t2 = itertools.tee(iterable)
|
|
return itertools.filterfalse(predicate, t1), filter(predicate, t2)
|
|
|
|
|
|
def bash_wrapper(command: str, path: str = ".") -> tuple[str, str, int]:
|
|
"""Execute a bash command and capture the output.
|
|
|
|
Args:
|
|
command (str): The bash command to be executed.
|
|
path (str): The current working directory, '.' by default
|
|
|
|
Returns:
|
|
Tuple[str, int]: A tuple containing the output of the command (stdout) as a string,
|
|
the error output (stderr) as a string (optional), and the return code as an integer.
|
|
"""
|
|
# This is a acceptable risk
|
|
process = Popen(command.split(), stdout=PIPE, stderr=PIPE, cwd=path) # noqa: S603
|
|
output, error = process.communicate()
|
|
|
|
return output.decode(), error.decode(), process.returncode
|