Source code for pyfracval.pca_kernels
"""PCA-specific JIT kernels for PyFracVAL.
JIT-compiled batch functions for PCA position calculation and overlap checking.
These kernels support the PCA stage in the FracVAL-style aggregation pipeline
:cite:p:`Moran2019FracVAL`.
Functions
---------
batch_calculate_positions_pca
JIT batch calculation of particle positions during PCA.
batch_check_overlaps_pca
JIT parallel overlap checker for PCA batch operations.
"""
import logging
import numpy as np
from numba import jit, prange
logger = logging.getLogger(__name__)
@jit(parallel=True, fastmath=True, cache=True, nopython=True)
[docs]
def batch_calculate_positions_pca(
vec_0: np.ndarray,
i_vec: np.ndarray,
j_vec: np.ndarray,
angles: np.ndarray,
) -> np.ndarray:
"""Calculate batch of positions on intersection circle for PCA.
Uses Numba parallel loops to compute multiple rotation positions simultaneously.
Parameters
----------
vec_0 : np.ndarray
[x0, y0, z0, r0] - center and radius of intersection circle
i_vec : np.ndarray
First basis vector (3D)
j_vec : np.ndarray
Second basis vector (3D)
angles : np.ndarray
Array of rotation angles (1D)
Returns
-------
np.ndarray
(N, 3) array of positions, one per angle
"""
n_angles = angles.shape[0]
positions = np.empty((n_angles, 3), dtype=np.float64)
x0, y0, z0, r0 = vec_0
# Parallel loop over angles
for i in prange(n_angles):
theta = angles[i]
cos_theta = np.cos(theta)
sin_theta = np.sin(theta)
# Calculate position: center + r0 * (cos(theta)*i_vec + sin(theta)*j_vec)
positions[i, 0] = x0 + r0 * (cos_theta * i_vec[0] + sin_theta * j_vec[0])
positions[i, 1] = y0 + r0 * (cos_theta * i_vec[1] + sin_theta * j_vec[1])
positions[i, 2] = z0 + r0 * (cos_theta * i_vec[2] + sin_theta * j_vec[2])
return positions
@jit(parallel=True, fastmath=True, cache=True, nopython=True)
[docs]
def batch_check_overlaps_pca(
coords_agg: np.ndarray,
radii_agg: np.ndarray,
candidate_positions: np.ndarray,
radius_new: float,
tolerance: float,
) -> np.ndarray:
"""Check overlap for batch of candidate positions (PCA).
Uses Numba parallel loops to evaluate multiple positions simultaneously.
Parameters
----------
coords_agg : np.ndarray
Current aggregate coordinates (n_agg, 3)
radii_agg : np.ndarray
Current aggregate radii (n_agg,)
candidate_positions : np.ndarray
Batch of candidate positions to test (n_candidates, 3)
radius_new : float
Radius of new particle
tolerance : float
Overlap tolerance
Returns
-------
np.ndarray
(n_candidates,) array of max overlap values for each position
"""
n_candidates = candidate_positions.shape[0]
n_agg = coords_agg.shape[0]
overlaps = np.empty(n_candidates, dtype=np.float64)
# Parallel loop over candidate positions
for idx in prange(n_candidates):
coord_new = candidate_positions[idx]
max_overlap = -np.inf
# For each candidate, check against all aggregate particles
for j in range(n_agg):
coord_agg = coords_agg[j]
radius_agg = radii_agg[j]
radius_sum = radius_new + radius_agg
# Compute squared distance
d_sq = 0.0
for dim in range(3):
diff = coord_new[dim] - coord_agg[dim]
d_sq += diff * diff
# Bounding sphere pre-check
radius_sum_sq = radius_sum * radius_sum
if d_sq > radius_sum_sq:
continue
# Compute overlap
dist = np.sqrt(d_sq)
overlap = 1.0 - dist / radius_sum
if overlap > max_overlap:
max_overlap = overlap
# Early termination check (can't break in prange, but helps inner loop)
if overlap > tolerance:
max_overlap = overlap
break
overlaps[idx] = max_overlap
return overlaps