laser.core package

Subpackages

Submodules

laser.core.cli module

Module that contains the command line app.

Why does this file exist, and why not put this in __main__?

You might be tempted to import things from __main__ later, but that will cause problems: the code will get executed twice:

  • When you run python -midmlaser python will execute __main__.py as a script. That means there will not be any idmlaser.__main__ in sys.modules.

  • When you import __main__ it will get executed again (as a module) because there”s no idmlaser.__main__ in sys.modules.

Also see (1) from https://click.palletsprojects.com/en/stable/setuptools/

laser.core.distributions module

Various probability distributions implemented using NumPy and Numba.

LASER based models generally move from pure NumPy to Numba-accelerated version of the core dynamics, e.g., transmission.

It would be a hassle to re-implement these functions for each desired distribution, so we provide these Numba-wrapped distributions here which can be passed in to other Numba compiled functions.

For example, a simple SIR model may want to parameterize the infectious period using a distribution. By passing in a Numba-wrapped distribution function, we can pick and parameterize a distribution based on configuration and sample from that distribution within the Numba-compiled SIR model without needing to re-implement the distribution logic within the SIR model itself.

A simple example of usage:

import laser.core.distributions as dist

# Create a Numba-wrapped beta distribution
beta_dist = dist.beta(2.0, 5.0)

# Assign the distribution to the model's infectious period distribution
# so the transmission component can sample from it during simulation
model.infectious_period_dist = beta_dist

Note that the distribution functions take two parameters, tick and node, which are currently unused but match the desired signature for disease model components that may need to sample from distributions based on the current simulation tick or node index. In other words, distributions with spatial or temporal variation could be implemented in the future.

Here are examples of Numba-wrapped distribution functions that could vary based on tick or tick + node:

# temporal variation only
cosine = np.cos(np.linspace(0, np.pi * 2, 365))

@nb.njit(nogil=True, cache=True)
def seasonal_distribution(tick: int, node: int) -> np.float32:
    # ignore node for this seasonal distribution
    day_of_year = tick % 365
    base_value = 42.0 + 3.14159 * cosine[day_of_year]
    # parameterize normal() with seasonal factor
    return np.float32(np.random.normal(base_value, 2.0))

# additional spatial variation
ramp = np.linspace(0, 2, 42)

@nb.njit(nogil=True, cache=True)
def ramped_distribution(tick: int, node: int) -> np.float32:
    day_of_year = tick % 365
    # use seasonal factor
    base_value = 42.0 + 3.14159 * cosine[day_of_year]
    # apply spatial ramp based on node index
    base_value *= ramp[node]
    # parameterize normal() with seasonal + spatial factor
    return np.float32(np.random.normal(base_value, 1.0))

Normally, these distributions—built in or custom—will be used once per agent as above. However, the sample_ints() and sample_floats() functions can be used to efficiently sample large arrays using multiple CPU cores in parallel.

laser.core.distributions.beta(a, b)[source]

Beta distribution.

$f(x; a, b) = frac {x^{a-1} (1-x)^{b-1}} {B(a, b)}$

where $B(a, b)$ is the beta function.

laser.core.distributions.binomial(n, p)[source]

Binomial distribution.

$f(k,n,p) = Pr(X = k) = binom {n} {k} p^k (1-p)^{n-k}$

where $n$ is the number of trials and $p$ is the probability of success [0, 1].

laser.core.distributions.constant_float(value)[source]

Constant distribution. Always returns the same floating point value.

laser.core.distributions.constant_int(value)[source]

Constant distribution. Always returns the same integer value.

laser.core.distributions.exponential(scale)[source]

Exponential distribution.

$f(x; frac {1} {beta}) = frac {1} {beta} e^{-frac {x} {beta}}$

where $beta$ is the scale parameter ($beta = 1 / lambda$).

laser.core.distributions.gamma(shape, scale)[source]

Gamma distribution.

$p(x) = x^{k-1} frac {e^{- x / theta}}{theta^k Gamma(k)}$

where $k$ is the shape, $theta$ is the scale, and $Gamma(k)$ is the gamma function.

laser.core.distributions.logistic(loc, scale)[source]

Logistic distribution.

$P(x) = frac {e^{-(x - mu) / s}} {s (1 + e^{-(x - mu) / s})^2}$

where $mu$ is the location parameter and $s$ is the scale parameter.

laser.core.distributions.lognormal(mean, sigma)[source]

Log-normal distribution.

$p(x) = frac {1} {sigma x sqrt {2 pi}} e^{- frac {(ln x - mu)^2} {2 sigma^2}}$

where $mu$ is the mean and $sigma$ is the standard deviation of the underlying normal distribution.

laser.core.distributions.negative_binomial(n, p)[source]

Negative binomial distribution.

$P(N; n, p) = frac {Gamma (N + n)} {N! Gamma (n)} p^n (1 - p)^N$

where $n$ is the number of successes, $p$ is the probability of success on each trial, $N + n$ is the number of trials, and $Gamma()$ is the gamma function. When $n$ is an integer,

$frac {Gamma (N + n)} {N! Gamma (n)} = binom {N + n - 1} {n - 1}$

which is the more common form of this term.

laser.core.distributions.normal(loc, scale)[source]

Normal (Gaussian) distribution.

$p(x) = frac {1} {sqrt {2 pi sigma^2}} e^{- frac {(x - mu)^2} {2 sigma^2}}$

where $mu$ is the mean and $sigma$ is the standard deviation.

laser.core.distributions.poisson(lam)[source]

Poisson distribution.

$f( k ; lambda ) = frac {lambda^k e^{- lambda}} {k!}$

where $lambda$ is the expected number of events in the given interval.

laser.core.distributions.sample_floats(fn, dest, tick=0, node=0)[source]

Fill an array with floating point values sampled from a Numba-wrapped distribution function.

Parameters:

  • fn (function) – Numba-wrapped distribution function returning float32 values.

  • dest (np.ndarray) – Pre-allocated destination float32 array to store samples.

  • tick (int, optional) – Current simulation tick (default is 0). Passed through to the distribution function.

  • node (int, optional) – Current node index (default is 0). Passed through to the distribution function.

