from functools import wraps from inspect import signature from itertools import starmap from typing import Callable from funcy import iterate, first, curry, map from pymonad.either import Left, Right, Either from pymonad.tools import curry as pmcurry from image_prediction.utils import get_logger logger = get_logger() def until(cond, func, *args, **kwargs): return first(filter(cond, iterate(func, *args, **kwargs))) def lift(fn): return curry(map)(fn) def starlift(fn): return curry(starmap)(fn) def bottom(*args, **kwargs): return False def top(*args, **kwargs): return True def left(fn): @wraps(fn) def inner(x): return Left(fn(x)) return inner def right(fn): @wraps(fn) def inner(x): return Right(fn(x)) return inner def wrap_left(fn, success_condition=top, error_message=None) -> Callable: return wrap_either(Left, Right, success_condition=success_condition, error_message=error_message)(fn) def wrap_right(fn, success_condition=top, error_message=None) -> Callable: return wrap_either(Right, Left, success_condition=success_condition, error_message=error_message)(fn) def wrap_either(success_type, failure_type, success_condition=top, error_message=None) -> Callable: @wraps(wrap_either) def wrapper(fn) -> Callable: n_params = len(signature(fn).parameters) @pmcurry(n_params) @wraps(fn) def wrapper(*args, **kwargs) -> Either: try: result = fn(*args, **kwargs) if success_condition(result): return success_type(result) else: return failure_type({"error": error_message, "result": result}) except Exception as err: logger.error(err) return failure_type({"error": error_message or err, "result": Void}) return wrapper return wrapper class Void: pass