【问题标题】:Efficiently compute n-body gravitation in python在 python 中有效地计算 n 体引力
【发布时间】:2019-03-04 20:26:45
【问题描述】:

我正在尝试计算 3 空间中 n 体问题的重力加速度(我使用的是辛欧拉)。

我有每个时间步长的位置和速度向量,并且正在使用以下(工作)代码来计算加速度并更新速度和位置。请注意,加速度是 3 空间中的向量,而不仅仅是幅度。

我想知道是否有更有效的方法来使用 numpy 计算它以避免循环。

def accelerations(positions, masses):
    '''Params:
    - positions: numpy array of size (n,3)
    - masses: numpy array of size (n,)
    Returns:
    - accelerations: numpy of size (n,3), the acceleration vectors in 3-space
    '''
    n_bodies = len(masses)
    accelerations = numpy.zeros([n_bodies,3]) # n_bodies * (x,y,z)

    # vectors from mass(i) to mass(j)
    D = numpy.zeros([n_bodies,n_bodies,3]) # n_bodies * n_bodies * (x,y,z)
    for i, j in itertools.product(range(n_bodies), range(n_bodies)):
        D[i][j] = positions[j]-positions[i]

    # Acceleration due to gravitational force between each pair of bodies
    A = numpy.zeros((n_bodies, n_bodies,3))
    for i, j in itertools.product(range(n_bodies), range(n_bodies)):
        if numpy.linalg.norm(D[i][j]) > epsilon:
            A[i][j] = gravitational_constant * masses[j] * D[i][j] \
            / numpy.linalg.norm(D[i][j])**3

    # Calculate net acceleration of each body (vectors in 3-space)
    accelerations = numpy.sum(A, axis=1) # sum of accel vectors for each body of shape (n_bodies,3)

    return accelerations

【问题讨论】:

  • 我已将此标记为重复,但要回答您的问题:stackoverflow.com/questions/20277982/…,(最终)docs.scipy.org/doc/scipy/reference/generated/…
  • 至于加速部分,您只需调用np.where 并结合一个“质量矩阵”即可:masses.reshape((1, -1))*masses.reshape((-1, 1))
  • 我认为 pdist 只是给出了向量的大小。由于我想在 3 空间中添加力,因此我还需要保留方向。
  • 最佳算法取决于 epsilon 和点质量的数量。对于至少中等规模的问题,我建议使用 kd-tree docs.scipy.org/doc/scipy/reference/generated/… 来查找彼此在给定范围内的点质量,并仅计算它们上的重力。这比计算无用的距离要快得多。对于其余部分,我会推荐一种简单的 Numba 方法。您能否提供合理的测试数据(位置、质量、epsilon),因为这会对运行时间产生巨大影响?

标签: python numpy simulation physics


【解决方案1】:

这是使用blas 的优化版本。 blas 具有用于对称或 Hermitian 矩阵上的线性代数的特殊例程。这些使用专门的打包存储,只保留上三角或下三角,而忽略(冗余)镜像条目。这样 blas 不仅节省了 ~ 一半的存储空间,而且还节省了 ~ 一半的触发器。

我已经放置了很多 cmets 以使其可读。

import numpy as np
import itertools
from scipy.linalg.blas import zhpr, dspr2, zhpmv

def acc_vect(pos, mas):
    n = mas.size
    d2 = pos@(-2*pos.T)
    diag = -0.5 * np.einsum('ii->i', d2)
    d2 += diag + diag[:, None]
    np.einsum('ii->i', d2)[...] = 1
    return np.nansum((pos[:, None, :] - pos) * (mas[:, None] * d2**-1.5)[..., None], axis=0)