Returns:

dest (np.ndarray) – The destination array filled with sampled values.

laser.core.distributions.sample_ints(fn, dest, tick=0, node=0)[source]

Fill an array with integer values sampled from a Numba-wrapped distribution function.

Parameters:

  • fn (function) – Numba-wrapped distribution function returning int32 values.

  • dest (np.ndarray) – Pre-allocated destination int32 array to store samples.

  • tick (int, optional) – Current simulation tick (default is 0). Passed through to the distribution function.

  • node (int, optional) – Current node index (default is 0). Passed through to the distribution function.

Returns:

dest (np.ndarray) – The destination array filled with sampled values.

laser.core.distributions.uniform(low, high)[source]

Uniform distribution.

$p(x) = frac {1} {b - a}$

where $a$ is the lower bound and $b$ is the upper bound, [$a$, $b$).

laser.core.distributions.weibull(a, lam)[source]

Weibull distribution.

$X = lambda (- ln ( U ))^{1 / a}$

where $a$ is the shape parameter and $lambda$ is the scale parameter.

laser.core.extension module

laser.core.extension.compiled(args)[source]

laser.core.laserframe module

laserframe.py

This module defines the LaserFrame class, which is used to manage dynamically allocated data for agents or nodes/patches. The LaserFrame class is similar to a database table or a Pandas DataFrame and supports scalar and vector properties.

Classes:

LaserFrame: A class to manage dynamically allocated data for agents or nodes/patches.

Usage Example:

`python laser_frame = LaserFrame(capacity=100) laser_frame.add_scalar_property('age', dtype=np.int32, default=0) laser_frame.add_vector_property('position', length=3, dtype=np.float32, default=0.0) start, end = laser_frame.add(10) laser_frame.sort(np.arange(10)[::-1]) laser_frame.squash(np.array([True, False, True, False, True, False, True, False, True, False])) `

ivar count:

The current count of active elements.

vartype count:

int

ivar capacity:

The maximum capacity of the frame.

vartype capacity:

int

Note

Since count can be less than capacity, properties return slices of the underlying arrays up to count by default so users do not have to include the slice themselves. I.e., if lf is a LaserFrame, then lf.age returns lf._age[0:lf.count] automatically. The full underlying array is always available as lf._age (or whatever the property name is). The slice returned is valid for all NumPy operations, including assignment, as well as for use with Numba compiled functions.

class laser.core.laserframe.LaserFrame(capacity: int, initial_count: int = -1, **kwargs)[source]

Bases: object

The LaserFrame class, similar to a db table or a Pandas DataFrame, holds dynamically allocated data for agents (generally 1-D or scalar) or for nodes|patches (e.g., 1-D for scalar value per patch or 2-D for time-varying per patch).

add(count: int) → tuple[int, int][source]

Adds the specified count to the current count of the LaserFrame.

This method increments the internal count by the given count, ensuring that the total does not exceed the frame’s capacity. If the addition would exceed the capacity, an assertion error is raised. This method is typically used to add new births during the simulation.

Parameters:

count (int) – The number to add to the current count.

Returns:

tuple[int, int] – A tuple containing the [start index, end index) after the addition.

Raises:

AssertionError – If the resulting count exceeds the frame’s capacity.

add_array_property(name: str, shape: tuple, dtype=<class 'numpy.uint32'>, default=0) → None[source]

Adds an array property to the object.

This method initializes a new property with the given name as a multi-dimensional NumPy array.

The array will have the given shape (note that there is no implied dimension of size self._capacity), datatype (default is np.uint32), and default value (default is 0).

Parameters:

  • name (str) – The name of the property to be added.

  • shape (tuple) – The shape of the array.

  • dtype (data-type, optional) – The desired data-type for the array, default is np.uint32.

  • default (scalar, optional) – The default value to fill the array with, default is 0.

Returns:

None

add_scalar_property(name: str, dtype=<class 'numpy.uint32'>, default=0) → None[source]

Add a scalar property to the class.

This method initializes a new scalar property for the class instance. The property is stored as a 1-D NumPy array (scalar / entry) with a specified data type and default value.

Parameters:

  • name (str) – The name of the scalar property to be added.

  • dtype (data-type, optional) – The desired data type for the property. Default is np.uint32.

  • default (scalar, optional) – The default value for the property. Default is 0.

Returns:

None

add_vector_property(name: str, length: int, dtype=<class 'numpy.uint32'>, default=0) → None[source]

Adds a vector property to the object.

This method initializes a new property with the given name as a 2-D NumPy array (vector per entry).

The array will have a shape of (length, self._capacity) and will be filled with the specified default value. The data type of the array elements is determined by the dtype parameter.

Parameters:

  • name (str) – The name of the property to be added.

  • length (int) – The length of the vector.

  • dtype (data-type, optional) – The desired data-type for the array, default is np.uint32.

  • default (scalar, optional) – The default value to fill the array with, default is 0.

Returns:

None

property capacity: int

Returns the capacity of the laser frame (total possible entries for dynamic properties).

Returns:

int – The capacity of the laser frame.

property count: int

Returns the current count (equivalent to len()).

Returns:

int – The current count value.

describe(target=None) → str[source]

Return a formatted string description of the laserframe object, including its attributes and their values.

Parameters:

target (string) – Optional string for the report header (generally the name of the LaserFrame variable, e.g., “People”. Unlike functions, we can’t get the name of a variable automatically).

Returns:

str –

A formatted string describing the laserframe object, including its capacity, count, and

details of its scalar, vector, and other properties.

classmethod load_snapshot(path, n_ppl, cbr, nt)[source]

Load a LaserFrame and optional extras from an HDF5 snapshot file.

Parameters:

  • path (str) – Path to the HDF5 snapshot file.

  • n_ppl (float or array-like) – Original total population (or per-node array) used to estimate births.

  • cbr (float or array-like) – Crude birth rate (per 1000/year).

  • nt (int) – Simulation duration (number of ticks).

