Performance utilities

Optional CPU acceleration (Numba, SciPy KD-tree) and parallel experiment helpers. Install with:

pip install 'ambr[perf]'

Flags

ambr.performance.HAS_NUMBA

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

ambr.performance.HAS_SCIPY

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

Scatter helpers (vectorized write path)

ambr.performance.apply_scatter_add(base: ndarray, positions: ndarray, delta: ndarray) → ndarray

Scatter-add with Numba acceleration when available.

Falls back to np.add.at for object dtypes or when Numba is missing. Always returns the array holding the result (may be a new buffer if a dtype upcast or contiguity copy was required). Callers must use the return value:

out = apply_scatter_add(column_copy, positions, delta)

ambr.performance.apply_scatter_write(base: ndarray, positions: ndarray, values: ndarray) → ndarray

Scatter-write (last write wins) with Numba when available.

Falls back to advanced indexing when Numba is missing or dtypes are object. Returns the array holding the result (use the return value).

ambr.performance.scatter_add_1d(base: ndarray, positions: ndarray, delta: ndarray) → ndarray

Accumulate delta into base at positions (duplicate-safe).

Same semantics as np.add.at(base, positions, delta) but often faster for irregular ABM scatter patterns on CPU (including Apple Silicon). Mutates and returns base. Prefer apply_scatter_add() at call sites.

ambr.performance.scatter_write_1d(base: ndarray, positions: ndarray, values: ndarray) → ndarray

Write values into base at positions (last write wins).

Mutates and returns base. Prefer apply_scatter_write() at call sites.

Spatial & parallel

class ambr.performance.SpatialIndex

Bases: object

Fast spatial indexing using KD-Tree for O(log n) neighbor queries.

Usage:

index = SpatialIndex() index.build(positions) # positions is Nx2 or Nx3 array neighbors = index.query_radius(point, radius) k_nearest = index.query_knn(point, k=5)

batch_query_radius(points: ndarray, radius: float) → List[List[int]]

Find neighbors for multiple query points.

Parameters:

• points – MxD array of query points

• radius – Search radius

Returns:

List of neighbor lists for each query point

build(positions: ndarray) → SpatialIndex

Build the spatial index from positions.

Parameters:

positions – Nx2 or NxD array of coordinates

Returns:

self for chaining

query_knn(point: ndarray, k: int = 5) → Tuple[ndarray, ndarray]

Find k nearest neighbors to query point.

Parameters:

• point – Query point coordinates

• k – Number of neighbors to find

Returns:

Tuple of (distances, indices)

query_pairs(radius: float) → set

Find all pairs of points within radius of each other.

Parameters:

radius – Maximum distance between pairs

Returns:

Set of (i, j) index pairs

query_radius(point: ndarray, radius: float) → List[int]

Find all points within radius of query point.

Parameters:

• point – Query point coordinates

• radius – Search radius

Returns:

List of indices of points within radius

class ambr.performance.ParallelRunner(model_class: Type, n_workers: int = None)

Bases: object

Run multiple simulations in parallel across CPU cores.

Usage:

runner = ParallelRunner(MyModel, n_workers=8) results = runner.run(param_list)

run(param_list: List[Dict[str, Any]], show_progress: bool = True) → List[Dict[str, Any]]

Run simulations in parallel.

Parameters:

• param_list – List of parameter dictionaries

• show_progress – Whether to show progress

Returns:

List of result dictionaries

run_with_seeds(base_params: Dict[str, Any], seeds: List[int], show_progress: bool = True) → List[Dict[str, Any]]

Run same parameters with different random seeds.

Parameters:

• base_params – Base parameter dictionary

• seeds – List of random seeds

• show_progress – Whether to show progress

Returns:

List of result dictionaries

ambr.performance.check_performance_deps() → Dict[str, bool]

Check which performance dependencies are available.