def acc_blas(pos, mas):
    n = mas.size
    # trick: use complex Hermitian to get the packed anti-symmetric
    # outer difference in the imaginary part of the zhpr answer
    # don't want to sum over dimensions yet, therefore must do them one-by-one
    trck = np.zeros((3, n * (n + 1) // 2), complex)
    for a, p in zip(trck, pos.T - 1j):
        zhpr(n, -2, p, a, 1, 0, 0, 1)
        # does  a  ->  a + alpha x x^H
        # parameters: n             --  matrix dimension
        #             alpha         --  real scalar
        #             x             --  complex vector
        #             ap            --  packed Hermitian n x n matrix a
        #                               i.e. an n(n+1)/2 vector
        #             incx          --  x stride
        #             offx          --  x offset
        #             lower         --  is storage of ap lower or upper
        #             overwrite_ap  --  whether to change a inplace
    # as a by-product we get pos pos^T:
    ppT = trck.real.sum(0) + 6
    # now compute matrix of squared distances ...
    # ... using (A-B)^2 = A^2 + B^2 - 2AB
    # ... that and the outer sum X (+) X.T equals X ones^T + ones X^T
    dspr2(n, -0.5, ppT[np.r_[0, 2:n+1].cumsum()], np.ones((n,)), ppT,
          1, 0, 1, 0, 0, 1)
    # does  a  ->  a + alpha x y^T + alpha y x^T    in packed symmetric storage
    # scale anti-symmetric differences by distance^-3
    np.divide(trck.imag, ppT*np.sqrt(ppT), where=ppT.astype(bool),
              out=trck.imag)
    # it remains to scale by mass and sum
    # this can be done by matrix multiplication with the vector of masses ...
    # ... unfortunately because we need anti-symmetry we need to work
    # with Hermitian storage, i.e. complex numbers, even though the actual
    # computation is only real:
    out = np.zeros((3, n), complex)
    for a, o in zip(trck, out):
        zhpmv(n, 0.5, a, mas*-1j, 1, 0, 0, o, 1, 0, 0, 1)
        # multiplies packed Hermitian matrix by vector
    return out.real.T

def accelerations(positions, masses, epsilon=1e-6, gravitational_constant=1.0):
    '''Params:
    - positions: numpy array of size (n,3)
    - masses: numpy array of size (n,)
    '''
    n_bodies = len(masses)
    accelerations = np.zeros([n_bodies,3]) # n_bodies * (x,y,z)

    # vectors from mass(i) to mass(j)
    D = np.zeros([n_bodies,n_bodies,3]) # n_bodies * n_bodies * (x,y,z)
    for i, j in itertools.product(range(n_bodies), range(n_bodies)):
        D[i][j] = positions[j]-positions[i]

    # Acceleration due to gravitational force between each pair of bodies
    A = np.zeros((n_bodies, n_bodies,3))
    for i, j in itertools.product(range(n_bodies), range(n_bodies)):
        if np.linalg.norm(D[i][j]) > epsilon:
            A[i][j] = gravitational_constant * masses[j] * D[i][j] \
            / np.linalg.norm(D[i][j])**3

    # Calculate net accleration of each body
    accelerations = np.sum(A, axis=1) # sum of accel vectors for each body

    return accelerations

from numpy.linalg import norm

def acc_pm(positions, masses, G=1):
    '''Params:
    - positions: numpy array of size (n,3)
    - masses: numpy array of size (n,)
    '''
    mass_matrix = masses.reshape((1, -1, 1))*masses.reshape((-1, 1, 1))
    disps = positions.reshape((1, -1, 3)) - positions.reshape((-1, 1, 3)) # displacements
    dists = norm(disps, axis=2)
    dists[dists == 0] = 1 # Avoid divide by zero warnings
    forces = G*disps*mass_matrix/np.expand_dims(dists, 2)**3
    return forces.sum(axis=1)/masses.reshape(-1, 1)

n = 500
pos = np.random.random((n, 3))
mas = np.random.random((n,))

from timeit import timeit

print(f"loops:      {timeit('accelerations(pos, mas)', globals=globals(), number=1)*1000:10.3f} ms")
print(f"pmende:     {timeit('acc_pm(pos, mas)', globals=globals(), number=10)*100:10.3f} ms")
print(f"vectorized: {timeit('acc_vect(pos, mas)', globals=globals(), number=10)*100:10.3f} ms")
print(f"blas:       {timeit('acc_blas(pos, mas)', globals=globals(), number=10)*100:10.3f} ms")

A = accelerations(pos, mas)
AV = acc_vect(pos, mas)
AB = acc_blas(pos, mas)
AP = acc_pm(pos, mas)

assert np.allclose(A, AV) and np.allclose(AB, AV) and np.allclose(AP, AV)

示例运行;与 OP 相比,我的纯 numpy 矢量化和@P Mende 的。

loops:        3213.130 ms
pmende:         41.480 ms
vectorized:     43.860 ms
blas:            7.726 ms

我们可以看到

1) P Mende 在矢量化方面比 I 稍好

2) blas 的速度是 ~5 倍;请注意,我的 blas 不是很好;我怀疑使用优化的 blas 可能会变得更好(不过,预计 numpy 在更好的 blas 上也会运行得更快)