Returns:

frame (LaserFrame) – Loaded LaserFrame object. results_r (np.ndarray or None): Optional 2D numpy array of recovered counts. pars (dict or None): Optional dictionary of parameters.

save_snapshot(path, results_r=None, pars=None)[source]

Save this LaserFrame and optional extras to an HDF5 snapshot file.

Parameters:

  • path (Path) – Destination file path

  • results_r (np.ndarray) – Optional 2D numpy array of recovered counts

  • pars (PropertySet or dict) – Optional PropertySet or dict of parameters

sort(indices, verbose: bool = False) → None[source]

Sorts the elements of the object’s numpy arrays based on the provided indices.

Parameters:

  • indices (np.ndarray) – An array of indices used to sort the numpy arrays. Must be of integer type and have the same length as the frame count (self._count).

  • verbose (bool, optional) – If True, prints the sorting progress for each numpy array attribute. Defaults to False.

Raises:

AssertionError – If indices is not an integer array or if its length does not match the frame count of active elements.

squash(indices, verbose: bool = False) → None[source]

Reduces the active count of the internal numpy arrays keeping only elements True in the provided boolean indices.

Parameters:

  • indices (np.ndarray) – A boolean array indicating which elements to keep. Must have the same length as the current frame active element count.

  • verbose (bool, optional) – If True, prints detailed information about the squashing process. Defaults to False.

Raises:

AssertionError – If indices is not a boolean array or if its length does not match the current frame active element count.

Returns:

None

laser.core.migration module

This module provides various functions to calculate migration networks based on different models, including the gravity model, competing destinations model, Stouffer’s model, and the radiation model.

Additionally, it includes a utility function to calculate the great-circle distance between two points on the Earth’s surface using the Haversine formula.

Functions:
gravity(pops: np.ndarray, distances: np.ndarray, k: float, a: float, b: float, c: float, max_frac: Union[float, None]=None, kwargs) -> np.ndarray:

Compute a gravity-model migration network based on origin and destination populations and pairwise distances.

row_normalizer(network: np.ndarray, max_rowsum: float) -> np.ndarray:

Normalize the rows of a given network matrix such that no row sum exceeds a specified maximum value.

competing_destinations(pops: np.ndarray, distances: np.ndarray, b: float, c: float, delta: float, params) -> np.ndarray:

Compute a migration network using the competing-destinations model, incorporating the influence of alternative destination nodes.

stouffer(pops: np.ndarray, distances: np.ndarray, k: float, a: float, b: float, include_home: bool, params) -> np.ndarray:

Compute a migration network using a modified Stouffer’s model.

radiation(pops: np.ndarray, distances: np.ndarray, k: float, include_home: bool, params) -> np.ndarray:

Compute a migration network using the radiation model, which models flows based on intervening population rather than physical distance.

distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:

Calculate the great-circle distance between two points on the Earth’s surface using the Haversine formula.

laser.core.migration.competing_destinations(pops, distances, k, a, b, c, delta, **params)[source]

Calculate the competing destinations model for a given set of populations and distances. (Fotheringham AS. Spatial flows and spatial patterns. Environment and planning A. 1984;16(4):529-543)

This function computes a network matrix based on the gravity model and then adjusts it using the competing destinations model. The adjustment is done by considering the interference from other destinations.

Mathematical formula:
Element-by-element:

$$ network_{i,j} = k times p_i^a times p_j^b / distance_{i,j}^c times sum_k {(p_k^b / distance_{j,k}^c text {small for k not in [i,j]})^{delta} } $$

As-implemented numpy math:

  • Compute all terms up to the sum_k using the gravity model

  • Construct the matrix inside the sum: p**b * distances**(1-c)

  • Sum on the second axis (k), and subtract off the diagonal (j=k terms):

    row_sums = np.sum(competition_matrix, axis=1) - np.diag(competition_matrix)

  • Now element-by-element, subtract k=i terms off the sum, exponentiate, and multiply the original network term:

    network[i][j] = network[i][j] * (row_sums[i] - competition_matrix[i][j]) ** delta

Parameters:

  • pops (numpy.ndarray) – Array of populations.

  • distances (numpy.ndarray) – Array of distances between locations.

  • k (float) – Scaling constant.

  • a (float) – Exponent for the population of the origin.

  • b (float) – Exponent parameter for populations in the gravity model.

  • c (float) – Exponent parameter for distances in the gravity model.

  • delta (float) – Exponent parameter for the competing destinations adjustment.

  • params (dict) – Additional parameters to be passed to the gravity model.

Returns:

network (numpy.ndarray) – Adjusted network matrix based on the competing destinations model.

laser.core.migration.distance(lat1, lon1, lat2=None, lon2=None)[source]

Calculate the great-circle distance between two points on the Earth’s surface. This function uses the Haversine formula to compute the distance between two points specified by their latitude and longitude in decimal degrees.

  • If lat2 and lon2 are not provided, they default to lat1 and lon1, respectively. This supports the default case of calculating the NxN matrix of distances between all pairs of points in (lat1, ` lon1`).

  • If all arguments are scalars, will return a single scalar distance, (lat1, lon1) to (lat2, lon2).

  • If lat2, lon2 are vectors, will return a vector of distances, (lat1, lon1) to each lat/lon in lat2, lon2.

  • If lat1, lon1 and lat2, lon2 are vectors, will return a matrix with shape (N, M) of distances where N is the length of lat1/lon1 and M is the length of lat2/lon2.

Parameters:

  • lat1 (float) – Latitude of the first point(s) in decimal degrees [-90, 90].

  • lon1 (float) – Longitude of the first point(s) in decimal degrees [-180, 180].

  • lat2 (float) – Latitude of the second point(s) in decimal degrees [-90, 90].

  • lon2 (float) – Longitude of the second point(s) in decimal degrees [-180, 180].

Returns:

distance (float) – The distance between the points in kilometers.

laser.core.migration.gravity(pops: ndarray, distances: ndarray, k: float, a: float, b: float, c: float, **kwargs)[source]

