File: Broken/Core/__init__.py¶

def flatten(
    *items: Any,
    cast: type = list,
    block: Optional[Collection] = (None, ""),
    unpack: Iterable[type] = (list, deque, tuple, map, Generator),
) -> Collection[Any]:
    """
    Flatten/unpack nested iterables (list, deque, tuple, map, Generator) to a plain 1D list
    - Removes common falsy values by default, modify with `block={None, False, "", [], ...}`

    Example:
        ```python
        # [1, 2, 3, 4, 5, 6, 7]
        flatten([1, [2, 3], 4, [5, [6, 7]]])

        # [0, 1, 2, True, False, "Hello"]
        flatten(range(3), (True, False), None, "Hello")
        ```

    Returns:
        `cast`ed object with all `unpack`ed `items` without any of the `block`ed values
    """
    def flatten(data):
        if bool(block):
            data = filter(lambda item: (item not in block), data)
        for item in data:
            if isinstance(item, unpack):
                yield from flatten(item)
                continue
            yield item
    return cast(flatten(items))

def every(
    *items: Any,
    cast: type = list,
    block: Collection[Any] = (None, ""),
) -> Optional[Collection]:
    """
    Returns the flattened items if not any element is in the block list, else None. Useful when
    a Model class has a list of optional arguments that doesn't add falsy values to a command

    Usage:
        ```python
        every(1, 2, 3) # [1, 2, 3]
        every(1, 2, 3, None) # None
        every("-arg, "") # None
        ```
    """
    items = flatten(*items, block=None, cast=cast)
    if any(item in block for item in items):
        return None
    return items

def shell(
    *args: Any,
    output: bool = False,
    Popen: bool = False,
    env: dict[str, str] = None,
    confirm: bool = False,
    skip: bool = False,
    echo: bool = True,
    **kwargs
) -> Optional[Union[
    subprocess.CompletedProcess,
    subprocess.Popen,
    str,
]]:
    """
    Enhanced subprocess runners with many additional features. Flattens the args, converts to str

    Example:
        ```python
        shell(["binary", "-m"], "arg1", None, "arg2", 3, confirm=True)
        ```
    """
    if (output and Popen):
        raise ValueError(log.error("Cannot use (output=True) and (Popen=True) at the same time"))

    args = tuple(map(str, flatten(args)))

    # Assert command won't fail due unknown binary
    if (not shell) and (not shutil.which(args[0])):
        raise FileNotFoundError(log.error(f"Binary doesn't exist or was not found on PATH ({args[0]})"))

    # Log the command being run, temp variables
    _log = (log.skip if skip else log.info)
    _cwd = f" @ ({kwargs.get('cwd', '') or Path.cwd()})"
    _log(f"{args}{_cwd}", echo=echo)
    if skip: return

    if kwargs.get("shell", False):
        args = ' '.join(args)
        log.warning((
            "Running command with (shell=True), be careful.. "
            "Consider using (confirm=True)"*(not confirm)
        ))

    if confirm and not click.confirm("• Confirm running the command above"):
        return

    # Update current environ for the command only
    kwargs["env"] = os.environ | (env or {})

    # Windows: preexec_fn is not supported, remove from kwargs
    if (os.name == "nt") and (kwargs.pop("preexec_fn", None)):
        log.minor("shell(preexec_fn=...) is not supported on Windows, ignoring..")

    if (output):
        return subprocess.check_output(args, **kwargs).decode("utf-8")

    if (Popen):
        return subprocess.Popen(args, **kwargs)

    return subprocess.run(args, **kwargs)

def apply(
    callback: Callable,
    iterable: Iterable, *,
    cast: Callable = list
) -> Collection:
    """Applies a callback to all items of an iterable, returning a $cast of the results"""
    return cast(map(callback, iterable))

def denum(item: Union[enum.Enum, Any]) -> Any:
    """De-enumerates an item: if it's an Enum, returns the value, else the item itself"""
    return (item.value if isinstance(item, enum.Enum) else item)

def pop_fill(data: Collection, fill: type, length: int) -> Collection:
    """Pop or fill until a data's length is met"""
    while (len(data) > length):
        data.pop()
    while (len(data) < length):
        data.append(fill())
    return data

@contextlib.contextmanager
def multi_context(*contexts: contextlib.AbstractContextManager) -> Generator:
    """Enter multiple contexts at once"""
    with contextlib.ExitStack() as stack:
        for context in flatten(contexts):
            stack.enter_context(context)
        yield None

@contextlib.contextmanager
def tempvars(**variables: str) -> Generator:
    """Temporarily sets environment variables inside a context"""
    original = os.environ.copy()
    os.environ.update(variables)
    try:
        log.info(f"Setting environment variables: {tuple(variables.items())}")
        yield None
    finally:
        log.info(f"Restoring environment variables: {tuple(variables.keys())}")
        os.environ.clear()
        os.environ.update(original)

