使用非重叠风将函数映射到容器

2024-03-29 15:18:28 发布

您现在位置:Python中文网/ 问答频道 /正文

我在Python中有一个容器(比如list),其中有N个元素:x_1, x_2, ... x_N。 我想将某个函数func()映射到k组中的元素,即使用大小为k的窗口。 我希望窗口不重叠:

x_1, x_2, x_3, x_4

变成:

func(x_1, x_2), func(x_3, x_4)

更具体地说:

l = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]


def func(values):
    return sum(values)


grouped_map(func, l, 2)
# [1, 5, 9, 13, 17]

最后,我还想使用NumPy数组(使用NumPy感知函数)来实现这一点。你知道吗


Tags: 函数numpy元素mapreturndef数组容器
1条回答
网友
1楼 · 发布于 2024-03-29 15:18:28

对于任何iterable,可以首先定义一个只进行拆分的grouped()函数来解决这个问题:

def grouped(items, size):
    iterators = [iter(items)] * size
    return zip(*iterators)


l = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
list(grouped(l, 2))
# [(0, 1), (2, 3), (4, 5), (6, 7), (8, 9)]

然后,可以使用规则的map(),例如:

def avg(values):
    return sum(values) / len(values)


list(map(sum, grouped(l, 2)))
# [1, 5, 9, 13, 17]

list(map(avg, grouped(l, 2)))
# [0.5, 2.5, 4.5, 6.5, 8.5]

对于NumPy阵列,仍然可以使用上面的机器,但是根据要使用的实际功能,可能有更快的替代方法。你知道吗

如果函数是矢量化的或可以接受np.ndarray输入,则可以使用:

def grouped_map(func, arr, size):
    return func(*(arr[i::size] for i in range(size))


arr = np.array(l)
grouped_map(lambda *x: avg(x), arr, 2)
# [0.5 2.5 4.5 6.5 8.5]

或者,如果函数支持axis参数:

def grouped_map_axis(func, arr, size):
    return func(arr.reshape(-1, size), axis=1)


# `np.mean()` is the equivalent of `avg()`
grouped_map_axis(np.mean, arr, 2)
# [0.5 2.5 4.5 6.5 8.5]

就时间而言,无论size,第一种方法仅对极少数项目具有竞争力。 对于较大的输入,基于NumPy的方法要快得多,如果size较小,则grouped_map()最快,而对于较大的size,则grouped_map_axis()更快。你知道吗

例如,对于size == 5

benchmark2

而对于size == 100

benchmark100


上面的图形是使用here的代码生成的,使用:

SIZE = 2  # or: SIZE = 100


def gen_input(n):
    return np.random.random(n * SIZE)


def equal_output(a, b):
    return np.all(np.isclose(a, b))


def my_grouped(arr):
    return np.array(list(map(avg, grouped(arr, SIZE))))


def my_grouped_map(arr):
    return grouped_map(lambda *x: avg(x), arr, SIZE)


def my_grouped_map_axis(arr):
    return grouped_map_axis(np.mean, arr, SIZE)


input_sizes = (5, 10, 50, 100, 500, 1000, 5000, 10000, 50000, 100000)  
funcs = my_grouped, my_grouped_map, my_grouped_map_axis


runtimes, input_sizes, labels, results = benchmark(
    funcs, gen_input=gen_input, equal_output=equal_output,
    input_sizes=input_sizes)


plot_benchmarks(runtimes, input_sizes, labels)

显然,当项目数不是size的倍数时,人们可能希望得到一个更健壮的解决方案,例如:

def grouped(items, size, truncate=False, fill=None):
    iterators = [iter(items)] * size
    if truncate:
        return zip(*iterators)
    else:
        return itertools.zip_longest(*iterators, fillvalue=fill)


def align_to_size(arr, size, truncate=False, fill=0):
    if len(arr) % size != 0:
        if truncate:
            arr = arr[:len(arr) // size * size]
        else:
            fill_arr = np.full(size - len(arr) % size, fill, dtype=arr.dtype)
            arr = np.concatenate([arr, fill_arr])
    return arr


def grouped_map(func, arr, size, truncate=False, fill=0):
    arr = align_to_size(arr, size, truncate, fill)
    return func(*(arr[i::size] for i in range(size)))


def grouped_map_axis(func, arr, size, truncate=False, fill=0):
    arr = align_to_size(arr, size, truncate, fill)
    return func(arr.reshape(-1, size), axis=1)

时间方面,对于size = 100并确保生成不对齐的输入:

def gen_input(n):
    return np.random.random(n * SIZE + 1)

benchmark3


对于NumPy解决方案,可以探索支持axis参数,但这超出了这个问题/答案的范围。你知道吗

相关问题 更多 >