Calculate a gravity model network.

This function computes a gravity model network based on the provided populations and distances. The gravity model estimates migration or interaction flows between populations using a mathematical formula that incorporates scaling, population sizes, and distances.

Mathematical formula::

$$ network_{i,j} = k cdot frac{p_i^a cdot p_j^b}{distance_{i,j}^c} $$

As implemented in NumPy:

network = k * (pops[:, np.newaxis] ** a) * (pops ** b) * (distances ** (-1 * c))
Parameters:

  • pops (numpy.ndarray) – 1D array of population sizes for each node.

  • distances (numpy.ndarray) – 2D array of distances between nodes. Must be symmetric, with self-distances (diagonal) handled.

  • k (float) – Scaling constant to adjust the overall magnitude of interaction flows.

  • a (float) – Exponent for the population size of the origin node.

  • b (float) – Exponent for the population size of the destination node.

  • c (float) – Exponent for the distance between nodes, controlling how distance impacts flows.

  • kwargs (dict) – Additional keyword arguments (not used in the current implementation).

Returns:

network (numpy.ndarray) – A 2D matrix representing the interaction network, where each element network[i, j] corresponds to the flow from node i to node j.

Example usage:

import numpy as np
from gravity_model import gravity

populations = np.array([1000, 500, 200])
distances = np.array([
    [0, 2, 3],
    [2, 0, 1],
    [3, 1, 0]
])

k = 0.5
a = 1.0
b = 1.0
c = 2.0

migration_network = gravity(populations, distances, k, a, b, c)

print("Migration Network:")
print(migration_network)

Notes

  • The diagonal of the distances matrix is set to 1 internally to avoid division by zero.

  • The diagonal of the output network is set to 0 to represent no self-loops.

  • Ensure the distances matrix is symmetric and non-negative.

laser.core.migration.radiation(pops, distances, k, include_home, **params)[source]

Calculate the migration network using the radiation model.

(Simini F, González MC, Maritan A, Barabási AL. A universal model for mobility and migration patterns. Nature. 2012;484(7392):96-100.)

Mathematical formula:
Element-by-element:

$$ network_{i,j} = k times p_i^a times (p_j / sum_k {p_k} )^b, $$ where the sum proceeds over all $k$ such that $distances_{i,k} leq distances_{i,j}$ the parameter include_home determines whether $p_i$ is included or excluded from the sum.

