本文共 4463 字，大约阅读时间需要 14 分钟。

numba用户手册

1.19。例子

1.19.1。曼德尔布罗

#! /usr/bin/env python# -*- coding: utf-8 -*-from __future__ import print_function, division, absolute_importfrom timeit import default_timer as timerfrom matplotlib.pylab import imshow, jet, show, ionimport numpy as npfrom numba import jit@jitdef mandel(x, y, max_iters):    """    Given the real and imaginary parts of a complex number,    determine if it is a candidate for membership in the Mandelbrot    set given a fixed number of iterations.    """    i = 0    c = complex(x,y)    z = 0.0j    for i in range(max_iters):        z = z*z + c        if (z.real*z.real + z.imag*z.imag) >= 4:            return i    return 255@jitdef create_fractal(min_x, max_x, min_y, max_y, image, iters):    height = image.shape[0]    width = image.shape[1]    pixel_size_x = (max_x - min_x) / width    pixel_size_y = (max_y - min_y) / height    for x in range(width):        real = min_x + x * pixel_size_x        for y in range(height):            imag = min_y + y * pixel_size_y            color = mandel(real, imag, iters)            image[y, x] = color    return imageimage = np.zeros((500 * 2, 750 * 2), dtype=np.uint8)s = timer()create_fractal(-2.0, 1.0, -1.0, 1.0, image, 20)e = timer()print(e - s)imshow(image)#jet()#ion()show()

1.19.2。移动平均

#!/usr/bin/env python"""A moving average function using @guvectorize."""import numpy as npfrom numba import guvectorize@guvectorize(['void(float64[:], intp[:], float64[:])'], '(n),()->(n)')def move_mean(a, window_arr, out):    window_width = window_arr[0]    asum = 0.0    count = 0    for i in range(window_width):        asum += a[i]        count += 1        out[i] = asum / count    for i in range(window_width, len(a)):        asum += a[i] - a[i - window_width]        out[i] = asum / countarr = np.arange(20, dtype=np.float64).reshape(2, 10)print(arr)print(move_mean(arr, 3))

1.19.3。多线程

下面的代码展示了使用功能时潜在的性能改进。例如，在四核计算机上，我得到以下结果打印出来：

numpy (1 thread)       145 msnumba (1 thread)       128 msnumba (4 threads)       35 ms

注意

在Python 3中，您可以使用标准的模块，而不是手动生成线程并分派任务。

#!/usr/bin/env pythonfrom __future__ import print_function, division, absolute_importimport mathimport threadingfrom timeit import repeatimport numpy as npfrom numba import jitnthreads = 4size = 10**6def func_np(a, b):    """    Control function using Numpy.    """    return np.exp(2.1 * a + 3.2 * b)@jit('void(double[:], double[:], double[:])', nopython=True, nogil=True)def inner_func_nb(result, a, b):    """    Function under test.    """    for i in range(len(result)):        result[i] = math.exp(2.1 * a[i] + 3.2 * b[i])def timefunc(correct, s, func, *args, **kwargs):    """    Benchmark *func* and print out its runtime.    """    print(s.ljust(20), end=" ")    # Make sure the function is compiled before we start the benchmark    res = func(*args, **kwargs)    if correct is not None:        assert np.allclose(res, correct), (res, correct)    # time it    print('{:>5.0f} ms'.format(min(repeat(lambda: func(*args, **kwargs),                                          number=5, repeat=2)) * 1000))    return resdef make_singlethread(inner_func):    """    Run the given function inside a single thread.    """    def func(*args):        length = len(args[0])        result = np.empty(length, dtype=np.float64)        inner_func(result, *args)        return result    return funcdef make_multithread(inner_func, numthreads):    """    Run the given function inside *numthreads* threads, splitting its    arguments into equal-sized chunks.    """    def func_mt(*args):        length = len(args[0])        result = np.empty(length, dtype=np.float64)        args = (result,) + args        chunklen = (length + numthreads - 1) // numthreads        # Create argument tuples for each input chunk        chunks = [[arg[i * chunklen:(i + 1) * chunklen] for arg in args]                  for i in range(numthreads)]        # Spawn one thread per chunk        threads = [threading.Thread(target=inner_func, args=chunk)                   for chunk in chunks]        for thread in threads:            thread.start()        for thread in threads:            thread.join()        return result    return func_mtfunc_nb = make_singlethread(inner_func_nb)func_nb_mt = make_multithread(inner_func_nb, nthreads)a = np.random.rand(size)b = np.random.rand(size)correct = timefunc(None, "numpy (1 thread)", func_np, a, b)timefunc(correct, "numba (1 thread)", func_nb, a, b)timefunc(correct, "numba (%d threads)" % nthreads, func_nb_mt, a,

转载地址：https://chunyou.blog.csdn.net/article/details/103963119 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！