"""Utility functions.
"""
from functools import reduce
import logging
logger = logging.getLogger('lavaflow')
# -----------------------------------------------------------------------------
# Helpers
[docs]def chain(x):
"""Function to flatten a list of lists.
Args:
x (list): list of lists of items
Returns:
items (list): list of items
Examples:
>>> x = [[1, 2], [3, 4]]
>>> chain(x)
[1, 2, 3, 4]
"""
items = []
for v in x:
items.extend(v)
return items
# -----------------------------------------------------------------------------
# Function wrappers
[docs]def compose(functions):
"""Compose an iterator of functions in sequence.
Args:
functions (iterable): an iterator of functions
Returns:
f (function): composition of functions
Examples:
>>> g = compose([lambda x: x + 1, lambda x: x * 2])
>>> g(1)
4
"""
def c(f, g):
return lambda x: f(g(x))
return reduce(c, reversed(functions), lambda x: x)
[docs]def dictify(f, x, y):
"""Wrap a function to take a dictionary as input and to map input key(s) to output key(s).
Args:
f (function): function to be wrapped
x (str|list): input key(s)
y (str|list): output key(s)
Returns:
g (function): new function
Examples:
>>> g = dictify(lambda a: a + 1, 'a', 'b')
>>> g({'a': 1})
{'a': 1, 'b': 2}
>>> g = dictify(lambda a, b: a + b, ['a', 'b'], 'c')
>>> g({'a': 1, 'b': 2})
{'a': 1, 'b': 2, 'c': 3}
>>> g = dictify(lambda a, b: (a + b, b - a), ['a', 'b'], ['c', 'd'])
>>> g({'a': 1, 'b': 2})
{'a': 1, 'b': 2, 'c': 3, 'd': 1}
"""
if isinstance(x, str):
x = [x]
def g(d):
u = f(*[d[k] for k in x])
if isinstance(y, str):
d[y] = u
else:
for k, v in zip(y, u):
d[k] = v
return d
return g
[docs]def windowify(f, i, x, j, y, chain_args):
"""Wrap a function to take a list of dictionaries as input and to map input key(s) to output key(s).
Args:
f (function): function to be wrapped
i (int|list): input item(s)
x (str|list): input key(s)
j (int|list): output item(s)
y (str|list): output key(s)
chain_args (bool): flatten inputs
Returns:
g (function): new function
Examples:
>>> g = windowify(lambda a0, a1, b0, b1: (a0 + a1, b0 + b1), [0, 2], ['a', 'b'], [1], ['c', 'd'], True)
>>> g([{'a': 1, 'b': 1}, {'a': 2, 'b': 2} , {'a': 3, 'b': 3}])
[{'a': 1, 'b': 1}, {'a': 2, 'b': 2, 'c': 4, 'd': 4}, {'a': 3, 'b': 3}]
>>> g = windowify(lambda a, b: (a[0] + a[1], b[0] + b[1]), [0, 2], ['a', 'b'], [1], ['c', 'd'], False)
>>> g([{'a': 1, 'b': 1}, {'a': 2, 'b': 2} , {'a': 3, 'b': 3}])
[{'a': 1, 'b': 1}, {'a': 2, 'b': 2, 'c': 4, 'd': 4}, {'a': 3, 'b': 3}]
"""
if isinstance(i, int):
i = [i]
if isinstance(x, str):
x = [x]
if isinstance(j, int):
j = [j]
def g(window):
if chain_args:
args = chain([[window[a][k] for a in i] for k in x])
else:
args = [[window[a][k] for a in i] for k in x]
u = f(*args)
for b in j:
if isinstance(y, str):
window[b][y] = u
else:
for k, v in zip(y, u):
window[b][k] = v
return window
return g