As-implemented numpy math:

  • Sort each row of the distance matrix (we’ll use ' below to indicate distance-sorted vectors)

  • Loop over “source nodes” i

    • Cumulative sum the sorted populations, ensuring appropriate handling when there are multiple destinations equidistant from the source

    • Subtract the source node population if include_home is False

    • Construct the row of the network matrix as

      $$ k times p_i times p_{j’} / (p_i + sum_{k’} {p_{k’}}) / (p_i + p_{j’} + sum_{k’} {p_{k’}}) $$

  • Unsort the rows of the network

Parameters:

  • pops (numpy.ndarray) – Array of population sizes for each node.

  • distances (numpy.ndarray) – 2D array of distances between nodes.

  • k (float) – Scaling factor for the migration rates.

  • include_home (bool) – Whether to include the home population in the calculations.

  • params (dict) – Additional parameters (currently not used).

Returns:

network (numpy.ndarray) – 2D array representing the migration network.

laser.core.migration.row_normalizer(network, max_rowsum)[source]

Normalizes the rows of a given network matrix such that no row sum exceeds a specified maximum value.

Parameters:

  • network (numpy.ndarray) – A 2D array representing the network matrix.

  • max_rowsum (float) – The maximum allowable sum for any row in the network matrix.

Returns:

network (numpy.ndarray) – The normalized network matrix where no row sum exceeds the specified maximum value.

laser.core.migration.stouffer(pops, distances, k, a, b, include_home, **params)[source]

Computes a migration network using a modified Stouffer’s model.

(Stouffer SA. Intervening opportunities: a theory relating mobility and distance. American sociological review. 1940;5(6):845-867)

Mathematical formula:
Element-by-element:

$$ network_{i,j} = k times p_i times p_j / ( (p_i + sum_k {p_k}) (p_i + p_j + sum_k {p_k}) ) $$ the parameter include_home determines whether $p_i$ is included or excluded from the sum

As-implemented numpy math:

  • Sort each row of the distance matrix (we’ll use ' below to indicate distance-sorted vectors)

  • Loop over “source nodes” i:

    • Cumulative sum the sorted populations, ensuring appropriate handling when there are multiple destinations equidistant from the source

    • Subtract the source node population if include_home is False

    • Construct the row of the network matrix as $k times p_i^a times (p_{j’} / sum_{k’} {p_{k’}})^b$

  • Unsort the rows of the network

Parameters:

  • pops (numpy.ndarray) – An array of population sizes.

  • distances (numpy.ndarray) – A 2D array where distances[i][j] is the distance from location i to location j.

  • k (float) – A scaling factor for the migration rates.

  • a (float) – Exponent applied to the population size of the origin.

  • b (float) – Exponent applied to the ratio of destination population to the sum of all populations at equal or lesser distances.

  • include_home (bool) – If True, includes the home population in the cumulative sum; otherwise, excludes it.

  • **params – Additional parameters (not used in the current implementation).

Returns:

network (numpy.ndarray) – A 2D array representing the migration network, where network[i][j] is the migration rate from location i to location j.

laser.core.migration.sum_populations_as_close_or_closer(sorted_pops, sorted_distance_row)[source]

laser.core.propertyset module

Implements a PropertySet class that can be used to store properties in a dictionary-like object.

class laser.core.propertyset.PropertySet(*bags: PropertySet | list | tuple | dict)[source]

Bases: object

A class that can be used to store properties in a dictionary-like object with .property access to properties.

Examples

Basic Initialization:

from laser.core import PropertySet ps = PropertySet() ps[‘infection_status’] = ‘infected’ ps[‘age’] = 35 print(ps.infection_status) # Outputs: ‘infected’ print(ps[‘age’]) # Outputs: 35

Combining two PropertySets:

ps1 = PropertySet({‘immunity’: ‘high’, ‘region’: ‘north’}) ps2 = PropertySet({‘infectivity’: 0.7}) combined_ps = ps1 + ps2 print(combined_ps.to_dict()) # Outputs: {‘immunity’: ‘high’, ‘region’: ‘north’, ‘infectivity’: 0.7}

Creating a PropertySet from a dictionary:

ps = PropertySet({‘mything’: 0.4, ‘that_other_thing’: 42}) print(ps.mything) # Outputs: 0.4 print(ps.that_other_thing) # Outputs: 42 print(ps.to_dict()) # Outputs: {‘mything’: 0.4, ‘that_other_thing’: 42}

Save and load:

ps.save(‘properties.json’) loaded_ps = PropertySet.load(‘properties.json’) print(loaded_ps.to_dict()) # Outputs the saved properties

Property access and length:

ps[‘status’] = ‘susceptible’ ps[‘exposure_timer’] = 5 print(ps[‘status’]) # Outputs: ‘susceptible’ print(len(ps)) # Outputs: 4

In-Place addition (added keys must not exist in the destination PropertySet):

ps += {‘new_timer’: 10, ‘susceptibility’: 0.75} print(ps.to_dict()) # Outputs: {‘mything’: 0.4, ‘that_other_thing’: 42, ‘status’: ‘susceptible’, ‘exposure_timer’: 5, ‘new_timer’: 10, ‘susceptibility’: 0.75}

In-place update (keys must already exist in the destination PropertySet):

ps <<= {‘exposure_timer’: 10, ‘infectivity’: 0.8} print(ps.to_dict()) # Outputs: {‘mything’: 0.4, ‘that_other_thing’: 42, ‘status’: ‘susceptible’, ‘exposure_timer’: 10, ‘infectivity’: 0.8}

In-place addition or update (no restriction on incoming keys):

ps |= {‘new_timer’: 10, ‘exposure_timer’: 8} print(ps.to_dict()) # Outputs: {‘mything’: 0.4, ‘that_other_thing’: 42, ‘status’: ‘susceptible’, ‘exposure_timer’: 8, ‘new_timer’: 10}

__add__(other)[source]

Add another PropertySet to this PropertySet.

This method allows the use of the + operator to combine two PropertySet instances.

Parameters:

other (PropertySet) – The other PropertySet instance to add.

Returns:

PropertySet (PropertySet) – A new PropertySet instance that combines the properties of both instances.

__contains__(key)[source]

Check if a key is in the property set.

Parameters:

key (str) – The key to check for existence in the property set.

Returns:

bool (bool) – True if the key exists in the property set, False otherwise.

__eq__(other)[source]

Check if two PropertySet instances are equal.

Parameters:

other (PropertySet) – The other PropertySet instance to compare.

Returns:

bool (bool) – True if the two instances are equal, False otherwise.

__getitem__(key)[source]

Retrieve the attribute of the object with the given key (e.g., ps[key]).

Parameters:

key (str) – The name of the attribute to retrieve.

Returns:

Any (any) – The value of the attribute with the specified key.

Raises:

AttributeError – If the attribute with the specified key does not exist.

__iadd__(other)[source]

Implements the in-place addition (+=) operator for the class.

This method allows the instance to be updated with attributes from another instance of the same class or from a dictionary. If other is an instance of the same class, its attributes are copied to the current instance. If other is a dictionary, its key-value pairs are added as attributes to the current instance.

Parameters:

other (Union[type(self), dict]) – The object or dictionary to add to the current instance.

Returns:

self (PropertySet) – The updated instance with the new attributes.

Raises:

  • AssertionError – If other is neither an instance of the same class nor a dictionary.

  • ValueError – If other contains keys already present in the PropertySet.

__ilshift__(other)[source]

Implements the <<= operator on PropertySet to override existing values with new values.

Parameters:

other (Union[type(self), dict]) – The object or dictionary with overriding values.

Returns:

self (PropertySet) – The updated instance with the overrides from other.

Raises:

  • AssertionError – If other is neither an instance of the same class nor a dictionary.

  • ValueError – If other contains keys not present in the PropertySet.

__ior__(other)[source]

Implements the |= operator on PropertySet to override existing values with new values.

Parameters:

other (Union[type(self), dict]) – The object or dictionary with overriding values.

Returns:

self (PropertySet) – The updated instance with all the values of self with new or overriding values from other.

Raises:

AssertionError – If other is neither an instance of the same class nor a dictionary.

__len__()[source]

Return the number of attributes in the instance.

This method returns the number of attributes stored in the instance’s __dict__ attribute, which represents the instance’s namespace.

Returns:

int (int) – The number of attributes in the instance.

__lshift__(other)[source]

Implements the << operator on PropertySet to override existing values with new values.

Parameters:

other (Union[type(self), dict]) – The object or dictionary with overriding values.

Returns:

PropertySet (PropertySet) – A new PropertySet with all the values of the first PropertySet with overrides from the second PropertySet.

Raises:

  • AssertionError – If other is neither an instance of the same class nor a dictionary.

  • ValueError – If other contains keys not present in the PropertySet.

__or__(other)[source]

Implements the | operator on PropertySet to add new or override existing values with new values.

Parameters:

other (Union[type(self), dict]) – The object or dictionary with overriding values.

Returns:

PropertySet (PropertySet) – A new PropertySet with all the values of the first PropertySet with new or overriding values from the second PropertySet.

Raises:

AssertionError – If other is neither an instance of the same class nor a dictionary.

__repr__() → str[source]

Return a string representation of the PropertySet instance.

The string representation includes the class name and the dictionary of the instance’s attributes.

Returns:

str – A string representation of the PropertySet instance.

__setitem__(key, value)[source]

Set the value of an attribute. This method allows setting an attribute of the instance using the dictionary-like syntax (e.g., ps[key] = value).

Parameters:

  • key (str) – The name of the attribute to set.

  • value (any) – The value to set for the attribute.

__str__() → str[source]

Returns a string representation of the object’s dictionary.

This method is used to provide a human-readable string representation of the object, which includes all the attributes stored in the object’s __dict__.

Returns:

str – A string representation of the object’s dictionary.

static load(filename)[source]

Load a PropertySet from a specified file.

Parameters:

filename (str) – The path to the file where the PropertySet is saved.

Returns:

PropertySet (PropertySet) – The PropertySet instance loaded from the file.

save(filename)[source]

Save the PropertySet to a specified file.

Parameters:

filename (str) – The path to the file where the PropertySet will be saved.

to_dict()[source]

Convert the PropertySet to a dictionary.

laser.core.random module

Functions for seeding and accessing the laser-core random number generator.

Using the seed() function here and the pseudo-random number generator (PRNG) returned from the prng() function in simulation code will guarantee that the same random number stream is generated and used during simulation runs using the same seed value (assuming no changes to code which add or remove PRNG calls or change the number of random draws requested). This is important for reproducibility and debugging purposes.

laser.core.random.get_seed() → uint32[source]

Return the seed used to initialize the pseudo-random number generator.

Returns:

uint32 – The seed value used to initialize the random number generators.

laser.core.random.prng() → Generator[source]

Return the global (to LASER) pseudo-random number generator.

laser.core.random.seed(seed) → Generator[source]

Initialize the pseudo-random number generator with a given seed.

This function sets the global pseudo-random number generator (_prng) to a new instance of numpy’s default random generator initialized with the provided seed. It also seeds Numba’s per-thread random number generators with the same seed.

Parameters:

seed (int) – The seed value to initialize the random number generators.

Returns:

numpy.random.Generator – The initialized pseudo-random number generator.

laser.core.utils module

This module provides utility functions for the laser-measles project.

Functions:
calc_capacity(birthrates: np.ndarray, initial_pop: np.ndarray, safety_factor: float = 1.0) -> np.ndarray:

Calculate the population capacity after a given number of ticks based on a constant birth rate.

grid(shape: Tuple[int, int], fill_value: float = 0.0) -> np.ndarray:

Create a 2D grid (numpy array) of the specified shape, filled with the given value.

laser.core.utils.calc_capacity(birthrates: ndarray, initial_pop: ndarray, safety_factor: float = 1.0) → ndarray[source]

Estimate the required capacity (number of agents) to model a population given birthrates over time.

Parameters:

  • birthrates (np.ndarray) – 2D array of shape (nsteps, nnodes) representing birthrates (CBR) per 1,000 individuals per year.

  • initial_pop (np.ndarray) – 1D array of length nnodes representing the initial population at each node.

  • safety_factor (float) – Safety factor to account for variability in population growth. Default is 1.0.

Returns:

np.ndarray – 1D array of length nnodes representing the estimated required capacity (number of agents) at each node.

laser.core.utils.grid(M=5, N=5, node_size_degs=0.08983, population_fn=None, origin_x=0, origin_y=0)[source]

Create an MxN grid of cells anchored at (lat, long) with populations and geometries.

Parameters:

  • M (int) – Number of rows (north-south).

  • N (int) – Number of columns (east-west).

  • node_size_degs (float) – Size of each cell in decimal degrees (default 0.08983 ≈ 10km at the equator).

  • population_fn (callable) – Function(row, col) returning population for a cell. Default is uniform random between 1,000 and 100,000.

  • origin_x (float) – longitude of the origin in decimal degrees (bottom-left corner) -180 <= origin_x < 180.

  • origin_y (float) – latitude of the origin in decimal degrees (bottom-left corner) -90 <= origin_y < 90.

Returns:

scenario (GeoDataFrame) – Columns are nodeid, population, geometry.

Module contents

class laser.core.LaserFrame(capacity: int, initial_count: int = -1, **kwargs)[source]

Bases: object

The LaserFrame class, similar to a db table or a Pandas DataFrame, holds dynamically allocated data for agents (generally 1-D or scalar) or for nodes|patches (e.g., 1-D for scalar value per patch or 2-D for time-varying per patch).

add(count: int) → tuple[int, int][source]

Adds the specified count to the current count of the LaserFrame.

This method increments the internal count by the given count, ensuring that the total does not exceed the frame’s capacity. If the addition would exceed the capacity, an assertion error is raised. This method is typically used to add new births during the simulation.

Parameters:

count (int) – The number to add to the current count.

Returns:

tuple[int, int] – A tuple containing the [start index, end index) after the addition.

Raises:

AssertionError – If the resulting count exceeds the frame’s capacity.

add_array_property(name: str, shape: tuple, dtype=<class 'numpy.uint32'>, default=0) → None[source]

Adds an array property to the object.

This method initializes a new property with the given name as a multi-dimensional NumPy array.

The array will have the given shape (note that there is no implied dimension of size self._capacity), datatype (default is np.uint32), and default value (default is 0).

Parameters:

  • name (str) – The name of the property to be added.

  • shape (tuple) – The shape of the array.

  • dtype (data-type, optional) – The desired data-type for the array, default is np.uint32.

  • default (scalar, optional) – The default value to fill the array with, default is 0.

Returns:

None

add_scalar_property(name: str, dtype=<class 'numpy.uint32'>, default=0) → None[source]

Add a scalar property to the class.

This method initializes a new scalar property for the class instance. The property is stored as a 1-D NumPy array (scalar / entry) with a specified data type and default value.

Parameters:

  • name (str) – The name of the scalar property to be added.

  • dtype (data-type, optional) – The desired data type for the property. Default is np.uint32.

  • default (scalar, optional) – The default value for the property. Default is 0.

Returns:

None

add_vector_property(name: str, length: int, dtype=<class 'numpy.uint32'>, default=0) → None[source]

Adds a vector property to the object.

This method initializes a new property with the given name as a 2-D NumPy array (vector per entry).

The array will have a shape of (length, self._capacity) and will be filled with the specified default value. The data type of the array elements is determined by the dtype parameter.

Parameters:

  • name (str) – The name of the property to be added.

  • length (int) – The length of the vector.

  • dtype (data-type, optional) – The desired data-type for the array, default is np.uint32.

  • default (scalar, optional) – The default value to fill the array with, default is 0.

Returns:

None

property capacity: int

Returns the capacity of the laser frame (total possible entries for dynamic properties).

Returns:

int – The capacity of the laser frame.

property count: int

Returns the current count (equivalent to len()).

Returns:

int – The current count value.

describe(target=None) → str[source]

Return a formatted string description of the laserframe object, including its attributes and their values.

Parameters:

target (string) – Optional string for the report header (generally the name of the LaserFrame variable, e.g., “People”. Unlike functions, we can’t get the name of a variable automatically).

Returns:

str –

A formatted string describing the laserframe object, including its capacity, count, and

details of its scalar, vector, and other properties.

classmethod load_snapshot(path, n_ppl, cbr, nt)[source]

Load a LaserFrame and optional extras from an HDF5 snapshot file.

Parameters:

  • path (str) – Path to the HDF5 snapshot file.

  • n_ppl (float or array-like) – Original total population (or per-node array) used to estimate births.

  • cbr (float or array-like) – Crude birth rate (per 1000/year).

  • nt (int) – Simulation duration (number of ticks).

Returns:

frame (LaserFrame) – Loaded LaserFrame object. results_r (np.ndarray or None): Optional 2D numpy array of recovered counts. pars (dict or None): Optional dictionary of parameters.

save_snapshot(path, results_r=None, pars=None)[source]

Save this LaserFrame and optional extras to an HDF5 snapshot file.

Parameters:

  • path (Path) – Destination file path

  • results_r (np.ndarray) – Optional 2D numpy array of recovered counts

  • pars (PropertySet or dict) – Optional PropertySet or dict of parameters

sort(indices, verbose: bool = False) → None[source]

Sorts the elements of the object’s numpy arrays based on the provided indices.

Parameters:

  • indices (np.ndarray) – An array of indices used to sort the numpy arrays. Must be of integer type and have the same length as the frame count (self._count).

  • verbose (bool, optional) – If True, prints the sorting progress for each numpy array attribute. Defaults to False.

Raises:

AssertionError – If indices is not an integer array or if its length does not match the frame count of active elements.

squash(indices, verbose: bool = False) → None[source]

Reduces the active count of the internal numpy arrays keeping only elements True in the provided boolean indices.

Parameters:

  • indices (np.ndarray) – A boolean array indicating which elements to keep. Must have the same length as the current frame active element count.

  • verbose (bool, optional) – If True, prints detailed information about the squashing process. Defaults to False.

Raises:

AssertionError – If indices is not a boolean array or if its length does not match the current frame active element count.

Returns:

None

class laser.core.PropertySet(*bags: PropertySet | list | tuple | dict)[source]

Bases: object

A class that can be used to store properties in a dictionary-like object with .property access to properties.

Examples

Basic Initialization:

from laser.core import PropertySet ps = PropertySet() ps[‘infection_status’] = ‘infected’ ps[‘age’] = 35 print(ps.infection_status) # Outputs: ‘infected’ print(ps[‘age’]) # Outputs: 35

Combining two PropertySets:

ps1 = PropertySet({‘immunity’: ‘high’, ‘region’: ‘north’}) ps2 = PropertySet({‘infectivity’: 0.7}) combined_ps = ps1 + ps2 print(combined_ps.to_dict()) # Outputs: {‘immunity’: ‘high’, ‘region’: ‘north’, ‘infectivity’: 0.7}

Creating a PropertySet from a dictionary:

ps = PropertySet({‘mything’: 0.4, ‘that_other_thing’: 42}) print(ps.mything) # Outputs: 0.4 print(ps.that_other_thing) # Outputs: 42 print(ps.to_dict()) # Outputs: {‘mything’: 0.4, ‘that_other_thing’: 42}

Save and load:

ps.save(‘properties.json’) loaded_ps = PropertySet.load(‘properties.json’) print(loaded_ps.to_dict()) # Outputs the saved properties

Property access and length:

ps[‘status’] = ‘susceptible’ ps[‘exposure_timer’] = 5 print(ps[‘status’]) # Outputs: ‘susceptible’ print(len(ps)) # Outputs: 4

In-Place addition (added keys must not exist in the destination PropertySet):

ps += {‘new_timer’: 10, ‘susceptibility’: 0.75} print(ps.to_dict()) # Outputs: {‘mything’: 0.4, ‘that_other_thing’: 42, ‘status’: ‘susceptible’, ‘exposure_timer’: 5, ‘new_timer’: 10, ‘susceptibility’: 0.75}

In-place update (keys must already exist in the destination PropertySet):

ps <<= {‘exposure_timer’: 10, ‘infectivity’: 0.8} print(ps.to_dict()) # Outputs: {‘mything’: 0.4, ‘that_other_thing’: 42, ‘status’: ‘susceptible’, ‘exposure_timer’: 10, ‘infectivity’: 0.8}

In-place addition or update (no restriction on incoming keys):

ps |= {‘new_timer’: 10, ‘exposure_timer’: 8} print(ps.to_dict()) # Outputs: {‘mything’: 0.4, ‘that_other_thing’: 42, ‘status’: ‘susceptible’, ‘exposure_timer’: 8, ‘new_timer’: 10}

__add__(other)[source]

Add another PropertySet to this PropertySet.

This method allows the use of the + operator to combine two PropertySet instances.

Parameters:

other (PropertySet) – The other PropertySet instance to add.

Returns:

PropertySet (PropertySet) – A new PropertySet instance that combines the properties of both instances.

__contains__(key)[source]

Check if a key is in the property set.

Parameters:

key (str) – The key to check for existence in the property set.

Returns:

bool (bool) – True if the key exists in the property set, False otherwise.

__eq__(other)[source]

Check if two PropertySet instances are equal.

Parameters:

other (PropertySet) – The other PropertySet instance to compare.

Returns:

bool (bool) – True if the two instances are equal, False otherwise.

__getitem__(key)[source]

Retrieve the attribute of the object with the given key (e.g., ps[key]).

Parameters:

key (str) – The name of the attribute to retrieve.

Returns:

Any (any) – The value of the attribute with the specified key.

Raises:

AttributeError – If the attribute with the specified key does not exist.

__iadd__(other)[source]

Implements the in-place addition (+=) operator for the class.

This method allows the instance to be updated with attributes from another instance of the same class or from a dictionary. If other is an instance of the same class, its attributes are copied to the current instance. If other is a dictionary, its key-value pairs are added as attributes to the current instance.

Parameters:

other (Union[type(self), dict]) – The object or dictionary to add to the current instance.

Returns:

self (PropertySet) – The updated instance with the new attributes.

Raises:

  • AssertionError – If other is neither an instance of the same class nor a dictionary.

  • ValueError – If other contains keys already present in the PropertySet.

__ilshift__(other)[source]

Implements the <<= operator on PropertySet to override existing values with new values.

Parameters:

other (Union[type(self), dict]) – The object or dictionary with overriding values.

Returns:

self (PropertySet) – The updated instance with the overrides from other.

Raises:

  • AssertionError – If other is neither an instance of the same class nor a dictionary.

  • ValueError – If other contains keys not present in the PropertySet.

__ior__(other)[source]

Implements the |= operator on PropertySet to override existing values with new values.

Parameters:

other (Union[type(self), dict]) – The object or dictionary with overriding values.

Returns:

self (PropertySet) – The updated instance with all the values of self with new or overriding values from other.

Raises:

AssertionError – If other is neither an instance of the same class nor a dictionary.

__len__()[source]

Return the number of attributes in the instance.

This method returns the number of attributes stored in the instance’s __dict__ attribute, which represents the instance’s namespace.

Returns:

int (int) – The number of attributes in the instance.

__lshift__(other)[source]

Implements the << operator on PropertySet to override existing values with new values.

Parameters:

other (Union[type(self), dict]) – The object or dictionary with overriding values.

Returns:

PropertySet (PropertySet) – A new PropertySet with all the values of the first PropertySet with overrides from the second PropertySet.

Raises:

  • AssertionError – If other is neither an instance of the same class nor a dictionary.

  • ValueError – If other contains keys not present in the PropertySet.

__or__(other)[source]

Implements the | operator on PropertySet to add new or override existing values with new values.

Parameters:

other (Union[type(self), dict]) – The object or dictionary with overriding values.

Returns:

PropertySet (PropertySet) – A new PropertySet with all the values of the first PropertySet with new or overriding values from the second PropertySet.

Raises:

AssertionError – If other is neither an instance of the same class nor a dictionary.

__repr__() → str[source]

Return a string representation of the PropertySet instance.

The string representation includes the class name and the dictionary of the instance’s attributes.

Returns:

str – A string representation of the PropertySet instance.

__setitem__(key, value)[source]

Set the value of an attribute. This method allows setting an attribute of the instance using the dictionary-like syntax (e.g., ps[key] = value).

Parameters:

  • key (str) – The name of the attribute to set.

  • value (any) – The value to set for the attribute.

__str__() → str[source]

Returns a string representation of the object’s dictionary.

This method is used to provide a human-readable string representation of the object, which includes all the attributes stored in the object’s __dict__.

Returns:

str – A string representation of the object’s dictionary.

static load(filename)[source]

Load a PropertySet from a specified file.

Parameters:

filename (str) – The path to the file where the PropertySet is saved.

Returns:

PropertySet (PropertySet) – The PropertySet instance loaded from the file.

save(filename)[source]

Save the PropertySet to a specified file.

Parameters:

filename (str) – The path to the file where the PropertySet will be saved.

to_dict()[source]

Convert the PropertySet to a dictionary.

class laser.core.SortedQueue(capacity: int, values: ndarray)[source]

Bases: object

A sorted (priority) queue implemented using NumPy arrays and sped-up with Numba.

Using the algorithm from the Python heapq module.

__init__ with an existing array of sorting values

__push__ with an index into sorting values

__pop__ returns the index of the lowest sorting value and its value

__len__() → int[source]

Return the number of elements in the sorted queue.

Returns:

int – The number of elements in the sorted queue.

peeki() → uint32[source]

Returns the index of the smallest value element in the sorted queue without removing it.

Raises:

IndexError – If the sorted queue is empty.

Returns:

np.uint32 – The index of the smallest value element.

peekiv() → tuple[uint32, Any][source]

Returns the index and value of the smallest value element in the sorted queue without removing it.

Returns:

tuple[np.uint32, Any] – A tuple containing the index and value of the smallest value element.

Raises:

IndexError – If the sorted queue is empty.

peekv() → Any[source]

Return the smallest value from the sorted queue without removing it.

Raises:

IndexError – If the sorted queue is empty.

Returns:

Any – The value with the smallest value in the sorted queue.

popi() → uint32[source]

Removes and returns the index of the smallest value element in the sorted queue.

This method first retrieves the index of the smallest value element using peeki(), then removes the element from the queue using pop(), and finally returns the index.

Returns:

np.uint32 – The index of the smallest value element in the sorted queue.

popiv() → tuple[uint32, Any][source]

Removes and returns the index and value of the smallest value element in the sorted queue.

This method first retrieves the index and value of the smallest value element using peekiv(), then removes the element from the queue using pop(), and finally returns the index and value.

Returns:

tuple[np.uint32, Any] – A tuple containing the index and value of the smallest value element.

popv() → Any[source]

Removes and returns the value at the front of the sorted queue.

This method first retrieves the value at the front of the queue without removing it by calling peekv(), and then removes the front element from the queue by calling pop(). The retrieved value is then returned.

Returns:

Any – The value at the front of the sorted queue.

push(index) → None[source]

Insert an element into the sorted queue.

This method adds an element at the back of the sorted queue and then ensures the heap property is maintained by sifting the element forward to its correct position.

Parameters:

index (int) – The index of the element to be added to the sorted queue.

Raises:

IndexError – If the sorted queue is full.

laser.core.compiled(args)[source]