3) 任何答案都比循环快得多

【讨论】:

  • 这很酷。我真的不明白你在这里做的数学,或者为什么它这么快。你有什么资源可以推荐来让这方面变得更聪明吗?
  • @PaulDickinson 我已经添加了一些一般性解释来解决为什么 blas 在这里要快得多的答案。有关详细信息,我不知道有任何魔法资源。数学只是一点线性代数。对于 blas 例程,必须查看 netlib.org/blasdocs.scipy.org/doc/scipy/reference/linalg.blas.html
  • 我更多地考虑复杂的厄米特人的情况。我听懂了,但是这个应用我不熟悉!
  • @PaulDickinson 啊,好吧,这只是一个恶作剧:说x 是x 坐标的向量。然后我们需要 1 外部差异 x-x[None, :],理想情况下在打包存储中,以及 2 外部产品 x@x[None, :](如果可能也打包),因为这些项出现在欧几里得距离中。现在,通过将虚数单位 1j 添加到 x,然后取 Hermitian 外积,我们得到一个 Hermititan 矩阵,它具有 i,k-entry (x_i + 1j)(x_k - 1j) = (x_i x_k + 1) + 1j(x_k - x_i),所以虚部给出了外差,实部给出了我们将外部产品直到偏移量。
【解决方案2】:

在您的原始帖子中对我的 cmets 进行跟进:

from numpy.linalg import norm

def accelerations(positions, masses):
    '''Params:
    - positions: numpy array of size (n,3)
    - masses: numpy array of size (n,)
    '''
    mass_matrix = masses.reshape((1, -1, 1))*masses.reshape((-1, 1, 1))
    disps = positions.reshape((1, -1, 3)) - positions.reshape((-1, 1, 3)) # displacements
    dists = norm(disps, axis=2)
    dists[dists == 0] = 1 # Avoid divide by zero warnings
    forces = G*disps*mass_matrix/np.expand_dims(dists, 2)**3
    return forces.sum(axis=1)/masses.reshape(-1, 1)

【讨论】:

  • 谢谢,pdist 和 np.where 都很值得了解!但这实际上并不能完全解决问题,因为输出应该是 3 空间中的向量,而不仅仅是它们的大小。这也是为什么在原始代码中有一个立方体而不是平方因子的原因。
  • @PaulDickinson 啊,对。我很抱歉。我已将答案更新为正确的形状。
  • (其实广播还是有问题的,我一会儿再更新。)
  • @PaulDickinson 现在应该可以了。大多数情况下,这是一种巧妙的重塑/广播练习。
  • 很好,这比原来的解决方案优雅多了!
【解决方案3】:

需要考虑的一些事项:

你只需要一半的距离;一旦你计算出D[i][j],就等于-D[j][i]

你可以df2 = df.apply(lambda x:gravitational_constant/x**3)

您可以生成一个数据框,记录每对物体的质量乘积。您只需执行一次,然后每次调用时都可以将其传递给accelearations

然后df.product(df2).product(mass_products).sum().div(masses) 给你加速。

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 2012-08-20
    • 2020-09-06
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2023-03-25
    • 2011-12-27
    相关资源
    最近更新 更多