heterodyne.core.kernels

High-Performance Computational Kernels for Heterodyne Scattering Analysis

This module provides Numba-accelerated computational kernels implementing the two-component heterodyne scattering model from He et al. PNAS 2024 (https://doi.org/10.1073/pnas.2401162121, Equations S-95 to S-98).

The kernels compute correlation functions, transport integrals, and chi-squared objectives for time-dependent nonequilibrium heterodyne scattering systems.

Created for: Rheo-SAXS-XPCS Heterodyne Analysis Authors: Wei Chen, Hongrui He Institution: Argonne National Laboratory

heterodyne.core.kernels.jit(*args, **kwargs)

Parameters:

• args (Any)

• kwargs (Any)

Return type:

Any

heterodyne.core.kernels.njit(*args, **kwargs)

Parameters:

• args (Any)

• kwargs (Any)

Return type:

Any

class heterodyne.core.kernels.DummyType

Bases: object

heterodyne.core.kernels.memory_efficient_cache(maxsize=128)

Memory-efficient LRU cache with automatic cleanup.

Features: - Least Recently Used eviction - Access frequency tracking - Configurable size limits - Cache statistics

Parameters:

maxsize (int) – Maximum cached items (0 disables caching)

Returns:

Function decorator with cache_info() and cache_clear() methods

Return type:

decorator

heterodyne.core.kernels.solve_least_squares_batch_numba(theory_batch, exp_batch)

Fallback implementation when Numba is not available.

heterodyne.core.kernels.compute_chi_squared_batch_numba(theory_batch, exp_batch, contrast_batch, offset_batch)

Fallback implementation when Numba is not available.

heterodyne.core.kernels.create_time_integral_matrix_numba(time_dependent_array)

Create time integral matrix for correlation calculations.

OPTIMIZED VERSION: Revolutionary vectorization using NumPy broadcasting Expected speedup: 5-10x through elimination of nested loops

Mathematical operation: matrix[i, j] = |cumsum[i] - cumsum[j]|

Vectorization strategy: 1. Compute cumulative sum once 2. Use broadcasting to create difference matrix in single operation 3. Apply absolute value vectorized operation 4. Exploit cache-friendly memory access patterns

heterodyne.core.kernels.refresh_kernel_functions()

Refresh kernel function assignments based on current Numba availability.

This function is useful in test environments where Numba availability may change dynamically during execution.

Returns:

True if Numba kernels are now available, False if using fallback functions

Return type:

bool

heterodyne.core.kernels.compute_sinc_squared_numba_flexible(x, prefactor=None)

Flexible sinc² function that handles both single values and matrices.

Parameters:

• x (float or np.ndarray) – Single value or matrix input

• prefactor (float, optional) – For matrix version (unused for single values)

Returns:

sinc²(x) result

Return type:

float or np.ndarray

heterodyne.core.kernels.compute_g1_correlation_vectorized(D_integral, wavevector_q_squared_half_dt)

Vectorized g1 correlation calculation from pre-computed transport coefficient integral.

This is the optimized 2-parameter version for performance-critical calculations where the transport coefficient integral ∫J(t)dt has been pre-computed.

Parameters:

• D_integral (array_like) – Pre-computed transport coefficient integral: ∫J(t’)dt’ from t1 to t2

• wavevector_q_squared_half_dt (float) – Pre-computed factor: q²/2 * Δt

Returns:

g1 correlation: exp(-q²/2 * Δt * D_integral)

Return type:

array_like

Notes

This is the primary g1 correlation function. The transport coefficient integral should be pre-computed via _create_time_integral_matrix_impl.

heterodyne.core.kernels.compute_g1_correlation_numba(D_integral, wavevector_q_squared_half_dt)

Vectorized g1 correlation calculation from pre-computed transport coefficient integral.

This is the optimized 2-parameter version for performance-critical calculations where the transport coefficient integral ∫J(t)dt has been pre-computed.

Parameters:

• D_integral (array_like) – Pre-computed transport coefficient integral: ∫J(t’)dt’ from t1 to t2

• wavevector_q_squared_half_dt (float) – Pre-computed factor: q²/2 * Δt

Returns:

g1 correlation: exp(-q²/2 * Δt * D_integral)

Return type:

array_like

Notes

This is the primary g1 correlation function. The transport coefficient integral should be pre-computed via _create_time_integral_matrix_impl.

heterodyne.core.kernels.compute_sinc_squared_numba(x, prefactor=None)

Flexible sinc² function that handles both single values and matrices.

Parameters:

• x (float or np.ndarray) – Single value or matrix input

• prefactor (float, optional) – For matrix version (unused for single values)

Returns:

sinc²(x) result

Return type:

float or np.ndarray

heterodyne.core.kernels.calculate_diffusion_coefficient_numba(t, D0, alpha, D_offset)

Calculate time-dependent transport coefficient.

Note: Function name retained for API compatibility, but calculates transport coefficient J(t), not traditional diffusion coefficient D.

Parameters:

• t (float or array) – Time (can be scalar or array)

• D0 (float) – Reference transport coefficient J₀ (labeled ‘D0’ for compatibility)

• alpha (float) – Transport coefficient time-scaling exponent

• D_offset (float) – Baseline transport coefficient J_offset

Returns:

J(t) = J₀ * t^alpha + J_offset (labeled as D(t) for compatibility)

Return type:

float or array

Notes

For negative alpha, uses physical limit approach to avoid NaN at t=0: - At t=0: D(0) = D_offset (physical limit) - For t > threshold: D(t) = D0 * t^alpha + D_offset

heterodyne.core.kernels.calculate_shear_rate_numba(t, gamma0, beta, gamma_offset)

Calculate time-dependent shear rate.

Parameters:

• t (float or array) – Time (can be scalar or array)

• gamma0 (float) – Reference shear rate

• beta (float) – Time-dependence exponent

• gamma_offset (float) – Baseline shear rate

Returns:

gamma_dot(t) = gamma0 * t^beta + gamma_offset

Return type:

float or array

Notes

For negative beta, uses physical limit approach to avoid NaN at t=0: - At t=0: gamma(0) = gamma_offset (physical limit) - For t > threshold: gamma(t) = gamma0 * t^beta + gamma_offset

heterodyne.core.kernels.get_kernel_info()

Get information about the current kernel configuration.

Returns:

Information about kernel availability and configuration

Return type:

dict[str, Any]