Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def wrapper(*args, **kwargs):
strip_args = map(strip_symbolic, args)
strip_kwargs = {k: strip_symbolic(v) for k,v in kwargs.items()}
if not args:
return dispatch_func(NoArgs(), **strip_kwargs)
return dispatch_func(*strip_args, **strip_kwargs)
def create_pipe_call(source, *args, **kwargs):
first, *rest = args
return Pipeable(Call(
"__call__",
strip_symbolic(source),
strip_symbolic(first),
*(Lazy(strip_symbolic(x)) for x in rest),
**{k: Lazy(strip_symbolic(v)) for k,v in kwargs.items()}
))
def create_pipe_call(source, *args, **kwargs):
first, *rest = args
return Pipeable(Call(
"__call__",
strip_symbolic(source),
strip_symbolic(first),
*(Lazy(strip_symbolic(x)) for x in rest),
**{k: Lazy(strip_symbolic(v)) for k,v in kwargs.items()}
))
def _call_strip_ascending(f):
if isinstance(f, Symbolic):
f = strip_symbolic(f)
if isinstance(f, Call) and f.func == "__neg__":
return f.args[0], False
return f, True
def case_when(__data, cases):
if isinstance(cases, Call):
cases = cases(__data)
# TODO: handle when receive list of (k,v) pairs for py < 3.5 compat?
stripped_cases = {strip_symbolic(k): strip_symbolic(v) for k,v in cases.items()}
n = len(__data)
out = np.repeat(None, n)
for k, v in reversed(list(stripped_cases.items())):
if callable(k):
result = _val_call(k, __data, n)
indx = np.where(result)[0]
val_res = _val_call(v, __data, n, indx)
out[indx] = val_res
elif k:
# e.g. k is just True, etc..
val_res = _val_call(v, __data, n)
out[:] = val_res
# by recreating an array, attempts to cast as best dtype
return np.array(list(out))