Something went wrong. Try again.
keyboard stuff
Something went wrong. Try again.
3.2 kB · 108 lines
Python
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109"""Utility functions."""import contextlibimport multiprocessingimport sysimport re
from milc import cli
TRIPLET_PATTERN = re.compile(r'^(\d+)\.(\d+)\.(\d+)')
maybe_exit_should_exit = Truemaybe_exit_reraise = False
# Controls whether or not early `exit()` calls should be madedef maybe_exit(rc): if maybe_exit_should_exit: sys.exit(rc) if maybe_exit_reraise: e = sys.exc_info()[1] if e: raise e
def maybe_exit_config(should_exit: bool = True, should_reraise: bool = False): global maybe_exit_should_exit global maybe_exit_reraise maybe_exit_should_exit = should_exit maybe_exit_reraise = should_reraise
def truthy(value, value_if_unknown=False): """Returns True if the value is truthy, False otherwise.
Deals with: True: 1, true, t, yes, y, on False: 0, false, f, no, n, off """ if value in {False, True}: return bool(value)
test_value = str(value).strip().lower()
if test_value in {"1", "true", "t", "yes", "y", "on"}: return True
if test_value in {"0", "false", "f", "no", "n", "off"}: return False
return value_if_unknown
@contextlib.contextmanagerdef parallelize(): """Returns a function that can be used in place of a map() call.
Attempts to use `mpire`, falling back to `multiprocessing` if it's not available. If parallelization is not requested, returns the original map() function. """
# Work out if we've already got a config value for parallel searching if cli.config.user.parallel_search is None: parallel_search = True else: parallel_search = cli.config.user.parallel_search
# Non-parallel searches use `map()` if not parallel_search: yield map return
# Prefer mpire's `WorkerPool` if it's available with contextlib.suppress(ImportError): from mpire import WorkerPool from mpire.utils import make_single_arguments with WorkerPool() as pool:
def _worker(func, *args): # Ensure we don't unpack tuples -- mpire's `WorkerPool` tries to do so normally so we tell it not to. for r in pool.imap_unordered(func, make_single_arguments(*args, generator=False), progress_bar=True): yield r
yield _worker return
# Otherwise fall back to multiprocessing's `Pool` with multiprocessing.Pool() as pool: yield pool.imap_unordered
def parallel_map(*args, **kwargs): """Effectively runs `map()` but executes it in parallel if necessary. """ with parallelize() as map_fn: # This needs to be enclosed in a `list()` as some implementations return # a generator function, which means the scope of the pool is closed off # before the results are returned. Returning a list ensures results are # materialised before any worker pool is shut down. return list(map_fn(*args, **kwargs))
def triplet_to_bcd(ver: str): m = TRIPLET_PATTERN.match(ver) if not m: return '0x00000000' return f'0x{int(m.group(1)):02d}{int(m.group(2)):02d}{int(m.group(3)):04d}'