laser_core package

Subpackages

• laser_core.demographics package
  • Submodules
  • laser_core.demographics.kmestimator module
    • KaplanMeierEstimator
      • KaplanMeierEstimator.cumulative_deaths
      • KaplanMeierEstimator.predict_age_at_death()
      • KaplanMeierEstimator.predict_year_of_death()
  • laser_core.demographics.pyramid module
    • AliasedDistribution
      • AliasedDistribution.alias
      • AliasedDistribution.probs
      • AliasedDistribution.sample()
      • AliasedDistribution.total
    • load_pyramid_csv()
  • laser_core.demographics.spatialpops module
    • distribute_population_skewed()
    • distribute_population_tapered()
  • Module contents
    • AliasedDistribution
      • AliasedDistribution.alias
      • AliasedDistribution.probs
      • AliasedDistribution.sample()
      • AliasedDistribution.total
    • KaplanMeierEstimator
      • KaplanMeierEstimator.cumulative_deaths
      • KaplanMeierEstimator.predict_age_at_death()
      • KaplanMeierEstimator.predict_year_of_death()
    • load_pyramid_csv()

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.extension module

laser_core.extension.compiled(args)

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:

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

class laser_core.laserframe.LaserFrame(capacity: int, initial_count: int = -1, **kwargs)

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]

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

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

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

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

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

Parameters:

target – 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)

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) results_r (np.ndarray or None) pars (dict or None)

save_snapshot(path, results_r=None, pars=None)

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

Parameters:

• path – Destination file path

• results_r – Optional 2D numpy array of recovered counts

• pars – Optional PropertySet or dict of parameters

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

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

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:

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:

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:

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)

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 – Additional parameters to be passed to the gravity model.

Returns:

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

laser_core.migration.distance(lat1, lon1, lat2, lon2)

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 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:

float – The distance between the two points in kilometers.

laser_core.migration.gravity(pops: ndarray, distances: ndarray, k: float, a: float, b: float, c: float, **kwargs)

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:

Additional keyword arguments (not used in the current implementation).

Returns:

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

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

# Parameters for the gravity model
k = 0.5
a = 1.0
b = 1.0
c = 2.0

# Compute the gravity model network
migration_network = gravity(populations, distances, k, a, b, c)

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

Notes:

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

• The diagonal of the output network matrix 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)

Calculate the migration network using the radiation model.

(Simini F, Gonza ́lez MC, Maritan A, Baraba ́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 – Additional parameters (currently not used).

Returns:

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

laser_core.migration.row_normalizer(network, max_rowsum)

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:

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)

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:

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)

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)

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())
{'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())
{'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())
{'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())
{'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())
{'mything': 0.4, 'that_other_thing': 42, 'status': 'susceptible', 'exposure_timer': 8, 'new_timer': 10}

__add__(other)

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 – A new PropertySet instance that combines the properties of both instances.

__contains__(key)

Check if a key is in the property set.

Parameters:

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

Returns:

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

__eq__(other)

Check if two PropertySet instances are equal.

Parameters:

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

Returns:

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

__getitem__(key)

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 – The value of the attribute with the specified key.

Raises:

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

__iadd__(other)

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 – 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)

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 – 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)

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 – 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__()

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 – The number of attributes in the instance.

__lshift__(other)

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:

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)

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:

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

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)

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.

Returns:

None

__str__() → str

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)

Load a PropertySet from a specified file.

Parameters:

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

Returns:

PropertySet – The PropertySet instance loaded from the file.

save(filename)

Save the PropertySet to a specified file.

Parameters:

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

Returns:

None

to_dict()

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

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

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

laser_core.random.seed(seed) → Generator

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.sortedqueue module

SortedQueue implementation using NumPy and Numba.

class laser_core.sortedqueue.SortedQueue(capacity: int, values: ndarray)

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

Return the number of elements in the sorted queue.

Returns:

int – The number of elements in the sorted queue.

peeki() → uint32

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]

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

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

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]

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

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

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.utils module

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

Functions:

calc_distances(latitudes: np.ndarray, longitudes: np.ndarray, verbose: bool = False) -> np.ndarray:

Calculate the pairwise distances between points given their latitudes and longitudes.

calc_capacity(population: np.uint32, nticks: np.uint32, cbr: np.float32, verbose: bool = False) -> np.uint32:

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

laser_core.utils.calc_capacity(population: uint32, nticks: uint32, cbr: float32, verbose: bool = False) → uint32

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

Parameters:

• population (np.uint32) – The initial population.

• nticks (np.uint32) – The number of ticks (time steps) to simulate.

• cbr (np.float32) – The constant birth rate per 1000 people per year.

• verbose (bool, optional) – If True, prints detailed population growth information. Defaults to False.

Returns:

np.uint32 – The estimated population capacity after the given number of ticks.

laser_core.utils.calc_distances(*args, **kwargs)

Module contents

class laser_core.LaserFrame(capacity: int, initial_count: int = -1, **kwargs)

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]

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

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

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

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

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

Parameters:

target – 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)

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) results_r (np.ndarray or None) pars (dict or None)

save_snapshot(path, results_r=None, pars=None)

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

Parameters:

• path – Destination file path

• results_r – Optional 2D numpy array of recovered counts

• pars – Optional PropertySet or dict of parameters

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

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

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)

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())
{'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())
{'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())
{'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())
{'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())
{'mything': 0.4, 'that_other_thing': 42, 'status': 'susceptible', 'exposure_timer': 8, 'new_timer': 10}

__add__(other)

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 – A new PropertySet instance that combines the properties of both instances.

__contains__(key)

Check if a key is in the property set.

Parameters:

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

Returns:

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

__eq__(other)

Check if two PropertySet instances are equal.

Parameters:

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

Returns:

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

__getitem__(key)

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 – The value of the attribute with the specified key.

Raises:

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

__iadd__(other)

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 – 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)

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 – 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)

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 – 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__()

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 – The number of attributes in the instance.

__lshift__(other)

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:

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)

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:

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

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)

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.

Returns:

None

__str__() → str

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)

Load a PropertySet from a specified file.

Parameters:

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

Returns:

PropertySet – The PropertySet instance loaded from the file.

save(filename)

Save the PropertySet to a specified file.

Parameters:

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

Returns:

None

to_dict()

Convert the PropertySet to a dictionary.

class laser_core.SortedQueue(capacity: int, values: ndarray)

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

Return the number of elements in the sorted queue.

Returns:

int – The number of elements in the sorted queue.

peeki() → uint32

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]

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

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

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]

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

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

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)