@contextlib.contextmanager
def block_modules(*modules: str) -> Generator:
    """Pretend a module isn't installed"""
    state = sys.modules.copy()
    try:
        for module in flatten(modules):
            sys.modules[module] = None
        yield None
    finally:
        for module in modules:
            if (module in state):
                sys.modules[module] = state[module]
                continue
            del sys.modules[module]

def smartproxy(object: Any) -> Any:
    """Returns a weakref proxy if the object is not already proxied"""
    from weakref import CallableProxyType, ProxyType, proxy

    if not isinstance(object, (CallableProxyType, ProxyType)):
        object = proxy(object)

    return object

def clamp(value: float, low: float=0, high: float=1) -> float:
    return max(low, min(value, high))

def nearest(number: Number, multiple: Number, *, cast=int, operator: Callable=round) -> Number:
    """Finds the nearest multiple of a base number, by default ints but works for floats too"""
    return cast(multiple * operator(number/multiple))

def list_get(data: list, index: int, default: Any=None) -> Optional[Any]:
    """Returns the item at 'index' or 'default' if out of range"""
    if (index >= len(data)):
        return default
    return data[index]

def hyphen_range(string: Optional[str], *, inclusive: bool=True) -> Iterable[int]:
    """
    Yields the numbers in a hyphenated CSV range, just like when selecting what pages to print
    - Accepts any of ("-", "..", "...", "_", "->") as a hyphenated range
    - Special values:
        - "all", returns infinite range from 0
        - "even", returns even numbers
        - "odd", returns odd numbers

    Example:
        ```python
        hyphen_range("2,3") # 2, 3
        hyphen_range("2-5") # 2, 3, 4, 5
        hyphen_range("1-3, 5") # 1, 2, 3, 5
        ```
    """
    if not bool(string):
        return None

    if (string == "all"):
        yield from itertools.count()
    elif (string == "even"):
        yield from itertools.count(0, 2)
    elif (string == "odd"):
        yield from itertools.count(1, 2)

    for part in string.split(","):
        if ("-" in part):
            start, end = map(int, re.split(r"_|-|\.\.|\.\.\.|\-\>", part))
            yield from range(start, end + int(inclusive))
            continue
        yield int(part)

def limited_ratio(
    number: Optional[float], *,
    limit: float = None
) -> Optional[tuple[int, int]]:
    """Same as Number.as_integer_ratio but with an optional upper limit and optional return"""
    if (number is None):
        return None

    num, den = number.as_integer_ratio()

    if limit and (den > limit or num > limit):
        normalize = limit/min(num, den)
        num *= normalize
        den *= normalize

    return (int(num), int(den))

def overrides(
    old: Any,
    new: Optional[Any],
    default: Optional[Any]=None,
    resets: Any=-1
) -> Optional[Any]:
    """Returns 'new' if is not None, else keeps 'old' value"""
    if (new == resets):
        return default
    if (new is None):
        return old
    return new

def install(*,
    packages: Union[str, Iterable[str]],
    pypi: Optional[Union[str, Iterable[str]]]=None,
    args: Optional[Union[str, Iterable[str]]]=None
) -> None:
    # Ensure arguments are tuples
    packages = flatten(packages, cast=tuple)
    pypi = flatten(pypi or packages, cast=tuple)
    args = flatten(args, cast=tuple)

    caller = inspect.currentframe().f_back.f_globals

    # Import the package and insert on the caller's globals
    def inject_packages():
        for package in packages:
            caller[package] = __import__(package)

    try:
        return inject_packages()
    except ImportError:
        log.info(f"Installing packages: {packages}..")

    for method in (
        (sys.executable, "-m", "uv", "pip", "install"),
        (sys.executable, "-m", "pip", "install")
    ):
        if shell(*method, *pypi, *args).returncode == 0:
            return inject_packages()

    raise RuntimeError(log.error(f"Failed to install packages: {packages}"))

def combinations(**options: Any) -> Iterable[DotMap]:
    """Returns a dictionary of key='this' of itertools.product"""
    for items in itertools.product(*options.values()):
        yield DotMap(zip(options.keys(), items))

def arguments() -> bool:
    """Returns True if any arguments are present on sys.argv"""
    return bool(sys.argv[1:])

def easyloop(method: Callable=None, *, period: float=0.0):
    """Wraps a method in an infinite loop called every 'period' seconds"""

    def decorator(method):

        @functools.wraps(method)
        def wrapper(*args, **kwargs):
            while True:
                method(*args, **kwargs)
                time.sleep(period)

        return wrapper

    if (method is None):
        return decorator

    return decorator(method)